guanlan-src/app_algorithmModel/app_algorithmModel_guanlan/windcms/pipeline.py

# -*- coding: utf-8 -*-
"""一条命令从原始数据到报告: 导入 (rar/7z | 解码 JSON 目录 | 通用波形目录) → 六层链 (扫线→能量占比→六层→融合) → 报告 → 知识库.
确定性管线处理数据; 模型 (千问) 只负责触发/读产物/推理/叙述. 每步计时与返回码落 out/analyze_log.json, 失败响亮."""
import os, sys, json, time, subprocess
from pathlib import Path
from src.windscada.config import ROOT
from src.windcms import data

# 六层链脚本需要真正的 Python 解释器 (含 pandas/scipy/pyarrow); 冻结版 (PyInstaller) 的 sys.executable 不是 python → 用 WINDCMS_SCRIPT_PY 或回退 venv
PY = os.environ.get('WINDCMS_SCRIPT_PY') or sys.executable    # 冻结包也用自己的解释器 (原写死 mac 路径)


def _run(step, cmd, env, log, cwd=ROOT, timeout=7200):
    t0 = time.time()
    p = subprocess.run(cmd, cwd=str(cwd), env=env, capture_output=True, text=True, errors='replace', timeout=timeout)
    # ★ 2026-09-17: `text=True` 下 stdout/stderr 可能是 None(极少数平台/编码路径) ⇒ 直接相加会抛
    #   "unsupported operand type(s) for +: 'NoneType' and 'str'", 于是**真正的失败原因被这句
    #   无意义的异常顶掉**(实测: oem_scan 步明确报了 rc=3 与缺什么口径, 经壳调用后看到的是这句)。
    out = (p.stdout or '') + (p.stderr or '')
    rec = dict(step=step, cmd=' '.join(map(str, cmd)), rc=p.returncode, seconds=round(time.time() - t0, 1),
               tail=out[-1500:])
    log.append(rec)
    print(f'[{step}] rc={p.returncode} {rec["seconds"]}s', flush=True)
    if p.returncode != 0:
        raise RuntimeError(f'步骤 {step} 失败 (rc={p.returncode}):\n{rec["tail"]}')
    return rec


def detect_input(path):
    p = Path(path)
    if p.is_file() and p.suffix.lower() in ('.rar', '.7z', '.zip'):
        return 'tcm_archive'
    if p.is_dir():
        if any(p.rglob('*_decode.json')):
            return 'tcm_decoded_json'
        if any(p.glob('*.npz')) or any(p.glob('*.csv')):
            return 'waveforms'
    return 'unknown'


def window_year_gate(m5_root, window, max_age_days=400):
    """B-1 年份闸 (2026-08-26 w1226 事故): wMMDD 窗名无年份, 历史包混入现役窗名会被字典序 sorted()[-1]
    当末窗 → 冲击分型基线倒挂全场 INSUFFICIENT. 现役窗名 (w+4位数字) 只准装最近 ~13 个月数据;
    更早的历史回填必须显式 histYYYY_ 前缀 (不匹配 w[0-9]{4} glob, 不进现役分析集). 缺时间列/缺 index 响亮 raise, 不静默放行."""
    import re
    import pandas as pd
    if not re.fullmatch(r'w\d{4}', str(window)):
        return None   # hist*/测试窗等显式非现役名, 不适用
    ix = Path(m5_root) / 'windows' / str(window) / 'index.parquet'
    if not ix.exists():
        raise RuntimeError(f'年份闸: {ix} 不存在 — ingest 未落 index, 不可判年份 (禁静默放行)')
    import pyarrow.parquet as pq
    cols = pq.read_schema(str(ix)).names
    tcol = next((c for c in ('trigger_time', 'time', 'timestamp') if c in cols), None)
    if tcol is None:
        raise RuntimeError(f'年份闸: {ix} 无时间列 (找过 trigger_time/time/timestamp, 实有 {cols[:8]}) — 不可判年份 (禁静默放行)')
    tmax = pd.to_datetime(pd.read_parquet(ix, columns=[tcol])[tcol]).max()
    age = (pd.Timestamp.now() - tmax).days
    if age > max_age_days:
        raise RuntimeError(
            f'年份闸: 窗 {window} 数据止于 {tmax:%Y-%m-%d} (距今 {age} 天 > {max_age_days}) — wMMDD 窗名无年份, '
            f'历史数据会被字典序当成末窗 (w1226 事故 2026-08-26). 历史回填请改名 hist{tmax.year}_{window} '
            f'(mv {ix.parent} 后重跑), 现役窗名只装最近 13 个月数据')
    return tmax


