Function execute_selfcheck_run

src/core/system_selfcheck.py:670–792

(
    run_id: int,
    *,
    mode: Optional[str] = None,
    source: Optional[str] = None,
    progress_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
    cancel_checker: Optional[Callable[[], bool]] = None,
)

def execute_selfcheck_run(
    run_id: int,
    *,
    mode: Optional[str] = None,
    source: Optional[str] = None,
    progress_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
    cancel_checker: Optional[Callable[[], bool]] = None,
) -> Dict[str, Any]:
    with get_db() as db:
        run = db.query(SelfCheckRun).filter(SelfCheckRun.id == int(run_id)).first()
        if not run:
            raise ValueError("自检任务不存在")
        mode_text = "full" if str(mode or run.mode or "").strip().lower() == "full" else "quick"
        source_text = str(source or run.source or "manual").strip().lower() or "manual"
        run.mode = mode_text
        run.source = source_text
        run.status = RUN_STATUS_RUNNING
        run.started_at = _utc_now()
        run.finished_at = None
        run.error_message = None
        run.result_data = {"checks": [], "repairs": [], "progress": {"completed": 0, "total": 0}}
        run.summary = "系统自检运行中"
        db.commit()

    started = time.perf_counter()
    checks: List[Dict[str, Any]] = []
    proxy_url = _resolve_selfcheck_proxy_url()
    check_funcs: List[Callable[[], Dict[str, Any]]] = [_check_environment, lambda: _check_network(proxy_url), lambda: _check_accounts_auth(mode_text, proxy_url)]
    if mode_text == "full":
        check_funcs.extend([_check_team_pool, _check_payment_pipeline, _check_data_consistency])

    total = len(check_funcs)
    failed_error = ""
    try:
        for idx, func in enumerate(check_funcs, start=1):
            if cancel_checker and cancel_checker():
                score_info = _compute_score(checks)
                elapsed_ms = int((time.perf_counter() - started) * 1000)
                with get_db() as db:
                    run = db.query(SelfCheckRun).filter(SelfCheckRun.id == int(run_id)).first()
                    if not run:
                        raise ValueError("自检任务不存在")
                    run.status = RUN_STATUS_CANCELLED
                    run.total_checks = score_info["total"]
                    run.passed_checks = score_info["passed"]
                    run.warning_checks = score_info["warns"]
                    run.failed_checks = score_info["failed"]
                    run.score = score_info["score"]
                    run.duration_ms = elapsed_ms
                    run.summary = f"任务已取消（已完成 {len(checks)}/{total}）"
                    run.error_message = None
                    run.finished_at = _utc_now()
                    data = _safe_dict(run.result_data)
                    data["checks"] = checks
                    data["progress"] = {"completed": len(checks), "total": total}
                    run.result_data = data
                    db.commit()
                    db.refresh(run)