#!/usr/bin/env python3
"""
serve_dashboard.py — small local web app for the job-search dashboard.

Run on the Mac from the Job Search folder:

    python3 "serve_dashboard.py"

Then open  http://<Mac-Tailscale-IP>:8787/  on the iPhone (or proxy it with
`tailscale serve --bg 8787` for a stable HTTPS URL — see README_dashboard.txt).

What it does:
  GET  /            dashboard built fresh from the xlsx on every request
  GET  /api/version xlsx mtime — the page polls this to auto-refresh
  POST /api/status  change an application's Stage or a lead's Odds
  POST /api/apply   move a Leads row to Job Tracker with today's date
  POST /api/dead    move a Leads row to the "Dead Leads" archive sheet
  POST /api/review_move  move a "Manual Review" row to Leads or Eng Leads
  POST /api/archive move a Job Tracker row to the "Archived" sheet
  POST /api/unarchive  move an Archived row back to Job Tracker
  POST /api/restart re-exec the server (picks up updated script code)

Writes go through tracker_io.py (shared with sweep sessions — see its
docstring for the full protocol):
  - .tracker.lock guards against concurrent writers (e.g. the daily sweep);
    if the tracker is busy/locked you get a clear error, nothing breaks
  - ONE rolling daily backup: _Backups/tracker-YYYY-MM-DD.xlsx, pruned to 5
    (no more per-session backup files in the main folder)
  - schema validation before every save — a corrupted workbook is never written
  - verified in-place save (Nextcloud-safe; see tracker_io.save_workbook)
  - dashboard.html is regenerated after each successful write

Background threads (daemons, restarted automatically by /api/restart):
  - Apify scrape timer: runs apify_jobs.py daily at 07:45 and 12:45 local.
    Each slot fires at most once per day (.apify_timer_state.json); slots
    missed by more than 3 h (server was down) are logged and skipped.
    A failed run is retried up to 2 more times at ~15-min intervals, still
    inside the slot's 3-h freshness window; a success ends the slot for
    the day. Every attempt and the final outcome land in .apify_log.txt.
  - JS-page verifier: every ~30 min, checks Manual Review rows whose Notes
    say a posting was unverifiable/JS-blocked, renders the posting + the
    company's board via Apify (verify_js_url.py), then either stamps the row
    "verified live <date> via JS render" or moves it to Dead Leads with the
    evidence. Also processes .verify_queue.json entries queued by sweep
    sessions. Daily budget cap: verifier.max_renders_per_day in
    .apify_config.json (shared with verify_js_url.py's CLI).
  Both log to .apify_log.txt and never take the server down with them.

Requires: openpyxl. If missing, this script prints the exact install command.
"""

import json
import os
import sys
import threading
import time
import subprocess
from datetime import date, datetime, timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

try:
    import openpyxl  # noqa: F401
except ImportError:
    sys.exit("\nopenpyxl is not installed. Run this once, then re-run me:\n\n"
             "    python3 -m pip install --user openpyxl\n")

sys.path.insert(0, str(Path(__file__).resolve().parent))
import generate_dashboard as gd  # noqa: E402  (same folder)
import verify_js_url as vjs  # noqa: E402  (same folder: Apify render + verdict helpers)
import tracker_io as tio  # noqa: E402  (same folder: lock/backup/validate/save)

PORT = 8787
def _jobdash_data_dir(_default):
    """Pi deployment: use the Nextcloud data-dir copy of the Job Search folder
    when jobdash_pi.json (written by deploy_to_pi.py) sits next to this script."""
    try:
        import json as _json
        return Path(_json.loads(
            (_default / "jobdash_pi.json").read_text())["data_dir"])
    except Exception:
        return _default
HERE = _jobdash_data_dir(Path(__file__).resolve().parent)
SCRIPT = Path(__file__).resolve()
XLSX = tio.XLSX
LOCK = tio.LOCK

# Unique per process — lets the page detect that a NEW process is answering
# after /api/restart (the xlsx mtime alone doesn't change on restart).
BOOT_ID = f"{os.getpid()}-{time.time()}"

VALID_STAGES = {s.lower() for s in gd.STAGES}
VALID_ODDS = {s.lower() for s in gd.ODDS}


# ---------- workbook plumbing (see tracker_io.py for the write protocol) ----------

TrackerBusy = tio.TrackerBusy


