diff --git a/tests/test_automation_ck_health.py b/tests/test_automation_ck_health.py index f523833..1e912c4 100644 --- a/tests/test_automation_ck_health.py +++ b/tests/test_automation_ck_health.py @@ -916,3 +916,81 @@ class TestSchedulerRunsJobsConcurrently: time.sleep(0.1) assert not scheduler._IN_FLIGHT, "失败后没清在途标记,作业会被永久跳过" assert calls, "作业应该真的跑过一次" + + +class TestRunIntervalIsScheduling: + """「多久跑一次」与「同一行多久重检」是**两个独立概念**。 + + 用户反馈:界面上那个「单账号重检间隔(小时)」很别扭,直接设「多久跑一次、 + 一次跑多少个」更符合直觉 —— 对,因为**调度频率以前根本改不了**: + + job.next_run_at = finished + timedelta(seconds=JOB_INTERVAL_SECONDS) # 硬编码 5 分钟 + + 于是不管界面上填什么,下一轮永远是「跑完 + 5 分钟」。 + """ + + def setup_method(self): + self.engine = get_test_engine() + self.session = make_test_session(self.engine) + reset_registry() + + def teardown_method(self): + reset_registry() + self.session.close() + + def _job(self, run_interval: int | None): + calls: list[int] = [] + + def probe(db, job): + calls.append(1) + return {"scanned": 1} + + register_job( + JobSpec( + key="probe", + title="probe", + interval_seconds=6 * 3600, # 同一行重检间隔(不参与调度) + batch_size=10, + concurrency=1, + run=probe, + **({"run_interval_seconds": run_interval} if run_interval else {}), + ) + ) + self.session.add( + AutomationJob( + key="probe", + title="probe", + interval_seconds=6 * 3600, + run_interval_seconds=run_interval or 300, + batch_size=10, + concurrency=1, + enabled=True, + ) + ) + self.session.commit() + return calls + + def _next_run_delay(self) -> float: + run_due_jobs(self.session) + job = ( + self.session.query(AutomationJob).filter(AutomationJob.key == "probe").one() + ) + assert job.last_run_at is not None and job.next_run_at is not None + return (job.next_run_at - job.last_run_at).total_seconds() + + def test_next_run_follows_run_interval_not_the_row_interval(self): + """下一轮按**运行间隔**排,不是按「同一行重检间隔」。""" + self._job(run_interval=1800) + delay = self._next_run_delay() + assert delay == 1800, f"应按运行间隔 1800 排,实际 {delay}" + assert delay != 6 * 3600, "不能再用「同一行重检间隔」当调度周期" + + def test_default_run_interval_keeps_old_behaviour(self): + """不填时默认 300 秒 —— 与改动前的硬编码行为一致(不会突然变快变慢)。""" + self._job(run_interval=None) + assert self._next_run_delay() == 300 + + def test_too_small_interval_is_clamped(self): + """太小的间隔是忙轮询,没有意义 —— 有下限。""" + self._job(run_interval=1) + assert self._next_run_delay() >= scheduler.MIN_RUN_INTERVAL_SECONDS diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 2058f51..690d69f 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -26,7 +26,7 @@ TESTS_DIR = Path(__file__).resolve().parent PROJECT_ROOT = TESTS_DIR.parent VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions" -HEAD_REVISION = "20261004_0050" +HEAD_REVISION = "20261004_0051" # 迁移新增、但不声明在模型里的复合查询索引。 EXTRA_INDEXES = ( diff --git a/web/backend/migrations/versions/20261004_0051_run_interval.py b/web/backend/migrations/versions/20261004_0051_run_interval.py new file mode 100644 index 0000000..c09f6c9 --- /dev/null +++ b/web/backend/migrations/versions/20261004_0051_run_interval.py @@ -0,0 +1,64 @@ +"""作业的「运行间隔」独立出来:`automation_jobs.run_interval_seconds` + +## 问题 + +界面上只有一个「单账号重检间隔(小时)」,但它其实承担了两个互不相干的概念, +而且**调度频率根本改不了**: + +| 概念 | 之前 | 之后 | +| --- | --- | --- | +| **多久跑一次** | **硬编码** `JOB_INTERVAL_SECONDS`(5 分钟)| `run_interval_seconds` | +| 同一行多久重检 | `interval_seconds` | `interval_seconds`(不变,只是不再兼任调度)| + +`_run_one` 里原本写着: + + job.next_run_at = finished + timedelta(seconds=JOB_INTERVAL_SECONDS) + +也就是不管在界面上填什么,下一轮永远是「跑完 + 5 分钟」。 +用户反馈「直接设置多久跑一次、一次跑多少个更符合直觉」—— 对。 + +默认 300 秒,与改动前的实际行为一致(不会因为这次迁移让已有作业突然变快/变慢)。 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision: str = "20261004_0051" +down_revision: str | None = "20261004_0050" +branch_labels: str | None = None +depends_on: str | None = None + +COLUMN = "run_interval_seconds" + + +def _column_exists(name: str) -> bool: + conn = op.get_bind() + count = conn.execute( + sa.text( + "SELECT COUNT(*) FROM information_schema.columns " + "WHERE table_schema = DATABASE() " + "AND table_name = 'automation_jobs' AND column_name = :name" + ), + {"name": name}, + ).scalar() + return bool(count) + + +def upgrade() -> None: + if not _column_exists(COLUMN): + op.add_column( + "automation_jobs", + sa.Column( + COLUMN, + sa.Integer(), + nullable=False, + server_default="300", + comment="多久跑一次(秒);与「同一行多久重检」是两回事", + ), + ) + + +def downgrade() -> None: + op.drop_column("automation_jobs", COLUMN) diff --git a/web/backend/models.py b/web/backend/models.py index 4ec4a5d..16b5687 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -897,6 +897,15 @@ class AutomationJob(Base): title: Mapped[str] = mapped_column(String(128), default="") # 默认关闭:按「渐进上线」原则,先观察再打开。 enabled: Mapped[bool] = mapped_column(Boolean, default=False, server_default="0") + # **多久跑一次**(秒)。这是调度频率,与"同一行多久重检"是两回事。 + # 以前调度频率是**硬编码**的 JOB_INTERVAL_SECONDS(5 分钟), + # UI 上那个"单账号重检间隔"其实是 `interval_seconds` —— + # 两个概念挤在一个输入框里,很别扭(用户指出)。 + run_interval_seconds: Mapped[int] = mapped_column( + Integer, default=300, server_default="300" + ) + # 同一**行**(账号)多久重检一次(秒)—— 由作业自己的选行逻辑使用, + # **不参与调度**。别和上面那个混。 interval_seconds: Mapped[int] = mapped_column( Integer, default=21600, server_default="21600" ) diff --git a/web/backend/routers/automation.py b/web/backend/routers/automation.py index 1b56a91..0a5c8bf 100644 --- a/web/backend/routers/automation.py +++ b/web/backend/routers/automation.py @@ -37,6 +37,8 @@ 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) @@ -47,6 +49,7 @@ def _job_out(job: AutomationJob) -> dict[str, Any]: "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, @@ -125,6 +128,8 @@ def update_job( 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 diff --git a/web/backend/services/automation/scheduler.py b/web/backend/services/automation/scheduler.py index b4058a9..af2eaa1 100644 --- a/web/backend/services/automation/scheduler.py +++ b/web/backend/services/automation/scheduler.py @@ -57,10 +57,14 @@ class JobSpec: key: str title: str + #: 同一**行**多久重检(由作业自己的选行逻辑用,不参与调度) interval_seconds: int batch_size: int concurrency: int run: Callable[[Session, AutomationJob], dict[str, Any]] + #: **多久跑一次**(调度频率)。默认 5 分钟,与改动前的实际行为一致。 + #: ⚠️ 必须放在**有默认值**的位置(dataclass 不允许默认值字段排在无默认值之前)。 + run_interval_seconds: int = 300 _REGISTRY: dict[str, JobSpec] = {} @@ -68,6 +72,9 @@ _REGISTRY: dict[str, JobSpec] = {} #: 同时最多几个作业在跑。作业之间互不依赖,并行是为了**慢作业不挡住快作业** #: (以前串行:一个作业跑 4 分钟,CK 巡检就得等 4 分钟)。 _JOB_MAX_WORKERS = 4 + +#: 运行间隔的下限:太短会变成忙轮询,没有意义 +MIN_RUN_INTERVAL_SECONDS = 30 _EXECUTOR: ThreadPoolExecutor | None = None _EXECUTOR_LOCK = threading.Lock() @@ -115,6 +122,7 @@ def ensure_jobs(db: Session) -> None: title=spec.title, enabled=False, interval_seconds=spec.interval_seconds, + run_interval_seconds=spec.run_interval_seconds, batch_size=spec.batch_size, concurrency=spec.concurrency, config=None, @@ -305,9 +313,12 @@ def _run_one(db: Session, job: AutomationJob, spec: JobSpec) -> dict[str, Any]: run.detail = counters.get("detail") or None job.last_run_at = finished - # 下一轮固定在一个 tick 之后;interval_seconds 是「同一行多久重检」, - # 由作业自己在选行时用,不参与调度。 - job.next_run_at = finished + timedelta(seconds=JOB_INTERVAL_SECONDS) + # 下一轮 = 跑完 + **运行间隔**(可配置)。 + # ⚠️ 别再用 `interval_seconds` —— 那是「同一行多久重检」, + # 由作业自己的选行逻辑使用,**不参与调度**(两个概念以前挤在一个字段里)。 + job.next_run_at = finished + timedelta( + seconds=max(int(job.run_interval_seconds or 0), MIN_RUN_INTERVAL_SECONDS) + ) job.last_status = status job.last_message = message job.updated_at = finished diff --git a/web/frontend/src/api/types.ts b/web/frontend/src/api/types.ts index a9a3898..0119cfb 100644 --- a/web/frontend/src/api/types.ts +++ b/web/frontend/src/api/types.ts @@ -876,6 +876,9 @@ export interface HuyaPasswordLoginSelectedRequest { has_cookie?: boolean; force_new_device?: boolean; max_attempts?: number; + /** 多久跑一次(秒)—— 调度频率 */ + run_interval_seconds?: number; + /** 同一账号多久重检(秒)—— 不参与调度 */ interval_seconds?: number; retry_delay_seconds?: number; } @@ -1031,6 +1034,8 @@ export interface AutomationRunItem { } export interface AutomationJobItem { + /** 多久跑一次(秒)—— 调度频率 */ + run_interval_seconds: number; key: string; title: string; enabled: boolean; @@ -1068,6 +1073,9 @@ export interface AutomationRunsResponse { export interface AutomationJobUpdate { enabled?: boolean; + /** 多久跑一次(秒)—— 调度频率 */ + run_interval_seconds?: number; + /** 同一账号多久重检(秒)—— 不参与调度 */ interval_seconds?: number; batch_size?: number; concurrency?: number; diff --git a/web/frontend/src/pages/AutomationPage.tsx b/web/frontend/src/pages/AutomationPage.tsx index 7fcf09f..b266abe 100644 --- a/web/frontend/src/pages/AutomationPage.tsx +++ b/web/frontend/src/pages/AutomationPage.tsx @@ -99,6 +99,11 @@ export default function AutomationPage() { job.key, { enabled: job.enabled, + // 「多久跑一次」与「同一行多久重检」是两件事,分开两个输入框 + run_interval_minutes: Math.max( + Math.round((job.run_interval_seconds ?? 300) / 60), + 1, + ), interval_hours: secondsToHours(job.interval_seconds), batch_size: job.batch_size, concurrency: job.concurrency, @@ -123,6 +128,10 @@ export default function AutomationPage() { try { await automationApi.updateJob(job.key, { enabled: values.enabled, + run_interval_seconds: Math.max( + Math.round((values.run_interval_minutes ?? 5) * 60), + 30, + ), interval_seconds: Math.max( Math.round((values.interval_hours ?? 6) * 3600), 300, @@ -312,10 +321,17 @@ export default function AutomationPage() { > + + +