From 3f9dc6643b69e58f9e6452d08de7a51a54525257 Mon Sep 17 00:00:00 2001 From: gitrusprus Date: Sun, 13 Sep 2026 20:22:38 +0300 Subject: [PATCH] Add reusable LLM parser profiles with multi-event extract. Support kind=llm profiles (instruction/schema), optional multi-event posts via #eN URLs, and recover stale running/queued parse jobs after worker crashes. Co-authored-by: Cursor --- .env.example | 3 + AGENTS.md | 2 +- README.md | 8 +- centers/analytics/api/app/models.py | 3 + centers/analytics/api/app/routers/admin.py | 20 +- centers/analytics/api/app/routers/internal.py | 3 +- .../api/app/routers/parser_profiles.py | 135 ++++- centers/analytics/api/app/schemas.py | 6 + .../analytics/api/app/services/job_stale.py | 36 ++ centers/analytics/api/app/services/jobs.py | 19 + .../analytics/api/app/services/migrations.py | 13 +- .../api/app/services/parser_builder.py | 117 ++++ .../analytics/api/app/services/scheduler.py | 44 +- .../analytics/frontend/src/api/entities.ts | 36 +- centers/analytics/frontend/src/types/admin.ts | 38 +- .../frontend/src/views/ParserProfilesView.vue | 525 +++++++++++++++--- .../frontend/src/views/ParsersView.vue | 28 +- centers/parsing/ARCHITECTURE.md | 34 +- .../workers/workers/adapters/telegram.py | 76 ++- .../workers/workers/heuristic_profile.py | 8 +- .../parsing/workers/workers/llm_extract.py | 176 ++++-- .../workers/sources/telegram_listener.py | 5 +- contracts/heuristic_profile.py | 53 ++ contracts/llm_profile.py | 86 +++ contracts/sources.py | 42 +- docs/architecture-overview.md | 8 +- docs/contracts.md | 15 +- 27 files changed, 1334 insertions(+), 205 deletions(-) create mode 100644 centers/analytics/api/app/services/job_stale.py create mode 100644 contracts/llm_profile.py diff --git a/.env.example b/.env.example index bc32015..ed113bd 100644 --- a/.env.example +++ b/.env.example @@ -27,6 +27,9 @@ ADMIN_JWT_SECRET=change-me-jwt-secret # DEEPSEEK_BASE_URL=https://api.deepseek.com # DEEPSEEK_MODEL=deepseek-chat +# Recover parse jobs stuck in running/queued after worker crash (seconds, default 900) +# STALE_JOB_SECONDS=900 + # CP adapter workers (set in docker-compose; override locally if needed) # ENABLED_ADAPTERS=telegram # WORKER_FAMILIES=telegram diff --git a/AGENTS.md b/AGENTS.md index 3029934..655cd44 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -74,7 +74,7 @@ Adapters return dicts matching `IngestEventItem` (`contracts/ingest.py`). Requir ### LLM extract (runtime, not dev) -`extract_mode: llm` uses DeepSeek via `workers/llm_extract.py`. Key: `DEEPSEEK_API_KEY` in `.env`. Do not confuse with Cursor dev agents. +`extract_mode: llm` uses DeepSeek via `workers/llm_extract.py` (batch). Reusable profiles in admin UI (`kind=heuristic|llm`) flatten into `extract_mode=profile` or `llm`. Key: `DEEPSEEK_API_KEY` in `.env`. Do not confuse with Cursor dev agents. ## Common tasks diff --git a/README.md b/README.md index 036c7c8..8927a2c 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,8 @@ docker compose up --build | Раздел | Путь | Описание | |--------|------|----------| | **Карта** | `/` | Интерактивная карта событий (публичный просмотр). CRUD ручных объектов — только для админа (ПКМ). Поддерживает `?eventId=` | -| **Парсеры** | `/parsers` | Адаптеры `telegram` / `crawl4ai` / `viina`, интервал, CRUD; дедуп по `source_url` | +| **Парсеры** | `/parsers` | Связка канал + профиль (`heuristic`→`extract_mode=profile`, `llm`→`extract_mode=llm`); дедуп по `source_url` | +| **Профили** | `/parser-profiles` | Reusable heuristic (правила) или llm (instruction/schema); целевые поля = Event | | **События** | `/events` | Фильтрация, пагинация, просмотр деталей, ссылка «На карте» для событий с координатами | | **Аналитика** | `/analytics` | KPI-карточки, график динамики ingest за 30 дней, топ населённых пунктов и регионов | | **ПИ** | `/consumers` | CRUD подписчиков distribution API, ротация ключей, тест среза через `/api/v1/events` | @@ -214,3 +215,8 @@ docker compose down ``` Данные PostgreSQL сохраняются в volume `pgdata`. + +## Вход в админку (/login): + +Логин: admin +Пароль: change-me diff --git a/centers/analytics/api/app/models.py b/centers/analytics/api/app/models.py index d8c6bd6..beb7d23 100644 --- a/centers/analytics/api/app/models.py +++ b/centers/analytics/api/app/models.py @@ -98,8 +98,11 @@ class ParserProfile(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) name: Mapped[str] = mapped_column(String(255), nullable=False) + # heuristic = static rules; llm = instruction + extract_schema at CP runtime + kind: Mapped[str] = mapped_column(String(50), default="heuristic", nullable=False, index=True) sample_post: Mapped[str] = mapped_column(Text, default="") heuristic_profile: Mapped[dict | None] = mapped_column(JSON, nullable=True) + llm_profile: Mapped[dict | None] = mapped_column(JSON, nullable=True) status: Mapped[str] = mapped_column(String(50), default="draft", index=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), diff --git a/centers/analytics/api/app/routers/admin.py b/centers/analytics/api/app/routers/admin.py index c88c320..e85bca8 100644 --- a/centers/analytics/api/app/routers/admin.py +++ b/centers/analytics/api/app/routers/admin.py @@ -87,7 +87,11 @@ def create_parse_job(payload: ParseJobCreate, db: Session = Depends(get_db)): raise HTTPException(status_code=404, detail="Profile not found") if not channel: raise HTTPException(status_code=404, detail="Channel not found") - if not profile.heuristic_profile: + kind = (profile.kind or "heuristic").strip().lower() + if kind == "llm": + if not profile.llm_profile: + raise HTTPException(status_code=400, detail="LLM profile has no llm_profile") + elif not profile.heuristic_profile: raise HTTPException(status_code=400, detail="Profile has no heuristic_profile") if not channel.is_active: raise HTTPException(status_code=400, detail="Channel is inactive") @@ -146,7 +150,10 @@ def retry_parse_job(job_id: int, db: Session = Depends(get_db)): job = db.query(ParseJob).filter(ParseJob.id == job_id).first() if not job: raise HTTPException(status_code=404, detail="Job not found") - if job.status in ("queued", "running"): + + from ..services.job_stale import is_stale_job + + if job.status in ("queued", "running") and not is_stale_job(job): raise HTTPException(status_code=409, detail="Job is already running or queued") job.status = "queued" @@ -218,7 +225,14 @@ def delete_parse_job(job_id: int, db: Session = Depends(get_db)): if not job: raise HTTPException(status_code=404, detail="Job not found") if job.status == "running": - raise HTTPException(status_code=409, detail="Cannot delete a running job") + from ..services.job_stale import is_stale_job + + if not is_stale_job(job): + raise HTTPException(status_code=409, detail="Cannot delete a running job") + # Stale running — allow delete after marking failed for audit trail + job.status = "failed" + job.last_error = "Deleted while stale running" + db.commit() db.delete(job) db.commit() diff --git a/centers/analytics/api/app/routers/internal.py b/centers/analytics/api/app/routers/internal.py index 7afe57d..b0a4185 100644 --- a/centers/analytics/api/app/routers/internal.py +++ b/centers/analytics/api/app/routers/internal.py @@ -52,7 +52,8 @@ def update_job_status( raise HTTPException(status_code=404, detail="Job not found") job.status = status - if status in ("completed", "failed"): + # Anchor staleness detection: running/queued start, and terminal finish + if status in ("running", "queued", "completed", "failed"): job.last_run_at = datetime.now(timezone.utc) job.last_error = error db.commit() diff --git a/centers/analytics/api/app/routers/parser_profiles.py b/centers/analytics/api/app/routers/parser_profiles.py index bb075aa..1afdc95 100644 --- a/centers/analytics/api/app/routers/parser_profiles.py +++ b/centers/analytics/api/app/routers/parser_profiles.py @@ -9,6 +9,7 @@ from pydantic import BaseModel, Field from sqlalchemy.orm import Session from contracts.heuristic_profile import HeuristicProfile +from contracts.llm_profile import LlmProfile from ..database import get_db from ..deps import verify_admin @@ -44,6 +45,22 @@ class PreviewRequest(BaseModel): class PreviewResponse(BaseModel): fields: dict[str, str] + matched: bool = True + missing_required: list[str] = Field(default_factory=list) + + +class PreviewLlmRequest(BaseModel): + sample_post: str = Field(min_length=1) + llm_profile: dict[str, Any] + + +class PreviewLlmResponse(BaseModel): + fields: dict[str, str] + events: list[dict[str, str]] = Field(default_factory=list) + matched: bool = True + missing_required: list[str] = Field(default_factory=list) + is_event: bool = True + matched_count: int = 0 def _validate_heuristic_profile(raw: dict[str, Any] | None) -> dict[str, Any] | None: @@ -55,9 +72,33 @@ def _validate_heuristic_profile(raw: dict[str, Any] | None) -> dict[str, Any] | raise HTTPException(status_code=400, detail=f"Invalid heuristic_profile: {exc}") from exc -def _profile_status(heuristic_profile: dict | None, explicit: str | None = None) -> str: +def _validate_llm_profile(raw: dict[str, Any] | None) -> dict[str, Any] | None: + if raw is None: + return None + try: + return LlmProfile.model_validate(raw).model_dump() + except Exception as exc: + raise HTTPException(status_code=400, detail=f"Invalid llm_profile: {exc}") from exc + + +def _normalize_kind(kind: str | None) -> str: + value = (kind or "heuristic").strip().lower() + if value not in ("heuristic", "llm"): + raise HTTPException(status_code=400, detail="kind must be heuristic or llm") + return value + + +def _profile_status( + *, + kind: str, + heuristic_profile: dict | None, + llm_profile: dict | None, + explicit: str | None = None, +) -> str: if explicit: return explicit + if kind == "llm": + return "ready" if llm_profile else "draft" return "ready" if heuristic_profile else "draft" @@ -66,6 +107,18 @@ def target_fields(): return {"fields": builder.get_target_fields()} +@router.get("/llm-defaults") +def llm_defaults(): + from contracts.llm_profile import DEFAULT_EXTRACT_SCHEMA, DEFAULT_INSTRUCTION + + return { + "instruction": DEFAULT_INSTRUCTION, + "extract_schema": DEFAULT_EXTRACT_SCHEMA, + "required_fields": [], + "multi_event": False, + } + + @router.post("/generate", response_model=GenerateResponse) async def generate_profile(payload: GenerateRequest): if not builder.deepseek_enabled(): @@ -87,6 +140,9 @@ async def generate_profile(payload: GenerateRequest): detail=f"DeepSeek generate failed: {exc}", ) from exc dumped = profile.model_dump() + if payload.current_profile and isinstance(payload.current_profile.get("required_fields"), list): + dumped["required_fields"] = payload.current_profile["required_fields"] + dumped = HeuristicProfile.model_validate(dumped).model_dump() preview = builder.preview_with_profile(payload.sample_post, dumped) return GenerateResponse( profile=dumped, @@ -102,10 +158,40 @@ def preview_profile(payload: PreviewRequest): raise HTTPException(status_code=400, detail="heuristic_profile is required") try: HeuristicProfile.model_validate(raw) - fields = builder.preview_with_profile(payload.sample_post, raw) + fields, matched, missing = builder.match_preview(payload.sample_post, raw) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - return PreviewResponse(fields=fields) + return PreviewResponse(fields=fields, matched=matched, missing_required=missing) + + +@router.post("/preview-llm", response_model=PreviewLlmResponse) +async def preview_llm_profile(payload: PreviewLlmRequest): + if not builder.deepseek_enabled(): + raise HTTPException( + status_code=503, + detail="DEEPSEEK_API_KEY is not set on ca-api. Add it to .env for LLM preview.", + ) + try: + LlmProfile.model_validate(payload.llm_profile) + events, fields, matched, missing, is_event, matched_count = await builder.preview_llm_extract( + payload.sample_post, + payload.llm_profile, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except Exception as exc: + raise HTTPException( + status_code=502, + detail=f"DeepSeek LLM preview failed: {exc}", + ) from exc + return PreviewLlmResponse( + fields=fields, + events=events, + matched=matched, + missing_required=missing, + is_event=is_event, + matched_count=matched_count, + ) @router.get("", response_model=list[ParserProfileRead]) @@ -115,12 +201,26 @@ def list_profiles(db: Session = Depends(get_db)): @router.post("", response_model=ParserProfileRead, status_code=201) def create_profile(payload: ParserProfileCreate, db: Session = Depends(get_db)): + kind = _normalize_kind(payload.kind) heuristic = _validate_heuristic_profile(payload.heuristic_profile) + llm = _validate_llm_profile(payload.llm_profile) + if kind == "heuristic" and llm and not heuristic: + # ignore stray llm blob when creating heuristic + llm = None + if kind == "llm" and heuristic and not llm: + heuristic = None profile = ParserProfile( name=payload.name.strip(), + kind=kind, sample_post=payload.sample_post or "", - heuristic_profile=heuristic, - status=_profile_status(heuristic, payload.status), + heuristic_profile=heuristic if kind == "heuristic" else None, + llm_profile=llm if kind == "llm" else None, + status=_profile_status( + kind=kind, + heuristic_profile=heuristic if kind == "heuristic" else None, + llm_profile=llm if kind == "llm" else None, + explicit=payload.status, + ), ) db.add(profile) db.commit() @@ -150,12 +250,33 @@ def update_profile( profile.name = payload.name.strip() if payload.sample_post is not None: profile.sample_post = payload.sample_post + if payload.kind is not None: + profile.kind = _normalize_kind(payload.kind) + kind = _normalize_kind(profile.kind) + if "heuristic_profile" in payload.model_fields_set: profile.heuristic_profile = _validate_heuristic_profile(payload.heuristic_profile) + if "llm_profile" in payload.model_fields_set: + profile.llm_profile = _validate_llm_profile(payload.llm_profile) + + # Keep only the blob matching kind + if kind == "heuristic": + profile.llm_profile = None + else: + profile.heuristic_profile = None + if payload.status is not None: profile.status = payload.status - elif "heuristic_profile" in payload.model_fields_set: - profile.status = _profile_status(profile.heuristic_profile) + elif ( + "heuristic_profile" in payload.model_fields_set + or "llm_profile" in payload.model_fields_set + or payload.kind is not None + ): + profile.status = _profile_status( + kind=kind, + heuristic_profile=profile.heuristic_profile, + llm_profile=profile.llm_profile, + ) db.commit() db.refresh(profile) diff --git a/centers/analytics/api/app/schemas.py b/centers/analytics/api/app/schemas.py index 1896388..98d0fc9 100644 --- a/centers/analytics/api/app/schemas.py +++ b/centers/analytics/api/app/schemas.py @@ -161,15 +161,19 @@ class IngestResponse(BaseModel): class ParserProfileCreate(BaseModel): name: str = Field(min_length=1, max_length=255) + kind: str = Field(default="heuristic", pattern="^(heuristic|llm)$") sample_post: str = "" heuristic_profile: dict[str, Any] | None = None + llm_profile: dict[str, Any] | None = None status: str | None = None class ParserProfileUpdate(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=255) + kind: str | None = Field(default=None, pattern="^(heuristic|llm)$") sample_post: str | None = None heuristic_profile: dict[str, Any] | None = None + llm_profile: dict[str, Any] | None = None status: str | None = None @@ -178,8 +182,10 @@ class ParserProfileRead(BaseModel): id: int name: str + kind: str = "heuristic" sample_post: str heuristic_profile: dict[str, Any] | None + llm_profile: dict[str, Any] | None = None status: str created_at: datetime diff --git a/centers/analytics/api/app/services/job_stale.py b/centers/analytics/api/app/services/job_stale.py new file mode 100644 index 0000000..e23c8d3 --- /dev/null +++ b/centers/analytics/api/app/services/job_stale.py @@ -0,0 +1,36 @@ +"""Stale parse-job recovery helpers (running/queued left behind after worker crash).""" + +from __future__ import annotations + +import os +from datetime import datetime, timezone + +from ..models import ParseJob + +# Jobs stuck in running/queued longer than this are considered abandoned. +STALE_JOB_SECONDS = int(os.getenv("STALE_JOB_SECONDS", "900")) + + +def _aware(dt: datetime | None) -> datetime | None: + if dt is None: + return None + if dt.tzinfo is None: + return dt.replace(tzinfo=timezone.utc) + return dt + + +def job_anchor_time(job: ParseJob) -> datetime | None: + """Best available timestamp for staleness (prefer last_run_at).""" + return _aware(job.last_run_at) or _aware(getattr(job, "created_at", None)) + + +def is_stale_job(job: ParseJob, now: datetime | None = None, *, ttl: int | None = None) -> bool: + if job.status not in ("running", "queued"): + return False + now = now or datetime.now(timezone.utc) + anchor = job_anchor_time(job) + if anchor is None: + # No timestamp — treat long-lived running as stale immediately for recovery + return job.status == "running" + limit = ttl if ttl is not None else STALE_JOB_SECONDS + return (now - anchor).total_seconds() >= limit diff --git a/centers/analytics/api/app/services/jobs.py b/centers/analytics/api/app/services/jobs.py index 8aa3a19..4cf3f73 100644 --- a/centers/analytics/api/app/services/jobs.py +++ b/centers/analytics/api/app/services/jobs.py @@ -33,6 +33,25 @@ def flatten_pair_config( limit: int = 100, ) -> dict: """Expand Profile + Channel into Redis/CP source_config.""" + kind = (profile.kind or "heuristic").strip().lower() + if kind == "llm": + from contracts.llm_profile import LlmProfile + + if not profile.llm_profile: + raise ValueError("ParserProfile.llm_profile is empty") + llm = LlmProfile.model_validate(profile.llm_profile) + cfg = TelegramSourceConfig( + channel=channel.channel.strip(), + limit=limit, + extract_mode="llm", + extract_schema=llm.extract_schema, + instruction=llm.instruction, + required_fields=list(llm.required_fields), + multi_event=bool(llm.multi_event), + sample_post=profile.sample_post or None, + ) + return cfg.model_dump() + if not profile.heuristic_profile: raise ValueError("ParserProfile.heuristic_profile is empty") cfg = TelegramSourceConfig( diff --git a/centers/analytics/api/app/services/migrations.py b/centers/analytics/api/app/services/migrations.py index f64b88e..1b669c6 100644 --- a/centers/analytics/api/app/services/migrations.py +++ b/centers/analytics/api/app/services/migrations.py @@ -23,11 +23,14 @@ def migrate_schema(engine: Engine) -> None: if "channel_id" not in columns: statements.append("ALTER TABLE parse_jobs ADD COLUMN channel_id INTEGER") - # create_all handles new tables; FKs on existing DBs may need indexes - if "parse_jobs" in tables: - # Re-inspect after potential adds is not needed for FK constraints here — - # create_all + nullable FKs are enough for MVP; optional constraints below. - pass + if "parser_profiles" in tables: + columns = {col["name"] for col in inspector.get_columns("parser_profiles")} + if "kind" not in columns: + statements.append( + "ALTER TABLE parser_profiles ADD COLUMN kind VARCHAR(50) NOT NULL DEFAULT 'heuristic'" + ) + if "llm_profile" not in columns: + statements.append("ALTER TABLE parser_profiles ADD COLUMN llm_profile JSON") if not statements: return diff --git a/centers/analytics/api/app/services/parser_builder.py b/centers/analytics/api/app/services/parser_builder.py index e8b4eaa..a2b9c67 100644 --- a/centers/analytics/api/app/services/parser_builder.py +++ b/centers/analytics/api/app/services/parser_builder.py @@ -14,6 +14,7 @@ from contracts.heuristic_profile import ( HeuristicProfile, TARGET_FIELDS, apply_profile, + match_profile, target_field_specs, ) @@ -47,6 +48,14 @@ def preview_with_profile(sample_post: str, profile: dict[str, Any] | HeuristicPr return apply_profile(sample_post, profile) +def match_preview( + sample_post: str, + profile: dict[str, Any] | HeuristicProfile, +) -> tuple[dict[str, str], bool, list[str]]: + matched, fields, missing = match_profile(sample_post, profile) + return fields, matched, missing + + def empty_preview_fields(preview: dict[str, str]) -> list[str]: return [name for name, value in preview.items() if not (value or "").strip()] @@ -191,3 +200,111 @@ async def generate_profile( content = data["choices"][0]["message"]["content"] return _parse_profile_response(content) + + +async def preview_llm_extract( + sample_post: str, + llm_profile: dict[str, Any], +) -> tuple[list[dict[str, str]], dict[str, str], bool, list[str], bool, int]: + """One-shot DeepSeek extract for CA preview (does not persist). + + Returns (events, fields, matched, missing_required, is_event, matched_count). + ``fields`` is the first event (or empty) for legacy UI compatibility. + """ + from contracts.llm_profile import DEFAULT_INSTRUCTION, LlmProfile, match_llm_required + + settings = deepseek_settings() + if not settings["api_key"]: + raise RuntimeError( + "DEEPSEEK_API_KEY is not set on ca-api. Add it to .env for LLM preview." + ) + + sample = (sample_post or "").strip() + if not sample: + raise ValueError("sample_post is required") + + profile = LlmProfile.model_validate(llm_profile) + schema = profile.extract_schema + instr = profile.instruction or DEFAULT_INSTRUCTION + schema_lines = "\n".join(f"- {k}: {v}" for k, v in schema.items()) + multi = bool(profile.multi_event) + + if multi: + user_prompt = ( + f"{instr}\n\n" + "If the text describes multiple distinct events (different places, " + "coords, or dates), return one object per event in \"events\".\n" + f"Fields per event:\n{schema_lines}\n\n" + 'Return JSON: {"is_event": true|false, "events": [{: }, ...]}\n' + "If there is no event, return is_event=false and events=[].\n\n" + f"Text:\n{sample[:12000]}" + ) + else: + user_prompt = ( + f"{instr}\n\n" + f"Fields to extract:\n{schema_lines}\n\n" + 'Return JSON: {"is_event": true|false, "fields": {: }}\n\n' + f"Text:\n{sample[:12000]}" + ) + + payload = { + "model": settings["model"], + "messages": [ + { + "role": "system", + "content": ( + "You extract structured event data for a geoint map. " + "Output valid JSON only, no markdown." + ), + }, + {"role": "user", "content": user_prompt}, + ], + "temperature": 0.1, + "response_format": {"type": "json_object"}, + } + + url = f"{settings['base_url']}/chat/completions" + async with httpx.AsyncClient(timeout=90.0) as client: + response = await client.post( + url, + headers={ + "Authorization": f"Bearer {settings['api_key']}", + "Content-Type": "application/json", + }, + json=payload, + ) + response.raise_for_status() + data = response.json() + + content = data["choices"][0]["message"]["content"] + parsed = _extract_json_object(content) + is_event = bool(parsed.get("is_event", True)) + empty_fields = {key: "" for key in schema} + + if not is_event: + return [], empty_fields, False, [], False, 0 + + events: list[dict[str, str]] = [] + if multi and isinstance(parsed.get("events"), list): + for item in parsed["events"]: + if isinstance(item, dict): + events.append({key: str(item.get(key) or "").strip() for key in schema}) + else: + fields_raw = parsed.get("fields") if isinstance(parsed.get("fields"), dict) else parsed + if not isinstance(fields_raw, dict): + fields_raw = {} + events.append({key: str(fields_raw.get(key) or "").strip() for key in schema}) + + if not events: + return [], empty_fields, False, [], False, 0 + + matched_events = [ev for ev in events if match_llm_required(ev, list(profile.required_fields))] + fields = events[0] + missing = [ + name + for name in profile.required_fields + if not str(fields.get(name) or "").strip() + ] + matched = match_llm_required(fields, list(profile.required_fields)) + return events, fields, matched, missing, True, len(matched_events) + diff --git a/centers/analytics/api/app/services/scheduler.py b/centers/analytics/api/app/services/scheduler.py index cfa0c87..591b583 100644 --- a/centers/analytics/api/app/services/scheduler.py +++ b/centers/analytics/api/app/services/scheduler.py @@ -5,6 +5,7 @@ from datetime import datetime, timezone from ..database import SessionLocal from ..models import ParseJob +from .job_stale import STALE_JOB_SECONDS, is_stale_job from .jobs import enqueue_parse_job logger = logging.getLogger(__name__) @@ -13,10 +14,47 @@ TICK_SECONDS = 30 RECURRING_STATUSES = ("completed", "failed") +def recover_stale_jobs(db, now: datetime) -> None: + """Mark abandoned running/queued jobs as failed; re-queue if still active.""" + stuck = ( + db.query(ParseJob) + .filter(ParseJob.status.in_(("running", "queued"))) + .all() + ) + for job in stuck: + if not is_stale_job(job, now): + continue + prev = job.status + job.status = "failed" + job.last_error = ( + f"Stale {prev} recovered after {STALE_JOB_SECONDS}s " + "(worker likely restarted)" + ) + job.last_run_at = now + db.commit() + logger.warning("Recovered stale job %s (was %s)", job.id, prev) + + if not job.is_active or job.interval_seconds <= 0: + continue + job.status = "queued" + job.last_error = None + db.commit() + try: + enqueue_parse_job(db, job) + logger.info("Re-queued recovered job %s", job.id) + except ValueError as exc: + job.status = "failed" + job.last_error = str(exc) + db.commit() + logger.warning("Skip re-queue recovered job %s: %s", job.id, exc) + + def run_scheduler_tick() -> None: db = SessionLocal() try: now = datetime.now(timezone.utc) + recover_stale_jobs(db, now) + jobs = ( db.query(ParseJob) .filter( @@ -61,5 +99,9 @@ def start_scheduler() -> threading.Event: stop_event = threading.Event() thread = threading.Thread(target=_scheduler_loop, args=(stop_event,), daemon=True) thread.start() - logger.info("Parse job scheduler started (tick every %ss)", TICK_SECONDS) + logger.info( + "Parse job scheduler started (tick every %ss, stale after %ss)", + TICK_SECONDS, + STALE_JOB_SECONDS, + ) return stop_event diff --git a/centers/analytics/frontend/src/api/entities.ts b/centers/analytics/frontend/src/api/entities.ts index da8dc79..9109126 100644 --- a/centers/analytics/frontend/src/api/entities.ts +++ b/centers/analytics/frontend/src/api/entities.ts @@ -1,6 +1,7 @@ import { ADMIN_BASE, request } from "./client"; import type { HeuristicProfile, + LlmProfile, ParseChannel, ParseChannelCreate, ParseChannelUpdate, @@ -10,7 +11,7 @@ import type { TargetField, } from "../types/admin"; -export type { HeuristicProfile, TargetField } from "../types/admin"; +export type { HeuristicProfile, LlmProfile, TargetField } from "../types/admin"; export function fetchChannels(): Promise { return request("/parse-channels", undefined, ADMIN_BASE); @@ -64,6 +65,10 @@ export function fetchTargetFields(): Promise<{ fields: TargetField[] }> { return request<{ fields: TargetField[] }>("/parser-profiles/target-fields", undefined, ADMIN_BASE); } +export function fetchLlmDefaults(): Promise { + return request("/parser-profiles/llm-defaults", undefined, ADMIN_BASE); +} + export function generateParserProfile(payload: { sample_post: string; hint?: string; @@ -94,10 +99,35 @@ export function generateParserProfile(payload: { export function previewParserProfile( sample_post: string, heuristic_profile: HeuristicProfile | Record, -): Promise<{ fields: Record }> { - return request<{ fields: Record }>( +): Promise<{ fields: Record; matched: boolean; missing_required: string[] }> { + return request<{ fields: Record; matched: boolean; missing_required: string[] }>( "/parser-profiles/preview", { method: "POST", body: JSON.stringify({ sample_post, heuristic_profile }) }, ADMIN_BASE, ); } + +export function previewLlmProfile( + sample_post: string, + llm_profile: LlmProfile | Record, +): Promise<{ + fields: Record; + events: Record[]; + matched: boolean; + missing_required: string[]; + is_event: boolean; + matched_count: number; +}> { + return request<{ + fields: Record; + events: Record[]; + matched: boolean; + missing_required: string[]; + is_event: boolean; + matched_count: number; + }>( + "/parser-profiles/preview-llm", + { method: "POST", body: JSON.stringify({ sample_post, llm_profile }) }, + ADMIN_BASE, + ); +} diff --git a/centers/analytics/frontend/src/types/admin.ts b/centers/analytics/frontend/src/types/admin.ts index 05cc647..0b88cd6 100644 --- a/centers/analytics/frontend/src/types/admin.ts +++ b/centers/analytics/frontend/src/types/admin.ts @@ -36,26 +36,52 @@ export interface ParseJobUpdate { is_active?: boolean; } +export type TargetField = { + name: string; + type: string; + description: string; +}; + +export type HeuristicProfile = { + version: number; + fields: Record>; + required_fields?: string[]; + notes?: string; +}; + +export type LlmProfile = { + instruction?: string | null; + extract_schema: Record; + required_fields?: string[]; + multi_event?: boolean; +}; + export interface ParserProfile { id: number; name: string; + kind: "heuristic" | "llm"; sample_post: string; heuristic_profile: Record | null; + llm_profile: LlmProfile | Record | null; status: string; created_at: string; } export interface ParserProfileCreate { name: string; + kind?: "heuristic" | "llm"; sample_post?: string; heuristic_profile?: Record | null; + llm_profile?: LlmProfile | Record | null; status?: string; } export interface ParserProfileUpdate { name?: string; + kind?: "heuristic" | "llm"; sample_post?: string; heuristic_profile?: Record | null; + llm_profile?: LlmProfile | Record | null; status?: string; } @@ -82,18 +108,6 @@ export interface ParseChannelUpdate { is_active?: boolean; } -export type TargetField = { - name: string; - type: string; - description: string; -}; - -export type HeuristicProfile = { - version: number; - fields: Record>; - notes?: string; -}; - export interface EventRecord { id: number; source_type: string; diff --git a/centers/analytics/frontend/src/views/ParserProfilesView.vue b/centers/analytics/frontend/src/views/ParserProfilesView.vue index 38255bb..5899562 100644 --- a/centers/analytics/frontend/src/views/ParserProfilesView.vue +++ b/centers/analytics/frontend/src/views/ParserProfilesView.vue @@ -1,14 +1,17 @@ @@ -228,9 +469,10 @@ onMounted(() => {

Профили парсера

- Generate строит статичные правила по образцу поста. Если Preview пустой по полю — - напишите подсказку и снова Generate (агент правит текущий JSON). Можно править JSON - вручную. Дальше профиль подключается к каналу на + Два вида профилей под фиксированные поля Event: + heuristic — статичные правила (Generate один раз, runtime без LLM); + llm — instruction + schema, DeepSeek на каждый пост в batch. + Связка с каналом — на Парсеры.

@@ -246,6 +488,13 @@ onMounted(() => { Название +