def mutate_workbook(mutator):
    """tracker_io.mutate (lock -> rolling daily backup -> load with formulas
    kept -> mutate -> schema-validate -> verified in-place save), then
    regenerate the static dashboard.html."""
    result = tio.mutate(mutator)
    gd.main()  # refresh static dashboard.html
    tio.nc_scan()  # Pi: index the regenerated dashboard.html
    return result


def header_map(ws):
    return {str(c.value).strip(): c.column for c in ws[1] if c.value}


def check_row(ws, row, company, col_company):
    actual = ws.cell(row=row, column=col_company).value
    if str(actual or "").strip().lower() != str(company or "").strip().lower():
        raise LookupError(
            "Tracker changed since this page loaded — pull down to refresh and retry.")


def first_empty_row(ws, col_company):
    r = 2
    while ws.cell(row=r, column=col_company).value not in (None, ""):
        r += 1
    return r


# ---------- actions ----------

LEAD_SHEETS = {"leads": "Leads", "eng": "Eng Leads", "review": "Manual Review"}


def do_status(payload):
    sheet = payload.get("sheet")
    row = int(payload.get("row", 0))
    company = payload.get("company", "")
    value = str(payload.get("value", "")).strip()
    if row < 2:
        raise ValueError("Bad row.")
    if sheet == "apps":
        if value.lower() not in VALID_STAGES:
            raise ValueError(f"Unknown stage: {value}")
        sheet_name, col_name = "Job Tracker", "Stage"
    elif sheet in LEAD_SHEETS:
        if value.lower() not in VALID_ODDS:
            raise ValueError(f"Unknown odds: {value}")
        sheet_name, col_name = LEAD_SHEETS[sheet], "Odds (honest)"
    else:
        raise ValueError("Bad sheet.")

    def m(wb):
        ws = wb[sheet_name]
        h = header_map(ws)
        check_row(ws, row, company, h["Company"])
        ws.cell(row=row, column=h[col_name], value=value)
        return {"ok": True}

    return mutate_workbook(m)


def do_apply(payload):
    row = int(payload.get("row", 0))
    company = payload.get("company", "")
    src = LEAD_SHEETS.get(payload.get("sheet", "leads"))
    if not src:
        raise ValueError("Bad sheet.")
    if row < 2:
        raise ValueError("Bad row.")

    def m(wb):
        leads = wb[src]
        jt = wb["Job Tracker"]
        lh = header_map(leads)
        jh = header_map(jt)
        check_row(leads, row, company, lh["Company"])

        def lv(name):
            c = lh.get(name)
            return leads.cell(row=row, column=c).value if c else None

        salary = str(lv("Salary") or "").strip()
        notes = str(lv("Notes") or "").strip()
        if salary and "Salary" not in jh:
            # legacy tracker without a Salary column — fold into Notes as before
            notes = f"Salary: {salary} · {notes}" if notes else f"Salary: {salary}"
            salary = ""

        dest = first_empty_row(jt, jh["Company"])
        newvals = {
            "Company": lv("Company"),
            "Link to JD": lv("Link"),
            "Date Applied": datetime.combine(date.today(), datetime.min.time()),
            "Stage": "Applied",
            "Role": lv("Role"),
            "Location": lv("Location"),
            "Salary": salary or None,
            "Notes": notes or None,
        }
        for name, val in newvals.items():
            c = jh.get(name)
            if c and val not in (None, ""):
                cell = jt.cell(row=dest, column=c, value=val)
                if name == "Date Applied":
                    cell.number_format = "yyyy-mm-dd"
        leads.delete_rows(row)
        return {"ok": True, "moved_to_row": dest}

    return mutate_workbook(m)


def do_dead(payload):
    """Move a lead row (Leads or Eng Leads) to the 'Dead Leads' archive sheet."""
    row = int(payload.get("row", 0))
    company = payload.get("company", "")
    src = LEAD_SHEETS.get(payload.get("sheet", "leads"))
    if not src:
        raise ValueError("Bad sheet.")
    if row < 2:
        raise ValueError("Bad row.")

    def m(wb):
        leads = wb[src]
        lh = header_map(leads)
        check_row(leads, row, company, lh["Company"])
        headers = [str(c.value).strip() if c.value else "" for c in leads[1]]
        if "Dead Leads" in wb.sheetnames:
            dead = wb["Dead Leads"]
        else:
            dead = wb.create_sheet("Dead Leads")
            dead.append(headers + ["Marked Dead"])
        vals = [leads.cell(row=row, column=i).value for i in range(1, len(headers) + 1)]
        dead.append(vals + [date.today().isoformat()])
        leads.delete_rows(row)
        return {"ok": True}

    return mutate_workbook(m)


