This commit is contained in:
lekss361 2026-04-27 20:51:44 +03:00
parent a56216aeb9
commit 12b1eb8169
5 changed files with 278 additions and 1 deletions

View file

@ -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)],

View file

@ -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()

View file

@ -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);

View file

@ -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<number | "">("");
const [logsRunId, setLogsRunId] = useState<number | "">("");
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<LogRow[]>(`/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}
</section>
<section style={{ ...cardStyle, marginTop: 16 }}>
<div
style={{
display: "flex",
alignItems: "center",
justifyContent: "space-between",
marginBottom: 12,
}}
>
<h2 style={{ margin: 0, fontSize: 16 }}>
Прогресс задач{" "}
<span style={{ fontWeight: 400, color: "#5b6066", fontSize: 13 }}>
· обновляется каждые 3 сек
</span>
</h2>
<input
type="number"
value={logsRunId}
onChange={(e) =>
setLogsRunId(e.target.value === "" ? "" : Number(e.target.value))
}
placeholder="run_id (опц.)"
style={{ ...inputStyle, width: 130 }}
/>
</div>
{!token ? (
<p style={{ color: "#5b6066", fontSize: 13 }}>
Введите Admin Token чтобы видеть лог.
</p>
) : logs.isLoading ? (
<p style={{ color: "#5b6066" }}>Загрузка</p>
) : logs.isError ? (
<p style={{ color: "#b3261e" }}>Ошибка: {String(logs.error)}</p>
) : (logs.data?.length ?? 0) === 0 ? (
<p style={{ color: "#5b6066", fontSize: 13 }}>
Нет событий{logsRunId !== "" ? ` для run_id=${logsRunId}` : ""}.
Когда worker запустится появятся записи.
</p>
) : (
<div
style={{
maxHeight: 360,
overflowY: "auto",
fontFamily: "ui-monospace, SFMono-Regular, Menlo, monospace",
fontSize: 12,
border: "1px solid #eef0f3",
borderRadius: 6,
background: "#0f172a",
color: "#e2e8f0",
padding: 10,
}}
>
{(logs.data ?? []).map((l) => {
const color =
l.level === "error"
? "#fca5a5"
: l.level === "warn"
? "#fcd34d"
: "#a3e635";
return (
<div key={l.log_id} style={{ marginBottom: 2 }}>
<span style={{ color: "#94a3b8" }}>
{l.ts?.slice(11, 19)}
</span>{" "}
<span style={{ color }}>{l.level.toUpperCase()}</span>{" "}
<span style={{ color: "#7dd3fc" }}>
[run-{l.run_id}
{l.stage ? ` · ${l.stage}` : ""}
{l.obj_id ? ` · obj=${l.obj_id}` : ""}]
</span>{" "}
<span>{l.message}</span>
</div>
);
})}
</div>
)}
</section>
<section style={{ ...cardStyle, marginTop: 16 }}>
<div
style={{

File diff suppressed because one or more lines are too long