"""Schedule cycle ↔ 5-field cron (Baota-compatible).""" from __future__ import annotations from typing import Any, TypedDict class ScheduleCycleConfig(TypedDict, total=False): type: str minute: int hour: int interval: int week: int monthDay: int customCron: str def _clamp_int(n: int | float, lo: int, hi: int) -> int: try: v = int(n) except (TypeError, ValueError): return lo return max(lo, min(hi, v)) SCHEDULE_MIN_INTERVAL_MINUTES = 1 def build_cron_from_cycle(config: dict[str, Any]) -> str: kind = config.get("type", "day") if kind == "custom": raw = str(config.get("customCron") or "").strip() if not raw: raise ValueError("自定义 cron 不能为空") parts = raw.split() if len(parts) != 5: raise ValueError("cron 表达式必须为 5 个字段") return raw minute = _clamp_int(config.get("minute", 0), 0, 59) hour = _clamp_int(config.get("hour", 0), 0, 23) if kind == "day": return f"{minute} {hour} * * *" if kind == "day-n": n = _clamp_int(config.get("interval", 1), 1, 365) return f"{minute} {hour} */{n} * *" if kind == "hour": return f"{minute} * * * *" if kind == "hour-n": n = _clamp_int(config.get("interval", 1), 1, 23) return f"{minute} */{n} * * *" if kind == "minute-n": n = _clamp_int(config.get("interval", 1), SCHEDULE_MIN_INTERVAL_MINUTES, 59) return f"*/{n} * * * *" if kind == "week": w = _clamp_int(config.get("week", 0), 0, 6) return f"{minute} {hour} * * {w}" if kind == "month": dom = _clamp_int(config.get("monthDay", 1), 1, 31) return f"{minute} {hour} {dom} * *" raise ValueError(f"未知的周期类型: {kind}") def _parse_step_or_fixed(field: str) -> tuple[str, int | None]: if field == "*": return "star", None if field.startswith("*/"): step = int(field[2:]) return "step", step return "fixed", int(field) def parse_cron_to_cycle(cron: str) -> dict[str, Any]: base: dict[str, Any] = { "type": "day", "minute": 0, "hour": 2, "interval": 2, "week": 1, "monthDay": 1, "customCron": "", } expr = (cron or "").strip() if not expr: return base parts = expr.split() if len(parts) != 5: return {**base, "type": "custom", "customCron": expr} min_f, hour_f, dom_f, mon_f, dow_f = parts if mon_f != "*": return {**base, "type": "custom", "customCron": expr} min_k, min_v = _parse_step_or_fixed(min_f) hour_k, hour_v = _parse_step_or_fixed(hour_f) dom_k, dom_v = _parse_step_or_fixed(dom_f) dow_k, dow_v = _parse_step_or_fixed(dow_f) if min_k == "step" and hour_k == "star" and dom_k == "star" and dow_k == "star": return {**base, "type": "minute-n", "interval": min_v} if min_k == "fixed" and hour_k == "star" and dom_k == "star" and dow_k == "star": return {**base, "type": "hour", "minute": min_v} if min_k == "fixed" and hour_k == "step" and dom_k == "star" and dow_k == "star": return {**base, "type": "hour-n", "minute": min_v, "interval": hour_v} if min_k == "fixed" and hour_k == "fixed" and dom_k == "star" and dow_k == "fixed": w = 0 if dow_f == "7" else dow_v return {**base, "type": "week", "minute": min_v, "hour": hour_v, "week": w} if min_k == "fixed" and hour_k == "fixed" and dom_k == "fixed" and dow_k == "star": return {**base, "type": "month", "minute": min_v, "hour": hour_v, "monthDay": dom_v} if min_k == "fixed" and hour_k == "fixed" and dom_k == "step" and dow_k == "star": return {**base, "type": "day-n", "minute": min_v, "hour": hour_v, "interval": dom_v} if min_k == "fixed" and hour_k == "fixed" and dom_k == "star" and dow_k == "star": return {**base, "type": "day", "minute": min_v, "hour": hour_v} return {**base, "type": "custom", "customCron": expr} _WEEK_LABELS = ["周日", "周一", "周二", "周三", "周四", "周五", "周六"] def format_cycle_label(cron_or_config: str | dict[str, Any]) -> str: config = parse_cron_to_cycle(cron_or_config) if isinstance(cron_or_config, str) else cron_or_config kind = config.get("type", "day") if kind == "custom": cc = config.get("customCron") or "" return f"自定义: {cc}" if cc else "—" minute = int(config.get("minute", 0)) hour = int(config.get("hour", 0)) time_s = f"{hour:02d}:{minute:02d}" if kind == "day": return f"当天 {time_s}" if kind == "day-n": return f"每 {config.get('interval', 1)} 天 {time_s}" if kind == "hour": return f"每小时 第 {minute} 分" if kind == "hour-n": return f"每 {config.get('interval', 1)} 小时 第 {minute} 分" if kind == "minute-n": return f"每 {config.get('interval', 1)} 分钟" if kind == "week": w = int(config.get("week", 0)) wd = _WEEK_LABELS[w] if 0 <= w < 7 else f"周{w}" return f"每周{wd} {time_s}" if kind == "month": return f"每月 {config.get('monthDay', 1)} 日 {time_s}" return "—"