def do_review_move(payload):
    """Move a 'Manual Review' row to Leads or Eng Leads (copied by header
    name, appended at the first empty row). '✕' on a review card goes
    through /api/dead with sheet='review' instead."""
    row = int(payload.get("row", 0))
    company = payload.get("company", "")
    dest_name = {"leads": "Leads", "eng": "Eng Leads"}.get(payload.get("dest"))
    if not dest_name:
        raise ValueError("Bad dest.")
    if row < 2:
        raise ValueError("Bad row.")

    def m(wb):
        if "Manual Review" not in wb.sheetnames:
            raise LookupError("No Manual Review sheet — refresh the page.")
        src = wb["Manual Review"]
        dst = wb[dest_name]
        sh = header_map(src)
        dh = header_map(dst)
        check_row(src, row, company, sh["Company"])
        dest = first_empty_row(dst, dh["Company"])
        for name, col in sh.items():
            val = src.cell(row=row, column=col).value
            if val not in (None, "") and name in dh:
                dst.cell(dest, dh[name], val)
        src.delete_rows(row)
        return {"ok": True, "moved_to": dest_name, "moved_to_row": dest}

    return mutate_workbook(m)


def do_archive(payload):
    """Move a Job Tracker row to an 'Archived' sheet (created on demand)."""
    row = int(payload.get("row", 0))
    company = payload.get("company", "")
    if row < 2:
        raise ValueError("Bad row.")

    def m(wb):
        jt = wb["Job Tracker"]
        jh = header_map(jt)
        check_row(jt, row, company, jh["Company"])
        headers = [str(c.value).strip() if c.value else "" for c in jt[1]]
        if "Archived" in wb.sheetnames:
            arch = wb["Archived"]
            ah = header_map(arch)
            # grow Archived with any Job Tracker columns it lacks (e.g. Salary)
            # so values are copied by header name, never by raw position
            for h in headers:
                if h and h not in ah:
                    c = arch.max_column + 1
                    arch.cell(1, c, h)
                    ah[h] = c
        else:
            arch = wb.create_sheet("Archived")
            for i, h in enumerate(headers + ["Archived"], start=1):
                arch.cell(1, i, h)
            ah = header_map(arch)
        dest = first_empty_row(arch, ah["Company"])
        for i, name in enumerate(headers, start=1):
            val = jt.cell(row=row, column=i).value
            if val not in (None, "") and name in ah:
                cell = arch.cell(dest, ah[name], val)
                if name == "Date Applied":
                    cell.number_format = "yyyy-mm-dd"
        arch.cell(dest, ah["Archived"], date.today().isoformat())
        jt.delete_rows(row)
        return {"ok": True}

    return mutate_workbook(m)


def _reexec():
    """Replace this process with a fresh run of the current script file.
    Picks up any updated code on disk (from Claude or the daily sweep)
    without needing the Mac to restart. Absolute paths, so cwd and the
    U+2024 folder name survive the exec."""
    try:
        tio.release_lock()
    except Exception:
        pass
    os.execv(sys.executable, [sys.executable, "-u", str(SCRIPT)])


def do_unarchive(payload):
    """Move an Archived row back to the first empty Job Tracker row."""
    row = int(payload.get("row", 0))
    company = payload.get("company", "")
    if row < 2:
        raise ValueError("Bad row.")

    def m(wb):
        if "Archived" not in wb.sheetnames:
            raise LookupError("No Archived sheet — refresh the page.")
        arch = wb["Archived"]
        jt = wb["Job Tracker"]
        ah = header_map(arch)
        jh = header_map(jt)
        check_row(arch, row, company, ah["Company"])
        dest = first_empty_row(jt, jh["Company"])
        for name, col in jh.items():
            src_col = ah.get(name)
            if not src_col:
                continue
            val = arch.cell(row=row, column=src_col).value
            if val not in (None, ""):
                cell = jt.cell(dest, col, val)
                if name == "Date Applied":
                    cell.number_format = "yyyy-mm-dd"
        arch.delete_rows(row)
        return {"ok": True, "restored_to_row": dest}

    return mutate_workbook(m)


