diff --git a/.env.example b/.env.example index 5b36051..1321538 100644 --- a/.env.example +++ b/.env.example @@ -1,18 +1,18 @@ -# Telnyx API credentials -# Sign up at https://portal.telnyx.com/ and get your API key -TELNYX_API_KEY=KEY_your_api_key_here +# Sinch API credentials +# Sign up at https://dashboard.sinch.com/ +# Go to Settings > Access Keys to create a key pair +SINCH_PROJECT_ID=your_project_id +SINCH_KEY_ID=your_key_id +SINCH_KEY_SECRET=your_key_secret -# Your Telnyx Fax Application connection ID -# Create a Fax Application in Mission Control Portal under Messaging > Fax -TELNYX_CONNECTION_ID=your_connection_id_here - -# Your Telnyx fax-enabled phone number (E.164 format) -# Purchase one in the Mission Control Portal under Numbers, assign to Fax App -TELNYX_FROM_NUMBER=+1XXXXXXXXXX +# Your Sinch fax-enabled phone number (E.164 format) +# Purchase one in the Sinch dashboard under Numbers +SINCH_FROM_NUMBER=+1XXXXXXXXXX # Destination fax number (E.164 format) +# Use +19898989898 for free test faxes (no charge, simulates delivery) FAX_TO_NUMBER=+1XXXXXXXXXX # ntfy notification URL (optional, comment out to disable) -NTFY_URL=https://nt.nevo.engineer/your-channel -NTFY_TOKEN=Bearer tk_your_token_here +# NTFY_URL=https://nt.nevo.engineer/your-channel +# NTFY_TOKEN=Bearer tk_your_token_here diff --git a/README.md b/README.md index a87f72c..016068d 100644 --- a/README.md +++ b/README.md @@ -1,13 +1,13 @@ # AutoFax -Automated hourly fax sender for insurance claim submissions. Sends your claim via Telnyx's fax API every hour and generates a printable HTML report documenting every attempt and its status -- useful for showing HR or the insurance company that their fax system is rejecting claims. +Automated hourly fax sender for insurance claim submissions. Sends your claim via Sinch's fax API every hour and generates a printable HTML report documenting every attempt and its status -- useful for showing HR or the insurance company that their fax system is rejecting claims. ## Quick Start (Windows GUI) If someone gave you `AutoFax.exe` on a flash drive: 1. Double-click `AutoFax.exe` -2. Fill in the Telnyx credentials (API Key, Connection ID, From Number) +2. Fill in the Sinch credentials (Project ID, Key ID, Key Secret, From Number) 3. Set the destination fax number 4. Click **Browse** and select your claim PDF 5. Click **Start Hourly Faxing** @@ -24,45 +24,24 @@ build_windows.bat This produces `dist/AutoFax.exe` -- a single self-contained file you can copy to a flash drive. -## Telnyx Account Setup +## Sinch Account Setup -### Automated (recommended) - -The setup wizard handles everything -- creates a fax application, buys a number, and writes your config: - -```bash -# Interactive -python3 setup_telnyx.py - -# Or non-interactive -python3 setup_telnyx.py --api-key KEY_xxx --to +18019382102 --area-code 801 -``` - -You just need a Telnyx API key: -1. Sign up at [portal.telnyx.com](https://portal.telnyx.com/) -2. Add a payment method -3. Go to **API Keys** and copy your key - -The wizard creates both `.env` (for CLI) and `autofax_config.json` (for GUI). - -### Manual - -1. Sign up at [portal.telnyx.com](https://portal.telnyx.com/) -2. Add a payment method -3. Go to **Messaging > Fax** and create a Fax Application (note the Connection ID) -4. Go to **Numbers** and purchase a fax-enabled number (~$1/month) -5. Assign the number to your Fax Application -6. Go to **API Keys** and copy your API key -7. Configure: +1. Sign up at [dashboard.sinch.com](https://dashboard.sinch.com/) +2. Go to **Numbers** and purchase a fax-enabled number +3. Go to **Settings > Access Keys** and create a key pair +4. Note your **Project ID** (shown at the top of the dashboard) +5. Configure: ```bash cp .env.example .env -# Edit .env with your Telnyx credentials and the destination fax number +# Edit .env with your Sinch credentials and the destination fax number ``` +**Test for free:** Use `+19898989898` as the destination number to simulate fax delivery without charges. + ## Linux/Server Setup (CLI + cron) -### 3. Add Your Claim +### 1. Add Your Claim Place your claim PDF in the `claims/` directory: @@ -70,7 +49,7 @@ Place your claim PDF in the `claims/` directory: cp /path/to/your/claim.pdf claims/ ``` -### 4. Install & Start +### 2. Install & Start ```bash chmod +x install.sh @@ -81,9 +60,9 @@ This will: - Create a Python virtual environment - Install dependencies - Set up an hourly cron job -- Schedule auto-cleanup after 7 days +- Auto-remove the cron after 7 days -### 5. Manual Test +### 3. Manual Test ```bash venv/bin/python autofax.py @@ -110,5 +89,5 @@ Optional push notifications via [ntfy](https://ntfy.sh). Configure in `.env` (CL ## Cost -Telnyx pricing: ~$0.007/page + ~$1/month for the phone number. -One week of hourly faxes (168 faxes x 5 pages = 840 pages) costs roughly $8-15. +Sinch pricing: $0.045/page flat rate, no monthly commitment. +30 faxes x 5 pages = 150 pages = ~$7 total. diff --git a/autofax.py b/autofax.py index e79cc8a..b0c1246 100755 --- a/autofax.py +++ b/autofax.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""AutoFax - Send a fax every hour via Telnyx and log results.""" +"""AutoFax - Send a fax every hour via Sinch and log results.""" import json import subprocess @@ -12,11 +12,11 @@ import requests import config -API = "https://api.telnyx.com/v2" +API = f"https://fax.api.sinch.com/v3/projects/{config.SINCH_PROJECT_ID}" -def headers(): - return {"Authorization": f"Bearer {config.TELNYX_API_KEY}"} +def auth(): + return (config.SINCH_KEY_ID, config.SINCH_KEY_SECRET) def find_claim_pdf() -> Path: @@ -27,45 +27,30 @@ def find_claim_pdf() -> Path: return pdfs[0] -def upload_media(pdf_path: Path) -> str: - """Upload PDF to Telnyx media storage. Returns the media_name.""" - media_name = f"autofax-claim-{pdf_path.stem}" - with open(pdf_path, "rb") as f: - resp = requests.post( - f"{API}/media", - headers={**headers(), "Content-Type": "application/pdf"}, - params={"media_name": media_name}, - data=f.read(), - ) - resp.raise_for_status() - return media_name - - def send_fax(pdf_path: Path) -> dict: - """Upload PDF and send a fax. Returns result dict.""" + """Send a fax via Sinch with direct file upload. Returns result dict.""" timestamp = datetime.now(timezone.utc).isoformat() try: - media_name = upload_media(pdf_path) - - resp = requests.post( - f"{API}/faxes", - headers={**headers(), "Content-Type": "application/json"}, - json={ - "connection_id": config.TELNYX_CONNECTION_ID, - "media_name": media_name, - "to": config.FAX_TO_NUMBER, - "from": config.TELNYX_FROM_NUMBER, - }, - ) + with open(pdf_path, "rb") as f: + resp = requests.post( + f"{API}/faxes", + auth=auth(), + files={"file": (pdf_path.name, f, "application/pdf")}, + data={ + "to": config.FAX_TO_NUMBER, + "from": config.SINCH_FROM_NUMBER, + }, + timeout=60, + ) resp.raise_for_status() - fax_data = resp.json().get("data", {}) + fax_data = resp.json() return { "timestamp": timestamp, - "status": fax_data.get("status", "queued"), + "status": fax_data.get("status", "QUEUED"), "fax_id": fax_data.get("id"), "to": config.FAX_TO_NUMBER, - "from": config.TELNYX_FROM_NUMBER, + "from": config.SINCH_FROM_NUMBER, "file": pdf_path.name, "error": None, } @@ -80,7 +65,7 @@ def send_fax(pdf_path: Path) -> dict: "status": "send_failed", "fax_id": None, "to": config.FAX_TO_NUMBER, - "from": config.TELNYX_FROM_NUMBER, + "from": config.SINCH_FROM_NUMBER, "file": pdf_path.name, "error": f"{e} | {error_body}", } @@ -90,7 +75,7 @@ def send_fax(pdf_path: Path) -> dict: "status": "send_failed", "fax_id": None, "to": config.FAX_TO_NUMBER, - "from": config.TELNYX_FROM_NUMBER, + "from": config.SINCH_FROM_NUMBER, "file": pdf_path.name, "error": str(e), } @@ -100,9 +85,9 @@ def check_fax_status(fax_id: str) -> dict | None: if not fax_id: return None try: - resp = requests.get(f"{API}/faxes/{fax_id}", headers=headers()) + resp = requests.get(f"{API}/faxes/{fax_id}", auth=auth(), timeout=15) resp.raise_for_status() - return resp.json().get("data", {}) + return resp.json() except Exception: return None @@ -125,7 +110,7 @@ def notify(subject: str, message: str, priority: str = "default"): hdrs = {"Title": subject, "Priority": priority} if config.NTFY_TOKEN: hdrs["Authorization"] = config.NTFY_TOKEN - requests.post(config.NTFY_URL, data=message.encode(), headers=hdrs) + requests.post(config.NTFY_URL, data=message.encode(), headers=hdrs, timeout=10) except Exception: pass @@ -134,17 +119,18 @@ def generate_report(entries: list[dict]): config.REPORTS_DIR.mkdir(parents=True, exist_ok=True) total = len(entries) - failed = sum(1 for e in entries if e["status"] not in ("queued", "sending", "sent", "delivered")) - succeeded = sum(1 for e in entries if e["status"] == "delivered") + success_statuses = {"QUEUED", "IN_PROGRESS", "COMPLETED", "delivered", "sent", "queued", "sending"} + failed = sum(1 for e in entries if e["status"] not in success_statuses) + succeeded = sum(1 for e in entries if e["status"] in ("COMPLETED", "delivered")) pending = total - failed - succeeded rows = "" for i, e in enumerate(entries, 1): status = e["status"] - if status in ("delivered", "sent"): + if status in ("COMPLETED", "delivered", "sent"): status_class = "success" - status_display = status.upper() - elif status in ("queued", "sending"): + status_display = "DELIVERED" if status == "COMPLETED" else status.upper() + elif status in ("QUEUED", "IN_PROGRESS", "queued", "sending"): status_class = "pending" status_display = status.upper() else: @@ -152,12 +138,11 @@ def generate_report(entries: list[dict]): status_display = "FAILED" error_col = e.get("error") or e.get("failure_reason") or "-" - ts = e["timestamp"] try: - dt = datetime.fromisoformat(ts) + dt = datetime.fromisoformat(e["timestamp"]) ts_display = dt.strftime("%b %d, %Y %I:%M %p UTC") except Exception: - ts_display = ts + ts_display = e["timestamp"] rows += f""" @@ -239,14 +224,14 @@ def generate_report(entries: list[dict]): def update_previous_statuses(entries: list[dict]) -> list[dict]: for entry in entries: - if entry["status"] in ("queued", "sending"): + if entry["status"] in ("QUEUED", "IN_PROGRESS", "queued", "sending"): status_data = check_fax_status(entry.get("fax_id")) if status_data: new_status = status_data.get("status", entry["status"]) if new_status != entry["status"]: entry["status"] = new_status - if new_status == "failed": - entry["failure_reason"] = status_data.get("failure_reason", "Unknown") + if new_status in ("FAILURE", "failed"): + entry["failure_reason"] = status_data.get("failureReason", "Unknown") return entries @@ -280,29 +265,28 @@ def main(): entries.append(result) # Wait briefly and check initial status - if result["fax_id"] and result["status"] == "queued": + if result["fax_id"] and result["status"] == "QUEUED": time.sleep(15) status_data = check_fax_status(result["fax_id"]) if status_data: result["status"] = status_data.get("status", result["status"]) - if result["status"] == "failed": - result["failure_reason"] = status_data.get("failure_reason", "Unknown") + if result["status"] == "FAILURE": + result["failure_reason"] = status_data.get("failureReason", "Unknown") save_log(entries) report_path = generate_report(entries) # Notify via ntfy status_labels = { - "delivered": "Fax Delivered", - "sent": "Fax Sent", - "queued": "Fax Queued", - "sending": "Fax Sending", + "COMPLETED": "Fax Delivered", + "IN_PROGRESS": "Fax Sending", + "QUEUED": "Fax Queued", } subject = status_labels.get(result["status"], "Fax Failed") msg = f"To: {result['to']}\nStatus: {result['status']}\nFile: {result['file']}" if result.get("error"): msg += f"\nError: {result['error']}" - priority = "high" if result["status"] not in ("queued", "sending", "sent", "delivered") else "default" + priority = "high" if result["status"] not in ("QUEUED", "IN_PROGRESS", "COMPLETED") else "default" notify(subject, msg, priority) print(f"Status: {result['status']}") diff --git a/config.py b/config.py index ac5976d..d1d2592 100644 --- a/config.py +++ b/config.py @@ -10,10 +10,11 @@ CLAIMS_DIR = BASE_DIR / "claims" REPORTS_DIR = BASE_DIR / "reports" LOG_FILE = REPORTS_DIR / "fax_log.json" -# Telnyx -TELNYX_API_KEY = os.environ["TELNYX_API_KEY"] -TELNYX_CONNECTION_ID = os.environ["TELNYX_CONNECTION_ID"] -TELNYX_FROM_NUMBER = os.environ["TELNYX_FROM_NUMBER"] +# Sinch +SINCH_PROJECT_ID = os.environ["SINCH_PROJECT_ID"] +SINCH_KEY_ID = os.environ["SINCH_KEY_ID"] +SINCH_KEY_SECRET = os.environ["SINCH_KEY_SECRET"] +SINCH_FROM_NUMBER = os.environ["SINCH_FROM_NUMBER"] FAX_TO_NUMBER = os.environ["FAX_TO_NUMBER"] # ntfy (optional) diff --git a/gui.py b/gui.py index f504b4b..d2a57d3 100644 --- a/gui.py +++ b/gui.py @@ -24,14 +24,13 @@ CONFIG_FILE = BASE_DIR / "autofax_config.json" CLAIMS_DIR = BASE_DIR / "claims" REPORTS_DIR = BASE_DIR / "reports" LOG_FILE = REPORTS_DIR / "fax_log.json" -API = "https://api.telnyx.com/v2" class AutoFaxApp: def __init__(self, root: tk.Tk): self.root = root self.root.title("AutoFax - Insurance Claim Faxer") - self.root.geometry("620x700") + self.root.geometry("620x720") self.root.resizable(False, False) self.running = False @@ -54,8 +53,9 @@ class AutoFaxApp: def save_config(self): self.config = { - "api_key": self.api_key_var.get().strip(), - "connection_id": self.conn_id_var.get().strip(), + "project_id": self.project_id_var.get().strip(), + "key_id": self.key_id_var.get().strip(), + "key_secret": self.key_secret_var.get().strip(), "from_number": self.from_var.get().strip(), "to_number": self.to_var.get().strip(), "pdf_path": self.pdf_var.get().strip(), @@ -85,17 +85,19 @@ class AutoFaxApp: main = tk.Frame(self.root, padx=15, pady=10) main.pack(fill="both", expand=True) - # ── Telnyx Settings ── - lf_telnyx = tk.LabelFrame(main, text="Telnyx Settings", padx=10, pady=5) - lf_telnyx.pack(fill="x", pady=(0, 8)) + # ── Sinch Settings ── + lf_sinch = tk.LabelFrame(main, text="Sinch Settings", padx=10, pady=5) + lf_sinch.pack(fill="x", pady=(0, 8)) - self.api_key_var = tk.StringVar() - self.conn_id_var = tk.StringVar() + self.project_id_var = tk.StringVar() + self.key_id_var = tk.StringVar() + self.key_secret_var = tk.StringVar() self.from_var = tk.StringVar() - self._add_field(lf_telnyx, "API Key:", self.api_key_var, show="*") - self._add_field(lf_telnyx, "Connection ID:", self.conn_id_var) - self._add_field(lf_telnyx, "From Number:", self.from_var, placeholder="+1XXXXXXXXXX") + self._add_field(lf_sinch, "Project ID:", self.project_id_var) + self._add_field(lf_sinch, "Key ID:", self.key_id_var) + self._add_field(lf_sinch, "Key Secret:", self.key_secret_var, show="*") + self._add_field(lf_sinch, "From Number:", self.from_var, placeholder="+1XXXXXXXXXX") # ── Fax Settings ── lf_fax = tk.LabelFrame(main, text="Fax Settings", padx=10, pady=5) @@ -173,20 +175,18 @@ class AutoFaxApp: def populate_fields(self): c = self.config - if c.get("api_key"): - self.api_key_var.set(c["api_key"]) - if c.get("connection_id"): - self.conn_id_var.set(c["connection_id"]) - if c.get("from_number"): - self.from_var.set(c["from_number"]) - if c.get("to_number"): - self.to_var.set(c["to_number"]) - if c.get("pdf_path"): - self.pdf_var.set(c["pdf_path"]) - if c.get("ntfy_url"): - self.ntfy_url_var.set(c["ntfy_url"]) - if c.get("ntfy_token"): - self.ntfy_token_var.set(c["ntfy_token"]) + for key, var in [ + ("project_id", self.project_id_var), + ("key_id", self.key_id_var), + ("key_secret", self.key_secret_var), + ("from_number", self.from_var), + ("to_number", self.to_var), + ("pdf_path", self.pdf_var), + ("ntfy_url", self.ntfy_url_var), + ("ntfy_token", self.ntfy_token_var), + ]: + if c.get(key): + var.set(c[key]) # ── Actions ───────────────────────────────────────────────────── @@ -201,10 +201,12 @@ class AutoFaxApp: def validate(self) -> bool: problems = [] - if not self.api_key_var.get().strip(): - problems.append("API Key is required") - if not self.conn_id_var.get().strip(): - problems.append("Connection ID is required") + if not self.project_id_var.get().strip(): + problems.append("Project ID is required") + if not self.key_id_var.get().strip(): + problems.append("Key ID is required") + if not self.key_secret_var.get().strip(): + problems.append("Key Secret is required") from_num = self.from_var.get().strip() if not from_num or from_num == "+1XXXXXXXXXX": problems.append("From Number is required") @@ -252,7 +254,6 @@ class AutoFaxApp: def run_loop(self): self._do_send() while self.running: - # Sleep in small increments so we can stop quickly for _ in range(3600): if not self.running: return @@ -262,47 +263,35 @@ class AutoFaxApp: # ── Fax logic ─────────────────────────────────────────────────── - def _headers(self): - return {"Authorization": f"Bearer {self.api_key_var.get().strip()}"} + def _auth(self): + return (self.key_id_var.get().strip(), self.key_secret_var.get().strip()) + + def _api_url(self): + return f"https://fax.api.sinch.com/v3/projects/{self.project_id_var.get().strip()}" def _do_send(self): pdf_path = Path(self.pdf_var.get().strip()) to_number = self.to_var.get().strip() from_number = self.from_var.get().strip() - conn_id = self.conn_id_var.get().strip() timestamp = datetime.now(timezone.utc).isoformat() try: - # Upload media - self.log(f"Uploading {pdf_path.name}...") - media_name = f"autofax-claim-{pdf_path.stem}" + self.log(f"Sending fax to {to_number}...") with open(pdf_path, "rb") as f: resp = requests.post( - f"{API}/media", - headers={**self._headers(), "Content-Type": "application/pdf"}, - params={"media_name": media_name}, - data=f.read(), + f"{self._api_url()}/faxes", + auth=self._auth(), + files={"file": (pdf_path.name, f, "application/pdf")}, + data={ + "to": to_number, + "from": from_number, + }, timeout=60, ) resp.raise_for_status() - - # Send fax - self.log(f"Sending fax to {to_number}...") - resp = requests.post( - f"{API}/faxes", - headers={**self._headers(), "Content-Type": "application/json"}, - json={ - "connection_id": conn_id, - "media_name": media_name, - "to": to_number, - "from": from_number, - }, - timeout=30, - ) - resp.raise_for_status() - fax_data = resp.json().get("data", {}) + fax_data = resp.json() fax_id = fax_data.get("id") - status = fax_data.get("status", "queued") + status = fax_data.get("status", "QUEUED") entry = { "timestamp": timestamp, @@ -315,21 +304,22 @@ class AutoFaxApp: } # Brief wait then check status - if fax_id and status == "queued": + if fax_id and status == "QUEUED": time.sleep(15) try: sr = requests.get( - f"{API}/faxes/{fax_id}", headers=self._headers(), timeout=15, + f"{self._api_url()}/faxes/{fax_id}", + auth=self._auth(), timeout=15, ) sr.raise_for_status() - sd = sr.json().get("data", {}) + sd = sr.json() entry["status"] = sd.get("status", status) - if entry["status"] == "failed": - entry["failure_reason"] = sd.get("failure_reason", "Unknown") + if entry["status"] == "FAILURE": + entry["failure_reason"] = sd.get("failureReason", "Unknown") except Exception: pass - self.log(f"Fax {entry['status'].upper()} (ID: {fax_id})") + self.log(f"Fax {entry['status']} (ID: {fax_id})") except requests.HTTPError as e: error_detail = "" @@ -359,7 +349,6 @@ class AutoFaxApp: } self.log(f"FAILED: {e}") - # Save to log and regenerate report self._save_entry(entry) self._generate_report() self._notify(entry) @@ -382,8 +371,8 @@ class AutoFaxApp: return try: status_labels = { - "delivered": "Fax Delivered", "sent": "Fax Sent", - "queued": "Fax Queued", "sending": "Fax Sending", + "COMPLETED": "Fax Delivered", "IN_PROGRESS": "Fax Sending", + "QUEUED": "Fax Queued", } subject = status_labels.get(entry["status"], "Fax Failed") msg = f"To: {entry['to']}\nStatus: {entry['status']}\nFile: {entry['file']}" @@ -393,7 +382,7 @@ class AutoFaxApp: ntfy_token = self.ntfy_token_var.get().strip() if ntfy_token and not ntfy_token.startswith("Bearer tk_..."): hdrs["Authorization"] = ntfy_token - if entry["status"] not in ("queued", "sending", "sent", "delivered"): + if entry["status"] not in ("QUEUED", "IN_PROGRESS", "COMPLETED"): hdrs["Priority"] = "high" requests.post(ntfy_url, data=msg.encode(), headers=hdrs, timeout=10) except Exception: @@ -412,17 +401,18 @@ class AutoFaxApp: return total = len(entries) - failed = sum(1 for e in entries if e["status"] not in ("queued", "sending", "sent", "delivered")) - succeeded = sum(1 for e in entries if e["status"] == "delivered") + success_statuses = {"QUEUED", "IN_PROGRESS", "COMPLETED"} + failed = sum(1 for e in entries if e["status"] not in success_statuses) + succeeded = sum(1 for e in entries if e["status"] == "COMPLETED") pending = total - failed - succeeded rows = "" for i, e in enumerate(entries, 1): status = e["status"] - if status in ("delivered", "sent"): - sc, sd = "success", status.upper() - elif status in ("queued", "sending"): - sc, sd = "pending", status.upper() + if status == "COMPLETED": + sc, sd = "success", "DELIVERED" + elif status in ("QUEUED", "IN_PROGRESS"): + sc, sd = "pending", status else: sc, sd = "failed", "FAILED" @@ -482,12 +472,11 @@ once per hour. FAILED = the receiving fax machine did not accept.

