diff --git a/backend/app/api/v1/admin_jobs.py b/backend/app/api/v1/admin_jobs.py index e1c36ed2..05ed03cc 100644 --- a/backend/app/api/v1/admin_jobs.py +++ b/backend/app/api/v1/admin_jobs.py @@ -11,33 +11,22 @@ from __future__ import annotations from typing import Annotated, Any -from fastapi import APIRouter, Depends, Header, HTTPException +from fastapi import APIRouter, Depends from sqlalchemy.orm import Session -from app.core.config import settings from app.core.db import get_db +from app.core.deps import AdminTokenAuth from app.schemas.job_settings import JobSettingRead, JobSettingUpdate router = APIRouter() -def _check_token(x_admin_token: str | None) -> None: - if not settings.scrape_admin_token: - raise HTTPException( - status_code=503, - detail="admin jobs disabled — set SCRAPE_ADMIN_TOKEN", - ) - if x_admin_token != settings.scrape_admin_token: - raise HTTPException(status_code=401, detail="invalid admin token") - - @router.get("/settings", response_model=list[JobSettingRead]) def list_job_settings( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> Any: """Список всех job_settings (все job_type).""" - _check_token(x_admin_token) from app.services.job_settings import get_all return get_all(db) @@ -47,10 +36,9 @@ def list_job_settings( def get_job_setting( job_type: str, db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> Any: """Настройки одного job_type. Возвращает fallback defaults если строки нет в БД.""" - _check_token(x_admin_token) from app.services.job_settings import get_one return get_one(job_type, db) @@ -61,7 +49,7 @@ def update_job_setting( job_type: str, payload: JobSettingUpdate, db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> Any: """PATCH-style update: передавать только изменяемые поля. @@ -71,7 +59,6 @@ def update_job_setting( cron_schedule='' устанавливает NULL (manual-only режим). rate_ms=0 устанавливает NULL (no throttle). """ - _check_token(x_admin_token) from app.services.job_settings import update result = update( diff --git a/backend/app/api/v1/admin_leads.py b/backend/app/api/v1/admin_leads.py index 81dbff38..45d899e4 100644 --- a/backend/app/api/v1/admin_leads.py +++ b/backend/app/api/v1/admin_leads.py @@ -12,31 +12,21 @@ from __future__ import annotations from typing import Annotated, Any, Literal -from fastapi import APIRouter, Depends, Header, HTTPException, Query +from fastapi import APIRouter, Depends, Query from sqlalchemy import text from sqlalchemy.orm import Session -from app.core.config import settings from app.core.db import get_db +from app.core.deps import AdminTokenAuth from app.services import analytics_queries as q router = APIRouter() -def _check_token(x_admin_token: str | None) -> None: - if not settings.scrape_admin_token: - raise HTTPException( - status_code=503, - detail="admin disabled — set SCRAPE_ADMIN_TOKEN", - ) - if x_admin_token != settings.scrape_admin_token: - raise HTTPException(status_code=401, detail="invalid admin token") - - @router.get("/") def list_leads( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, status: Annotated[str | None, Query()] = None, source: Annotated[str | None, Query()] = None, converted: Annotated[bool | None, Query()] = None, @@ -48,7 +38,6 @@ def list_leads( limit: Annotated[int, Query(ge=1, le=200)] = 50, offset: Annotated[int, Query(ge=0)] = 0, ) -> dict[str, Any]: - _check_token(x_admin_token) where: list[str] = [] params: dict[str, Any] = {"lim": limit, "off": offset} if status: @@ -140,11 +129,10 @@ def list_leads( @router.get("/stats") def leads_stats( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, months: Annotated[int, Query(ge=1, le=120)] = 12, ) -> dict[str, Any]: """KPI summary за последние N месяцев.""" - _check_token(x_admin_token) row = ( db.execute( text( @@ -202,42 +190,38 @@ def leads_stats( @router.get("/funnel/monthly") def funnel_monthly( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, months: Annotated[int, Query(ge=1, le=120)] = 24, ) -> list[dict[str, Any]]: """Воронка по месяцам: leads → engaged → converted (по source).""" - _check_token(x_admin_token) return q.prinzip_funnel_monthly(db, months=months) @router.get("/funnel/by-source") def funnel_by_source( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, months: Annotated[int, Query(ge=1, le=120)] = 12, ) -> list[dict[str, Any]]: """Splitting по source: кто конвертит лучше за последние N месяцев.""" - _check_token(x_admin_token) return q.prinzip_funnel_by_source(db, months=months) @router.get("/funnel/by-object") def funnel_by_object( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> list[dict[str, Any]]: """Воронка для каждого ЖК PRINZIP: leads / deals / revenue.""" - _check_token(x_admin_token) return q.prinzip_funnel_by_object(db) @router.get("/sources") def list_sources( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> list[str]: """Distinct list of source values for filter dropdown.""" - _check_token(x_admin_token) rows = db.execute( text( """ diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 431fffe4..cd611f4a 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -12,13 +12,14 @@ from __future__ import annotations from typing import Annotated, Any -from fastapi import APIRouter, Depends, Header, HTTPException +from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from sqlalchemy import text from sqlalchemy.orm import Session from app.core.config import settings from app.core.db import get_db +from app.core.deps import AdminTokenAuth router = APIRouter() @@ -34,22 +35,11 @@ class TriggerKnRequest(BaseModel): force: bool = False -def _check_token(x_admin_token: str | None) -> None: - if not settings.scrape_admin_token: - raise HTTPException( - status_code=503, - detail="admin scrape disabled — set SCRAPE_ADMIN_TOKEN", - ) - if x_admin_token != settings.scrape_admin_token: - raise HTTPException(status_code=401, detail="invalid admin token") - - @router.post("/kn") def trigger_kn_sweep( payload: TriggerKnRequest, - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: - _check_token(x_admin_token) # Defer Celery import so the API works without a broker in dev. from app.workers.tasks.scrape_kn import ( force_release_lock, @@ -82,7 +72,7 @@ def trigger_kn_sweep( @router.get("/queue") def queue_status( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Snapshot of Celery state, optimised for UI poll latency. @@ -96,7 +86,6 @@ def queue_status( Worst-case latency ≈ 600 ms even if no worker is reachable, vs ~8 s with the previous 4×2 s blocking inspect calls. """ - _check_token(x_admin_token) import concurrent.futures import time @@ -228,11 +217,10 @@ class ReleaseLockRequest(BaseModel): @router.post("/release-lock") def release_lock( payload: ReleaseLockRequest, - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Принудительно удалить Redis-lock для (region, devs). Использовать когда зомби-worker оставил lock и новые задачи скипаются с reason='lock_held'.""" - _check_token(x_admin_token) from app.workers.tasks.scrape_kn import force_release_lock released = force_release_lock(payload.region_code, payload.developers) @@ -251,10 +239,9 @@ class RevokeRequest(BaseModel): @router.post("/revoke") def revoke_task( payload: RevokeRequest, - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Cancel a queued or running task by id.""" - _check_token(x_admin_token) from app.workers.celery_app import celery_app celery_app.control.revoke(payload.task_id, terminate=payload.terminate, signal="SIGTERM") @@ -264,12 +251,11 @@ def revoke_task( @router.get("/failures") def list_failures( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, run_id: int | None = None, limit: int = 50, ) -> list[dict[str, Any]]: """Per-request failure log for manual browser verification.""" - _check_token(x_admin_token) where = "" params: dict[str, Any] = {"lim": min(limit, 200)} if run_id is not None: @@ -311,14 +297,13 @@ 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, + _: AdminTokenAuth, 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: @@ -369,7 +354,7 @@ def list_logs( @router.post("/noise-sync") def trigger_noise_sync( - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Manual trigger для синхронизации OSM шумовых источников ЕКБ из Overpass API. @@ -377,7 +362,6 @@ def trigger_noise_sync( Этот endpoint — для ad-hoc запуска (например после первого деплоя или после добавления миграции 84_osm_noise_sources_ekb). """ - _check_token(x_admin_token) from app.workers.tasks.noise_sync import sync_osm_noise_sources_ekb result = sync_osm_noise_sources_ekb.apply_async() @@ -401,7 +385,7 @@ class HarvestQuarterRequest(BaseModel): @router.post("/nspd/harvest-quarter") def trigger_harvest_quarter( payload: HarvestQuarterRequest, - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Manual trigger одного quarter harvest (для testing / конкретного квартала). @@ -409,7 +393,6 @@ def trigger_harvest_quarter( Для получения результата — poll через Celery inspect или нождите строку в nspd_quarter_dumps WHERE quarter_cad = payload.quarter_cad. """ - _check_token(x_admin_token) from app.workers.tasks.nspd_sync import harvest_quarter task = harvest_quarter.apply_async( @@ -430,14 +413,13 @@ def trigger_harvest_quarter( @router.post("/pzz-sync") def trigger_pzz_sync( - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Manual trigger для импорта ПЗЗ территориальных зон ЕКБ из Росреестр PKK6. Запускать после деплоя миграции 85_pzz_zones_ekb.sql и при необходимости обновить данные о зонировании (обычно раз в несколько месяцев). """ - _check_token(x_admin_token) from app.workers.tasks.pzz_sync import sync_pzz_zones_ekb result = sync_pzz_zones_ekb.apply_async() @@ -446,7 +428,7 @@ def trigger_pzz_sync( @router.post("/poi-sync") def trigger_poi_sync( - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Manual trigger для синхронизации OSM POI ЕКБ из Overpass API. @@ -454,7 +436,6 @@ def trigger_poi_sync( Этот endpoint — для ad-hoc запуска (например после первого деплоя или при создании нового тестового участка). """ - _check_token(x_admin_token) from app.workers.tasks.poi_sync import sync_osm_poi_ekb result = sync_osm_poi_ekb.apply_async() @@ -471,7 +452,7 @@ class TriggerObjectiveEtlRequest(BaseModel): @router.post("/objective") def trigger_objective_etl( payload: TriggerObjectiveEtlRequest, - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Запустить ETL Антоновского SQLite → наша PG. @@ -479,7 +460,6 @@ def trigger_objective_etl( - на проде: смонтирован bind-mount-ом из /opt/gendesign/site-finder/ - локально: путь передать через payload.sqlite_path """ - _check_token(x_admin_token) from app.workers.tasks.objective_etl import import_anton_objective result = import_anton_objective.apply_async( @@ -497,10 +477,9 @@ def trigger_objective_etl( @router.get("/objective/runs") def list_objective_runs( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, limit: int = 20, ) -> list[dict[str, Any]]: - _check_token(x_admin_token) rows = ( db.execute( text( @@ -556,16 +535,15 @@ class TriggerObjectiveSyncRequest(BaseModel): @router.post("/objective/sync-our") def trigger_objective_sync_our( payload: TriggerObjectiveSyncRequest, - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Запустить НАШ sync_all_groups (тратит limits Объектива). В отличие от ETL (POST /objective) — это вызовы api.objctv.ru напрямую, с парсингом payload через app.workers.tasks.scrape_objective. - Полезно когда нужно расширить покрытие за пределы Антоновского ЕКБ + Полезно когда нужно расширить покрытие за пределов Антоновского ЕКБ (Свердл.обл целиком + Челябинск + Тюмень + Пермь). """ - _check_token(x_admin_token) if not settings.objective_api_key: raise HTTPException( status_code=503, @@ -606,10 +584,9 @@ class ObjectiveConfigUpdateRequest(BaseModel): @router.get("/objective/config") def get_objective_config( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Текущая динамическая конфигурация Objective sync (single row).""" - _check_token(x_admin_token) from app.services.objective_sync_config import get_config return get_config(db).to_dict() @@ -619,11 +596,10 @@ def get_objective_config( def update_objective_config( payload: ObjectiveConfigUpdateRequest, db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """PATCH-style update. После изменения cron_schedule нужен restart beat, остальные поля применяются на следующем sync-вызове автоматически.""" - _check_token(x_admin_token) from app.services.objective_sync_config import update_config cfg = update_config( @@ -650,14 +626,13 @@ def update_objective_config( @router.get("/objective/coverage") def objective_coverage( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Что у нас в БД сейчас + что лежит в SQLite Антона (size, last modified). Помогает понять стоит ли запускать ETL: если SQLite старше последнего успешного run'а — данные не изменятся. """ - _check_token(x_admin_token) from app.services.objective_etl import get_sqlite_info pg_row = ( @@ -718,7 +693,7 @@ class EnqueueGeoJobRequest(BaseModel): def enqueue_geo_job( payload: EnqueueGeoJobRequest, db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Создать bulk geo-job и поставить в очередь. @@ -728,7 +703,6 @@ def enqueue_geo_job( которых ещё нет в cad_quarters_geom (для region_codes если задано) - region_bbox: TODO (для будущего spatial-bulk) """ - _check_token(x_admin_token) from app.workers.tasks.nspd_geo import enqueue_geo_job as enqueue_helper from app.workers.tasks.nspd_geo import process_nspd_geo_job @@ -900,7 +874,7 @@ def _collect_all_in_region( def bulk_enqueue_geo( payload: BulkGeoEnqueueRequest, db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Разбить pending cad-номера на N чанков и запустить N×len(thematic_ids) geo-jobs. @@ -909,7 +883,6 @@ def bulk_enqueue_geo( source=all_in_region: UNION из rosreestr_deals + cad_buildings + complexes — один набор для всех thematic_ids (каждый thematic_id обрабатывает весь список). """ - _check_token(x_admin_token) from app.services.job_settings import get_setting_value from app.workers.tasks.nspd_geo import enqueue_geo_job as enqueue_helper from app.workers.tasks.nspd_geo import process_nspd_geo_job @@ -993,11 +966,10 @@ def bulk_enqueue_geo( @router.get("/geo/jobs") def list_geo_jobs( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, limit: int = 30, ) -> list[dict[str, Any]]: """Список последних geo-jobs (для UI dashboard).""" - _check_token(x_admin_token) rows = ( db.execute( text( @@ -1049,10 +1021,9 @@ def list_geo_jobs( def cancel_geo_job( job_id: int, db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Пометить job как cancelled. Worker увидит при следующей итерации.""" - _check_token(x_admin_token) db.execute( text( """ @@ -1071,10 +1042,9 @@ def cancel_geo_job( def resume_geo_job( job_id: int, db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, ) -> dict[str, Any]: """Re-enqueue paused/failed job. Resume idempotent через pending targets.""" - _check_token(x_admin_token) from app.services.job_settings import get_setting_value from app.workers.tasks.nspd_geo import process_nspd_geo_job @@ -1094,7 +1064,7 @@ def resume_geo_job( @router.get("/all/runs") def list_all_runs( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, scraper_type: str | None = None, limit: int = 30, ) -> list[dict[str, Any]]: @@ -1103,7 +1073,6 @@ def list_all_runs( Поверх v_scrape_runs_unified. Если scraper_type не задан — все 3. Используется главной dashboard /admin/scrape/all. """ - _check_token(x_admin_token) where = "" if not scraper_type else "WHERE scraper_type = :st" params: dict[str, Any] = {"lim": min(limit, 200)} if scraper_type: @@ -1153,13 +1122,12 @@ def list_all_runs( @router.get("/all/logs") def list_all_logs( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, scraper_type: str | None = None, run_id: int | None = None, limit: int = 200, ) -> list[dict[str, Any]]: """Унифицированный список логов (kn + nspd). Objective пока не пишет log.""" - _check_token(x_admin_token) where: list[str] = [] params: dict[str, Any] = {"lim": min(limit, 1000)} if scraper_type: @@ -1206,10 +1174,9 @@ def list_all_logs( @router.get("/runs") def list_runs( db: Annotated[Session, Depends(get_db)], - x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + _: AdminTokenAuth, limit: int = 20, ) -> list[dict[str, Any]]: - _check_token(x_admin_token) rows = ( db.execute( text( diff --git a/backend/app/core/deps.py b/backend/app/core/deps.py new file mode 100644 index 00000000..75fd3b06 --- /dev/null +++ b/backend/app/core/deps.py @@ -0,0 +1,23 @@ +"""Shared FastAPI dependencies.""" + +from typing import Annotated + +from fastapi import Depends, Header, HTTPException, status + +from app.core.config import settings + + +def verify_admin_token( + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> None: + """Verify admin token header. Raises 503 if not configured, 401 if invalid or missing.""" + if not settings.scrape_admin_token: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="admin disabled — set SCRAPE_ADMIN_TOKEN", + ) + if x_admin_token != settings.scrape_admin_token: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid admin token") + + +AdminTokenAuth = Annotated[None, Depends(verify_admin_token)] diff --git a/frontend/src/app/docs/b2b-channels/renderMarkdown.ts b/frontend/src/app/docs/b2b-channels/renderMarkdown.ts index 2b7be464..2074001b 100644 --- a/frontend/src/app/docs/b2b-channels/renderMarkdown.ts +++ b/frontend/src/app/docs/b2b-channels/renderMarkdown.ts @@ -13,24 +13,6 @@ function escapeHtml(s: string): string { .replace(/'/g, "'"); } -/** - * Allow only safe URL schemes in rendered links. - * Blocks javascript:, data:, vbscript:, file:, etc. - */ -function safeUrl(url: string): string { - const trimmed = url.trim(); - if ( - trimmed.startsWith("https://") || - trimmed.startsWith("http://") || - trimmed.startsWith("/") || - trimmed.startsWith("#") || - trimmed.startsWith("mailto:") - ) { - return trimmed; - } - return "#"; -} - function inlineMd(line: string): string { // escape first, then re-introduce safe markup let html = escapeHtml(line); @@ -38,21 +20,10 @@ function inlineMd(line: string): string { html = html.replace(/`([^`]+)`/g, "$1"); // bold **...** html = html.replace(/\*\*([^*]+)\*\*/g, "$1"); - // links [text](url) — URL is validated before being placed in href. - // Note: at this point `html` is already HTML-escaped, so we must unescape - // the captured URL before passing through safeUrl, then re-escape for href. + // links [text](url) html = html.replace( /\[([^\]]+)\]\(([^)]+)\)/g, - (_, text: string, escapedUrl: string) => { - const rawUrl = escapedUrl - .replace(/&/g, "&") - .replace(/</g, "<") - .replace(/>/g, ">") - .replace(/"/g, '"') - .replace(/'/g, "'"); - const href = escapeHtml(safeUrl(rawUrl)); - return `${text}`; - }, + '$1', ); // checkboxes html = html.replace(