# ---------- background: Apify scrape timer ----------

APIFY_LOG = HERE / ".apify_log.txt"
TIMER_STATE = HERE / ".apify_timer_state.json"
VERIFY_QUEUE = HERE / ".verify_queue.json"
SCRAPE_SLOTS = ("07:45", "12:45")
SCRAPE_CATCHUP_MIN = 180          # fire a missed slot only within 3 h of it
SCRAPE_RETRY_MAX = 2              # extra attempts after a failed run (3 total)
SCRAPE_RETRY_DELAY_MIN = 15       # minutes between attempts
VERIFY_INTERVAL_SECS = 30 * 60

_log_lock = threading.Lock()


def alog(tag, msg):
    """Append a timestamped line to .apify_log.txt; never raises."""
    line = f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} [{tag}] {msg}\n"
    try:
        with _log_lock, open(APIFY_LOG, "a") as f:
            f.write(line)
    except Exception:
        pass


def _load_json(path, default):
    try:
        return json.loads(path.read_text())
    except Exception:
        return default


def _slots_due(now, state, slots=SCRAPE_SLOTS, catchup_min=SCRAPE_CATCHUP_MIN):
    """Pure slot logic (unit-tested): given local `now` and persisted state
    {"date","ran","retries"}, return (new_state, [(slot, "run"|"skip_late"), ...]).
    Each slot fires at most once per calendar day; a slot more than
    `catchup_min` minutes in the past (server was down) is skipped, logged.
    A new day drops any leftover retry bookkeeping with the old state."""
    today = now.date().isoformat()
    if state.get("date") != today:
        state = {"date": today, "ran": []}
    actions = []
    for slot in slots:
        if slot in state["ran"]:
            continue
        h, m = map(int, slot.split(":"))
        slot_dt = now.replace(hour=h, minute=m, second=0, microsecond=0)
        mins_past = (now - slot_dt).total_seconds() / 60
        if mins_past < 0:
            continue
        state["ran"].append(slot)
        actions.append((slot, "run" if mins_past <= catchup_min else "skip_late"))
    return state, actions


def _retries_due(now, state, catchup_min=SCRAPE_CATCHUP_MIN):
    """Pure retry logic (unit-tested): pop scheduled retries whose time has
    come. Returns (new_state, [(slot, attempt_no, "run"|"give_up_late"), ...]).
    A retry that comes due after the slot's freshness window (slot time +
    `catchup_min`) is abandoned, not run. Popped entries are only re-added
    by _record_result if the new attempt fails and attempts remain."""
    actions = []
    retries = state.get("retries") or {}
    for slot in list(retries):
        entry = retries[slot]
        try:
            due = datetime.fromisoformat(entry["due"])
            prev_attempt = int(entry["attempt"])
        except Exception:
            del retries[slot]  # corrupt entry — drop it
            continue
        if now < due:
            continue
        del retries[slot]
        h, m = map(int, slot.split(":"))
        slot_dt = now.replace(hour=h, minute=m, second=0, microsecond=0)
        late = (now - slot_dt).total_seconds() / 60 > catchup_min
        actions.append((slot, prev_attempt + 1, "give_up_late" if late else "run"))
    state["retries"] = retries
    return state, actions


def _record_result(now, state, slot, attempt, returncode,
                   retry_max=SCRAPE_RETRY_MAX, delay_min=SCRAPE_RETRY_DELAY_MIN):
    """Pure outcome logic (unit-tested). Returns (new_state, log_msg).
    Exit 0 or exhausted attempts end the slot for today (it stays in
    state["ran"], so it never double-runs); a retryable failure schedules
    the next attempt `delay_min` minutes out."""
    total = 1 + retry_max
    retries = state.setdefault("retries", {})
    retries.pop(slot, None)
    if returncode == 0:
        return state, (f"slot {slot}: attempt {attempt}/{total} succeeded — "
                       "slot complete for today")
    if attempt < total:
        due = now + timedelta(minutes=delay_min)
        retries[slot] = {"attempt": attempt, "due": due.isoformat()}
        return state, (f"slot {slot}: attempt {attempt}/{total} failed "
                       f"(exit {returncode}); retrying at ~{due.strftime('%H:%M')}")
    return state, (f"slot {slot}: attempt {attempt}/{total} failed "
                   f"(exit {returncode}) — giving up until tomorrow's slot")


