diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index c1a71dae..59506012 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -272,6 +272,56 @@ def list_failures( ] +@router.get("/logs") +def list_logs( + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + run_id: int | None = None, + since_id: int | None = None, + limit: int = 200, +) -> list[dict[str, Any]]: + """Per-run progress events. Use since_id to poll incrementally: + pass the highest log_id seen → returns only newer rows.""" + _check_token(x_admin_token) + where: list[str] = [] + params: dict[str, Any] = {"lim": min(limit, 1000)} + if run_id is not None: + where.append("run_id = :rid") + params["rid"] = run_id + if since_id is not None: + where.append("log_id > :sid") + params["sid"] = since_id + where_sql = ("WHERE " + " AND ".join(where)) if where else "" + rows = ( + db.execute( + text( + f""" + SELECT log_id, run_id, ts, level, stage, obj_id, message + FROM kn_scrape_log + {where_sql} + ORDER BY log_id DESC + LIMIT :lim + """ + ), + params, + ) + .mappings() + .all() + ) + return [ + { + "log_id": r["log_id"], + "run_id": r["run_id"], + "ts": r["ts"].isoformat() if r["ts"] else None, + "level": r["level"], + "stage": r["stage"], + "obj_id": r["obj_id"], + "message": r["message"], + } + for r in rows + ] + + @router.get("/runs") def list_runs( db: Annotated[Session, Depends(get_db)], diff --git a/backend/app/services/scrapers/domrf_kn.py b/backend/app/services/scrapers/domrf_kn.py index 783f2b4f..ed2829be 100644 --- a/backend/app/services/scrapers/domrf_kn.py +++ b/backend/app/services/scrapers/domrf_kn.py @@ -883,6 +883,43 @@ async def probe_endpoint( } +def log_progress( + db: Session, + run_id: int, + message: str, + *, + level: str = "info", + stage: str | None = None, + obj_id: int | None = None, +) -> None: + """Append a progress event to kn_scrape_log so the admin UI can display + live status. Best-effort: SQL errors are swallowed and logged so a flaky + log table never aborts an actual sweep.""" + try: + db.execute( + text( + """ + INSERT INTO kn_scrape_log (run_id, level, stage, obj_id, message) + VALUES (:rid, :lvl, :stg, :oid, :msg) + """ + ), + { + "rid": run_id, + "lvl": level, + "stg": stage, + "oid": obj_id, + "msg": message[:500], + }, + ) + db.commit() + except Exception as e: + logger.warning("log_progress failed: %s", e) + try: + db.rollback() + except Exception: + pass + + async def run_region_sweep( region_code: int, developers: list[str] | None = None, @@ -921,6 +958,13 @@ async def run_region_sweep( db.commit() place = place_override or _place_str(region_code) + log_progress( + db, + run_id, + f"Старт sweep региона {region_code} (place={place}, devs={developers or '*'}," + f" extras={extras}, photos={download_photos_binary})", + stage="start", + ) try: async with BrowserSession( @@ -934,7 +978,19 @@ async def run_region_sweep( for status in statuses: rows = await fetch_objects_for_status(sess, place, status) all_objects.extend(rows) + log_progress( + db, + run_id, + f"objStatus={status}: получено {len(rows)} объектов", + stage="fetch_objects", + ) logger.info("place=%s total objects across %s = %d", place, statuses, len(all_objects)) + log_progress( + db, + run_id, + f"Всего объектов: {len(all_objects)} (по статусам {statuses})", + stage="fetch_objects", + ) # local developer filter if developers: @@ -946,6 +1002,12 @@ async def run_region_sweep( and o["developer"].get("companyGroup") in wanted_groups ] logger.info("after developers filter: %d objects", len(all_objects)) + log_progress( + db, + run_id, + f"После фильтра devs={developers}: {len(all_objects)} объектов", + stage="filter_devs", + ) # --- flats per object -------------------------------------------- all_flats: list[dict[str, Any]] = [] @@ -963,7 +1025,20 @@ async def run_region_sweep( len(all_objects), len(all_flats), ) + log_progress( + db, + run_id, + f"Квартиры: {i + 1}/{len(all_objects)} объектов," + f" {len(all_flats)} квартир", + stage="fetch_flats", + ) logger.info("total flats: %d (across %d objs)", len(all_flats), len(all_objects)) + log_progress( + db, + run_id, + f"Квартиры собраны: {len(all_flats)} шт по {len(all_objects)} объектам", + stage="fetch_flats", + ) # --- extras (sale_graph, sales_agg, infra, photos) per object ---- extras_counts = { @@ -1041,7 +1116,24 @@ async def run_region_sweep( ) # Periodic commit so failures are visible in admin UI even mid-run db.commit() + log_progress( + db, + run_id, + f"Extras: {i + 1}/{len(all_objects)} объектов |" + f" sale_graph={extras_counts['sale_graph_rows']}" + f" agg={extras_counts['sales_agg_rows']}" + f" infra={extras_counts['infra_rows']}" + f" photos={extras_counts['photos_rows']}" + f" downloaded={extras_counts['photos_downloaded']}", + stage="extras", + ) logger.info("extras done: %s", extras_counts) + log_progress( + db, + run_id, + f"Extras готовы: {extras_counts}", + stage="extras", + ) request_count = sess.request_count @@ -1063,6 +1155,12 @@ async def run_region_sweep( {"oc": obj_count, "fc": flat_count, "rq": request_count, "rid": run_id}, ) db.commit() + log_progress( + db, + run_id, + f"Готово ✅ objects={obj_count} flats={flat_count} requests={request_count}", + stage="done", + ) return { "run_id": run_id, "region_code": region_code, @@ -1087,6 +1185,13 @@ async def run_region_sweep( {"err": str(e)[:2000], "rid": run_id}, ) db.commit() + log_progress( + db, + run_id, + f"Ошибка ❌ {type(e).__name__}: {str(e)[:300]}", + level="error", + stage="failed", + ) raise finally: db.close() diff --git a/data/sql/54_schema_scrape_log.sql b/data/sql/54_schema_scrape_log.sql new file mode 100644 index 00000000..757e7360 --- /dev/null +++ b/data/sql/54_schema_scrape_log.sql @@ -0,0 +1,20 @@ +-- Per-run progress log for kn-API sweeps. +-- Workers append rows during run_region_sweep so the admin UI can show live +-- progress (started region X, fetched N objects, processing obj Y, ...). +-- Idempotent. + +CREATE TABLE IF NOT EXISTS kn_scrape_log ( + log_id BIGSERIAL PRIMARY KEY, + run_id BIGINT REFERENCES kn_scrape_runs(run_id) ON DELETE CASCADE, + ts TIMESTAMPTZ NOT NULL DEFAULT NOW(), + level TEXT NOT NULL CHECK (level IN ('info', 'warn', 'error')), + stage TEXT, + obj_id INT, + message TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_kn_scrape_log_run_ts + ON kn_scrape_log (run_id, ts DESC); + +CREATE INDEX IF NOT EXISTS idx_kn_scrape_log_ts + ON kn_scrape_log (ts DESC); diff --git a/frontend/src/app/admin/scrape/page.tsx b/frontend/src/app/admin/scrape/page.tsx index 89e38fe9..aa761421 100644 --- a/frontend/src/app/admin/scrape/page.tsx +++ b/frontend/src/app/admin/scrape/page.tsx @@ -43,6 +43,16 @@ interface QueueStatus { scheduled: QueueTask[]; } +interface LogRow { + log_id: number; + run_id: number | null; + ts: string | null; + level: "info" | "warn" | "error"; + stage: string | null; + obj_id: number | null; + message: string; +} + interface FailureRow { failure_id: number; run_id: number | null; @@ -68,6 +78,7 @@ export default function ScrapeAdminPage() { const [downloadPhotos, setDownloadPhotos] = useState(false); const [showFailures, setShowFailures] = useState(false); const [failuresRunId, setFailuresRunId] = useState(""); + const [logsRunId, setLogsRunId] = useState(""); const queryClient = useQueryClient(); const runs = useQuery({ @@ -159,6 +170,19 @@ export default function ScrapeAdminPage() { revokeMutation.mutate({ task_id: taskId, terminate: false }); }; + const logs = useQuery({ + queryKey: ["scrape-logs", logsRunId, token], + queryFn: () => { + const q = + logsRunId !== "" ? `?run_id=${logsRunId}&limit=200` : "?limit=200"; + return apiFetch(`/api/v1/admin/scrape/logs${q}`, { + headers: { "X-Admin-Token": token }, + }); + }, + enabled: !!token, + refetchInterval: 3000, + }); + const failures = useQuery({ queryKey: ["scrape-failures", failuresRunId, token], queryFn: () => { @@ -576,6 +600,84 @@ export default function ScrapeAdminPage() { ) : null} +
+
+

+ Прогресс задач{" "} + + · обновляется каждые 3 сек + +

+ + setLogsRunId(e.target.value === "" ? "" : Number(e.target.value)) + } + placeholder="run_id (опц.)" + style={{ ...inputStyle, width: 130 }} + /> +
+ {!token ? ( +

+ Введите Admin Token чтобы видеть лог. +

+ ) : logs.isLoading ? ( +

Загрузка…

+ ) : logs.isError ? ( +

Ошибка: {String(logs.error)}

+ ) : (logs.data?.length ?? 0) === 0 ? ( +

+ Нет событий{logsRunId !== "" ? ` для run_id=${logsRunId}` : ""}. + Когда worker запустится — появятся записи. +

+ ) : ( +
+ {(logs.data ?? []).map((l) => { + const color = + l.level === "error" + ? "#fca5a5" + : l.level === "warn" + ? "#fcd34d" + : "#a3e635"; + return ( +
+ + {l.ts?.slice(11, 19)} + {" "} + {l.level.toUpperCase()}{" "} + + [run-{l.run_id} + {l.stage ? ` · ${l.stage}` : ""} + {l.obj_id ? ` · obj=${l.obj_id}` : ""}] + {" "} + {l.message} +
+ ); + })} +
+ )} +
+