yml2213
55f899c615
用户反馈：「单账号重检间隔(小时) 很别扭，直接设置多久跑一次、一次跑多少个更符合直觉」。
**对，而且比"别扭"更严重：调度频率以前根本改不了。**
## 问题：两个概念挤在一个字段里
`_run_one` 里原本写着：
job.next_run_at = finished + timedelta(seconds=JOB_INTERVAL_SECONDS) # 硬编码 300 秒
也就是不管界面上填什么，下一轮永远是「跑完 + 5 分钟」。
而界面那个「单账号重检间隔(小时)」对应的是 `interval_seconds` ——
它只被作业**自己的选行逻辑**使用（"这行多久该重检"），**完全不参与调度**。
于是出现最坏的情况：**看起来能配调度，其实配的是一个不相干的参数**。
上一轮我还拿它算过吞吐（"10 分钟一轮"），算错了 —— 已在调度器注释里纠正。
## 拆法
| 概念 | 字段 | 界面 |
| --- | --- | --- |
| **多久跑一次** | `run_interval_seconds`（迁移 0051，默认 300）| 「多久跑一次(分钟)」|
| 同一账号多久重检 | `interval_seconds`（不变）| 「同一账号多久重检(小时)」|
- `next_run_at = finished + run_interval_seconds`（有下限 30 秒，避免忙轮询）
- 接口：`GET` 返回新字段、`PATCH` 可改
- 前端：两个输入框分开，并把原来那段绕口的 tooltip
（"它不决定调度频率 —— 调度周期见页面顶部说明"）改掉 ——
那句话本身就是这个设计缺陷的自白
## 默认 300 秒
与改动前**实际行为一致** —— 不会因为这次迁移让已有作业突然变快或变慢。
想调快就自己填（比如「账号资料定时补全」填 5 分钟 + 每轮 400 条 ≈ 3 小时补完）。
## ⚠️ 顺带说明一个你截图里可能看到的现象
截图里「账号资料定时补全」显示的是 **0.5 小时 / 30 条** —— 那是我**最初的默认值**。
因为 `ensure_jobs` **只创建不更新**，我后面改成 10 分钟/400 **不会作用到已存在的行**。
这是刻意的（否则会覆盖运维的手工调整），但意味着**默认值改了要手动在界面上改**，
或者删掉那行让 `ensure_jobs` 重建。现在有了「多久跑一次」这个框，直接填就行。
## 测试（+3）
- 下一轮按**运行间隔**排，不是按「同一行重检间隔」（两个值刻意设成不同来钉住）
- 不填时默认 300 秒（保持旧行为）
- 过小的间隔被夹到下限
## 验证
TZ=Asia/Shanghai **639 passed / 6 skipped**；ruff 全绿；前端 build 通过、lint 0 errors。
285 lines
10 KiB
Python
285 lines
10 KiB
Python
"""自动化作业接口。
|
||
|
||
设计上「先能看、再能开」：查看需要 ``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)
|