def _run_scrape(slot, attempt):
    """One apify_jobs.py attempt. Logs it; returns the exit code (a launch
    error or timeout counts as a failure, never crashes the timer)."""
    total = 1 + SCRAPE_RETRY_MAX
    alog("scrape", f"slot {slot}: running apify_jobs.py (attempt {attempt}/{total})")
    try:
        r = subprocess.run([sys.executable, str(SCRIPT.parent / "apify_jobs.py")],
                           cwd=HERE, capture_output=True, text=True,
                           timeout=1200)
        rc, out = r.returncode, ((r.stdout or "") + (r.stderr or "")).strip()
    except Exception as e:
        rc, out = -1, f"launch/timeout failure: {e!r}"
    alog("scrape", f"slot {slot}: attempt {attempt}/{total} exit {rc}; "
                   f"{out[-800:] or 'no output'}")
    tio.nc_scan()  # Pi: index new .apify_results/.apify_log files
    return rc


def scrape_timer_loop():
    """Daemon: run apify_jobs.py at each SCRAPE_SLOTS time, once per day,
    retrying failures up to SCRAPE_RETRY_MAX times at ~15-min intervals."""
    while True:
        try:
            now = datetime.now()
            state, actions = _slots_due(now, _load_json(TIMER_STATE, {}))
            state, retry_actions = _retries_due(now, state)
            if actions or retry_actions:
                TIMER_STATE.write_text(json.dumps(state))  # mark BEFORE running
            for slot, what in actions:
                if what == "skip_late":
                    alog("scrape", f"slot {slot} missed by >{SCRAPE_CATCHUP_MIN} min "
                                   "(server was down?) — skipped for today")
                    continue
                rc = _run_scrape(slot, 1)
                state, msg = _record_result(datetime.now(), state, slot, 1, rc)
                TIMER_STATE.write_text(json.dumps(state))
                alog("scrape", msg)
            for slot, attempt, what in retry_actions:
                if what == "give_up_late":
                    alog("scrape", f"slot {slot}: retry due after the "
                                   f"{SCRAPE_CATCHUP_MIN}-min freshness window "
                                   "(server was down?) — giving up for today")
                    continue
                rc = _run_scrape(slot, attempt)
                state, msg = _record_result(datetime.now(), state, slot, attempt, rc)
                TIMER_STATE.write_text(json.dumps(state))
                alog("scrape", msg)
        except Exception as e:
            alog("scrape", f"timer error (server unaffected): {e!r}")
        time.sleep(30)


# ---------- background: JS-page verifier ----------

# Verdicts we couldn't write because the tracker was busy — retried next pass
# instead of paying for a re-render.
_pending_verdicts = []


def _apply_verdict(cand, verdict, evidence):
    """Write a verdict into the tracker (locates the row by Link, so it's
    safe even if rows moved since we read them). Uses mutate_workbook, so it
    takes .tracker.lock, makes the daily backup and regenerates dashboard.html."""
    today = date.today().isoformat()

    def m(wb):
        if "Manual Review" not in wb.sheetnames:
            raise LookupError("no Manual Review sheet")
        ws = wb["Manual Review"]
        h = header_map(ws)
        row = None
        for r in range(2, ws.max_row + 1):
            if str(ws.cell(r, h["Link"]).value or "").strip() == cand["link"]:
                row = r
                break
        if row is None:
            raise LookupError("row no longer in Manual Review")
        notes_cell = ws.cell(row, h["Notes"])
        old = str(notes_cell.value or "").strip()
        if verdict == "verified":
            stamp = f"verified live {today} via JS render — {evidence}."
            notes_cell.value = f"{stamp} {old}".strip()
            if "Verified Live" in h and not ws.cell(row, h["Verified Live"]).value:
                ws.cell(row, h["Verified Live"], today)
        elif verdict == "dead":
            headers = [str(c.value).strip() if c.value else "" for c in ws[1]]
            if "Dead Leads" in wb.sheetnames:
                dead = wb["Dead Leads"]
            else:
                dead = wb.create_sheet("Dead Leads")
                dead.append(headers + ["Marked Dead"])
            vals = [ws.cell(row, i).value for i in range(1, len(headers) + 1)]
            ni = headers.index("Notes")
            vals[ni] = (f"moved by JS-render verifier {today}: {evidence}. "
                        f"{vals[ni] or ''}").strip()
            dead.append(vals + [today])
            ws.delete_rows(row)
        else:  # inconclusive — annotate, leave for Sohaib
            stamp = (f"JS render {today} inconclusive — {evidence}; "
                     f"left for manual judgment.")
            notes_cell.value = f"{stamp} {old}".strip()
        return {"ok": True}

    mutate_workbook(m)


