1日2回のはずの分類バッチが1回しか動かない — 成果物から実行回数を復元する

"""未処理キューの残高から、実効スループットが計画を下回っていないかを判定する"""
 
def drain_rate(backlog: list[int]) -> float:
    """1日あたりの平均減少量。正なら消化、負なら滞留している"""
    deltas = [backlog[i] - backlog[i + 1] for i in range(len(backlog) - 1)]
    return sum(deltas) / len(deltas)
 
 
def days_to_empty(backlog: list[int]) -> float | None:
    rate = drain_rate(backlog)
    return None if rate <= 0 else backlog[-1] / rate
 
 
def plan_ratio(planned_per_run: int, runs_per_day: int, backlog: list[int]) -> dict:
    """計画スループットに対する実測の比率。0.5 付近なら回数の欠落を疑う"""
    planned = planned_per_run * runs_per_day
    observed = drain_rate(backlog)
    return {
        "planned_per_day": planned,
        "observed_per_day": round(observed, 1),
        "ratio": round(observed / planned, 2) if planned else None,
    }
 
 
if __name__ == "__main__":
    # 8日分の残高（毎朝同じ時刻に計測・期間中の新規流入なし）
    backlog = [1180, 1108, 1036, 962, 890, 818, 744, 672]
    print("平均消化 /日:", round(drain_rate(backlog), 1))
    print("枯渇まで:", round(days_to_empty(backlog), 1), "日")
    print(plan_ratio(planned_per_run=72, runs_per_day=2, backlog=backlog))

平均消化 /日: 72.6
枯渇まで: 9.3 日
{'planned_per_day': 144, 'observed_per_day': 72.6, 'ratio': 0.5}

