guanlan-src/app_ontology/app_ontology_guanlan/sop/ledger_dates.py

# -*- coding: utf-8 -*-
"""大部件维修台账 日期区间解析器 + 三条硬校验 (2026-08-26).

**为什么要有**: `rudong.ops.maint_major_parts` 的 `维修时间_dt` 列 **20/20 全部 NaT** —— 日期解析
完全失败却无人报错, 下游任何用该列的分析拿到的都是空值 (**静默失效**)。而回测的 ground truth 精度
直接取决于锚日期: 解析失败 → 自动流程全废; 更坏的是**宽松解析吃进脏值** → 造出错误的提前期。

台账实况 (20 行, 人工录入, 格式极不统一):
    '2024年1月5日-2023年1月7日'      ← 结束年份笔误, 早于开始
    '2024年5月13日至2024年5月19日'    ← "至" 而非 "-"
    '2024/9/19至2024-10-13'          ← 一行内混用 / 与 -
    '2025/8/23 -2025-8-28'           ← 空格 + 混用分隔符
    '2025年12月4日-2025年12月70'      ← 日越界 (70 号)
    nan                              ← 缺失

## 三条硬校验 (违反即 raise, 不静默降级)
1. **解析成功率** — 非空行必须 100% 解析出 (start, end); 失败即 raise 并列出原文
2. **区间单调** — end >= start; 违反即 raise (捕 "2024年1月5日-2023年1月7日" 类年份笔误)
3. **日期合法** — 年 2000~2100 ∧ 月 1~12 ∧ 日在该月实际天数内; 违反即 raise (捕 "12月70")

**为什么 raise 而不是标 NaT**: 静默降级正是本模块要消灭的病 —— 原 `维修时间_dt` 就是"解析不了就给
NaT", 结果 20/20 全空却无人察觉。宁可跑不动, 不可假装跑通 (memory criterion-required-input-no-silent-get)。

用法:
    from src.sop.ledger_dates import parse_range, parse_ledger_column
    parse_range('2025/8/23 -2025-8-28')      # → (Timestamp('2025-08-23'), Timestamp('2025-08-28'))
    parse_ledger_column(df['维修时间'])       # → DataFrame[start, end, raw]; 任一校验失败 raise
"""
from __future__ import annotations

import calendar
import re

import pandas as pd

# 分隔符: 中文"至" / 连字符 / 波浪号 / "到"; 两侧允许空白
_SEP = re.compile(r'\s*(?:至|到|~|—|–|-)\s*')
# 单个日期: 2024年1月5日 | 2024/9/19 | 2024-10-13 | 2025年12月4日(无"日")
_ONE = re.compile(r'(\d{4})\s*[年/\-]\s*(\d{1,2})\s*[月/\-]\s*(\d{1,2})\s*日?')

YEAR_MIN, YEAR_MAX = 2000, 2100


class LedgerDateError(ValueError):
    """台账日期解析/校验失败 — 响亮抛出, 不静默降级。"""


def _check_ymd(y, m, d, raw, which):
    if not (YEAR_MIN <= y <= YEAR_MAX):
        raise LedgerDateError(f'[校验3 日期合法] {which} 年份 {y} 越界 ({YEAR_MIN}~{YEAR_MAX}) — 原文 {raw!r}')
    if not (1 <= m <= 12):
        raise LedgerDateError(f'[校验3 日期合法] {which} 月份 {m} 越界 (1~12) — 原文 {raw!r}')
    dmax = calendar.monthrange(y, m)[1]
    if not (1 <= d <= dmax):
        raise LedgerDateError(f'[校验3 日期合法] {which} 日 {d} 越界 ({y}年{m}月 共 {dmax} 天) — 原文 {raw!r}')


def parse_range(raw, *, allow_single=True):
    """解析一个维修时间单元格 → (start, end)。三条硬校验全在此处执行, 违反即 raise。

    allow_single=True 时, 只有一个日期的单元格返回 (d, d) (维修当天完成)。
    """
    if raw is None or (isinstance(raw, float) and pd.isna(raw)):
        return None
    s = str(raw).strip().strip("'\"")
    if not s or s.lower() == 'nan':
        return None
    hits = _ONE.findall(s)
    if not hits:
        raise LedgerDateError(f'[校验1 解析成功率] 无法解析出任何日期 — 原文 {raw!r}')
    if len(hits) == 1:
        if not allow_single:
            raise LedgerDateError(f'[校验1 解析成功率] 只解析出 1 个日期但要求区间 — 原文 {raw!r}')
        y, m, d = (int(x) for x in hits[0])
        _check_ymd(y, m, d, raw, '单日')
        t = pd.Timestamp(year=y, month=m, day=d)
        return t, t
    if len(hits) > 2:
        raise LedgerDateError(f'[校验1 解析成功率] 解析出 {len(hits)} 个日期 (期望 2) — 原文 {raw!r}')
    (y1, m1, d1), (y2, m2, d2) = ((int(x) for x in h) for h in hits)
    _check_ymd(y1, m1, d1, raw, '开始')
    _check_ymd(y2, m2, d2, raw, '结束')
    a = pd.Timestamp(year=y1, month=m1, day=d1)
    b = pd.Timestamp(year=y2, month=m2, day=d2)
    if b < a:
        raise LedgerDateError(
            f'[校验2 区间单调] 结束 {b:%Y-%m-%d} 早于开始 {a:%Y-%m-%d} — 原文 {raw!r}; '
            f'台账实见此类为**年份笔误** (如 2024年1月5日-2023年1月7日, 结束年应为 2024)')
    return a, b


def parse_ledger_column(col, *, strict=True):
    """整列解析 → DataFrame[raw, start, end, n_days]。

    strict=True (默认): 任一非空行解析或校验失败 → raise, 并在消息里汇总全部问题行 (一次看全, 不用逐条试)。
    strict=False: 问题行的 start/end 置 NaT 并在返回的 `.attrs['errors']` 里列出 —— **仅供人工排查用**,
                  禁止用于判据输入 (那就退回静默降级了)。
    """
    rows, errs = [], []
    for i, raw in enumerate(col):
        try:
            r = parse_range(raw)
        except LedgerDateError as e:
            errs.append(f'  行{i}: {e}')
            rows.append(dict(raw=raw, start=pd.NaT, end=pd.NaT, n_days=None))
            continue
        if r is None:
            rows.append(dict(raw=raw, start=pd.NaT, end=pd.NaT, n_days=None))
        else:
            rows.append(dict(raw=raw, start=r[0], end=r[1], n_days=int((r[1] - r[0]).days) + 1))
    if errs and strict:
        raise LedgerDateError(f'台账日期校验失败 {len(errs)} 行 (共 {len(col)} 行):\n' + '\n'.join(errs))
    out = pd.DataFrame(rows)
    out.attrs['errors'] = errs
    n_nonnull = sum(1 for x in col if not (x is None or (isinstance(x, float) and pd.isna(x))
                                           or str(x).strip().lower() in ('', 'nan')))
    out.attrs['parse_rate'] = (out.start.notna().sum() / n_nonnull) if n_nonnull else 1.0
    return out