diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 30e2b6ae..c1a71dae 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -91,6 +91,25 @@ def queue_status( from app.workers.celery_app import celery_app + # 0) Reap zombies: workers killed by SIGKILL (OOM, container restart, + # deploy mid-run) leave kn_scrape_runs rows stuck in 'running' forever + # because the except/finally block never fires. Anything older than + # 60 minutes without finishing is almost certainly dead. + db.execute( + text( + """ + UPDATE kn_scrape_runs + SET status = 'zombie', + finished_at = NOW(), + error = 'auto-marked: no heartbeat 60+ min' + WHERE status = 'running' + AND finished_at IS NULL + AND started_at < NOW() - INTERVAL '60 minutes' + """ + ) + ) + db.commit() + # 1) Active runs straight from DB — no broker round-trip. active_rows = ( db.execute( diff --git a/backend/app/api/v1/photos.py b/backend/app/api/v1/photos.py index ab06f637..7b3c5d7d 100644 --- a/backend/app/api/v1/photos.py +++ b/backend/app/api/v1/photos.py @@ -2,13 +2,11 @@ GET /api/v1/photos/{obj_id}/{file_id}?size=thumb|full -- size=thumb (default): WebP 320×240 from local cache. If thumb is missing - but the original is present locally, generate one on the fly. If neither - is present, redirect to the upstream DOM.РФ URL. -- size=full: original PNG/JPG if cached locally, else redirect upstream. - -This isolates the frontend grid from upstream latency and from Next.js dev-mode -optimizer cold-hit cost. +- size=thumb (default): WebP 320×240 from local cache. If neither the thumb + nor the original exists locally, lazily fetch the original from DOM.РФ via + httpx + Basic auth, save it, generate the thumb, and serve. Subsequent + requests hit the disk cache and return in single-digit ms. +- size=full: original PNG/JPG if cached locally, else 302 redirect upstream. """ from __future__ import annotations @@ -17,6 +15,7 @@ import logging from pathlib import Path from typing import Annotated, Literal +import httpx from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import FileResponse, RedirectResponse, Response from sqlalchemy import text @@ -28,10 +27,15 @@ from app.services.photos.thumbs import make_thumbnail logger = logging.getLogger(__name__) router = APIRouter() +# Same Basic auth used by the kn-API scraper (1:qwe). Photo CDN accepts it +# without the TLS-fingerprint challenge that the JSON API enforces. +_UPSTREAM_AUTH = "Basic MTpxd2U=" +_UPSTREAM_TIMEOUT = httpx.Timeout(8.0, connect=4.0) +_PHOTOS_DIR = Path("data/raw/domrf_photos") _PHOTO_LOOKUP_SQL = text( """ - SELECT local_path, thumb_path, photo_url + SELECT local_path, thumb_path, photo_url, photo_name, size_bytes FROM domrf_kn_photos WHERE obj_id = :obj AND obj_file_id = :fid LIMIT 1 @@ -39,6 +43,32 @@ _PHOTO_LOOKUP_SQL = text( ) +def _ext_from_name(name: str | None) -> str: + n = (name or "").lower() + return ".jpg" if n.endswith((".jpg", ".jpeg")) else ".png" + + +def _fetch_upstream(url: str) -> bytes | None: + """Fetch upstream image bytes. Returns None on any error (logged).""" + try: + with httpx.Client(timeout=_UPSTREAM_TIMEOUT, follow_redirects=True) as c: + r = c.get(url, headers={"Authorization": _UPSTREAM_AUTH}) + if r.status_code == 200 and r.content: + return r.content + logger.warning("upstream photo %s status=%s len=%s", url, r.status_code, len(r.content)) + except Exception as e: + logger.warning("upstream photo %s failed: %s", url, e) + return None + + +def _persist_original(obj_id: int, file_id: str, name: str | None, data: bytes) -> Path: + obj_dir = _PHOTOS_DIR / str(obj_id) + obj_dir.mkdir(parents=True, exist_ok=True) + dst = obj_dir / f"{file_id}{_ext_from_name(name)}" + dst.write_bytes(data) + return dst + + @router.get("/{obj_id}/{file_id}") def get_photo( db: Annotated[Session, Depends(get_db)], @@ -53,35 +83,56 @@ def get_photo( local_path = row["local_path"] thumb_path = row["thumb_path"] upstream = row["photo_url"] + photo_name = row["photo_name"] headers = {"Cache-Control": "public, max-age=604800, immutable"} - if size == "full": + # ── size=thumb ────────────────────────────────────────────────────────── + if size == "thumb": + if thumb_path and Path(thumb_path).exists(): + return FileResponse(thumb_path, media_type="image/webp", headers=headers) + + # Generate from existing local original. if local_path and Path(local_path).exists(): - return FileResponse(local_path, headers=headers) + generated = make_thumbnail(Path(local_path)) + if generated and generated.exists(): + db.execute( + text( + "UPDATE domrf_kn_photos SET thumb_path = :tp" + " WHERE obj_id = :o AND obj_file_id = :f" + ), + {"tp": str(generated), "o": obj_id, "f": file_id}, + ) + db.commit() + return FileResponse(str(generated), media_type="image/webp", headers=headers) + + # Lazy: fetch upstream, persist original + thumb, then serve. + if upstream: + data = _fetch_upstream(upstream) + if data: + src = _persist_original(obj_id, file_id, photo_name, data) + generated = make_thumbnail(src) + if generated and generated.exists(): + db.execute( + text( + "UPDATE domrf_kn_photos" + " SET local_path = :lp, thumb_path = :tp," + " downloaded_at = NOW()" + " WHERE obj_id = :o AND obj_file_id = :f" + ), + {"lp": str(src), "tp": str(generated), "o": obj_id, "f": file_id}, + ) + db.commit() + return FileResponse(str(generated), media_type="image/webp", headers=headers) + + # Last resort — redirect to upstream so the page still renders. if upstream: return RedirectResponse(upstream, status_code=302) - raise HTTPException(status_code=404, detail="full photo unavailable") + raise HTTPException(status_code=404, detail="photo unavailable") - # size=thumb - if thumb_path and Path(thumb_path).exists(): - return FileResponse(thumb_path, media_type="image/webp", headers=headers) - - # Generate thumb on the fly from cached original + # ── size=full ─────────────────────────────────────────────────────────── if local_path and Path(local_path).exists(): - generated = make_thumbnail(Path(local_path)) - if generated and generated.exists(): - db.execute( - text( - "UPDATE domrf_kn_photos SET thumb_path = :tp" - " WHERE obj_id = :o AND obj_file_id = :f" - ), - {"tp": str(generated), "o": obj_id, "f": file_id}, - ) - db.commit() - return FileResponse(str(generated), media_type="image/webp", headers=headers) - - # Last resort — redirect to upstream so the page still renders + return FileResponse(local_path, headers=headers) if upstream: return RedirectResponse(upstream, status_code=302) - raise HTTPException(status_code=404, detail="photo unavailable") + raise HTTPException(status_code=404, detail="full photo unavailable") diff --git a/backend/app/workers/tasks/scrape_kn.py b/backend/app/workers/tasks/scrape_kn.py index bc17ce7f..1a55e783 100644 --- a/backend/app/workers/tasks/scrape_kn.py +++ b/backend/app/workers/tasks/scrape_kn.py @@ -6,15 +6,41 @@ import asyncio import logging import random import time +from contextlib import contextmanager from pathlib import Path from typing import Any +import redis + from app.core.config import settings from app.services.scrapers.domrf_kn import run_region_sweep from app.workers.celery_app import celery_app logger = logging.getLogger(__name__) +# Lock TTL longer than the longest expected sweep (442-obj sweep with extras +# can run ~30-60 min). 2 hours is a safe upper bound — after that we assume +# the worker died and let a new run proceed. +_LOCK_TTL_SECONDS = 2 * 60 * 60 + + +@contextmanager +def _region_lock(region_code: int, developers: list[str] | None): + """Redis singleton lock so beat double-fire or two workers can't run the + same (region, devs) sweep concurrently. SET NX EX is atomic.""" + devs_key = ",".join(developers) if developers else "*" + key = f"scrape:kn:lock:{region_code}:{devs_key}" + r = redis.Redis.from_url(settings.redis_url) + acquired = r.set(key, "1", nx=True, ex=_LOCK_TTL_SECONDS) + try: + yield bool(acquired) + finally: + if acquired: + try: + r.delete(key) + except Exception: + pass + @celery_app.task(bind=True, name="tasks.scrape_kn.scrape_kn_region", max_retries=2) def scrape_kn_region( @@ -43,21 +69,36 @@ def scrape_kn_region( logger.warning("storage_state %s not found — bootstrap will run cold", state_path) state_path = None - logger.info( - "scrape_kn_region start region=%s devs=%s extras=%s photos=%s state=%s", - region_code, - developers, - extras, - download_photos, - state_path, - ) - return asyncio.run( - run_region_sweep( - region_code=region_code, - developers=developers, - load_state=state_path, - fetch_flats=fetch_flats, - extras=extras, - download_photos_binary=download_photos, + with _region_lock(region_code, developers) as got_lock: + if not got_lock: + logger.warning( + "scrape_kn_region SKIPPED region=%s devs=%s — another sweep is" + " already running (Redis lock held)", + region_code, + developers, + ) + return { + "skipped": True, + "reason": "lock_held", + "region_code": region_code, + "developers": developers, + } + + logger.info( + "scrape_kn_region start region=%s devs=%s extras=%s photos=%s state=%s", + region_code, + developers, + extras, + download_photos, + state_path, + ) + return asyncio.run( + run_region_sweep( + region_code=region_code, + developers=developers, + load_state=state_path, + fetch_flats=fetch_flats, + extras=extras, + download_photos_binary=download_photos, + ) ) - )