# -*- coding: utf-8 -*-
"""contract_gate.py — CL-1 清洗调用门: apply_contract_gate(df, contract, section) (docs/SOP.md §2.3/§3.1).

清洗入口强制经由此函数 (campaign 94 个脚本零读契约 = DP-10 主证的修复本体)。
执行即落盘 outputs/<farm>/sop/clean_gate.json — artifact 证调用 (sop_check CL-1 断言)。

语义:
- sentinel → NaN (契约 sentinel 列表, 计数)
- valid_range 界外 → NaN (非 clamp: 越物理界 = 无效测量, clamp 会扭曲统计;
  历史教训: 唐兴振动 raw max 308973 入 mean 冒充头条 [#36]; 振动统计另须 median, 见 SOP.md §3.1)
- 死列检测: 全局常值/全NaN; 给 machine_col 时按跨机判 (单机常值≠死列, memory dec_3p2mw)
- ffill: 契约字段标 ffill: true (事件驱动稀疏状态码 [#14]) 的列前向填充, 留痕
"""

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]
from src import paths as P
import json
import hashlib
from datetime import datetime
from pathlib import Path

ROOT = _install_root(__file__)


def _sha256(p, buf=1 << 20):
    h = hashlib.sha256()
    with open(p, 'rb') as f:
        for blk in iter(lambda: f.read(buf), b''):
            h.update(blk)
    return h.hexdigest()


def _load_contract(contract):
    if isinstance(contract, dict):
        return contract, '<dict>'
    import yaml
    p = Path(contract)
    with open(p, encoding='utf-8') as f:
        return yaml.safe_load(f), str(p)


def apply_contract_gate(df, contract, section, farm=None, out_dir=None, machine_col=None,
                        persist=True, time_col=None):
    """按契约清洗 df 并落盘 clean_gate.json。返回 (df_clean, report)。

    df: pandas.DataFrame; contract: yaml 路径或 dict; section: 契约 section 名 (如 turbine_sec);
    farm: outputs/<farm>/sop/ 落盘 (与 out_dir 二选一); machine_col: 跨机死列判定用机号列。
    persist: True(默认) 同时落清洗快照 outputs/<farm>/cleaned/<section>.parquet(zstd) +
             <section>_fingerprint.json(sha256/行数/窗/契约版本) 供审计 (审计=可复现∧指纹匹配, §3/§9.1.1);
             秒级 GB 层探查跑可传 persist=False 省盘。time_col: 有则指纹记数据窗 min/max。
    """
    import pandas as pd
    doc, cpath = _load_contract(contract)
    sec = doc.get(section) or {}
    cols = sec.get('columns', sec.get('fields', {})) or {}
    n_in = len(df)
    d = df.copy()
    clip_counts, sentinel_counts, ffill_cols, dead = {}, {}, [], []

    for col, spec in cols.items():
        if not isinstance(spec, dict) or col not in d.columns:
            continue
        s = pd.to_numeric(d[col], errors='coerce') if d[col].dtype == object else d[col]
        # 1) sentinel → NaN
        for sent in (spec.get('sentinel') or []):
            cnt = int((s == sent).sum())
            if cnt:
                s = s.where(s != sent)
                sentinel_counts[col] = sentinel_counts.get(col, 0) + cnt
        # 2) valid_range 界外 → NaN
        vr = spec.get('valid_range') or spec.get('valid_range_canonical')
        if isinstance(vr, (list, tuple)) and len(vr) == 2:
            bad = ((s < vr[0]) | (s > vr[1]))
            cnt = int(bad.sum())
            clip_counts[col] = cnt
            if cnt:
                s = s.where(~bad)
        # 3) ffill 声明 (事件驱动稀疏)
        if spec.get('ffill'):
            s = s.ffill()
            ffill_cols.append(col)
        d[col] = s

    # 4) 死列检测 (跨机判)
    for col in d.columns:
        if col == machine_col:
            continue
        s = d[col]
        nn = s.notna().sum()
        if nn == 0:
            dead.append({'col': col, 'reason': 'all-NaN'})
            continue
        try:
            nu = s.nunique(dropna=True)
        except TypeError:
            continue
        if nu <= 1:
            if machine_col and machine_col in d.columns:
                per_m = d.groupby(machine_col)[col].nunique(dropna=True)
                if (per_m <= 1).all():
                    dead.append({'col': col, 'reason': f'跨机全常值 ({int(per_m.shape[0])} 机)'})
                # 单机常值不判死 (memory dec_3p2mw)
            else:
                dead.append({'col': col, 'reason': '全局常值 (无machine_col, 未跨机验, 降置信)'})

    report = {
        'gate': 'CL-1', 'contract': cpath, 'section': section,
        'ts': datetime.now().isoformat(timespec='seconds'),
        'clip_counts': clip_counts or {'_note': '契约无 valid_range 列或零界外'},
        'sentinel_to_nan': sentinel_counts or {'_note': '零哨兵残留'},
        'dead_columns': dead,
        'ffill_applied': ffill_cols,
        'rows_in_out': {'in': int(n_in), 'out': int(len(d))},
        'contract_range_cols': [c for c, sp in cols.items()
                                if isinstance(sp, dict) and (sp.get('valid_range') or sp.get('valid_range_canonical'))],
    }
    target = None
    if out_dir:
        target = Path(out_dir)
    elif farm:
        target = P.sop(farm)
    if target:
        target.mkdir(parents=True, exist_ok=True)
        (target / 'clean_gate.json').write_text(
            json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8')

    # 清洗快照 + 指纹 (用户 2026-06-17 审计留存): 审计=可复现 ∧ 指纹匹配, 对齐 §9.1.1 输入指纹
    fp = None
    if target and persist:
        try:
            cleaned_dir = target.parent / 'cleaned'
            cleaned_dir.mkdir(parents=True, exist_ok=True)
            pq = cleaned_dir / f'{section}.parquet'
            d.to_parquet(pq, compression='zstd', index=False)
            win = None
            if time_col and time_col in d.columns:
                tc = pd.to_datetime(d[time_col], errors='coerce')
                win = {'min': str(tc.min()), 'max': str(tc.max())}
            try:
                rel = pq.relative_to(ROOT).as_posix()
            except ValueError:
                rel = str(pq)
            fp = {
                'artifact': 'cleaned_fingerprint', 'section': section, 'parquet': rel,
                'sha256': _sha256(pq), 'rows': int(len(d)), 'cols': int(d.shape[1]),
                'window': win, 'contract': cpath,
                'n_machines': int(d[machine_col].nunique(dropna=True))
                if machine_col and machine_col in d.columns else None,
                'ts': report['ts'],
            }
            (cleaned_dir / f'{section}_fingerprint.json').write_text(
                json.dumps(fp, ensure_ascii=False, indent=2), encoding='utf-8')
        except Exception as e:
            report['persist_error'] = str(e)[:200]
    report['cleaned_fingerprint'] = fp
    return d, report