#!/usr/bin/env python3
"""
Full Teamtailor export -> /root/tt-export
  jobs.jsonl          every job, all four statuses
  candidates.jsonl    every candidate + their CV downloaded to cvs/<id>.pdf
  applications.jsonl  the join: candidate_id + job_id + stage + dates

Resumable:Every page is checkpointed, so a restart continues where it stopped.
CV links are presigned for 60s, so each page's PDFs are pulled immediately
after that page is fetched, never collected for later.
"""
import json, os, sys, time, urllib.request, urllib.error, threading
from concurrent.futures import ThreadPoolExecutor

OUT = "/root/tt-export"
CVS = f"{OUT}/cvs"
KEY = open(f"{OUT}/.ttkey").read().strip()
VER = "20240904"
BASE = "https://api.teamtailor.com/v1"
os.makedirs(CVS, exist_ok=True)

import subprocess

def notify(msg):
    """Ping Telegram through Ultron's existing helper (token lives outside the
    repo). Never let a failed notification kill the export."""
    try:
        subprocess.run(
            ["node", "-e",
             'require("/root/ultron/scripts/notify.js").send(process.argv[1])', "--", msg],
            cwd="/root/ultron", timeout=30,
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    except Exception:
        pass

lock = threading.Lock()
def log(m):
    line = f"[{time.strftime('%H:%M:%S')}] {m}"
    with lock:
        print(line, flush=True)

def api(path, tries=6):
    url = path if path.startswith("http") else BASE + path
    for i in range(tries):
        try:
            r = urllib.request.Request(url, headers={
                "Authorization": f"Token token={KEY}", "X-Api-Version": VER,
                "Accept": "application/vnd.api+json"})
            with urllib.request.urlopen(r, timeout=90) as resp:
                return json.loads(resp.read())
        except urllib.error.HTTPError as e:
            if e.code in (429, 500, 502, 503, 504) and i < tries - 1:
                wait = min(60, 3 * (2 ** i))
                log(f"  HTTP {e.code}, backing off {wait}s")
                time.sleep(wait); continue
            raise
        except Exception:
            if i == tries - 1: raise
            time.sleep(3 * (i + 1))

def state_get(k, d=None):
    try: return json.load(open(f"{OUT}/state.json")).get(k, d)
    except Exception: return d

def state_set(k, v):
    try: s = json.load(open(f"{OUT}/state.json"))
    except Exception: s = {}
    s[k] = v
    tmp = f"{OUT}/state.json.tmp"
    json.dump(s, open(tmp, "w")); os.replace(tmp, f"{OUT}/state.json")

def appender(name):
    f = open(f"{OUT}/{name}", "a", encoding="utf-8")
    def w(obj):
        f.write(json.dumps(obj, ensure_ascii=False) + "\n"); f.flush()
    return w

A = lambda r: r.get("attributes", {})

# ---------------------------------------------------------------- jobs
def do_jobs():
    if state_get("jobs_done"): return log("jobs: already done, skipping")
    w = appender("jobs.jsonl"); n = 0
    for st in (None, "archived", "unlisted", "draft"):
        q = "/jobs?page%5Bsize%5D=30" if st is None else f"/jobs?filter%5Bstatus%5D={st}&page%5Bsize%5D=30"
        while q:
            d = api(q)
            for j in d.get("data", []):
                a = A(j)
                w({"job_id": j["id"], "job_title": a.get("title") or a.get("internal-name"),
                   "status": st or "published", "internal_name": a.get("internal-name"),
                   "employment_type": a.get("employment-type"), "recruiter_email": a.get("recruiter-email"),
                   "created_at": a.get("created-at"), "end_date": a.get("end-date"),
                   "tags": a.get("tags"), "pitch": a.get("pitch")})
                n += 1
            q = (d.get("links") or {}).get("next")
    state_set("jobs_done", True); log(f"jobs: {n} written")

# ------------------------------------------------------- candidates + CVs
def fetch_cv(item):
    cid, url = item
    path = f"{CVS}/{cid}.pdf"
    if os.path.exists(path) and os.path.getsize(path) > 0:
        return ("skip", cid)
    try:
        with urllib.request.urlopen(url, timeout=90) as r:
            b = r.read()
        if not b: return ("empty", cid)
        tmp = path + ".part"
        open(tmp, "wb").write(b); os.replace(tmp, path)
        return ("ok", cid)
    except Exception as e:
        return (f"fail:{getattr(e,'code',type(e).__name__)}", cid)

def do_candidates():
    page = state_get("cand_page", 1)
    w = appender("candidates.jsonl")
    wf = appender("cv_failures.jsonl")
    got = state_get("cand_count", 0); cvs = state_get("cv_count", 0)
    while True:
        d = api(f"/candidates?page%5Bsize%5D=30&page%5Bnumber%5D={page}")
        rows = d.get("data", [])
        if not rows:
            break
        total_pages = (d.get("meta") or {}).get("page-count")
        pending = []
        for c in rows:
            a = A(c); cid = c["id"]
            w({"candidate_id": cid, "first_name": a.get("first-name"), "last_name": a.get("last-name"),
               "email": a.get("email"), "phone": a.get("phone"), "city": a.get("city"),
               "state": a.get("state"), "country": a.get("country"), "zip": a.get("zip"),
               "address": a.get("address"), "linkedin_url": a.get("linkedin-url"),
               "linkedin_profile": a.get("linkedin-profile"), "resume_summary": a.get("resume-summary"),
               "pitch": a.get("pitch"), "tags": a.get("tags"), "sourced": a.get("sourced"),
               "referred": a.get("referred"), "referring_site": a.get("referring-site"),
               "created_at": a.get("created-at"), "updated_at": a.get("updated-at"),
               "unsubscribed": a.get("unsubscribed"),
               "consent_future_jobs_at": a.get("consent-future-jobs-at"),
               "restricted_at": a.get("restricted-at"),
               "has_cv": bool(a.get("resume"))})
            got += 1
            if a.get("resume"):
                pending.append((cid, a["resume"]))
        # 60-second signed links: download NOW, in parallel, before they die.
        if pending:
            with ThreadPoolExecutor(max_workers=10) as ex:
                for status, cid in ex.map(fetch_cv, pending):
                    if status in ("ok", "skip"): cvs += 1
                    else: wf({"candidate_id": cid, "reason": status})
        state_set("cand_page", page + 1); state_set("cand_count", got); state_set("cv_count", cvs)
        if page % 10 == 0 or page == 1:
            log(f"candidates page {page}/{total_pages}  rows={got}  cvs={cvs}")
        if total_pages:
            pct = int(100 * page / total_pages)
            last = state_get("cand_pct", 0)
            if pct >= last + 10:
                state_set("cand_pct", pct - pct % 10)
                gb = sum(os.path.getsize(os.path.join(CVS, x))
                         for x in os.listdir(CVS)) / 1024**3
                notify(f"📥 Teamtailor export — candidates {pct}%\n"
                       f"{got:,} people · {cvs:,} CVs · {gb:.1f} GB\n"
                       f"page {page}/{total_pages}")
        page += 1
    state_set("cand_done", True); log(f"candidates: {got} rows, {cvs} CVs")

# -------------------------------------------------------- applications
def do_apps():
    page = state_get("app_page", 1)
    w = appender("applications.jsonl")
    got = state_get("app_count", 0)
    while True:
        d = api(f"/job-applications?include=job,stage&page%5Bsize%5D=30&page%5Bnumber%5D={page}")
        rows = d.get("data", [])
        if not rows: break
        total_pages = (d.get("meta") or {}).get("page-count")
        inc = {(r["type"], r["id"]): r for r in d.get("included", [])}
        for ap in rows:
            a = A(ap); rel = ap.get("relationships", {})
            rid = lambda n: ((rel.get(n) or {}).get("data") or {}).get("id")
            sid = rid("stage")
            w({"application_id": ap["id"], "candidate_id": rid("candidate"),
               "job_id": rid("job"), "stage_id": sid,
               "stage": A(inc.get(("stages", sid), {})).get("name") if sid else None,
               "applied_at": a.get("created-at"), "rejected_at": a.get("rejected-at"),
               "stage_changed_at": a.get("changed-stage-at"),
               "referring_site": a.get("referring-site"), "match": a.get("match"),
               "cover_letter": a.get("cover-letter")})
            got += 1
        state_set("app_page", page + 1); state_set("app_count", got)
        if page % 25 == 0 or page == 1:
            log(f"applications page {page}/{total_pages}  rows={got}")
        if total_pages:
            pct = int(100 * page / total_pages)
            last = state_get("app_pct", 0)
            if pct >= last + 25:
                state_set("app_pct", pct - pct % 25)
                notify(f"🔗 Teamtailor export — applications {pct}% ({got:,} rows)")
        page += 1
    state_set("apps_done", True); log(f"applications: {got} rows")

def main():
        t0 = time.time()
        log("=== export start ===")
        if not state_get("started_pinged"):
            notify("🚀 Teamtailor export started on the VPS.\n"
                   "~131 jobs · 80k candidates · 98k applications · ~36 GB of CVs.\n"
                   "I'll ping every 10%.")
            state_set("started_pinged", True)
        do_jobs()
        if not state_get("apps_done"): do_apps()      # fast, do it first
        else: log("applications: already done, skipping")
        if not state_get("cand_done"): do_candidates()  # slow (CV downloads)
        else: log("candidates: already done, skipping")
        ncv = len(os.listdir(CVS))
        gb = sum(os.path.getsize(os.path.join(CVS, x)) for x in os.listdir(CVS)) / 1024**3
        fails = 0
        try: fails = sum(1 for _ in open(f"{OUT}/cv_failures.jsonl"))
        except Exception: pass
        notify(f"✅ Teamtailor export DONE in {(time.time()-t0)/3600:.1f} h\n"
               f"{state_get('cand_count',0):,} candidates\n"
               f"{state_get('app_count',0):,} applications\n"
               f"{ncv:,} CVs ({gb:.1f} GB)\n"
               f"{fails} CV failures (retryable)")
        log(f"=== DONE in {(time.time()-t0)/3600:.2f} h ===")


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        # Report, then exit non-zero so systemd restarts us. Every phase is
        # checkpointed in state.json, so the restart resumes rather than redoes.
        notify(f"⚠️ Teamtailor export crashed: {type(e).__name__}: {str(e)[:200]}\n"
               f"Auto-restarting in 30s and resuming from the last checkpoint.")
        log(f"CRASH {type(e).__name__}: {e}")
        raise