"""実行スロットの欠落検知（成果物側からの逆算）"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
 
JST = ZoneInfo("Asia/Tokyo")
 
 
@dataclass(frozen=True)
class Slot:
    day: str        # "2026-08-26"
    label: str      # "04:30"
    start: datetime
    end: datetime   # 猶予込みの締切
 
 
def expected_slots(days: list[str], times: list[str], grace_min: int = 90) -> list[Slot]:
    """予定表から、猶予つきの受け入れ窓を作る"""
    out = []
    for d in days:
        y, m, dd = map(int, d.split("-"))
        for t in times:
            hh, mm = map(int, t.split(":"))
            start = datetime(y, m, dd, hh, mm, tzinfo=JST)
            out.append(Slot(d, t, start, start + timedelta(minutes=grace_min)))
    return sorted(out, key=lambda s: s.start)
 
 
def observed_runs(records: list[dict]):
    """成果物に付いた run_id と完了時刻を、実行の痕跡として取り出す"""
    seen = set()
    for r in records:
        rid = r.get("run_id")
        if not rid or rid in seen:
            continue
        seen.add(rid)
        yield datetime.fromisoformat(r["finished_at"]).astimezone(JST)
 
 
def reconcile(slots: list[Slot], runs: list[datetime]) -> dict:
    """窓に痕跡を割り当て、余った窓と余った痕跡を分けて返す"""
    remaining = sorted(runs)
    missed, matched = [], []
    for s in slots:
        hit = next((r for r in remaining if s.start <= r <= s.end), None)
        if hit is None:
            missed.append(s)
        else:
            matched.append((s, hit))
            remaining.remove(hit)
    return {"missed": missed, "matched": matched, "orphans": remaining}
 
 
if __name__ == "__main__":
    days = [f"2026-08-{d:02d}" for d in range(20, 28)]
    slots = expected_slots(days, ["04:30", "16:30"])
 
    # 成果物から拾えた完了時刻（04:30 枠のぶんだけが並んでいた）
    records = [
        {"run_id": f"r{i}", "finished_at": f"{d}T04:52:00+09:00"}
        for i, d in enumerate(days)
    ]
 
    res = reconcile(slots, list(observed_runs(records)))
    print(f"期待スロット: {len(slots)}  実行痕跡: {len(res['matched'])}  欠落: {len(res['missed'])}")
 
    by_label: dict[str, int] = {}
    for s in res["missed"]:
        by_label[s.label] = by_label.get(s.label, 0) + 1
    for label, n in sorted(by_label.items()):
        print(f"  欠落が集中したスロット {label}: {n}/{len(days)} 日")

期待スロット: 16  実行痕跡: 8  欠落: 8
  欠落が集中したスロット 16:30: 8/8 日

"""予定時刻からの遅れの実測から、受け入れ窓の猶予を決める"""
import math
 
 
def suggest_grace(delays_min: list[int], safety: float = 3.0, step: int = 15) -> int:
    """中央値ではなく最大値を基準にする。猶予は誤報を防ぐための余白なので、
    平均的な回ではなく、いちばん遅かった回に合わせる必要がある"""
    worst = max(delays_min)
    return int(math.ceil(worst * safety / step) * step)
 
 
def grace_fits(grace_min: int, times: list[str]) -> bool:
    """窓が隣の枠へ食い込んでいないか。食い込むと痕跡の取り違えが起きる"""
    mins = sorted(int(t[:2]) * 60 + int(t[3:]) for t in times)
    gaps = [b - a for a, b in zip(mins, mins[1:])] + [1440 - mins[-1] + mins[0]]
    return grace_min < min(gaps)
 
 
if __name__ == "__main__":
    delays = [2, 3, 2, 4, 9, 2, 3, 26]   # 予定時刻から起動までの遅れ（分）
    g = suggest_grace(delays)
    print("いちばん遅かった回:", max(delays), "分")
    print("推奨する猶予:", g, "分")
    print("枠どうしが重ならないか:", grace_fits(g, ["04:30", "16:30"]))

いちばん遅かった回: 26 分
推奨する猶予: 90 分
枠どうしが重ならないか: True

"""起動したこと自体を、処理内容と切り離して記録する"""
import sqlite3
import uuid
from datetime import datetime, timezone
 
DDL = """
CREATE TABLE IF NOT EXISTS run_heartbeat (
    run_id      TEXT PRIMARY KEY,
    slot_label  TEXT NOT NULL,     -- "04:30" など予定表側の識別子
    started_at  TEXT NOT NULL,
    finished_at TEXT,
    item_count  INTEGER            -- 0 も正当な値として記録する
);
"""
 
 
def open_ledger(path: str = "runs.db") -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute(DDL)
    conn.commit()
    return conn
 
 
def begin_run(conn: sqlite3.Connection, slot_label: str) -> str:
    """Gemini を呼ぶ前に呼ぶ。ここで書かないと不発火と区別できなくなる"""
    run_id = uuid.uuid4().hex
    conn.execute(
        "INSERT INTO run_heartbeat (run_id, slot_label, started_at) VALUES (?, ?, ?)",
        (run_id, slot_label, datetime.now(timezone.utc).isoformat()),
    )
    conn.commit()
    return run_id
 
 
def end_run(conn: sqlite3.Connection, run_id: str, item_count: int) -> None:
    conn.execute(
        "UPDATE run_heartbeat SET finished_at = ?, item_count = ? WHERE run_id = ?",
        (datetime.now(timezone.utc).isoformat(), item_count, run_id),
    )
    conn.commit()
 
 
def missing_slots(conn: sqlite3.Connection, day: str, expected: list[str]) -> list[str]:
    rows = conn.execute(
        "SELECT slot_label FROM run_heartbeat WHERE started_at LIKE ? || '%'", (day,)
    ).fetchall()
    seen = {r[0] for r in rows}
    return [s for s in expected if s not in seen]

"""begin_run だけ書かれて finished_at が埋まっていない回を拾う"""
from datetime import datetime, timedelta, timezone
 
 
def stuck_runs(conn: sqlite3.Connection, older_than_min: int = 120) -> list[tuple[str, str]]:
    """不発火とは別種の事故なので、欠落とは分けて数える"""
    cutoff = (datetime.now(timezone.utc) - timedelta(minutes=older_than_min)).isoformat()
    rows = conn.execute(
        "SELECT run_id, slot_label FROM run_heartbeat "
        "WHERE finished_at IS NULL AND started_at < ?",
        (cutoff,),
    ).fetchall()
    return [(r[0], r[1]) for r in rows]