From 0202720096963f72dbbc8395495854a3b976041b Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 23 May 2026 14:54:25 +0000 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20in-app=20scheduler=20=E2=80=94?= =?UTF-8?q?=20UI-managed=20schedule=20(replaces=20SSH=20crontab)=20(#482)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/api/v1/admin.py | 91 ++++++++ tradein-mvp/backend/app/main.py | 25 ++- tradein-mvp/backend/app/schemas/trade_in.py | 29 ++- tradein-mvp/backend/app/services/scheduler.py | 203 ++++++++++++++++++ .../backend/data/sql/052_scrape_schedules.sql | 46 ++++ tradein-mvp/backend/tests/test_scheduler.py | 51 +++++ 6 files changed, 443 insertions(+), 2 deletions(-) create mode 100644 tradein-mvp/backend/app/services/scheduler.py create mode 100644 tradein-mvp/backend/data/sql/052_scrape_schedules.sql create mode 100644 tradein-mvp/backend/tests/test_scheduler.py diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index fa9599a4..9fa51729 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -5,6 +5,7 @@ Auth: Caddy basic_auth gate'ит /trade-in/api/v1/admin/* — application layer from __future__ import annotations +import json import logging import time from typing import Annotated, Literal @@ -16,6 +17,7 @@ from sqlalchemy.orm import Session from app.core.config import settings from app.core.db import SessionLocal, get_db +from app.schemas.trade_in import ScheduleConfig, ScheduleConfigUpdate from app.services import cian_session as cian_session_svc from app.services import scrape_runs as runs_mod from app.services.geocoder import geocode @@ -541,3 +543,92 @@ def cancel_avito_city_sweep( """Отменить running sweep. Cooperative: pipeline проверяет scrape_runs.status каждый anchor.""" cancelled = runs_mod.mark_cancelled(db, run_id) return {"ok": True, "run_id": run_id, "cancelled": cancelled} + + +# ── In-app scheduler endpoints (Stage 4e) ──────────────────────────────────── + + +@router.get("/scrape/schedules", response_model=list[ScheduleConfig]) +def list_schedules(db: Annotated[Session, Depends(get_db)]) -> list[ScheduleConfig]: + """Список всех schedules. Сейчас single row 'avito_city_sweep', extensible.""" + rows = db.execute( + text( + """ + SELECT id, source, enabled, window_start_hour, window_end_hour, + default_params, last_run_id, last_run_at, next_run_at, + created_at, updated_at + FROM scrape_schedules + ORDER BY source + """ + ), + ).mappings().all() + return [ + ScheduleConfig( + id=r["id"], + source=r["source"], + enabled=r["enabled"], + window_start_hour=r["window_start_hour"], + window_end_hour=r["window_end_hour"], + default_params=r["default_params"] or {}, + last_run_id=r["last_run_id"], + last_run_at=r["last_run_at"].isoformat() if r["last_run_at"] else None, + next_run_at=r["next_run_at"].isoformat() if r["next_run_at"] else None, + updated_at=r["updated_at"].isoformat() if r["updated_at"] else None, + ) + for r in rows + ] + + +@router.put("/scrape/schedules/{source}", response_model=ScheduleConfig) +def update_schedule( + source: str, + payload: ScheduleConfigUpdate, + db: Annotated[Session, Depends(get_db)], +) -> ScheduleConfig: + """UPDATE existing schedule (create если не существует, через INSERT ON CONFLICT).""" + from app.services.scheduler import compute_next_run_at + + # Compute new next_run_at если window изменился — recompute, иначе keep existing + next_at = compute_next_run_at(payload.window_start_hour, payload.window_end_hour) + + row = db.execute( + text( + """ + INSERT INTO scrape_schedules + (source, enabled, window_start_hour, window_end_hour, default_params, next_run_at) + VALUES (:source, :enabled, :ws, :we, CAST(:params AS jsonb), :next_at) + ON CONFLICT (source) DO UPDATE SET + enabled = EXCLUDED.enabled, + window_start_hour = EXCLUDED.window_start_hour, + window_end_hour = EXCLUDED.window_end_hour, + default_params = EXCLUDED.default_params, + next_run_at = EXCLUDED.next_run_at, + updated_at = NOW() + RETURNING id, source, enabled, window_start_hour, window_end_hour, + default_params, last_run_id, last_run_at, next_run_at, updated_at + """ + ), + { + "source": source, + "enabled": payload.enabled, + "ws": payload.window_start_hour, + "we": payload.window_end_hour, + "params": json.dumps(payload.default_params, ensure_ascii=False), + "next_at": next_at, + }, + ).mappings().fetchone() + db.commit() + + assert row is not None, f"scrape_schedules upsert returned no row for source={source!r}" + return ScheduleConfig( + id=row["id"], + source=row["source"], + enabled=row["enabled"], + window_start_hour=row["window_start_hour"], + window_end_hour=row["window_end_hour"], + default_params=row["default_params"] or {}, + last_run_id=row["last_run_id"], + last_run_at=row["last_run_at"].isoformat() if row["last_run_at"] else None, + next_run_at=row["next_run_at"].isoformat() if row["next_run_at"] else None, + updated_at=row["updated_at"].isoformat() if row["updated_at"] else None, + ) diff --git a/tradein-mvp/backend/app/main.py b/tradein-mvp/backend/app/main.py index 3f0f4c5e..a32cb039 100644 --- a/tradein-mvp/backend/app/main.py +++ b/tradein-mvp/backend/app/main.py @@ -6,7 +6,10 @@ Standalone версия, выделена из основного gendesign repo from __future__ import annotations +import asyncio import logging +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager import sentry_sdk from fastapi import FastAPI @@ -15,6 +18,9 @@ from fastapi.middleware.cors import CORSMiddleware from app.api.v1 import admin, brand, geocode, search, trade_in from app.core.config import settings from app.core.ratelimit import RateLimitMiddleware +from app.services.scheduler import scheduler_loop + +logger = logging.getLogger(__name__) logging.basicConfig( level=logging.INFO, @@ -32,10 +38,27 @@ if settings.glitchtip_dsn: ) logging.getLogger("app.main").info("GlitchTip monitoring enabled") + +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: + # Startup: launch scheduler background task + scheduler_task = asyncio.create_task(scheduler_loop()) + logger.info("FastAPI lifespan: scheduler task spawned") + yield + # Shutdown: cancel scheduler + scheduler_task.cancel() + try: + await scheduler_task + except asyncio.CancelledError: + pass + logger.info("FastAPI lifespan: scheduler cancelled cleanly") + + app = FastAPI( title="Trade-In MVP API", - description="Оценка вторичного жилья (выкупная стоимость) — копия trade-in feature из gendesign", + description="Оценка вторичного жилья (выкупная стоимость) — копия trade-in feature из gendesign", # noqa: E501 version="0.1.0", + lifespan=lifespan, ) app.add_middleware( diff --git a/tradein-mvp/backend/app/schemas/trade_in.py b/tradein-mvp/backend/app/schemas/trade_in.py index 6656285b..7ae83e1b 100644 --- a/tradein-mvp/backend/app/schemas/trade_in.py +++ b/tradein-mvp/backend/app/schemas/trade_in.py @@ -6,7 +6,7 @@ POST /api/v1/trade-in/estimate → AggregatedEstimate from __future__ import annotations from datetime import date, datetime -from typing import Literal +from typing import Any, Literal from uuid import UUID from pydantic import BaseModel, Field @@ -128,3 +128,30 @@ class IMVBenchmarkResponse(BaseModel): # comparison vs our estimate our_median_price: int | None = None diff_pct: float | None = None # (our - imv) / imv * 100 + + +# ── In-app scheduler (Stage 4e) ───────────────────────────────────────────── + + +class ScheduleConfig(BaseModel): + """Текущая конфигурация schedule для source.""" + + id: int + source: str + enabled: bool + window_start_hour: int = Field(ge=0, le=23) + window_end_hour: int = Field(ge=0, le=23) + default_params: dict[str, Any] = Field(default_factory=dict) + last_run_id: int | None = None + last_run_at: str | None = None # ISO + next_run_at: str | None = None # ISO + updated_at: str | None = None # ISO + + +class ScheduleConfigUpdate(BaseModel): + """PUT body для update_schedule endpoint.""" + + enabled: bool = True + window_start_hour: int = Field(default=2, ge=0, le=23) + window_end_hour: int = Field(default=5, ge=0, le=23) + default_params: dict[str, Any] = Field(default_factory=dict) diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py new file mode 100644 index 00000000..9469d51f --- /dev/null +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -0,0 +1,203 @@ +"""In-app scheduler — periodically triggers configured scrapes based on scrape_schedules table. + +Запускается в FastAPI lifespan (app.main lifespan context manager) при startup. +Логика: +1. Каждые 60 sec query scrape_schedules WHERE enabled=true AND next_run_at <= NOW(). +2. Для каждого: проверить нет ли уже running run этого source — если нет, trigger. +3. После trigger — recompute next_run_at в следующих суток. +4. На old runs (orphaned status='running' > 6h без heartbeat) — пометить как 'zombie'. +""" +from __future__ import annotations + +import asyncio +import logging +import random +from datetime import UTC, datetime, time, timedelta +from typing import Any + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.core.db import SessionLocal +from app.services import scrape_runs as runs_mod +from app.services.scrape_pipeline import run_avito_city_sweep + +logger = logging.getLogger(__name__) + +# Loop interval — check каждую минуту +SCHEDULER_TICK_SEC = 60 +ZOMBIE_THRESHOLD_HOURS = 6 + + +def compute_next_run_at( + window_start_hour: int, window_end_hour: int, *, now: datetime | None = None +) -> datetime: + """Pick random datetime в window [start, end) UTC, для СЛЕДУЮЩИХ суток после now. + + Если window_end_hour <= window_start_hour → cross-midnight window + (например 22→3 → окно 22:00-23:59 ИЛИ 00:00-02:59). + """ + now = now or datetime.now(tz=UTC) + tomorrow = (now + timedelta(days=1)).date() + + if window_end_hour > window_start_hour: + # Обычное окно (например 2..5 → 02:00-04:59) + start_seconds = window_start_hour * 3600 + end_seconds = window_end_hour * 3600 + rand_seconds = random.randint(start_seconds, end_seconds - 1) + return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta( + seconds=rand_seconds + ) + else: + # Cross-midnight (22..3 → 22:00-23:59 + 00:00-02:59) + # Длина окна = (24-start) + end часов + total_seconds = ((24 - window_start_hour) + window_end_hour) * 3600 + rand_seconds = random.randint(0, total_seconds - 1) + # Если rand попадает в первую часть (start..24) + first_half = (24 - window_start_hour) * 3600 + if rand_seconds < first_half: + # Текущая дата (если ещё не наступило окно) или next day + base_date = now.date() if now.hour < window_start_hour else tomorrow + return datetime.combine(base_date, time(0, 0), tzinfo=UTC) + timedelta( + seconds=window_start_hour * 3600 + rand_seconds + ) + else: + # Во второй части (0..end), следующего дня + offset = rand_seconds - first_half + return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta(seconds=offset) + + +def has_running_run(db: Session, source: str) -> bool: + """Есть ли активный run для source (status='running').""" + row = db.execute( + text( + """ + SELECT 1 FROM scrape_runs + WHERE source = :source AND status = 'running' + LIMIT 1 + """ + ), + {"source": source}, + ).fetchone() + return row is not None + + +def reap_zombies(db: Session) -> int: + """Mark scrape_runs as 'zombie' если heartbeat не обновлялся > ZOMBIE_THRESHOLD_HOURS hours.""" + zombie_interval = f"{ZOMBIE_THRESHOLD_HOURS} hours" + result = db.execute( + text( + """ + UPDATE scrape_runs + SET status = 'zombie', finished_at = NOW() + WHERE status = 'running' + AND (heartbeat_at IS NULL + OR heartbeat_at < NOW() - CAST(:interval AS interval)) + RETURNING id + """ + ), + {"interval": zombie_interval}, + ) + rows = result.fetchall() + db.commit() + if rows: + logger.warning("scheduler: reaped %d zombie runs: %s", len(rows), [r.id for r in rows]) + return len(rows) + + +async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать новый scrape_runs + launch run_avito_city_sweep в asyncio.create_task. + + Returns run_id (или None если skip — есть running run). + """ + if has_running_run(db, schedule_row["source"]): + logger.info("scheduler: skip — already running for source=%s", schedule_row["source"]) + return None + + params = schedule_row.get("default_params") or {} + run_id = runs_mod.create_run(db, source=schedule_row["source"], params=params) + + # Update schedule: last_run_id, last_run_at, recompute next_run_at + next_at = compute_next_run_at( + schedule_row["window_start_hour"], + schedule_row["window_end_hour"], + ) + db.execute( + text( + """ + UPDATE scrape_schedules + SET last_run_id = :run_id, last_run_at = NOW(), + next_run_at = :next_at, updated_at = NOW() + WHERE source = :source + """ + ), + {"run_id": run_id, "next_at": next_at, "source": schedule_row["source"]}, + ) + db.commit() + + # Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick) + async def _run() -> None: + run_db = SessionLocal() + try: + await run_avito_city_sweep( + run_db, + run_id=run_id, + pages_per_anchor=int(params.get("pages_per_anchor", 3)), + detail_top_n=int(params.get("detail_top_n", 20)), + request_delay_sec=float(params.get("request_delay_sec", 7.0)), + enrich_houses=bool(params.get("enrich_houses", True)), + radius_m=int(params.get("radius_m", 1500)), + ) + except Exception: + logger.exception("scheduler: run_avito_city_sweep crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + # Keep reference to avoid GC before task completes (RUF006) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info( + "scheduler: triggered city_sweep run_id=%d next_run_at=%s", run_id, next_at.isoformat() + ) + return run_id + + +def get_due_schedules(db: Session) -> list[dict[str, Any]]: + """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW()).""" + rows = db.execute( + text( + """ + SELECT id, source, enabled, window_start_hour, window_end_hour, + default_params, last_run_id, last_run_at, next_run_at + FROM scrape_schedules + WHERE enabled = true + AND (next_run_at IS NULL OR next_run_at <= NOW()) + """ + ), + ).mappings().all() + return [dict(r) for r in rows] + + +async def scheduler_loop() -> None: + """Бесконечный async loop — tick каждые SCHEDULER_TICK_SEC секунд.""" + logger.info("scheduler: started (tick=%ds)", SCHEDULER_TICK_SEC) + # Initial sleep 30s чтобы дать FastAPI startup завершиться + await asyncio.sleep(30) + while True: + try: + db = SessionLocal() + try: + # Reap zombies first + reap_zombies(db) + # Process due schedules + due = get_due_schedules(db) + for sch in due: + if sch["source"] == "avito_city_sweep": + await trigger_avito_city_sweep_run(db, sch) + else: + logger.warning("scheduler: unknown source=%s, skip", sch["source"]) + finally: + db.close() + except Exception: + logger.exception("scheduler: tick failed") + await asyncio.sleep(SCHEDULER_TICK_SEC) diff --git a/tradein-mvp/backend/data/sql/052_scrape_schedules.sql b/tradein-mvp/backend/data/sql/052_scrape_schedules.sql new file mode 100644 index 00000000..3d354268 --- /dev/null +++ b/tradein-mvp/backend/data/sql/052_scrape_schedules.sql @@ -0,0 +1,46 @@ +-- 052_scrape_schedules.sql +-- In-app scheduler config — заменяет cron-script (avito-city-sweep.sh) для UI-managed scheduling. +-- Background loop в FastAPI lifespan periodically (every 60s) checks этой таблицы. + +BEGIN; + +CREATE TABLE IF NOT EXISTS scrape_schedules ( + id bigserial PRIMARY KEY, + source text NOT NULL UNIQUE, -- 'avito_city_sweep' + enabled boolean NOT NULL DEFAULT true, + window_start_hour int NOT NULL DEFAULT 2, -- UTC hour 0-23 inclusive + window_end_hour int NOT NULL DEFAULT 5, -- UTC hour 0-23 inclusive (если меньше start → ночное окно cross-midnight) + default_params jsonb NOT NULL DEFAULT '{}'::jsonb, + -- {"pages_per_anchor":3,"detail_top_n":20, + -- "request_delay_sec":7,"enrich_houses":true,"radius_m":1500} + last_run_id bigint REFERENCES scrape_runs(id), + last_run_at timestamptz, + next_run_at timestamptz, -- UTC; computed после каждого fire + created_at timestamptz NOT NULL DEFAULT NOW(), + updated_at timestamptz NOT NULL DEFAULT NOW(), + + CONSTRAINT window_start_range CHECK (window_start_hour BETWEEN 0 AND 23), + CONSTRAINT window_end_range CHECK (window_end_hour BETWEEN 0 AND 23) +); + +CREATE INDEX IF NOT EXISTS scrape_schedules_next_run_idx + ON scrape_schedules (next_run_at) + WHERE enabled = true; + +-- Seed: default schedule для avito_city_sweep (02:00-05:00 UTC = 05:00-08:00 МСК) +INSERT INTO scrape_schedules (source, enabled, window_start_hour, window_end_hour, default_params) +VALUES ( + 'avito_city_sweep', + true, + 2, + 5, + '{"pages_per_anchor": 3, "detail_top_n": 20, "request_delay_sec": 7.0, "enrich_houses": true, "radius_m": 1500}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS 'In-app scheduler config (заменяет cron-script setup).'; +COMMENT ON COLUMN scrape_schedules.window_start_hour IS 'UTC hour 0-23. Next_run_at пикается random в [window_start_hour, window_end_hour].'; +COMMENT ON COLUMN scrape_schedules.window_end_hour IS 'UTC. Если меньше start → cross-midnight (например 22→3 = 22:00-03:00 UTC).'; +COMMENT ON COLUMN scrape_schedules.next_run_at IS 'Когда scheduler должен trigger next run (UTC). Recomputed после каждого fire.'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_scheduler.py b/tradein-mvp/backend/tests/test_scheduler.py new file mode 100644 index 00000000..2f04b0ed --- /dev/null +++ b/tradein-mvp/backend/tests/test_scheduler.py @@ -0,0 +1,51 @@ +"""Offline smoke for scheduler logic.""" +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from datetime import datetime, timezone + +import pytest + +from app.services.scheduler import ZOMBIE_THRESHOLD_HOURS, compute_next_run_at + + +def test_compute_next_run_in_normal_window() -> None: + now = datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc) + next_at = compute_next_run_at(2, 5, now=now) + assert next_at.date() == datetime(2026, 5, 24).date() + assert 2 <= next_at.hour <= 4 # window [2, 5) UTC + + +def test_compute_next_run_cross_midnight() -> None: + now = datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc) + # Window 22..3 = 22:00-23:59 + 00:00-02:59 + next_at = compute_next_run_at(22, 3, now=now) + # Должен попасть либо в [22,23] либо в [0,3) range + hour = next_at.hour + assert hour >= 22 or hour < 3 + + +def test_compute_next_run_window_1_hour() -> None: + now = datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc) + # Минимальное окно [3, 4) → ровно 3-й час + next_at = compute_next_run_at(3, 4, now=now) + assert next_at.hour == 3 + + +def test_zombie_threshold_constant() -> None: + assert ZOMBIE_THRESHOLD_HOURS > 0 + + +def test_compute_next_run_is_future() -> None: + """next_run_at всегда в будущем (завтра) — не в прошлом.""" + now = datetime(2026, 5, 23, 3, 30, tzinfo=timezone.utc) + next_at = compute_next_run_at(2, 5, now=now) + assert next_at > now + + +def test_compute_next_run_timezone_aware() -> None: + """Результат всегда timezone-aware (UTC).""" + now = datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc) + next_at = compute_next_run_at(2, 5, now=now) + assert next_at.tzinfo is not None