def _zip_has_decoded_json(path) -> bool:
    """压缩包里是否已经是**解码后**的 `*_decode.json` (2026-09-16)。

    为什么要有这个判断: `detect_input()` 把任何 .zip/.rar/.7z 都归为 `tcm_archive`, 而那条路要调
    `scripts/rudong_tcm_ingest_raw.py` (解原始 base64+XML 导出) —— 该脚本**没随包** ⇒ 现场直接
    把 CMS 导出的 zip 指过来必然崩在 `FileNotFoundError`。但其实"zip 里就是 `*_decode.json`"时
    根本不需要它: 本包的 `rudong_tcm_index.py` / `rudong_tcm_spectra.py` **支持把 zip 当 --root**
    (直接按成员读), 走它们即可。这里只做"是不是解码后包"的判定, 不假装支持原始导出。
    """
    import zipfile
    try:
        with zipfile.ZipFile(path) as zf:
            return any(not i.is_dir() and i.filename.endswith('_decode.json') for i in zf.infolist())
    except Exception:
        return False


def ingest(cfg, path, window, log, env):
    kind = detect_input(path)
    if kind == 'tcm_archive' and _zip_has_decoded_json(path):
        # 解码后的导出包 (zip 形态): 直接走索引/谱, 不必经缺失的 rudong_tcm_ingest_raw.py
        print(f'[ingest] {Path(path).name}: 内含 *_decode.json → 按"已解码导出"处理 (zip 当 root)', flush=True)
        kind = 'tcm_decoded_json'
    if kind == 'tcm_archive':
        raw = ROOT / 'scripts/rudong_tcm_ingest_raw.py'
        if not raw.is_file():
            raise RuntimeError(
                f'输入 {path} 是**原始** TCM 导出包 (rar/7z, 需先解 base64+XML), 而解码脚本 '
                f'{raw.name} 未随包 (振动线分支产物)。两条可走的路: '
                f'① 现场若能直接给"含 *_decode.json 的导出"(zip 或目录), 把它放 data/raw/<场>/windcms/ 下, '
                f'用 scripts/vib_raw_build.py 摄入 (本包支持 zip 当 root); '
                f'② 向振动线索取 scripts/rudong_tcm_ingest_raw.py 后再跑本步。')
        _run('ingest:tcm_archive', [PY, str(raw), '--rar', str(path), '--window', window], env, log)
    elif kind == 'tcm_decoded_json':
        out = cfg['m5'] / 'windows' / window
        out.mkdir(parents=True, exist_ok=True)
        _run('ingest:index', [PY, str(ROOT / 'scripts/rudong_tcm_index.py'), '--root', str(path), '--out', str(out / 'index.parquet')], env, log)
        _run('ingest:spectra', [PY, str(ROOT / 'scripts/rudong_tcm_spectra.py'), '--root', str(path), '--out', str(out / 'spectra')], env, log)
    elif kind == 'waveforms':
        from src.windcms import ingest_waveforms
        t0 = time.time()
        n = ingest_waveforms.ingest_dir(cfg, Path(path), window)
        log.append(dict(step='ingest:waveforms', cmd=f'ingest_waveforms.ingest_dir({path}, {window})', rc=0, seconds=round(time.time() - t0, 1), tail=f'{n} records'))
    else:
        raise RuntimeError(f'无法识别输入格式: {path} (支持 TCM rar/7z/zip 导出包 · 含 *_decode.json 的目录 · 含 npz/csv 波形的目录)')
    window_year_gate(cfg['m5'], window)
    return kind