def _review_candidates():
    """Read Manual Review rows whose Notes flag them unverifiable/JS-blocked
    and that don't already carry a verifier verdict. Read-only, no lock."""
    out = []
    wb = openpyxl.load_workbook(XLSX, read_only=True)
    try:
        if "Manual Review" not in wb.sheetnames:
            return out
        rows = list(wb["Manual Review"].iter_rows(values_only=True))
        if not rows:
            return out
        idx = {str(v).strip(): i for i, v in enumerate(rows[0]) if v}
        for r in rows[1:]:
            def cell(name):
                i = idx.get(name)
                return str(r[i] or "").strip() if i is not None and i < len(r) else ""
            if not cell("Company"):
                continue
            notes, link = cell("Notes"), cell("Link")
            if (link.startswith("http")
                    and vjs.NEEDS_VERIFY_RE.search(notes)
                    and not vjs.ALREADY_RE.search(notes)):
                out.append({"company": cell("Company"), "role": cell("Role"),
                            "link": link, "notes": notes})
    finally:
        wb.close()
    return out


def verifier_pass():
    """One sweep: retry pending verdicts, verify flagged Manual Review rows,
    process .verify_queue.json — all inside the daily render budget."""
    cfg = vjs.load_config()
    cap = int(cfg.get("verifier", {}).get("max_renders_per_day", 10))

    # 0) verdicts that couldn't be written last pass (tracker was busy)
    still = []
    for cand, verdict, evidence in _pending_verdicts:
        try:
            _apply_verdict(cand, verdict, evidence)
            alog("verify", f"{cand['company']}: {verdict} (applied on retry)")
        except TrackerBusy:
            still.append((cand, verdict, evidence))
        except LookupError as e:
            alog("verify", f"{cand['company']}: verdict dropped — {e}")
        except Exception as e:
            alog("verify", f"{cand['company']}: retry failed: {e!r}")
    _pending_verdicts[:] = still

    rendered = {}  # url -> item, reused between rows and queue this pass

    # 1) flagged Manual Review rows
    try:
        candidates = _review_candidates()
    except Exception as e:
        alog("verify", f"could not read tracker: {e!r}")
        candidates = []
    pending_links = {p[0]["link"] for p in _pending_verdicts}
    for cand in candidates:
        if cand["link"] in pending_links:
            continue
        urls = [cand["link"]]
        burl = vjs.board_url(cand["link"])
        if burl.rstrip("/") != cand["link"].rstrip("/"):
            urls.append(burl)
        st = vjs.load_state()
        if st["renders"] + len(urls) > cap:
            alog("verify", f"daily render cap {cap} reached — "
                           f"{cand['company']} waits for tomorrow")
            break
        st["renders"] += len(urls)
        vjs.save_state(st)
        try:
            items, cost, run_id, status = vjs.render(urls, cfg)
        except Exception as e:
            alog("verify", f"{cand['company']}: render failed: {e!r}")
            continue
        alog("verify", f"{cand['company']}: rendered {len(urls)} url(s) "
                       f"run={run_id} {status} cost=${cost:.4f} "
                       f"({st['renders']}/{cap} renders today)")
        for it in items:
            rendered[str(it.get('url', '')).rstrip('/')] = it
        detail = items[0]
        board = items[1] if len(items) > 1 else items[0]
        verdict, evidence = vjs.decide(cand["role"], board.get("text", ""),
                                       detail.get("text", ""))
        try:
            _apply_verdict(cand, verdict, evidence)
            alog("verify", f"{cand['company']}: {verdict} — {evidence}")
        except TrackerBusy:
            _pending_verdicts.append((cand, verdict, evidence))
            alog("verify", f"{cand['company']}: tracker busy — verdict "
                           f"{verdict} queued for next pass (no re-render)")
        except Exception as e:
            alog("verify", f"{cand['company']}: could not write verdict: {e!r}")

    # 2) sweep-queued URLs (.verify_queue.json)
    queue = _load_json(VERIFY_QUEUE, [])
    if not isinstance(queue, list):
        return
    changed = False
    for entry in queue:
        if not isinstance(entry, dict) or not entry.get("url") or entry.get("result"):
            continue
        url = entry["url"]
        it = rendered.get(url.rstrip("/"))
        cost = 0.0
        if it is None:
            st = vjs.load_state()
            if st["renders"] + 1 > cap:
                alog("verify", f"daily render cap {cap} reached — queue entry "
                               f"{url} waits for tomorrow")
                break
            st["renders"] += 1
            vjs.save_state(st)
            try:
                items, cost, run_id, status = vjs.render([url], cfg)
                it = items[0]
                alog("verify", f"queue: rendered {url} run={run_id} {status} "
                               f"cost=${cost:.4f} ({st['renders']}/{cap} today)")
            except Exception as e:
                entry["result"] = {"status": "error", "error": str(e),
                                   "processed_at": datetime.now().isoformat(timespec='seconds')}
                alog("verify", f"queue: {url} failed: {e!r}")
                changed = True
                continue
        text = it.get("text", "")
        entry["result"] = {
            "status": "ok",
            "title": it.get("title", ""),
            "dead_signals": vjs.dead_signals(text),
            "text_excerpt": text[:4000],
            "cost_usd": cost,
            "processed_at": datetime.now().isoformat(timespec='seconds'),
        }
        changed = True
    if changed:
        try:
            VERIFY_QUEUE.write_text(json.dumps(queue, indent=2))
        except Exception as e:
            alog("verify", f"could not write {VERIFY_QUEUE.name}: {e!r}")


