"""自动化作业接口。 设计上「先能看、再能开」：查看需要 ``automation:view``（运营也有），改配置与 手动触发需要 ``automation:manage``（仅超管）。作业默认关闭，打开后由进程内 调度器按间隔执行。 """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.orm import Session from ..database import get_db from ..deps import require_permission from ..models import Account, AutomationJob, AutomationRun, LoginTask, User from ..services.audit_service import record_audit from ..services.automation import ( ensure_jobs, registered_jobs, run_job_now, scheduler_status, ) router = APIRouter(prefix="/api/automation", tags=["自动化"]) MAX_BATCH_SIZE = 2000 MIN_INTERVAL_SECONDS = 300 class JobUpdate(BaseModel): """作业配置更新；只传要改的字段。""" enabled: bool | None = None interval_seconds: int | None = Field(default=None, ge=MIN_INTERVAL_SECONDS) # 多久跑一次（秒）。下限 30 秒，避免变成忙轮询。 run_interval_seconds: int | None = Field(default=None, ge=30) batch_size: int | None = Field(default=None, ge=1, le=MAX_BATCH_SIZE) concurrency: int | None = Field(default=None, ge=1, le=16) def _job_out(job: AutomationJob) -> dict[str, Any]: return { "key": job.key, "title": job.title, "enabled": bool(job.enabled), "interval_seconds": int(job.interval_seconds or 0), "run_interval_seconds": int(job.run_interval_seconds or 0), "batch_size": int(job.batch_size or 0), "concurrency": int(job.concurrency or 0), "last_run_at": job.last_run_at.isoformat() if job.last_run_at else None, "next_run_at": job.next_run_at.isoformat() if job.next_run_at else None, "last_status": job.last_status or "", "last_message": job.last_message or "", "registered": job.key in registered_jobs(), } def _run_out(run: AutomationRun) -> dict[str, Any]: return { "id": run.id, "job_key": run.job_key, "started_at": run.started_at.isoformat() if run.started_at else None, "finished_at": run.finished_at.isoformat() if run.finished_at else None, "status": run.status or "", "scanned": int(run.scanned or 0), "valid": int(run.valid or 0), "invalid": int(run.invalid or 0), "skipped": int(run.skipped or 0), "message": run.message or "", "detail": run.detail or None, } @router.get("/jobs") def list_jobs( db: Session = Depends(get_db), current: User = Depends(require_permission("automation:view")), ): """列出作业配置、最近一次结果与调度器状态。""" ensure_jobs(db) jobs = db.query(AutomationJob).order_by(AutomationJob.key.asc()).all() latest: dict[str, AutomationRun] = {} for run in ( db.query(AutomationRun).order_by(AutomationRun.id.desc()).limit(200).all() ): latest.setdefault(run.job_key, run) # 巡检与重登是两个作业，但通过数据库联动：巡检把行标成 invalid，重登作业的 # 候选条件正是这个字段。这里把「当前有多少条等着重登」一并返回，页面才能把 # 这层上下游关系显示出来，而不是让人以为两个开关互不相干。 from ..services.automation.jobs.ck_relogin import count_pending_relogin pending_relogin = count_pending_relogin(db) return { "scheduler": scheduler_status(), "pending_relogin": pending_relogin, "jobs": [ { **_job_out(job), "latest_run": _run_out(latest[job.key]) if job.key in latest else None, } for job in jobs ], } @router.patch("/jobs/{key}") def update_job( key: str, payload: JobUpdate, db: Session = Depends(get_db), current: User = Depends(require_permission("automation:manage")), ): """改开关 / 间隔 / 批量 / 并发。打开开关时排定下一轮时间。""" ensure_jobs(db) job = db.query(AutomationJob).filter(AutomationJob.key == key).first() if job is None: raise HTTPException(status_code=404, detail="作业不存在") changed: dict[str, Any] = {} if payload.enabled is not None and bool(payload.enabled) != bool(job.enabled): job.enabled = bool(payload.enabled) changed["enabled"] = job.enabled # 刚打开时排一轮：不早于一个 tick，避免打开瞬间就压满 CPU。 job.next_run_at = datetime.now(UTC) + _tick_delta() if job.enabled else None if payload.run_interval_seconds is not None: job.run_interval_seconds = int(payload.run_interval_seconds) if payload.interval_seconds is not None: changed["interval_seconds"] = payload.interval_seconds job.interval_seconds = payload.interval_seconds if payload.batch_size is not None: changed["batch_size"] = payload.batch_size job.batch_size = payload.batch_size if payload.concurrency is not None: changed["concurrency"] = payload.concurrency job.concurrency = payload.concurrency if changed: job.updated_at = datetime.now(UTC) record_audit( db, current, action="automation:update", target=key, detail=changed, ) db.commit() return _job_out(job) @router.post("/jobs/{key}/run") def run_job( key: str, db: Session = Depends(get_db), current: User = Depends(require_permission("automation:manage")), ): """立刻跑一轮（不看 enabled）。用于正式打开开关前先试跑。""" ensure_jobs(db) try: outcome = run_job_now(db, key) except KeyError: raise HTTPException(status_code=404, detail="作业不存在") from None record_audit( db, current, action="automation:run", target=key, detail={ "status": outcome.get("status"), "scanned": outcome.get("scanned"), "valid": outcome.get("valid"), "invalid": outcome.get("invalid"), "message": outcome.get("message"), }, ) db.commit() return outcome # 失败分型 → 页面展示用的中文说明。「转人工」的两种要写清为什么， # 否则客服不知道该做什么。 MANUAL_KIND_LABELS = { "captcha": "极验滑块：需要人工过一次验证", "credential": "凭证问题：需要核对账号密码 / 邮箱", "": "重试次数用尽：需要人工确认账号是否还可用", } def _manual_kind_label(kind: str) -> str: return MANUAL_KIND_LABELS.get(kind, f"待确认（{kind}）") @router.get("/manual-queue") def manual_queue( limit: int = Query(default=20, ge=1, le=100), db: Session = Depends(get_db), current: User = Depends(require_permission("automation:view")), ): """待人工处理的重登失败账号，按失败原因分组。 自动重登只处理「还能自动救」的失败；极验滑块、凭证问题、以及重试次数用尽的 必须交给人 —— 这个接口就是那份清单（规划 §7.4.6）。 **2026-10-04 起直接读生命周期状态 `status == "manual"`**：这个判据原本在这里 用 4 个条件自己拼（分型属人工 / 有错 / 无下次排期 / 试过），生命周期状态机 上线后同一件事只在 `account_lifecycle.derive_status` 里定义一次。 """ from ..services.account_lifecycle import MANUAL_KINDS, STATUS_MANUAL # 队列单位是**账号**。 「最近尝试时间」仍来自登录记录（那是尝试的属性）， # 用相关子查询取最后一条。 latest_attempt_at = ( select(LoginTask.attempt_at) .where(LoginTask.account_id == Account.id) .order_by(LoginTask.id.desc()) .limit(1) .scalar_subquery() ) rows = ( db.query(Account, latest_attempt_at.label("last_attempt_at")) .filter(Account.status == STATUS_MANUAL) .order_by(latest_attempt_at.desc(), Account.id.desc()) .all() ) grouped: dict[str, dict] = {} for account, last_attempt_at in rows: kind = account.relogin_failure_kind or "" # 可重试的类型却已经没有下次排期 = 次数用尽，单独成组。 # 否则它会混进「验证码失败」组里，客服看不出「这个已经不会再自动试了」。 if kind not in MANUAL_KINDS: kind = "" group = grouped.setdefault( kind, { "kind": kind, "label": _manual_kind_label(kind), "count": 0, "samples": [], }, ) group["count"] += 1 if len(group["samples"]) < limit: group["samples"].append( { "account_id": account.id, "account_username": account.username, "relogin_attempts": int(account.relogin_attempts or 0), "last_error": account.relogin_last_error or "", "attempt_at": last_attempt_at.isoformat() if last_attempt_at else None, } ) groups = sorted(grouped.values(), key=lambda item: -item["count"]) return {"total": len(rows), "groups": groups} @router.get("/runs") def list_runs( job_key: str | None = Query(default=None), limit: int = Query(default=50, ge=1, le=200), db: Session = Depends(get_db), current: User = Depends(require_permission("automation:view")), ): """运行历史，最近的在前。""" query = db.query(AutomationRun) if job_key: query = query.filter(AutomationRun.job_key == job_key) runs = query.order_by(AutomationRun.id.desc()).limit(limit).all() return {"items": [_run_out(run) for run in runs]} def _tick_delta(): from datetime import timedelta from ..services.automation.scheduler import TICK_SECONDS return timedelta(seconds=TICK_SECONDS)