Nexus/server/utils/schedule_cycle.py

"""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 "—"