backend/fetlife_queue.py

"""Backend fetlife queue service."""
from __future__ import annotations

import os
import secrets
import time
from collections import defaultdict
from datetime import UTC, datetime
from typing import Any
from sqlmodel import Session, select
from models import FetlifeCrawlJob, FetlifeKinkMeta, FetlifeSupervisorRun



def queue_fetlife_job(self, job_type: str, target_key: str, url: str, discovered_from: str = "") -> str:
    job_id = f"fetlife_job_{job_type}_{self.stable_fragment(target_key + '|' + url)}"
    with Session(self.engine) as session:
        job = session.get(FetlifeCrawlJob, job_id)
        if not job:
            session.add(
                FetlifeCrawlJob(
                    id=job_id,
                    job_type=job_type,
                    target_key=target_key,
                    url=url,
                    state="queued",
                    discovered_from=discovered_from,
                    updated_at=self._now_iso(),
                )
            )
            session.commit()
    return job_id

def start_fetlife_supervisor_run(self, profile: str = "", pid: int = 0) -> str:
    run_id = f"fetlife_supervisor_{int(time.time())}_{secrets.token_hex(3)}"
    now = self._now_iso()
    with Session(self.engine) as session:
        session.add(
            FetlifeSupervisorRun(
                id=run_id,
                profile=profile,
                pid=pid,
                state="running",
                started_at=now,
                heartbeat_at=now,
            )
        )
        session.commit()
    return run_id

def heartbeat_fetlife_supervisor(
    self,
    run_id: str,
    *,
    current_job: dict[str, Any] | None = None,
    claimed: int = 0,
    ok: int = 0,
    failed: int = 0,
    challenge: int = 0,
    restart: int = 0,
    last_error: str = "",
) -> None:
    with Session(self.engine) as session:
        run = session.get(FetlifeSupervisorRun, run_id)
        if not run:
            return
        run.heartbeat_at = self._now_iso()
        run.jobs_claimed += claimed
        run.jobs_ok += ok
        run.jobs_failed += failed
        run.challenge_events += challenge
        run.restart_count += restart
        if current_job:
            run.current_job_id = str(current_job.get("id", ""))
            run.current_job_type = str(current_job.get("job_type", ""))
            run.current_target_key = str(current_job.get("target_key", ""))
        if last_error:
            run.last_error = last_error[:1000]
        session.commit()

def finish_fetlife_supervisor_run(self, run_id: str, state: str = "stopped", last_error: str = "") -> None:
    with Session(self.engine) as session:
        run = session.get(FetlifeSupervisorRun, run_id)
        if not run:
            return
        run.state = state
        run.completed_at = self._now_iso()
        run.heartbeat_at = run.completed_at
        if last_error:
            run.last_error = last_error[:1000]
        session.commit()

def _process_is_live(self, pid: int) -> bool:
    if pid <= 0:
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True

def fetlife_supervisor_status(self) -> dict[str, Any]:
    with Session(self.engine) as session:
        runs = session.exec(select(FetlifeSupervisorRun).order_by(FetlifeSupervisorRun.started_at.desc())).all()
        now = datetime.now(UTC)
        changed = False
        for run in runs:
            if run.state != "running":
                continue
            stale_heartbeat = False
            try:
                heartbeat = datetime.fromisoformat(run.heartbeat_at) if run.heartbeat_at else None
                stale_heartbeat = bool(heartbeat and (now - heartbeat).total_seconds() > 900)
            except ValueError:
                stale_heartbeat = True
            if stale_heartbeat or not self._process_is_live(run.pid):
                run.state = "stale"
                run.completed_at = self._now_iso()
                if not run.last_error:
                    run.last_error = "marked stale after supervisor process exited"
                changed = True
        if changed:
            session.commit()
            runs = session.exec(select(FetlifeSupervisorRun).order_by(FetlifeSupervisorRun.started_at.desc())).all()
    latest = runs[0] if runs else None
    active = [run for run in runs if run.state == "running"]

    def serialize(run: FetlifeSupervisorRun) -> dict[str, Any]:
        elapsed_hours = 0.0
        try:
            started = datetime.fromisoformat(run.started_at).timestamp() if run.started_at else 0.0
            ended = datetime.fromisoformat((run.completed_at or run.heartbeat_at)).timestamp() if (run.completed_at or run.heartbeat_at) else time.time()
            elapsed_hours = max((ended - started) / 3600.0, 0.001)
        except ValueError:
            elapsed_hours = 0.001
        return {
            "id": run.id,
            "profile": run.profile,
            "pid": run.pid,
            "state": run.state,
            "started_at": run.started_at,
            "heartbeat_at": run.heartbeat_at,
            "completed_at": run.completed_at,
            "current_job_id": run.current_job_id,
            "current_job_type": run.current_job_type,
            "current_target_key": run.current_target_key,
            "jobs_claimed": run.jobs_claimed,
            "jobs_ok": run.jobs_ok,
            "jobs_failed": run.jobs_failed,
            "challenge_events": run.challenge_events,
            "restart_count": run.restart_count,
            "jobs_per_hour": round(run.jobs_ok / elapsed_hours, 2),
            "last_error": run.last_error,
        }

    return {
        "active": [serialize(run) for run in active[:5]],
        "latest": serialize(latest) if latest else None,
        "history": [serialize(run) for run in runs[:10]],
    }