def verifier_loop():
    """Daemon: verifier_pass every ~30 min; first pass ~2 min after boot.
    Each pass ends with tracker_io.housekeeping() (prunes old backups,
    processed queue entries, stale apify result files — all >7 days)."""
    time.sleep(120)
    while True:
        try:
            verifier_pass()
        except Exception as e:
            alog("verify", f"verifier error (server unaffected): {e!r}")
        try:
            msg = tio.housekeeping()
            if not msg.endswith("0 backup(s), 0 queue entr(ies), 0 result file(s)"):
                alog("housekeeping", msg)
        except Exception:
            pass
        tio.nc_scan()  # Pi: index queue/log written this pass
        time.sleep(VERIFY_INTERVAL_SECS)


# ---------- http ----------

class Handler(BaseHTTPRequestHandler):
    def _send(self, code, body, ctype):
        data = body.encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(data)

    def _json(self, code, obj):
        self._send(code, json.dumps(obj), "application/json")

    def do_GET(self):
        path = self.path.split("?")[0]
        if path == "/api/version":
            try:
                self._json(200, {"v": str(XLSX.stat().st_mtime), "boot": BOOT_ID})
            except OSError:
                self._json(200, {"v": "missing", "boot": BOOT_ID})
        elif path in ("/", "/index.html", "/dashboard.html"):
            try:
                self._send(200, gd.build(), "text/html; charset=utf-8")
            except Exception as e:
                self._send(500, f"<h1>Could not read tracker</h1><p>{e}</p>",
                           "text/html; charset=utf-8")
        else:
            self._json(404, {"error": "Not found"})

    do_HEAD = do_GET

    def do_POST(self):
        try:
            n = int(self.headers.get("Content-Length") or 0)
            payload = json.loads(self.rfile.read(n) or b"{}")
        except Exception:
            self._json(400, {"error": "Bad JSON."})
            return
        try:
            if self.path == "/api/status":
                self._json(200, do_status(payload))
            elif self.path == "/api/apply":
                self._json(200, do_apply(payload))
            elif self.path == "/api/dead":
                self._json(200, do_dead(payload))
            elif self.path == "/api/review_move":
                self._json(200, do_review_move(payload))
            elif self.path == "/api/archive":
                self._json(200, do_archive(payload))
            elif self.path == "/api/unarchive":
                self._json(200, do_unarchive(payload))
            elif self.path == "/api/restart":
                self._json(200, {"ok": True, "restarting": True})
                threading.Timer(0.5, _reexec).start()
            else:
                self._json(404, {"error": "Not found"})
        except TrackerBusy:
            self._json(423, {"error": "Tracker is being updated (daily sweep?) — "
                                       "try again in a few seconds."})
        except PermissionError:
            self._json(423, {"error": "Tracker file is locked — close it in "
                                       "Excel/Numbers and retry."})
        except LookupError as e:
            self._json(409, {"error": str(e)})
        except (ValueError, KeyError) as e:
            self._json(400, {"error": f"Bad request: {e}"})
        except Exception as e:
            self._json(500, {"error": f"Update failed: {e}"})

    def log_message(self, *args):
        pass


