"""Telegram bot: send a job URL (or paste a JD) → get a tailored resume PDF. Third touchpoint alongside the Streamlit UI and the Chrome extension. It reuses the SAME pipeline (hardcoded default resume + uncapped extraction + structured placement + Tectonic compile) — no ATS logic lives here, only Telegram I/O and orchestration. Wiring: api_server.py exposes POST /telegram/webhook, which hands each update to `process_update()` in a background task (so Telegram gets an instant 200 while the ~1 min generation runs and the PDF is sent asynchronously). Required env (HF Space secrets): TELEGRAM_BOT_TOKEN — from @BotFather TELEGRAM_ALLOWED_USER_IDS — comma-separated numeric Telegram user IDs (allowlist) TELEGRAM_WEBHOOK_SECRET — optional; validates the webhook came from Telegram """ from __future__ import annotations import os import re import time import socket import logging import tempfile import shutil import requests log = logging.getLogger("telegram_bot") # HF Spaces' IPv6 egress to api.telegram.org black-holes (TCP connects but reads # hang → "Read timed out"), while IPv4 works fine. Force IPv4 for all urllib3 # (requests) connections so the bot can actually reach Telegram. Safe: IPv4 works # for every host we call. try: import urllib3.util.connection as _u3c _u3c.allowed_gai_family = lambda: socket.AF_INET except Exception: # noqa: BLE001 pass _API = "https://api.telegram.org/bot{token}/{method}" _URL_RE = re.compile(r"https?://\S+") _SEND_RETRIES = 3 def _proxies(): """Optional proxy for reaching api.telegram.org when HF egress is blocked. Uses TELEGRAM_PROXY, else the first SCRAPER_PROXIES / SCRAPER_PROXY entry.""" p = os.getenv("TELEGRAM_PROXY", "").strip() if not p: raw = os.getenv("SCRAPER_PROXIES", "") or os.getenv("SCRAPER_PROXY", "") p = (raw.replace("\n", ",").split(",")[0].strip()) if raw else "" return {"http": p, "https": p} if p else None def diagnose() -> dict: """Report whether the Space can actually reach api.telegram.org (direct and, if configured, via proxy). Returns reachability only — never the token.""" res = {"has_token": bool(_token()), "proxy_configured": bool(_proxies())} tok = _token() if not tok: return res url = _API.format(token=tok, method="getMe") attempts = [("direct", None)] if _proxies(): attempts.append(("proxy", _proxies())) for label, prox in attempts: t = time.time() try: r = requests.get(url, timeout=(5, 12), proxies=prox) res[label] = {"ok": bool(r.ok), "status": r.status_code, "elapsed_s": round(time.time() - t, 1)} except Exception as exc: # noqa: BLE001 res[label] = {"ok": False, "error": str(exc)[:160], "elapsed_s": round(time.time() - t, 1)} return res def _token() -> str: return os.getenv("TELEGRAM_BOT_TOKEN", "").strip() def is_configured() -> bool: return bool(_token()) def _allowed_ids() -> set: raw = os.getenv("TELEGRAM_ALLOWED_USER_IDS", "") ids = set() for part in re.split(r"[,\s]+", raw or ""): part = part.strip() if part.lstrip("-").isdigit(): ids.add(int(part)) return ids def is_allowed(user_id) -> bool: """Allowlist-only: if no IDs are configured, NOBODY is allowed (fail closed).""" ids = _allowed_ids() if not ids: log.warning("TELEGRAM_ALLOWED_USER_IDS not set — denying all (fail-closed).") return False try: return int(user_id) in ids except (TypeError, ValueError): return False # ── Telegram Bot API helpers ────────────────────────────────────────────────── def _api(method: str, **payload): tok = _token() if not tok: return None url = _API.format(token=tok, method=method) prox = _proxies() last = None for attempt in range(_SEND_RETRIES): try: return requests.post(url, json=payload, timeout=(10, 30), proxies=prox) except Exception as exc: # noqa: BLE001 last = exc time.sleep(1.5 * (attempt + 1)) log.warning("telegram %s failed after %d tries: %s", method, _SEND_RETRIES, last) return None def send_message(chat_id, text: str): return _api("sendMessage", chat_id=chat_id, text=text, disable_web_page_preview=True) def send_document(chat_id, file_path: str, caption: str = ""): tok = _token() if not tok or not os.path.exists(file_path): return None url = _API.format(token=tok, method="sendDocument") prox = _proxies() last = None for attempt in range(_SEND_RETRIES): try: with open(file_path, "rb") as fh: return requests.post( url, data={"chat_id": chat_id, "caption": caption[:1024]}, files={"document": (os.path.basename(file_path), fh, "application/pdf")}, timeout=(10, 180), proxies=prox, ) except Exception as exc: # noqa: BLE001 last = exc time.sleep(1.5 * (attempt + 1)) log.warning("telegram sendDocument failed after %d tries: %s", _SEND_RETRIES, last) return None _HELP = ( "👋 Send me a *job link* (or paste the full job description) and I'll send " "back your tailored, ATS-optimized resume PDF.\n\n" "• Company/ATS links (Greenhouse, Lever, Ashby, Naukri) usually work directly.\n" "• If I can't read a LinkedIn/Indeed link, just copy the description text and " "send that instead.\n\n" "Commands: /start, /help" ) # ── Core orchestration ────────────────────────────────────────────────────────── def process_update(update: dict) -> None: """Handle one Telegram update end-to-end. Safe to run in a background task.""" try: msg = update.get("message") or update.get("edited_message") or {} chat = msg.get("chat") or {} chat_id = chat.get("id") user_id = (msg.get("from") or {}).get("id") text = (msg.get("text") or "").strip() if chat_id is None: return if not is_allowed(user_id): send_message(chat_id, "⛔ This bot is private and not authorized for this account.") return if not text or text.lower() in ("/start", "/help", "start", "help"): send_message(chat_id, _HELP) return # 1. Resolve the JD: a URL we fetch, or pasted description text. url_match = _URL_RE.search(text) job_title = "" if url_match: url = url_match.group(0).rstrip(").,") send_message(chat_id, "🔍 Reading the job posting…") from src.jd_from_url import fetch_jd_from_url res = fetch_jd_from_url(url) if not res.get("ok"): send_message( chat_id, "⚠️ I couldn't read that link (sites like LinkedIn/Indeed block " "server access). Please copy the *job description text* and send " "it to me directly — I'll tailor your resume from that.", ) return jd_text = res["jd_text"] job_title = res.get("job_title", "") elif len(text) >= 200: jd_text = text # treat a long paste as the JD itself else: send_message( chat_id, "Send me a *job link*, or paste the full *job description* " "(at least a paragraph) and I'll tailor your resume.", ) return # 2. Generate via the SAME pipeline (Maximum ATS Mode, hardcoded resume). send_message(chat_id, "⚙️ Generating your ATS-optimized resume… (up to ~1–2 min)") out_dir = tempfile.mkdtemp(prefix="tg_resume_") try: from src.default_resume import get_default_resume_latex from src.latex_resume import optimize_latex_resume try: from src.candidate_vault import user_blocked_terms blocked = list(user_blocked_terms()) except Exception: # noqa: BLE001 blocked = [] report = optimize_latex_resume( get_default_resume_latex(), jd_text, maximum_ats_mode=True, blocked_terms=blocked, compile_pdf=True, out_dir=out_dir, job_title=job_title, ) pct = report.get("pct", 0) or 0 pdf_path = report.get("pdf_path") if report.get("compiled") and pdf_path and os.path.exists(pdf_path): cap = f"✅ Tailored resume — ~{pct}% JD keyword coverage." if job_title: cap = f"✅ {job_title} — ~{pct}% JD keyword coverage." send_document(chat_id, pdf_path, caption=cap) else: # Compile failed — still hand over the .tex so nothing is lost. tex = report.get("tex", "") if tex: tex_path = os.path.join(out_dir, "resume.tex") with open(tex_path, "w", encoding="utf-8") as f: f.write(tex) send_document(chat_id, tex_path, caption="⚠️ PDF compile failed — here's the .tex " "(compile at overleaf.com).") else: send_message(chat_id, "❌ Sorry, generation failed. Please try again.") finally: shutil.rmtree(out_dir, ignore_errors=True) except Exception as exc: # noqa: BLE001 - never let a background task crash silently log.exception("process_update failed: %s", exc) try: cid = ((update.get("message") or {}).get("chat") or {}).get("id") if cid is not None: send_message(cid, f"❌ Something went wrong: {str(exc)[:200]}") except Exception: # noqa: BLE001 pass