AgentLab/backend/app/data/source_maintenance.py

"""数据源的定时维护：探查 → 剖析需要的表 → 取值观察（docs/design/next-round/03-数据接入与目录维护.md B4，目录-D5）。

**为什么要有它。** 以前只能手动探查、手动剖析，目录只在点按钮那一刻是对的；结构变化、取值漂移要靠定期观察才看得到。

**定时只有一套**（总方案 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

    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 _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"]