"""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)}