def requeue_stale_fetlife_jobs(self, max_age_minutes: int = 20) -> int:
    cutoff = datetime.now(UTC).timestamp() - (max_age_minutes * 60)
    reclaimed = 0
    with Session(self.engine) as session:
        jobs = session.exec(select(FetlifeCrawlJob).where(FetlifeCrawlJob.state == "in_progress")).all()
        for job in jobs:
            try:
                updated_ts = datetime.fromisoformat(job.updated_at).timestamp() if job.updated_at else 0
            except ValueError:
                updated_ts = 0
            if updated_ts <= cutoff:
                job.state = "retry_later"
                job.last_error = "reclaimed stale in_progress job"
                job.updated_at = self._now_iso()
                reclaimed += 1
        if reclaimed:
            session.commit()
    return reclaimed

def claim_fetlife_jobs(self, limit: int = 10) -> list[dict[str, Any]]:
    self.requeue_stale_fetlife_jobs()
    priority = {
        "fetish_popular_page": 0,
        "fetish_detail": 1,
        "fetish_similar": 2,
        "fetish_pictures": 3,
        "fetish_kinksters": 4,
        "user_profile": 5,
    }
    type_limits = {
        "fetish_popular_page": max(1, min(4, limit)),
        "fetish_detail": max(12, limit // 2),
        "fetish_similar": max(8, limit // 4),
        "fetish_pictures": max(8, limit // 4),
        "fetish_kinksters": max(6, limit // 8),
        "user_profile": max(4, limit // 12),
    }
    with Session(self.engine) as session:
        meta_by_fetish_id = {
            row.fetish_id: float(row.popularity or 0.0)
            for row in session.exec(select(FetlifeKinkMeta)).all()
        }
        jobs = session.exec(select(FetlifeCrawlJob)).all()
        jobs = [
            job
            for job in jobs
            if job.state in {"queued", "retry_later"}
        ]
        jobs.sort(
            key=lambda job: (
                0 if job.job_type == "fetish_popular_page" else 1 if meta_by_fetish_id.get(job.target_key, 0.0) > 0 else 2,
                -meta_by_fetish_id.get(job.target_key, 0.0),
                priority.get(job.job_type, 99),
                job.updated_at,
                job.id,
            )
        )
        claimed: list[dict[str, Any]] = []
        claimed_by_type: dict[str, int] = defaultdict(int)
        for job in jobs:
            if len(claimed) >= limit:
                break
            if claimed_by_type[job.job_type] >= type_limits.get(job.job_type, limit):
                continue
            job.state = "in_progress"
            job.attempts += 1
            job.updated_at = self._now_iso()
            claimed_by_type[job.job_type] += 1
            claimed.append(
                {
                    "id": job.id,
                    "job_type": job.job_type,
                    "target_key": job.target_key,
                    "url": job.url,
                    "attempts": job.attempts,
                    "discovered_from": job.discovered_from,
                }
            )
        session.commit()
        return claimed

def complete_fetlife_job(
    self,
    job_id: str,
    state: str,
    *,
    content_hash: str = "",
    last_error: str = "",
) -> None:
    with Session(self.engine) as session:
        job = session.get(FetlifeCrawlJob, job_id)
        if not job:
            return
        job.state = state
        job.content_hash = content_hash
        job.last_error = last_error[:1000]
        job.updated_at = self._now_iso()
        session.commit()

def fetlife_job_stats(self) -> dict[str, Any]:
    with self._sqlite() as conn:
        total = int(conn.execute("SELECT COUNT(*) AS n FROM fetlifecrawljob").fetchone()["n"])
        by_state = {row["state"]: int(row["n"]) for row in conn.execute("SELECT state, COUNT(*) AS n FROM fetlifecrawljob GROUP BY state").fetchall()}
        by_type = {row["job_type"]: int(row["n"]) for row in conn.execute("SELECT job_type, COUNT(*) AS n FROM fetlifecrawljob GROUP BY job_type").fetchall()}
    return {"total": total, "by_state": by_state, "by_type": by_type}

def fetlife_jobs(self, limit: int = 100, state: str | None = None) -> list[dict[str, Any]]:
    with Session(self.engine) as session:
        jobs = session.exec(select(FetlifeCrawlJob).order_by(FetlifeCrawlJob.updated_at.desc())).all()
    if state:
        jobs = [job for job in jobs if job.state == state]
    return [
        {
            "id": job.id,
            "job_type": job.job_type,
            "target_key": job.target_key,
            "url": job.url,
            "state": job.state,
            "attempts": job.attempts,
            "last_error": job.last_error,
            "content_hash": job.content_hash,
            "discovered_from": job.discovered_from,
            "updated_at": job.updated_at,
        }
        for job in jobs[:limit]
    ]

def fetlife_queue_preview(self, limit: int = 20) -> list[dict[str, Any]]:
    priority = {
        "fetish_popular_page": 0,
        "fetish_detail": 1,
        "fetish_similar": 2,
        "fetish_pictures": 3,
        "fetish_kinksters": 4,
        "user_profile": 5,
    }
    with Session(self.engine) as session:
        meta_by_fetish_id = {
            row.fetish_id: float(row.popularity or 0.0)
            for row in session.exec(select(FetlifeKinkMeta)).all()
        }
        jobs = [
            job
            for job in session.exec(select(FetlifeCrawlJob)).all()
            if job.state in {"queued", "retry_later"}
        ]
    jobs.sort(
        key=lambda job: (
            0 if job.job_type == "fetish_popular_page" else 1 if meta_by_fetish_id.get(job.target_key, 0.0) > 0 else 2,
            -meta_by_fetish_id.get(job.target_key, 0.0),
            priority.get(job.job_type, 99),
            job.updated_at,
            job.id,
        )
    )
    return [
        {
            "id": job.id,
            "job_type": job.job_type,
            "target_key": job.target_key,
            "url": job.url,
            "state": job.state,
            "attempts": job.attempts,
            "priority": priority.get(job.job_type, 99),
            "popularity": meta_by_fetish_id.get(job.target_key, 0.0),
            "last_error": job.last_error,
            "updated_at": job.updated_at,
        }
        for job in jobs[:limit]
    ]

def queue_fetlife_priority_debt(self, limit: int = 20) -> dict[str, Any]:
    debt = self.fetlife_coverage_debt(limit=limit)
    queued: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    for lane in ("starter_priority", "missing_similar", "missing_images"):
        for item in debt.get(lane, []):
            fetish_id = str(item["fetish_id"])
            base = f"https://fetlife.com/fetishes/{fetish_id}"
            if item.get("needs_similar") and ("fetish_similar", fetish_id) not in seen:
                seen.add(("fetish_similar", fetish_id))
                queued.append({"job_id": self.queue_fetlife_job("fetish_similar", fetish_id, f"{base}/similar", discovered_from=f"data_health:{lane}"), "lane": lane, "job_type": "fetish_similar", **item})
            if item.get("needs_images") and ("fetish_pictures", fetish_id) not in seen:
                seen.add(("fetish_pictures", fetish_id))
                queued.append({"job_id": self.queue_fetlife_job("fetish_pictures", fetish_id, f"{base}/pictures", discovered_from=f"data_health:{lane}"), "lane": lane, "job_type": "fetish_pictures", **item})
            if len(queued) >= limit:
                break
        if len(queued) >= limit:
            break
    return {"queued": queued, "count": len(queued)}