# ── Helpers ───────────────────────────────────────────────────── def update_status_display(self): - """Update the status label with attempt count.""" if LOG_FILE.exists(): try: entries = json.loads(LOG_FILE.read_text()) total = len(entries) - failed = sum(1 for e in entries if e["status"] not in ("queued", "sending", "sent", "delivered")) + failed = sum(1 for e in entries if e["status"] not in ("QUEUED", "IN_PROGRESS", "COMPLETED")) if total > 0: extra = f" | {total} sent, {failed} failed" current = self.status_label.cget("text").split(" |")[0] @@ -500,7 +489,6 @@ once per hour. FAILED = the receiving fax machine did not accept.

if not report.exists(): messagebox.showinfo("No Report", "No fax attempts recorded yet.") return - # Cross-platform open if sys.platform == "win32": os.startfile(str(report)) elif sys.platform == "darwin": diff --git a/setup_telnyx.py b/setup_telnyx.py deleted file mode 100755 index 9a0412b..0000000 --- a/setup_telnyx.py +++ /dev/null @@ -1,259 +0,0 @@ -#!/usr/bin/env python3 -""" -Telnyx setup wizard - automates account provisioning for AutoFax. - -Run this once after signing up at portal.telnyx.com and getting an API key. -It will: - 1. Create a Fax Application - 2. Search for and purchase a fax-capable phone number - 3. Assign the number to the Fax Application - 4. Write the .env file (or autofax_config.json for the GUI) - -Usage: - python setup_telnyx.py # interactive - python setup_telnyx.py --api-key KEY_xxx --to +18019382102 # non-interactive -""" - -import argparse -import json -import sys -import time -from pathlib import Path - -import requests - -API = "https://api.telnyx.com/v2" -BASE_DIR = Path(__file__).parent - - -def api(method, path, token, **kwargs): - """Make a Telnyx API call. Raises on HTTP error with details.""" - resp = requests.request( - method, - f"{API}{path}", - headers={ - "Authorization": f"Bearer {token}", - "Content-Type": "application/json", - }, - **kwargs, - ) - if not resp.ok: - detail = resp.text[:500] - print(f" API error ({resp.status_code}): {detail}", file=sys.stderr) - resp.raise_for_status() - return resp.json() - - -def create_fax_application(token: str, name: str = "AutoFax") -> dict: - """Create a Fax Application. Returns the app data including connection ID.""" - print(f"Creating Fax Application '{name}'...") - data = api("POST", "/fax_applications", token, json={ - "application_name": name, - "webhook_event_url": "https://example.com/fax-status", - "active": True, - }) - app = data["data"] - print(f" Created: {app['application_name']} (ID: {app['id']})") - return app - - -def search_fax_numbers(token: str, area_code: str = None, limit: int = 5) -> list: - """Search for available fax-capable phone numbers.""" - print("Searching for fax-capable numbers...") - params = { - "filter[country_code]": "US", - "filter[features]": "fax", - "filter[phone_number_type]": "local", - "filter[limit]": limit, - } - if area_code: - params["filter[national_destination_code]"] = area_code - - data = api("GET", "/available_phone_numbers", token, params=params) - numbers = data.get("data", []) - if not numbers: - print(" No numbers found. Try a different area code.") - return numbers - - -def order_number(token: str, phone_number: str, connection_id: str) -> dict: - """Purchase a phone number and assign it to the fax application.""" - print(f"Ordering {phone_number}...") - data = api("POST", "/number_orders", token, json={ - "phone_numbers": [{"phone_number": phone_number}], - "connection_id": connection_id, - }) - order = data["data"] - print(f" Order ID: {order['id']} | Status: {order.get('status', 'pending')}") - - # Wait for order to complete - order_id = order["id"] - for attempt in range(12): - time.sleep(5) - check = api("GET", f"/number_orders/{order_id}", token) - status = check["data"].get("status", "pending") - if status == "success": - print(f" Number purchased successfully!") - return check["data"] - elif status in ("failed",): - print(f" Order failed: {check['data']}", file=sys.stderr) - sys.exit(1) - print(f" Waiting... ({status})") - - print(" Order is still processing. Check the Telnyx portal.") - return order - - -def assign_number_to_app(token: str, phone_number_id: str, connection_id: str): - """Assign a purchased number to the fax application.""" - print(f"Assigning number to fax application...") - api("PATCH", f"/phone_numbers/{phone_number_id}", token, json={ - "connection_id": connection_id, - }) - print(" Done.") - - -def write_env(api_key: str, connection_id: str, from_number: str, to_number: str): - """Write the .env file for the CLI workflow.""" - env_path = BASE_DIR / ".env" - content = f"""# Generated by setup_telnyx.py -TELNYX_API_KEY={api_key} -TELNYX_CONNECTION_ID={connection_id} -TELNYX_FROM_NUMBER={from_number} -FAX_TO_NUMBER={to_number} - -# ntfy notification URL (optional) -# NTFY_URL=https://nt.nevo.engineer/autofax -# NTFY_TOKEN=Bearer tk_your_token_here -""" - env_path.write_text(content) - print(f"\n Wrote {env_path}") - - -def write_gui_config(api_key: str, connection_id: str, from_number: str, to_number: str): - """Write the GUI config file.""" - config_path = BASE_DIR / "autofax_config.json" - config = { - "api_key": api_key, - "connection_id": connection_id, - "from_number": from_number, - "to_number": to_number, - "pdf_path": "", - "ntfy_url": "", - "ntfy_token": "", - } - config_path.write_text(json.dumps(config, indent=2)) - print(f" Wrote {config_path}") - - -def pick_number(numbers: list) -> str: - """Let the user pick from available numbers.""" - print("\nAvailable numbers:") - for i, n in enumerate(numbers, 1): - pn = n.get("phone_number", "?") - region = n.get("region_information", [{}]) - loc = "" - if region: - loc = f" ({region[0].get('region_name', '')}, {region[0].get('region_type', '')})" - cost = n.get("cost_information", {}).get("monthly_cost", "?") - print(f" [{i}] {pn}{loc} - ${cost}/mo") - - while True: - try: - choice = input(f"\nSelect a number [1-{len(numbers)}]: ").strip() - idx = int(choice) - 1 - if 0 <= idx < len(numbers): - return numbers[idx]["phone_number"] - except (ValueError, EOFError): - pass - print("Invalid choice, try again.") - - -def main(): - parser = argparse.ArgumentParser(description="Set up Telnyx for AutoFax") - parser.add_argument("--api-key", help="Telnyx API key (starts with KEY_)") - parser.add_argument("--to", help="Destination fax number in E.164 format") - parser.add_argument("--area-code", help="Preferred area code for your fax number") - parser.add_argument("--app-name", default="AutoFax", help="Fax application name") - args = parser.parse_args() - - print("=" * 50) - print(" AutoFax - Telnyx Setup Wizard") - print("=" * 50) - print() - - # Get API key - api_key = args.api_key - if not api_key: - print("You need a Telnyx API key. Get one at:") - print(" https://portal.telnyx.com/ -> API Keys") - print() - api_key = input("Paste your API key: ").strip() - - if not api_key.startswith("KEY"): - print("Warning: API key usually starts with 'KEY'. Continuing anyway...") - - # Get destination number - to_number = args.to - if not to_number: - to_number = input("Destination fax number (E.164, e.g. +18019382102): ").strip() - - # Verify API key works - print("\nVerifying API key...") - try: - api("GET", "/balance", api_key) - print(" API key is valid.") - except Exception: - print("ERROR: Invalid API key or account issue.", file=sys.stderr) - sys.exit(1) - - # Step 1: Create fax application - print() - app = create_fax_application(api_key, args.app_name) - connection_id = app["id"] - - # Step 2: Search for a number - print() - numbers = search_fax_numbers(api_key, area_code=args.area_code) - if not numbers: - # Try without area code filter - numbers = search_fax_numbers(api_key) - if not numbers: - print("ERROR: No fax numbers available. Try again later.", file=sys.stderr) - sys.exit(1) - - # Step 3: Pick and order a number - if args.area_code and len(numbers) == 1: - chosen = numbers[0]["phone_number"] - print(f"\nAuto-selected: {chosen}") - else: - chosen = pick_number(numbers) - - print() - order_number(api_key, chosen, connection_id) - - # Step 4: Write config files - print() - print("Writing configuration...") - write_env(api_key, connection_id, chosen, to_number) - write_gui_config(api_key, connection_id, chosen, to_number) - - print() - print("=" * 50) - print(" Setup complete!") - print("=" * 50) - print() - print(f" Fax Application: {app['application_name']}") - print(f" Connection ID: {connection_id}") - print(f" From Number: {chosen}") - print(f" To Number: {to_number}") - print() - print("Next steps:") - print(" 1. Place your claim PDF in the claims/ directory") - print(" 2. Linux: ./install.sh") - print(" 3. Windows: Double-click AutoFax.exe") - print() - - -if __name__ == "__main__": - main()