def tailscale_ip():
    for cmd in (["tailscale", "ip", "-4"],
                ["/Applications/Tailscale.app/Contents/MacOS/Tailscale", "ip", "-4"]):
        try:
            out = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
            if out.returncode == 0 and out.stdout.strip():
                return out.stdout.strip().splitlines()[0]
        except Exception:
            pass
    return None


def _maybe_run_pi_deploy():
    """One-shot Raspberry Pi migration hook (macOS only). If the trigger file
    ".deploy_to_pi_requested" exists, delete it FIRST — so a crash or failed
    deploy can never loop — then run deploy_to_pi.py in a background thread,
    appending all its output to .pi_deploy_log.txt. Never raises: a broken
    deploy cannot take the dashboard down. The Pi's copy of this file inherits
    the hook harmlessly (non-darwin -> immediate no-op), which also means a
    Nextcloud-synced trigger file can't make the Pi try to deploy to itself."""
    trigger = HERE / ".deploy_to_pi_requested"
    try:
        if sys.platform != "darwin" or not trigger.exists():
            return
        trigger.unlink()  # BEFORE running: one shot, never a loop
        script = HERE / "deploy_to_pi.py"
        if not script.exists():
            print("pi-deploy: trigger found but deploy_to_pi.py is missing — skipped.")
            return

        def _run():
            try:
                with open(HERE / ".pi_deploy_log.txt", "a") as f:
                    f.write(f"\n===== pi deploy {datetime.now().isoformat(timespec='seconds')} "
                            f"(triggered by server restart) =====\n")
                    f.flush()
                    subprocess.run([sys.executable, "-u", str(script)], cwd=HERE,
                                   stdout=f, stderr=subprocess.STDOUT, timeout=3600)
            except Exception as e:
                try:
                    with open(HERE / ".pi_deploy_log.txt", "a") as f:
                        f.write(f"pi-deploy hook error: {e!r}\n")
                except Exception:
                    pass

        threading.Thread(target=_run, daemon=True, name="pi-deploy").start()
        print("Pi deploy started in the background — progress in .pi_deploy_log.txt")
    except Exception as e:
        print(f"pi-deploy hook error (server unaffected): {e!r}")


if __name__ == "__main__":
    if not XLSX.exists():
        sys.exit(f"Tracker not found: {XLSX}")
    ip = tailscale_ip()
    url = f"http://{ip}:{PORT}/" if ip else f"http://<Mac-Tailscale-IP>:{PORT}/"
    print(f"Job-search dashboard:  {url}")
    print("Stable HTTPS instead:  tailscale serve --bg 8787   (see README_dashboard.txt)")
    print(f"Background: Apify scrape at {' & '.join(SCRAPE_SLOTS)}; JS verifier "
          f"every {VERIFY_INTERVAL_SECS // 60} min. Log: {APIFY_LOG.name}")
    print("Ctrl-C to stop.")
    _maybe_run_pi_deploy()  # one-shot: only if .deploy_to_pi_requested exists
    if sys.platform == "darwin" and (HERE / ".pi_is_primary").exists():
        print("NOTE: the Raspberry Pi is primary (.pi_is_primary present) — the")
        print("Apify scrape timer and JS verifier stay OFF on this Mac to avoid")
        print("double API spend and conflicting tracker writes. Delete")
        print(".pi_is_primary and restart to re-enable them here.")
    else:
        threading.Thread(target=scrape_timer_loop, daemon=True,
                         name="apify-scrape-timer").start()
        threading.Thread(target=verifier_loop, daemon=True,
                         name="js-verifier").start()
    ThreadingHTTPServer.allow_reuse_address = True
    ThreadingHTTPServer(("", PORT), Handler).serve_forever()