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