From f5c94cf12d5b2f9aaae2d1947eeb303dde82e3c7 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 16:48:02 +0800 Subject: [PATCH] =?UTF-8?q?W3-a=20=E5=8F=98=E5=8C=96=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E7=9A=84=E6=8E=A5=E5=8F=A3=E4=B8=8E=E5=A4=84=E7=90=86=E3=80=81?= =?UTF-8?q?=E5=AE=9A=E6=97=B6=E7=BB=B4=E6=8A=A4=EF=BC=88next-round=2003=20?= =?UTF-8?q?B3=E3=80=81B4=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 变化没有统一的记录：进不了审阅，发不了通知，口径账本的召回也没有可以订阅的接口；目录只能手动探查、手动剖析。现在： - api/source_changes.py（骨架已在 datasources 通配之前登记，搭建者可用，访问密钥不放）：GET /api/datasources/{id}/changes （缺省开着的和知道了的，按「使用次数权重 × 严重程度」排，带各级别计数）、GET /changes/{change_id}、POST /changes/{change_id}/resolve。处理动作落到目录已有的写入接口：清理目录里说到被删的列的项（整份写、乐观锁）、加入码值 （apply_patch 配 _merge_codes，含义待填写，依据写变化编号）、重新起草这张表（模型用不了时只按表结构起草）、重新剖析、 知道了、误报（必须写理由）；每个动作写 resolution，动了目录的记目录版本，建了任务的记任务编号。 - data/source_changes.py 给召回：list_changes（按 seq 游标拉）、affecting（按源和表反查）、resolve、triggers_recall（只认 结构类、处理时改了目录的、触及有效记录条件的新码值）。 - 通知 source.changed 的发送函数 notify/service.notify_source_changed：只推还要人看的「要处理」级变化，按源汇总，消息里 只有源名、各级别条数和链接（D9）；探查、剖析任务结束时有新的要处理变化就调。这一行的订阅范围第 4 波有了界面才打开， 现在推不出去（照实）。 - 定时维护（data/source_maintenance.py，X4：就是一条 target_kind=source_maintenance 的定时，不另建表）：fire 按顺序建 探查任务 → 剖析任务（范围 needed / observe / maintenance 按选的步骤），任务上记 trigger=schedule、fire_at、schedule_id； 剖析等探查进了同源队才建，靠 SourceLanes 串行；剖析没开、数据源停用 / 不存在 / 是上传表格时照实记没执行，同源还有 探查或剖析在做时记跳过；触发记录的 inputs 写明建了哪几个任务、哪几步没执行。check_target 核对目标。维护步骤存在 options.catalog_maintenance，经 GET / PUT /api/datasources/{id}/maintenance 改，数据源表单不显示它（api/datasources.py 的 _MANAGED_OPTIONS 加了这一个键）。 测试：test_source_changes（列表与计数、误报要理由、处理过的 409、清理目录与加入码值记目录版本、剖析没开照实 422、 去重、游标拉取与反查、召回触发、通知只有源名条数链接）、test_source_maintenance（按顺序串行、trigger 与 fire_at、 剖析没开记未执行、停用 / 不存在 / 忙、调度循环只补最近一次且积压只记一条、步骤接口与核对目标）。 --- backend/app/api/datasources.py | 5 +- backend/app/api/source_changes.py | 122 +++++++++++- backend/app/data/catalog_profile_jobs.py | 11 ++ backend/app/data/introspect_jobs.py | 11 ++ backend/app/data/source_changes.py | 24 ++- backend/app/data/source_maintenance.py | 235 +++++++++++++++++++++- backend/app/notify/service.py | 30 +++ backend/tests/test_source_changes.py | 239 +++++++++++++++++++++++ backend/tests/test_source_maintenance.py | 210 ++++++++++++++++++++ 9 files changed, 874 insertions(+), 13 deletions(-) create mode 100644 backend/tests/test_source_changes.py create mode 100644 backend/tests/test_source_maintenance.py diff --git a/backend/app/api/datasources.py b/backend/app/api/datasources.py index e386e8fc..b1450b87 100644 --- a/backend/app/api/datasources.py +++ b/backend/app/api/datasources.py @@ -140,8 +140,9 @@ def _import_mode(row: DataSource) -> str | None: #: options 里由别的接口维护的项：数据源表单不显示、不提交它们，整份回写 options 时原样留着（同模型接入 extra 里的 #: last_check）。数据目录的起草要求只经 PUT /catalog/guidance 改（api/catalog.py）：表单把认不得的项放进「高级参数」 #: 的单行输入框，多行的要求会被压成一行 -#: 探查范围（introspect_scope）同理：只经 PUT /introspect-scope 改，表单整份回写 options 时原样留着 -_MANAGED_OPTIONS = (CATALOG_DRAFT_GUIDANCE_OPTION, source_objects.SCOPE_OPTION) +#: 探查范围（introspect_scope）同理：只经 PUT /introspect-scope 改，表单整份回写 options 时原样留着；定时维护的步骤 +#: （catalog_maintenance）只经 PUT /maintenance 改（api/source_changes.py） +_MANAGED_OPTIONS = (CATALOG_DRAFT_GUIDANCE_OPTION, source_objects.SCOPE_OPTION, "catalog_maintenance") def _shown_options(options: dict[str, Any] | None) -> dict[str, Any]: diff --git a/backend/app/api/source_changes.py b/backend/app/api/source_changes.py index e41f05c1..cc95b76b 100644 --- a/backend/app/api/source_changes.py +++ b/backend/app/api/source_changes.py @@ -1,9 +1,125 @@ -"""数据源的变化记录（docs/design/next-round/03-数据接入与目录维护.md B3）：结构变化、取值变化，查看与处理。 +"""数据源的变化记录（docs/design/next-round/03-数据接入与目录维护.md B3）：结构变化、取值变化，查看与处理；以及同一前缀下 +数据源定时维护要做哪几步（B4）。 -挂在 /api/datasources/{id}/changes 下，要在 datasources 的通配 /{source_id} 之前登记（main.py）。第 3 波 W3-a 填。 +挂在 /api/datasources/{id}/… 下，要在 datasources 的通配 /{source_id} 之前登记（main.py）： + +- GET /{id}/changes 变化列表（缺省开着的和知道了的；status=all 全部）：按「使用次数权重 × 严重程度」排， + 同分新的在前；带各级别的计数（数据源卡片的角标「N 个变化要处理」按它，第 4 波） +- GET /{id}/changes/{change_id} 一条变化 +- POST /{id}/changes/{change_id}/resolve 处理：{action, note?, codes?}。动作落到目录已有的写入接口并记目录版本 + （data/source_changes.handle）；误报必须写理由（422）；已经处理过的 409 +- GET /{id}/maintenance 定时维护要做哪几步、剖析开没开、指向这个源的维护定时 +- PUT /{id}/maintenance 改要做哪几步：{steps: [introspect | profile_needed | observe]}。定时本身（cron、时区、开关） + 在 /api/schedules 里建（target_kind=source_maintenance），这里只管步骤 + +访问密钥默认不放（auth/keys.py 的 KEY_ROUTES 没有登记这几条）。界面（审阅页「变化」标签页、数据源角标、定时维护设置）在 +第 4 波。 """ from __future__ import annotations -from fastapi import APIRouter +from typing import Any + +from fastapi import APIRouter, Depends, Header, HTTPException, Query +from pydantic import BaseModel, Field +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.deps import actor_of +from app.data import catalog, source_changes as sc, source_maintenance +from app.db.base import get_session +from app.db.models import DataSource router = APIRouter(prefix="/api/datasources", tags=["source-changes"]) + +#: 一页最多几条 +LIMIT_MAX = 200 + + +async def _source(session: AsyncSession, source_id: str) -> DataSource: + row = await session.get(DataSource, source_id) + if row is None: + raise HTTPException(404, "数据源不存在，可能已被删除") + return row + + +@router.get("/{source_id}/changes") +async def list_changes(source_id: str, status: str = Query("pending"), kind: list[str] | None = Query(None), + table: str | None = Query(None), limit: int = Query(100, ge=1, le=LIMIT_MAX), + offset: int = Query(0, ge=0), session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """{items: [变化（agentlab-change/1）], total, counts: {pending, open, by_severity}}。""" + if status not in ("pending", "all", *sc.STATUSES): + raise HTTPException(422, f"不认识的状态：{status}。可选：pending、all、{'、'.join(sc.STATUSES)}") + if kind and (bad := [k for k in kind if k not in sc.KINDS]): + raise HTTPException(422, f"不认识的变化种类：{'、'.join(bad)}") + row = await _source(session, source_id) + rows, total = await sc.list_for_source(session, row, status=status, kinds=kind, table=table, limit=limit, + offset=offset) + return {"items": [sc.change_out(r) for r in rows], "total": total, "counts": await sc.counts(session, row.id)} + + +@router.get("/{source_id}/changes/{change_id}") +async def get_change(source_id: str, change_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + row = await _source(session, source_id) + found = await sc.load(session, row.id, change_id) + if found is None: + raise HTTPException(404, "这条变化不存在，可能已随数据源删除") + return {"change": sc.change_out(found)} + + +class ResolveIn(BaseModel): + action: str + #: 理由：误报必须写 + note: str | None = Field(default=None, max_length=sc.NOTE_MAX) + #: 加入码值时的 {码值: 含义}（含义可以空）；不给就加这条变化测到的新码值 + codes: dict[str, str] | None = None + + +@router.post("/{source_id}/changes/{change_id}/resolve") +async def resolve_change(source_id: str, change_id: str, payload: ResolveIn, x_actor: str | None = Header(default=None), + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """处理一条变化：clean_catalog 清理目录里说到这些列的项、add_codes 加入码值、redraft 重新起草这张表、reprofile 重新 + 剖析、acknowledge 知道了、dismiss 误报（必须写理由）。返回 {change, catalog_version?, job?}。目录被别人刚改过 409。""" + row = await _source(session, source_id) + found = await sc.load(session, row.id, change_id) + if found is None: + raise HTTPException(404, "这条变化不存在，可能已随数据源删除") + try: + return await sc.handle(session, row, found, payload.action, actor=actor_of(x_actor), note=payload.note, + codes=payload.codes) + except sc.ChangeInvalid as e: + await session.rollback() + raise HTTPException(422, str(e)) from e + except sc.ChangeFinished as e: + await session.rollback() + raise HTTPException(409, str(e)) from e + except catalog.CatalogConflict as e: + await session.rollback() + raise HTTPException(409, "这张表的数据目录刚被别人改过，请刷新后再处理") from e + + +@router.get("/{source_id}/maintenance") +async def get_maintenance(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """{steps, profile_enabled, schedules}：要做哪几步、剖析开没开（没开时剖析、观察两步到点照实记没执行）、指向这个源的 + 维护定时。""" + from app.data import catalog_profile + + row = await _source(session, source_id) + return {"steps": source_maintenance.steps_of(row.options), + "profile_enabled": catalog_profile.profile_settings(row.options).enabled, + "schedules": await source_maintenance.schedules_for(session, row.id)} + + +class MaintenanceIn(BaseModel): + steps: list[Any] + + +@router.put("/{source_id}/maintenance") +async def put_maintenance(source_id: str, payload: MaintenanceIn, x_actor: str | None = Header(default=None), + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + row = await _source(session, source_id) + if (row.origin or "manual") in ("upload", "records"): + raise HTTPException(409, source_maintenance.NOT_MAINTAINED) + if problem := source_maintenance.steps_problem(payload.steps): + raise HTTPException(422, problem) + row.options = source_maintenance.with_steps(row.options, [str(s) for s in payload.steps], actor=actor_of(x_actor)) + await session.commit() + return await get_maintenance(source_id, session) diff --git a/backend/app/data/catalog_profile_jobs.py b/backend/app/data/catalog_profile_jobs.py index 4f31887d..df7cf3df 100644 --- a/backend/app/data/catalog_profile_jobs.py +++ b/backend/app/data/catalog_profile_jobs.py @@ -904,6 +904,16 @@ class _ProfileRun(runtime.JobRun): await self._write(_close, self.job_id, job_status, job_error, task_status, task_error) except Exception: # noqa: BLE001 logger.exception("数据剖析任务 %s 收尾时写库出错", self.job_id) + await self._notify() + + async def _notify(self) -> None: + """这次取值观察新记了「要处理」的变化（撤销了已列全）：按源推一条 source.changed。""" + from app.data import source_changes + + async with SessionLocal() as session: + job = await session.get(CatalogProfileJob, self.job_id) + if job is not None and job.scope in OBSERVING_SCOPES: + await source_changes.notify_after_job(job.source_id, self.job_id) def stop_text(self) -> str: return STOPPED_ERROR if self.stop_reason == "user" else INTERRUPTED_ERROR @@ -966,6 +976,7 @@ class _ProfileRun(runtime.JobRun): break await self._settle() await self._write(self._finish) + await self._notify() # ---------------------------------------------------------------- 进度回调（引擎调） diff --git a/backend/app/data/introspect_jobs.py b/backend/app/data/introspect_jobs.py index 137f2b6a..d1f0b993 100644 --- a/backend/app/data/introspect_jobs.py +++ b/backend/app/data/introspect_jobs.py @@ -572,6 +572,16 @@ class _IntrospectRun(runtime.JobRun): await self._write(_close, self.job_id, job_status, job_error, task_error) except Exception: # noqa: BLE001 logger.exception("探查任务 %s 收尾时写库出错", self.job_id) + await self._notify() + + async def _notify(self) -> None: + """这次新记了「要处理」的结构变化：按源推一条 source.changed（data/source_changes.notify_after_job）。""" + from app.data import source_changes + + async with SessionLocal() as session: + job = await session.get(IntrospectJob, self.job_id) + if job is not None: + await source_changes.notify_after_job(job.source_id, self.job_id) async def _run(self) -> None: if not await self._write(self._claim): @@ -600,6 +610,7 @@ class _IntrospectRun(runtime.JobRun): await self._write(self._save_batch, source_id, schema, batch, got) await self._settle() await self._write(self._finish, source_id) + await self._notify() # ---------------------------------------------------------------- 库操作 diff --git a/backend/app/data/source_changes.py b/backend/app/data/source_changes.py index 3e43d61c..cb9de4f5 100644 --- a/backend/app/data/source_changes.py +++ b/backend/app/data/source_changes.py @@ -325,6 +325,27 @@ def triggers_recall(change: Mapping[str, Any]) -> bool: return kind == CODES_NEW and bool((change.get("affects") or {}).get("rules_touched")) +async def notify_after_job(source_id: str, job_id: str) -> int: + """一次探查、剖析任务结束时：这次新记了还开着的「要处理」级变化，就在后台按源推一条 source.changed（不等它推完）。返回 + 这次新记了几条要处理的。出错只记日志：通知是附加的。""" + from app.db.base import SessionLocal + + try: + async with SessionLocal() as session: + n = int((await session.execute(select(func.count()).select_from(SourceChange).where( + SourceChange.source_id == source_id, SourceChange.severity == "action", + SourceChange.status == "open", + func.json_extract(SourceChange.detected_by, "$.job_id") == job_id))).scalar() or 0) + if n: + from app.notify import service + + service.spawn(service.notify_source_changed(source_id)) + return n + except Exception: # noqa: BLE001 + logger.exception("任务 %s 结束后推送数据源变化的通知时出错", job_id) + return 0 + + # ========================================================================== # 处理 # ========================================================================== @@ -487,6 +508,7 @@ __all__ = [ "ACTIONS", "ACTION_STATUS", "CODES_NEW", "CODES_SHIFT", "COVERAGE_DROP", "ChangeFinished", "ChangeInvalid", "ChangeNotFound", "DATE_GAP", "DATE_STALLED", "Detected", "KINDS", "KIND_LABEL", "NewChange", "PENDING", "ROWS_DROP", "ROWS_JUMP", "SCHEMA", "SEVERITIES", "SEVERITY_RANK", "STATUSES", "VALUE_KINDS", "VOLUME_DROP", "affecting", - "change_out", "counts", "fingerprint", "handle", "list_changes", "list_for_source", "load", "record", "record_all", + "change_out", "counts", "fingerprint", "handle", "list_changes", "list_for_source", "load", "notify_after_job", + "record", "record_all", "resolve", "summary", "triggers_recall", "without_paths", ] diff --git a/backend/app/data/source_maintenance.py b/backend/app/data/source_maintenance.py index 24efde1a..a2bba31e 100644 --- a/backend/app/data/source_maintenance.py +++ b/backend/app/data/source_maintenance.py @@ -1,21 +1,242 @@ -"""数据源的定时维护：探查 → 剖析需要的表 → 取值观察（docs/design/next-round/03-数据接入与目录维护.md B4）。 +"""数据源的定时维护：探查 → 剖析需要的表 → 取值观察（docs/design/next-round/03-数据接入与目录维护.md B4，目录-D5）。 -scheduler/service.py 按目标种类导入本模块（_TARGET_MODULES）：fire 到点发起一次，check_target 建任务时核对目标。 -第 3 波 W3-a 填；填之前照实记「当前版本还不能按时发起这类目标」，不发起。 +**为什么要有它。** 以前只能手动探查、手动剖析，目录只在点按钮那一刻是对的；结构变化、取值漂移要靠定期观察才看得到。 + +**定时只有一套**（总方案 X4）：维护就是一条 Schedule，target_kind=source_maintenance、target_id=数据源 id。认领、错过的只补 +最近一次、积压只记一条、重叠跳过、触发记录、通知都是 scheduler 那一套（scheduler/service.py 按目标种类导入本模块：fire 到点 +发起一次、check_target 建任务时核对目标），不另建表、不另写 cron。按数据源开、缺省关（没有这条定时就不维护），只在设定的 +时段跑（cron 自己定）。 + +**做哪几步**存在数据源的 options.catalog_maintenance（{steps: [...], updated_by, updated_at}，由 PUT +/api/datasources/{id}/maintenance 维护，数据源表单不显示、不提交它）：introspect 探查结构、profile_needed 剖析需要的表、 +observe 取值观察，可多选，没设过是三步都做。 + +**到点**（fire，持着 scheduler 的 launch_lock）：按顺序建探查任务 → 剖析任务（范围 needed / observe / maintenance，按选的 +步骤），任务上记 trigger=schedule 和 fire_at，靠 SourceLanes 串行：剖析排在同一次触发的探查后面（等探查进了同源队才建剖析 +任务），开始时按那时的表结构和目录挑表（挑出 0 张也照建，任务以 0 张表结束、提醒里写明为什么）。预算、止损照剖析设置。 +照实记没执行的：数据源不存在、已停用、是上传表格或记录表时整次不执行（failed_to_start）；剖析没开时剖析、观察两步不执行、 +探查照做；上一次的维护（或别的探查、剖析）还没做完时这次跳过（skipped_overlap）。每次触发记下建了哪几个任务、哪几步没执行、 +为什么（触发记录的 inputs：{steps, jobs, skipped}）。 + +AGENTLAB_SCHEDULER 关着时调度循环不起，到点不会触发（拷来的数据目录不替原库查业务库）；「立即运行」是人点的，照常执行。 """ from __future__ import annotations +import asyncio +import logging +from collections.abc import Mapping from datetime import datetime from typing import Any +from sqlalchemy import select, update + +from app.db.base import SessionLocal, utcnow + +logger = logging.getLogger(__name__) + +#: options 里维护设置的键 +MAINTENANCE_OPTION = "catalog_maintenance" +STEPS = ("introspect", "profile_needed", "observe") +STEP_LABEL = {"introspect": "探查结构", "profile_needed": "剖析需要的表", "observe": "取值观察"} +DEFAULT_STEPS = STEPS +#: 等同一次触发的探查任务进同源队最多等几秒（进了队，剖析任务才排得到它后面） +LANE_WAIT_S = 5.0 + +#: 先记下这次触发、还没建完任务时的说法：进程在这中间没了，留下的记录照实说没有执行 +INTERRUPTED = "服务在建维护任务前停止，这次维护没有执行" +SOURCE_GONE = "数据源不存在，可能已被删除" +SOURCE_DISABLED = "数据源「{name}」已停用，这次维护没有执行" +NOT_MAINTAINED = "上传的表格、记录表的结构随版本或表单走，不需要定时维护" +NO_STEPS = "没有选要做的维护步骤（探查结构、剖析需要的表、取值观察至少选一步）" +BUSY = "数据源「{name}」上一次的维护（或别的探查、剖析）还没做完：{what}，这次触发已跳过" +NOT_RUN = "{step}没有执行：{why}" + + +def steps_of(options: Mapping[str, Any] | None) -> list[str]: + """这个源定时维护要做哪几步（按 STEPS 的顺序）；没设过是三步都做。""" + raw = (options or {}).get(MAINTENANCE_OPTION) + steps = raw.get("steps") if isinstance(raw, Mapping) else None + if not isinstance(steps, list): + return list(DEFAULT_STEPS) + picked = {s for s in steps if s in STEPS} + return [s for s in STEPS if s in picked] + + +def steps_problem(raw: Any) -> str | None: + """PUT /maintenance 交来的步骤合不合规：不合规返回一句中文。""" + if not isinstance(raw, list) or not all(isinstance(s, str) for s in raw): + return "维护步骤要是一组文字：introspect（探查结构）、profile_needed（剖析需要的表）、observe（取值观察）" + unknown = [s for s in raw if s not in STEPS] + if unknown: + return f"不认识的维护步骤：{'、'.join(unknown)}。可选：{'、'.join(f'{k}（{v}）' for k, v in STEP_LABEL.items())}" + if not raw: + return NO_STEPS + return None + + +def with_steps(options: Mapping[str, Any] | None, steps: list[str], *, actor: str | None) -> dict[str, Any]: + """options 换上新的维护步骤（不改别的项）。""" + ordered = [s for s in STEPS if s in set(steps)] + return {**dict(options or {}), + MAINTENANCE_OPTION: {"steps": ordered, "updated_by": actor, "updated_at": utcnow().isoformat()}} + + +def _origin_problem(row: Any) -> str | None: + return NOT_MAINTAINED if (getattr(row, "origin", None) or "manual") in ("upload", "records") else None + + +async def check_target(session: Any, target_id: str) -> str | None: + """建定时任务、改目标时核对：数据源要在、不是上传表格或记录表、选了至少一步。停用了、没开剖析不拦（到点照实记没执行）。""" + from app.db.models import DataSource + + row = await session.get(DataSource, target_id) + if row is None: + return SOURCE_GONE + if problem := _origin_problem(row): + return problem + if not steps_of(row.options): + return NO_STEPS + return None + + +async def _busy(session: Any, source_id: str) -> str | None: + """同源有排队或进行中的探查、剖析任务：说明是哪一种。""" + from app.data import catalog_profile_jobs, introspect_jobs + + if await introspect_jobs.active_job(session, source_id) is not None: + return "正在探查结构" + if await catalog_profile_jobs.active_job(session, source_id) is not None: + return "正在剖析" + return None + + +async def _entered(source_id: str, job_id: str) -> None: + """等这个探查任务进了同源队（或者已经不在跑了）：剖析任务要排在它后面。最多等 LANE_WAIT_S 秒。""" + from app.core import job_runtime as runtime + from app.data import introspect_jobs + + loop = asyncio.get_running_loop() + deadline = loop.time() + LANE_WAIT_S + while loop.time() < deadline: + if runtime.lanes.ahead(source_id, job_id) is not None or not introspect_jobs.manager.running(job_id): + return + await asyncio.sleep(0.01) + logger.warning("定时维护：探查任务 %s 过了 %.0f 秒还没进同源队，剖析任务照建", job_id, LANE_WAIT_S) + + +async def _settle_fire(fire_id: str, status: str, reason: str | None, inputs: dict[str, Any]) -> Any: + from app.db.models import ScheduleFire + + async with SessionLocal() as session: + await session.execute(update(ScheduleFire).where(ScheduleFire.id == fire_id) + .values(status=status, reason=reason, inputs=inputs)) + await session.commit() + return await session.get(ScheduleFire, fire_id) + async def fire(*, schedule_id: str, fire_at: datetime, kind: str, actor: str | None, target_id: str, name: str, tz: str, cron_expr: str) -> Any: + """到点维护一次（持着 launch_lock 调用）：建探查任务、剖析任务并在后台开始，记下触发。同一次触发已经记过就交回那一条。""" + from app.db.models import DataSource, ScheduleFire from app.scheduler import service - reason = service.TARGET_UNSUPPORTED.format(kind=service.TARGET_LABEL["source_maintenance"]) - return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", reason) + async with SessionLocal() as session: + existing = (await session.execute(select(ScheduleFire).where( + ScheduleFire.schedule_id == schedule_id, ScheduleFire.fire_at == fire_at, + ScheduleFire.kind == kind))).scalar_one_or_none() + if existing is not None: + return existing + row = await session.get(DataSource, target_id) + if row is None: + return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", SOURCE_GONE) + if problem := _origin_problem(row): + return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", problem) + if row.enabled is False: + return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", + SOURCE_DISABLED.format(name=row.name)) + steps = steps_of(row.options) + if not steps: + return await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", NO_STEPS) + if what := await _busy(session, row.id): + return await service.record_fire(schedule_id, fire_at, kind, "skipped_overlap", + BUSY.format(name=row.name, what=what), {"steps": steps}) + placeholder = await service.record_fire(schedule_id, fire_at, kind, "failed_to_start", INTERRUPTED, + {"steps": steps}) + if placeholder is None: + return None + try: + jobs, skipped = await _start(target_id, steps, actor=actor or service.actor_label(name), + fire_at=fire_at.isoformat(), schedule_id=schedule_id) + except Exception: # noqa: BLE001 - 建任务出意外不能让调度循环停下：照实记下，详细信息进日志 + logger.exception("定时维护（%s）建任务时出错", schedule_id) + return await _settle_fire(placeholder.id, "failed_to_start", "建维护任务时出错，详细信息见服务日志", + {"steps": steps}) + reason = "；".join(NOT_RUN.format(step=STEP_LABEL[s["step"]], why=s["reason"]) for s in skipped) or None + return await _settle_fire(placeholder.id, "started" if jobs else "failed_to_start", reason, + {"steps": steps, "jobs": jobs, "skipped": skipped}) -async def check_target(session: Any, target_id: str) -> str | None: - return "当前版本还不能按时发起「数据源维护」这类目标" +async def _start(source_id: str, steps: list[str], *, actor: str, fire_at: str, + schedule_id: str) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: + """按顺序建任务并开始：探查（introspect）→ 剖析 / 观察（一个剖析任务：范围按选的步骤）。返回 (建了的任务, 没执行的步骤 + 和原因)。""" + from app.data import catalog_profile, catalog_profile_jobs, introspect_jobs, table_versions + from app.db.models import DataSource + + jobs: list[dict[str, Any]] = [] + skipped: list[dict[str, str]] = [] + introspect_id: str | None = None + if "introspect" in steps: + async with SessionLocal() as session: + row = await session.get(DataSource, source_id) + try: + job, n = await introspect_jobs.prepare(session, row, actor=actor, trigger="schedule", fire_at=fire_at, + schedule_id=schedule_id) + except (introspect_jobs.IntrospectRefused, introspect_jobs.IntrospectBusy, introspect_jobs.IntrospectFailed, + introspect_jobs.ScopeTooLarge) as e: + skipped.append({"step": "introspect", "reason": str(e)}) + else: + introspect_jobs.manager.start(job.id) + introspect_id = job.id + jobs.append({"step": "introspect", "kind": "introspect", "id": job.id, "tables": n}) + wanted = [s for s in steps if s in ("profile_needed", "observe")] + if not wanted: + return jobs, skipped + if introspect_id is not None: + await _entered(source_id, introspect_id) + scope = "maintenance" if len(wanted) == 2 else ("needed" if wanted == ["profile_needed"] else "observe") + async with SessionLocal() as session: + row = await session.get(DataSource, source_id) + try: + settings = catalog_profile.ensure_enabled(row) + except catalog_profile.ProfileDisabled as e: + skipped += [{"step": s, "reason": str(e)} for s in wanted] + return jobs, skipped + try: + view = await table_versions.resolve_source(session, row) + job, tasks = await catalog_profile_jobs.create_job( + session, row, view, tables=None, scope=scope, settings=settings, actor=actor, trigger="schedule", + fire_at=fire_at, schedule_id=schedule_id, allow_empty=True) + except (catalog_profile_jobs.ProfileBusy, catalog_profile_jobs.ProfileNothing, + table_versions.SnapshotMissing) as e: + skipped += [{"step": s, "reason": str(e)} for s in wanted] + return jobs, skipped + catalog_profile_jobs.manager.start(job.id) + jobs.append({"step": "+".join(wanted), "kind": "profile", "id": job.id, "scope": scope, "tables": len(tasks)}) + return jobs, skipped + + +async def schedules_for(session: Any, source_id: str) -> list[dict[str, Any]]: + """指向这个数据源的维护定时（数据源设置里「定时维护」的快捷入口读它，第 4 波）：[{id, name, cron, timezone, enabled, + next_run_at}]。""" + from app.db.models import Schedule + + rows = (await session.execute(select(Schedule).where( + Schedule.target_kind == "source_maintenance", Schedule.target_id == source_id) + .order_by(Schedule.created_at))).scalars() + return [{"id": r.id, "name": r.name, "cron": r.cron, "timezone": r.timezone, "enabled": r.enabled, + "next_run_at": r.next_run_at.isoformat() if r.next_run_at else None} for r in rows] + + +__all__ = ["DEFAULT_STEPS", "MAINTENANCE_OPTION", "STEPS", "STEP_LABEL", "check_target", "fire", "schedules_for", + "steps_of", "steps_problem", "with_steps"] diff --git a/backend/app/notify/service.py b/backend/app/notify/service.py index b2b1f218..b56a82f3 100644 --- a/backend/app/notify/service.py +++ b/backend/app/notify/service.py @@ -357,3 +357,33 @@ async def notify_action(action_id: str, event: str) -> int: "subject": {"action_id": action_id, "run_id": rec.run_id, "task_id": task_id}} return await _push(channels, message, subject_kind="action", subject_id=action_id, event=event, run_id=rec.run_id, task_id=task_id) + + +async def notify_source_changed(source_id: str) -> int: + """数据源有要处理的变化（source.changed，03 B3 ④、目录-D9）：只推还要人看（开着、知道了还没处理）的「要处理」级变化， + 按源汇总；消息里只有源名、各级别条数和链接，不放表名、列名、取值。探查、剖析任务结束时有新的「要处理」变化才调。 + + 这一行的 scopes 第 4 波有了「变化」标签页才打开：现在 _channels 推不出去，这里返回 0（照实：没有界面看不了的通知不发）。""" + from app.data import source_changes + from app.db.models import DataSource + + event = "source.changed" + async with SessionLocal() as session: + row = await session.get(DataSource, source_id) + if row is None: + return 0 + counts = await source_changes.counts(session, source_id) + by = counts["by_severity"] + if not by.get("action"): + return 0 + channels = await _channels(session, event, None, []) + if not channels: + return 0 + base = await _base_url(session) + lines = [("数据源", f"「{row.name}」"), ("要处理", f"{by['action']} 条"), ("提醒", f"{by.get('warning', 0)} 条"), + ("说明", f"{by.get('info', 0)} 条")] + message = {"event": event, "status": "action", "title": f"数据源「{row.name[:80]}」有 {by['action']} 个变化要处理", + "lines": lines, "links": [("去看看", f"{base}/data/catalog/{source_id}")] if base else [], + "subject": {"source_id": source_id}} + return await _push(channels, message, subject_kind="source", subject_id=source_id, event=event, run_id=None, + task_id=None) diff --git a/backend/tests/test_source_changes.py b/backend/tests/test_source_changes.py new file mode 100644 index 00000000..1eb0cf60 --- /dev/null +++ b/backend/tests/test_source_changes.py @@ -0,0 +1,239 @@ +"""数据源的变化记录：接口、处理动作、给召回的拉取、通知（next-round 03 B3）。 + +SQLite 夹具（虚构的店铺库）探查建基线、写目录，再删一列、重新探查得到真实的结构变化；取值变化（新码值）直接经 +source_changes.record 记一条。守的是： + +- GET /changes 缺省列开着的和知道了的，带各级别计数；状态、种类筛选；不认识的状态 422； +- 处理：误报必须写理由（422），处理过的再处理 409；清理目录把影响面里的项从目录里去掉、记目录版本；加入码值走 + apply_patch（含义待填写、依据写变化编号）、记目录版本；知道了只改状态；剖析没开时「重新剖析」照实 422； +- 同一个变化再次测到只更新 last_seen_at、seen_count； +- list_changes 按 seq 游标拉、affecting 按表反查、triggers_recall 只认结构类和触及规则的新码值； +- source.changed 的推送：订阅还没打开时推不出去；打开后（测试里临时打开）消息里只有源名、条数和链接。 +""" +from __future__ import annotations + +import json +import sqlite3 +import uuid + +import httpx +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.data import catalog, source_changes as sc +from app.db.base import SessionLocal +from app.db.models import DataSource +from app.db.models_sources import SourceChange +from app.main import app +from app.notify import core, service +from tests.fixtures.sources import drop_source, introspect_now + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +@pytest.fixture +async def shop(client, tmp_path): + path = tmp_path / "shop.db" + db = sqlite3.connect(path) + db.executescript(""" + CREATE TABLE orders (id INTEGER PRIMARY KEY, status VARCHAR(10), channel VARCHAR(8), amount NUMERIC(10, 2)); + INSERT INTO orders VALUES (1, 'paid', 'web', 3), (2, 'void', 'shop', 4); + """) + db.commit() + db.close() + name = f"chg_{uuid.uuid4().hex[:8]}" + r = await client.post("/api/datasources", json={"name": name, "kind": "sqlite", "database": str(path), + "readonly": True}) + sid = r.json()["id"] + await introspect_now(client, sid) + async with SessionLocal() as session: + await catalog.write_entry(session, sid, "orders", { + "label": catalog.make_item("订单", "human"), + "columns": {"status": {"codes": catalog.make_item({"paid": "已支付", "void": "作废"}, "human"), + "label": catalog.make_item("状态", "human")}, + "channel": {"codes": catalog.make_item({"web": "线上"}, "profile", "proposed")}}, + }, if_version=0, actor="测试员") + db = sqlite3.connect(path) + db.execute("ALTER TABLE orders DROP COLUMN status") + db.commit() + db.close() + await introspect_now(client, sid) + yield {"id": sid, "name": name} + await drop_source(sid) + + +async def _change(sid: str, kind: str) -> SourceChange: + async with SessionLocal() as session: + return (await session.execute(select(SourceChange).where( + SourceChange.source_id == sid, SourceChange.kind == kind))).scalars().first() + + +async def _codes_new(sid: str, *, rules: bool = False) -> SourceChange: + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + change, new = await sc.record(session, row, sc.NewChange( + kind=sc.CODES_NEW, table="orders", column="channel", severity="warning", + after={"codes": ["app"], "shares": {"app": 0.4}}, basis={"method": "full", "key": "full"}, + affects={"catalog": [], "templates": [], "rules_touched": rules}, key=("full", ("app",))), + sc.Detected(job_kind="profile", job_id="job-x", trigger="schedule")) + await session.commit() + assert new + return change + + +async def test_list_counts_and_filters(client, shop): + sid = shop["id"] + r = await client.get(f"/api/datasources/{sid}/changes") + assert r.status_code == 200, r.text + body = r.json() + [removed] = [c for c in body["items"] if c["kind"] == "column_removed"] + assert removed["schema"] == "agentlab-change/1" and removed["column"] == "status" and removed["status"] == "open" + assert removed["severity"] == "action" and "删除了列" in removed["summary"] + assert body["counts"]["pending"] == body["total"] and body["counts"]["by_severity"]["action"] >= 1 + r = await client.get(f"/api/datasources/{sid}/changes", params={"kind": ["column_added"]}) + assert r.json()["items"] == [] + assert (await client.get(f"/api/datasources/{sid}/changes", params={"status": "nope"})).status_code == 422 + one = (await client.get(f"/api/datasources/{sid}/changes/{removed['id']}")).json()["change"] + assert one["id"] == removed["id"] + + +async def test_dismiss_needs_a_reason_and_finished_changes_are_final(client, shop): + sid = shop["id"] + change = await _change(sid, "column_removed") + url = f"/api/datasources/{sid}/changes/{change.id}/resolve" + r = await client.post(url, json={"action": "dismiss"}) + assert r.status_code == 422 and "理由" in r.json()["detail"] + r = await client.post(url, json={"action": "dismiss", "note": "测试库里故意删的列"}, headers={"X-Actor": "tester"}) + assert r.status_code == 200, r.text + got = r.json()["change"] + assert got["status"] == "dismissed" and got["resolution"]["note"] == "测试库里故意删的列" + assert got["resolution"]["action"] == "dismiss" and got["resolution"]["by"] == "tester" + assert (await client.post(url, json={"action": "acknowledge"})).status_code == 409 + counts = (await client.get(f"/api/datasources/{sid}/changes")).json()["counts"] + assert counts["by_severity"]["action"] == 0, "误报不再算要处理" + + +async def test_clean_catalog_removes_the_items_and_records_the_catalog_version(client, shop): + sid = shop["id"] + change = await _change(sid, "column_removed") + r = await client.post(f"/api/datasources/{sid}/changes/{change.id}/resolve", json={"action": "clean_catalog"}) + assert r.status_code == 200, r.text + body = r.json() + async with SessionLocal() as session: + entry = await catalog.read_entry(session, sid, "orders") + assert "status" not in (entry.notes.get("columns") or {}), "说到被删的列的项都清掉了" + assert entry.notes["label"]["value"] == "订单", "别的项不动" + assert body["catalog_version"] == entry.version == body["change"]["resolution"]["catalog_version"] + assert body["change"]["status"] == "resolved" + + +async def test_add_codes_goes_through_apply_patch(client, shop): + sid = shop["id"] + change = await _codes_new(sid) + r = await client.post(f"/api/datasources/{sid}/changes/{change.id}/resolve", json={"action": "add_codes"}) + assert r.status_code == 200, r.text + async with SessionLocal() as session: + entry = await catalog.read_entry(session, sid, "orders") + item = entry.notes["columns"]["channel"]["codes"] + assert item["value"] == {"web": "线上", "app": ""}, "码值是补充：原来的留着，新码值含义待填写" + assert (item["source"], item["status"]) == ("human", "confirmed") + assert change.id[:6] in item["reason"], "确认时的依据写这条变化的编号" + assert r.json()["catalog_version"] == entry.version + removed = await _change(sid, "column_removed") + r = await client.post(f"/api/datasources/{sid}/changes/{removed.id}/resolve", json={"action": "add_codes"}) + assert r.status_code == 422, "只有新码值能加入码值" + + +async def test_acknowledge_and_reprofile_when_profiling_is_off(client, shop): + sid = shop["id"] + change = await _codes_new(sid) + url = f"/api/datasources/{sid}/changes/{change.id}/resolve" + r = await client.post(url, json={"action": "reprofile"}) + assert r.status_code == 422 and "剖析" in r.json()["detail"], "剖析没开：照实说做不了" + r = await client.post(url, json={"action": "acknowledge"}) + assert r.status_code == 200 and r.json()["change"]["status"] == "acknowledged" + counts = (await client.get(f"/api/datasources/{sid}/changes")).json()["counts"] + assert counts["pending"] > counts["open"], "知道了还没处理的照样算要人看" + + +async def test_same_change_detected_again_only_bumps_seen_count(shop): + sid = shop["id"] + first = await _codes_new(sid) + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + again, new = await sc.record(session, row, sc.NewChange( + kind=sc.CODES_NEW, table="orders", column="channel", severity="action", key=("full", ("app",))), + sc.Detected(job_kind="profile")) + await session.commit() + assert not new and again.id == first.id and again.seen_count == 2 and again.severity == "action" + + +async def test_cursor_pull_affecting_and_recall_triggers(shop): + sid, name = shop["id"], shop["name"] + codes = await _codes_new(sid, rules=True) + async with SessionLocal() as session: + everything = await sc.list_changes(session, source_id=sid) + seqs = [c["seq"] for c in everything] + assert seqs == sorted(seqs) and len(seqs) >= 2 + rest = await sc.list_changes(session, source_id=sid, since_seq=seqs[0]) + assert [c["seq"] for c in rest] == seqs[1:], "按 seq 游标：大于 since_seq 的" + hit = await sc.affecting(session, name, ["ORDERS"]) + assert {c["id"] for c in hit} == {c["id"] for c in everything}, "表名不分大小写" + assert await sc.affecting(session, name, ["refunds"]) == [] + assert await sc.affecting(session, name, ["orders"], since=seqs[-1]) == [] + by = {c["kind"]: c for c in everything} + assert sc.triggers_recall(by["column_removed"]), "结构类触发召回" + assert sc.triggers_recall(next(c for c in everything if c["id"] == codes.id)), "触及有效记录条件的新码值触发召回" + assert not sc.triggers_recall({"kind": "volume_drop", "affects": {}}), "纯取值漂移只算数据质量提醒" + + +async def test_source_changed_message_has_only_name_counts_and_link(shop, monkeypatch): + sid, name = shop["id"], shop["name"] + assert await service.notify_source_changed(sid) == 0, "订阅还没打开（第 4 波）：推不出去" + got: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + got.append(request) + return httpx.Response(200, json={"ok": True}) + + monkeypatch.setattr(core, "_transport", httpx.MockTransport(handler)) + monkeypatch.setattr(core, "RETRY_DELAYS", (0, 0, 0)) + monkeypatch.setitem(service.EVENTS, "source.changed", service.Event("数据源有变化要处理", ("global",))) + async with SessionLocal() as session: + from app.core.crypto import encrypt + from app.db.models import NotifyChannel + + channel = NotifyChannel(name=f"变化-{uuid.uuid4().hex[:6]}", scope="global", format="generic", + url=encrypt("https://hooks.example.com/agentlab/changes-7a1c"), + events=["source.changed"]) + session.add(channel) + await session.commit() + channel_id = channel.id + try: + assert await service.notify_source_changed(sid) == 1 + [request] = got + body = json.loads(request.content) + text = body["title"] + body["text"] + assert body["event"] == "source.changed" and name in body["title"] + assert "要处理" in text and "orders" not in text and "status" not in text and "app" not in text, \ + "不放表名、列名、取值" + assert body["subject"] == {"source_id": sid} + finally: + async with SessionLocal() as session: + from app.db.models import NotifyChannel + + row = await session.get(NotifyChannel, channel_id) + await session.delete(row) + await session.commit() + + +def test_maintenance_option_is_hidden_from_the_datasource_form(): + from app.api import datasources + from app.data import source_maintenance + + assert source_maintenance.MAINTENANCE_OPTION in datasources._MANAGED_OPTIONS diff --git a/backend/tests/test_source_maintenance.py b/backend/tests/test_source_maintenance.py new file mode 100644 index 00000000..1f7af490 --- /dev/null +++ b/backend/tests/test_source_maintenance.py @@ -0,0 +1,210 @@ +"""数据源的定时维护（app/data/source_maintenance.py，next-round 03 B4、总方案 X4）。 + +定时只有一套：维护就是一条 target_kind=source_maintenance 的 Schedule，认领、补跑、积压只记一条都是 scheduler 那一套。 +这里守： + +- 到点按顺序建探查任务 → 剖析任务（范围按选的步骤），任务上记 trigger=schedule 和 fire_at；剖析排在同一次触发的探查后面 + （同源队串行：探查做完剖析才开始）；触发记录写明建了哪几个任务； +- 照实记没执行的：剖析没开时剖析、观察两步不执行（探查照做）；数据源停用、不存在、是上传表格时整次不执行；上一次的探查 + 还没做完时这次跳过； +- 调度循环的认领：错过的只补最近一次、积压只记一条； +- 维护步骤的接口：不合规 422，合规存进 options、数据源表单不显示它；核对目标（check_target）。 +""" +from __future__ import annotations + +import sqlite3 +import uuid +from datetime import timedelta + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete, select + +from app.data import catalog, catalog_profile_jobs, introspect_jobs, source_maintenance as sm +from app.db.base import SessionLocal, utcnow +from app.db.models import CatalogProfileJob, DataSource, Schedule, ScheduleFire +from app.db.models_sources import IntrospectJob +from app.main import app +from app.scheduler import loop, service +from tests.fixtures.sources import drop_source, introspect_now + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def _source(client, path, *, profile: bool = True) -> str: + db = sqlite3.connect(path) + db.executescript(""" + CREATE TABLE orders (id INTEGER PRIMARY KEY, status VARCHAR(10), paid_on DATE); + INSERT INTO orders VALUES (1, 'paid', '2026-10-01'), (2, 'void', '2026-10-02'); + """) + db.commit() + db.close() + r = await client.post("/api/datasources", json={"name": f"mnt_{uuid.uuid4().hex[:8]}", "kind": "sqlite", + "database": str(path), "readonly": True}) + sid = r.json()["id"] + await introspect_now(client, sid) + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + if profile: + row.options = {**(row.options or {}), "catalog_profile": {"enabled": True}} + await session.commit() + await catalog.write_entry(session, sid, "orders", { + "columns": {"status": {"codes": catalog.make_item({"paid": "已支付", "void": "作废"}, "human")}}, + }, if_version=0, actor="测试员") + return sid + + +async def _schedule(sid: str, **over) -> str: + async with SessionLocal() as session: + fields = {"name": f"维护-{uuid.uuid4().hex[:6]}", "cron": "0 3 * * *", "timezone": "Asia/Shanghai", + **service.target_columns("source_maintenance", sid), **over} + row = Schedule(**fields) + session.add(row) + await session.commit() + return row.id + + +@pytest.fixture +async def made(): + ids: list[str] = [] + yield ids + async with SessionLocal() as session: + await session.execute(delete(Schedule).where(Schedule.id.in_(ids))) + await session.commit() + + +async def _wait_jobs(fire: ScheduleFire) -> None: + for job in (fire.inputs or {}).get("jobs") or []: + manager = introspect_jobs.manager if job["kind"] == "introspect" else catalog_profile_jobs.manager + await manager.wait(job["id"], timeout=60) + + +async def test_fire_builds_introspect_then_profile_in_order(client, tmp_path, made): + sid = await _source(client, tmp_path / "a.db") + try: + schedule_id = await _schedule(sid) + made.append(schedule_id) + at = utcnow().replace(microsecond=0) + fire = await service.fire_once(schedule_id, at, "manual", actor="测试员") + assert fire.status == "started" and fire.reason is None, fire.reason + jobs = fire.inputs["jobs"] + assert [j["kind"] for j in jobs] == ["introspect", "profile"] and jobs[1]["scope"] == "maintenance" + assert fire.inputs["steps"] == list(sm.STEPS) + await _wait_jobs(fire) + async with SessionLocal() as session: + intro = await session.get(IntrospectJob, jobs[0]["id"]) + prof = await session.get(CatalogProfileJob, jobs[1]["id"]) + assert intro.trigger == "schedule" and intro.scope["fire_at"] == at.isoformat() + assert intro.scope["schedule_id"] == schedule_id and intro.created_by == "测试员" + assert catalog_profile_jobs.trigger_of(prof) == {"trigger": "schedule", "fire_at": at.isoformat(), + "schedule_id": schedule_id} + assert prof.started_at >= intro.finished_at, "同源串行：探查做完，剖析才开始" + assert prof.status in ("succeeded", "partial"), prof.error + # 同一次触发再来一遍：交回已有的那一条，不再建任务 + again = await service.fire_once(schedule_id, at, "manual", actor="测试员") + assert again.id == fire.id + finally: + await drop_source(sid) + + +async def test_profile_off_records_steps_not_run_and_still_introspects(client, tmp_path, made): + sid = await _source(client, tmp_path / "b.db", profile=False) + try: + schedule_id = await _schedule(sid) + made.append(schedule_id) + fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0), "manual") + assert fire.status == "started" + assert [j["kind"] for j in fire.inputs["jobs"]] == ["introspect"] + assert {s["step"] for s in fire.inputs["skipped"]} == {"profile_needed", "observe"} + assert "剖析需要的表没有执行" in fire.reason and "取值观察没有执行" in fire.reason and "数据剖析" in fire.reason + await _wait_jobs(fire) + # 只选观察一步：剖析没开就整次没执行 + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + row.options = sm.with_steps(row.options, ["observe"], actor="测试员") + await session.commit() + fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0) + timedelta(seconds=1), "manual") + assert fire.status == "failed_to_start" and "取值观察没有执行" in fire.reason + finally: + await drop_source(sid) + + +async def test_disabled_missing_and_busy_sources(client, tmp_path, made): + sid = await _source(client, tmp_path / "c.db") + try: + schedule_id = await _schedule(sid) + made.append(schedule_id) + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + row.enabled = False + await session.commit() + fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0), "manual") + assert fire.status == "failed_to_start" and "已停用" in fire.reason + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + row.enabled = True + session.add(IntrospectJob(source_id=sid, status="queued", scope={})) + await session.commit() + fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0) + timedelta(seconds=2), "manual") + assert fire.status == "skipped_overlap" and "正在探查结构" in fire.reason + async with SessionLocal() as session: + await session.execute(delete(IntrospectJob).where(IntrospectJob.source_id == sid, + IntrospectJob.status == "queued")) + await session.commit() + finally: + await drop_source(sid) + fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0) + timedelta(seconds=3), "manual") + assert fire.status == "failed_to_start" and fire.reason == sm.SOURCE_GONE + + +async def test_loop_claims_and_only_the_latest_missed_fire_runs(client, tmp_path, made): + sid = await _source(client, tmp_path / "d.db") + try: + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + row.options = sm.with_steps(row.options, ["introspect"], actor="测试员") + await session.commit() + now = utcnow().replace(second=0, microsecond=0) + # 每 10 分钟一次，停机错过了好几次：只补最近一次，更早的积压只记一条 + schedule_id = await _schedule(sid, cron="*/10 * * * *", timezone="UTC", + next_run_at=now - timedelta(minutes=40), misfire_grace_s=3600) + made.append(schedule_id) + fires = await loop.fire_due(now + timedelta(minutes=1)) + assert sorted(f.status for f in fires) == ["skipped_misfire", "started"], [(f.kind, f.status) for f in fires] + started = next(f for f in fires if f.status == "started") + assert [j["kind"] for j in started.inputs["jobs"]] == ["introspect"] + await _wait_jobs(started) + async with SessionLocal() as session: + sched = await session.get(Schedule, schedule_id) + count = len(list((await session.execute(select(ScheduleFire).where( + ScheduleFire.schedule_id == schedule_id))).scalars())) + assert sched.next_run_at > now and count == 2 + finally: + await drop_source(sid) + + +async def test_steps_api_and_check_target(client, tmp_path): + sid = await _source(client, tmp_path / "e.db") + try: + url = f"/api/datasources/{sid}/maintenance" + body = (await client.get(url)).json() + assert body["steps"] == list(sm.STEPS) and body["profile_enabled"] is True and body["schedules"] == [] + assert (await client.put(url, json={"steps": ["nope"]})).status_code == 422 + assert (await client.put(url, json={"steps": []})).status_code == 422 + r = await client.put(url, json={"steps": ["observe", "introspect"]}) + assert r.status_code == 200 and r.json()["steps"] == ["introspect", "observe"], "按固定顺序存" + [shown] = [s["options"] for s in (await client.get("/api/datasources")).json() if s["id"] == sid] + assert sm.MAINTENANCE_OPTION not in shown, "数据源表单不显示、不提交它" + async with SessionLocal() as session: + assert await sm.check_target(session, sid) is None + assert await sm.check_target(session, "missing") == sm.SOURCE_GONE + row = await session.get(DataSource, sid) + row.origin = "upload" + await session.commit() + assert await sm.check_target(session, sid) == sm.NOT_MAINTAINED + finally: + await drop_source(sid)