def analyze(cfg, input_path=None, window=None, steps=None, bands='ALL'):
    """全链. steps 子集可选: ingest, oem_scan, energy_share, model_run, fusion, report, kb."""
    out = cfg['out']
    out.mkdir(parents=True, exist_ok=True)
    log = []
    env = dict(os.environ)
    env.setdefault('MPLBACKEND', 'Agg')
    t_all = time.time()
    try:
        if input_path:
            if not window:
                raise RuntimeError('导入需要 --window 名 (如 w0905)')
            ingest(cfg, input_path, window, log, env)
        wins = list(data.windows(cfg))
        env['M5_WINDOWS'] = ','.join(wins)
        env['SCAN_BANDS'] = bands
        want = set(steps or ['oem_scan', 'energy_share', 'model_run', 'fusion', 'report', 'kb'])
        # ★ 2026-09-17: 缺脚本要给"人看得懂、且知道去哪儿要料"的错, 而不是让 python 抛一句
        #   `can't open file 'scripts/rudong_model_run.py'`（实测就是这句 —— 现场看到它没法判断
        #   这是"包缺件"还是"我敲错了"）。四步脚本属振动线分支产物, 未随包; 缺了就指名要说清楚。
        def _step(name: str, script: str):
            p = ROOT / 'scripts' / script
            if not p.is_file():
                raise RuntimeError(
                    f'六层链的 {name} 步脚本未随包: {p.relative_to(ROOT).as_posix()}（振动线分支产物）。'
                    f'调用约定见 docs/振动六层链_接口规格与缺口_v0.1.md §1, 取料单见 docs/向振动线取料单_v0.1.md; '
                    f'自查: python scripts/chain_gap_check.py。'
                    f'（本包已逆向实现 fusion 步的一件产物: scripts/rudong_fusion_run.py 可重出 fleet_scalar_z）')
            return [PY, str(p)]

        if 'oem_scan' in want:
            _run('oem_scan', _step('oem_scan', 'rudong_tcm_oem_scan.py'), env, log)
        if 'energy_share' in want:
            _run('energy_share', _step('energy_share', 'rudong_line_energy_share.py'), env, log)
        if 'model_run' in want:
            _run('model_run', _step('model_run', 'rudong_model_run.py'), env, log)
        if 'fusion' in want:
            _run('fusion', _step('fusion', 'rudong_fusion_run.py'), env, log)
        if 'report' in want:
            # 标量缓存按窗集合命名, 新窗自动失效; 谱元数据进程内缓存由新进程重建
            _run('report', [PY, str(ROOT / 'scripts/windcms.py'), 'report', '--farm', cfg.get('key', 'rudong')], env, log)
        if 'kb' in want:
            _run('kb', [PY, str(ROOT / 'scripts/windcms.py'), 'kb', '--farm', cfg.get('key', 'rudong')], env, log)
        status = 'ok'
    except Exception as e:
        log.append(dict(step='FAILED', cmd='', rc=1, seconds=0, tail=str(e)[-1500:]))
        status = 'failed'
    summary = dict(status=status, windows=list(data.windows(cfg)), seconds=round(time.time() - t_all, 1), finished=time.strftime('%Y-%m-%d %H:%M:%S'), steps=log)
    # ★ 2026-09-17 用户令 2: 运行日志只在 logs/ —— 原先写 `out/analyze_log.json`(产物仓),
    #   于是产物仓里躺着运行日志(反向审计把它当产物、log_audit 也要报了)。改落
    #   `logs/build/<场>/windcms/analyze_log.json`; 产物仓里的旧件已被清理。
    #   日志里有中文命令回显: 缺 encoding 会按系统 locale(cp936) 写 → 中文 Windows 崩
    from src import logfile as _lf
    logp = _lf.audit_log('windcms_analyze.json')   # 审计/报告类 JSON 按口径落 logs/audit/
    logp.parent.mkdir(parents=True, exist_ok=True)
    with open(logp, 'w', encoding='utf-8', newline='\n') as f:
        json.dump(summary, f, ensure_ascii=False, indent=1)
    print(f'analyze {status}: {summary["seconds"]}s → {logp}', flush=True)
    if status != 'ok':
        # ★ 2026-09-17: 失败时把**原因**打到控制台, 而不是只留一句 "analyze failed" 让人去翻 JSON。
        #   实测场景: 四步脚本未随包时, 原来只在日志里能看到 python 的 FileNotFoundError,
        #   现场看到的是"重算失败"四个字, 无从判断是包缺件还是操作有误。
        bad = [s for s in log if s.get('step') == 'FAILED'] + [s for s in log if s.get('rc') not in (0, None)]
        for s in bad[:2]:
            print('   [X] ' + str(s.get('step')) + ': ' + str(s.get('tail', '')).strip()[:400], flush=True)
        print(f'   详情: {logp}', flush=True)
    return summary