112 lines
4.4 KiB
Python
112 lines
4.4 KiB
Python
"""
|
|
FTB (California Franchise Tax Board) hold-line dialer.
|
|
|
|
Usage:
|
|
conda run -n cf python -m app.dialers.ftb
|
|
|
|
Reads from environment (see .env.example):
|
|
TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_FROM_NUMBER
|
|
OSPREY_BASE_URL — public HTTPS base, e.g. https://circuitforge.tech
|
|
OSPREY_BRIDGE_NUMBER — your cell in E.164 format, e.g. +15105551234
|
|
OSPREY_IVR_SEQUENCE — DTMF string to navigate FTB IVR (see below)
|
|
OSPREY_RETRY_INTERVAL — seconds between retry attempts (default: 180)
|
|
OSPREY_MAX_RETRIES — give up after this many rejections (default: 20)
|
|
|
|
OSPREY_IVR_SEQUENCE format:
|
|
Comma-separated DTMF digits. Each comma = 0.5s pause.
|
|
Example: "1,,,2,,,0" means press 1, wait 1.5s, press 2, wait 1.5s, press 0.
|
|
Set this to the FTB IVR key sequence that reaches a human agent.
|
|
|
|
TODO: Fill in the correct FTB IVR sequence below once confirmed.
|
|
The FTB main line is 800-852-5711. # gitleaks:allow — public government phone number
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
import dotenv
|
|
from twilio.rest import Client
|
|
|
|
dotenv.load_dotenv()
|
|
|
|
# ── Twilio credentials ────────────────────────────────────────────────────────
|
|
ACCOUNT_SID = os.environ["TWILIO_ACCOUNT_SID"]
|
|
AUTH_TOKEN = os.environ["TWILIO_AUTH_TOKEN"]
|
|
FROM_NUMBER = os.environ["TWILIO_FROM_NUMBER"] # your Twilio number, E.164
|
|
|
|
# ── Call targets ──────────────────────────────────────────────────────────────
|
|
FTB_NUMBER = "+18008525711"
|
|
BASE_URL = os.environ["OSPREY_BASE_URL"].rstrip("/")
|
|
STATE_FILE = os.environ.get("OSPREY_STATE_FILE", "/tmp/osprey_call_state.txt")
|
|
|
|
# ── Retry config ──────────────────────────────────────────────────────────────
|
|
RETRY_INTERVAL = int(os.environ.get("OSPREY_RETRY_INTERVAL", "180")) # seconds
|
|
MAX_RETRIES = int(os.environ.get("OSPREY_MAX_RETRIES", "20"))
|
|
|
|
|
|
def _read_state() -> tuple[str, int]:
|
|
"""Read last call status + duration from state file written by /status webhook."""
|
|
try:
|
|
with open(STATE_FILE) as f:
|
|
parts = f.read().strip().split(":")
|
|
return parts[0], int(parts[1]) if len(parts) > 1 else 0
|
|
except FileNotFoundError:
|
|
return "unknown", 0
|
|
|
|
|
|
def _clear_state() -> None:
|
|
try:
|
|
os.remove(STATE_FILE)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def dial(client: Client) -> str:
|
|
"""Initiate one outbound call to FTB. Returns the CallSid."""
|
|
call = client.calls.create(
|
|
to=FTB_NUMBER,
|
|
from_=FROM_NUMBER,
|
|
url=f"{BASE_URL}/webhooks/twilio/twiml",
|
|
status_callback=f"{BASE_URL}/webhooks/twilio/status",
|
|
status_callback_event=["completed", "busy", "failed", "no-answer"],
|
|
status_callback_method="POST",
|
|
# machine_detection="DetectMessageEnd", # uncomment to skip voicemail
|
|
)
|
|
print(f"[Osprey] Dialing FTB — CallSid: {call.sid}")
|
|
return call.sid
|
|
|
|
|
|
def run() -> None:
|
|
client = Client(ACCOUNT_SID, AUTH_TOKEN)
|
|
_clear_state()
|
|
|
|
for attempt in range(1, MAX_RETRIES + 1):
|
|
print(f"\n[Osprey] Attempt {attempt}/{MAX_RETRIES}")
|
|
dial(client)
|
|
|
|
# Poll for status — Twilio posts to /status within a few seconds of call end.
|
|
# A call that stays alive means we navigated past the rejection.
|
|
for _ in range(90): # wait up to 90s for status
|
|
time.sleep(1)
|
|
status, duration = _read_state()
|
|
if status not in ("unknown", "initiated", "ringing", "in-progress"):
|
|
break
|
|
|
|
status, duration = _read_state()
|
|
rejected = status in ("completed", "busy", "failed", "no-answer") and duration < 30
|
|
|
|
if not rejected:
|
|
print(f"[Osprey] In queue! (status={status}, duration={duration}s) — bridging your phone.")
|
|
print("[Osprey] Pick up when your phone rings.")
|
|
return
|
|
|
|
print(f"[Osprey] Rejected (status={status}, duration={duration}s). "
|
|
f"Retrying in {RETRY_INTERVAL}s...")
|
|
_clear_state()
|
|
time.sleep(RETRY_INTERVAL)
|
|
|
|
print(f"[Osprey] Gave up after {MAX_RETRIES} attempts. Try again later.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|