guanlan-src/app_ontology/app_ontology_guanlan/ontology/monthly_roll.py

# -*- coding: utf-8 -*-
"""P3-c 月度滚动: 新数据 → L1 重跑 → 本体增量 → 沙盘刷新 → 变化摘要 (只报变化).

跑法: .venv/bin/python -m src.ontology.monthly_roll [--dry]
--dry = 只检测与出摘要, 不重跑 L1 (数据未更新时的空转应零变化零动作)."""
from src import paths as P
import copy, json, pathlib, subprocess, sys
import pandas as pd

ONT = P.ont()
ST = P.store()
PY = P.python_exe()                                          # 本安装目录的 venv (跨平台), 原写死 mac 路径


def data_month():
    lub = pd.read_parquet(ST / 'mblub_monthly.parquet')
    return str(lub.month.max())


def ont_month():
    o = json.loads((ONT / 'objects.json').read_text(encoding='utf-8'))
    mw = o.get('mwindow/rudong/climatology', {}).get('props', {})
    roll = o.get('meta/roll_state', {}).get('props', {})
    return roll.get('data_month')


def diff_summary(before, after):
    """两版本体 diff → 变化摘要 (只报变化; 判级变化逐条, 计数类汇总)."""
    chg = []
    b_ids, a_ids = set(before), set(after)
    new_ids = a_ids - b_ids
    gone = b_ids - a_ids
    from collections import Counter
    if new_ids:
        chg.append(f"新增对象 {len(new_ids)}: " + str(dict(Counter(after[i]['type'] for i in new_ids))))
    if gone:
        chg.append(f"消失对象 {len(gone)}: {sorted(gone)[:8]}")
    for i in (a_ids & b_ids):
        if after[i]['type'] != 'Verdict':
            continue
        lb = before[i].get('props', {}).get('level')
        la = after[i].get('props', {}).get('level')
        if lb != la:
            chg.append(f"判级变化 {i}: {lb} → {la}")
    return chg


def roll(dry=False):
    before = json.loads((ONT / 'objects.json').read_text(encoding='utf-8'))
    dm, om = data_month(), ont_month()
    actions = []
    if dm == om:
        actions.append(f'数据月 {dm} 无更新 — 空转 (L1 不重跑)')
    elif not dry:
        # L1 重跑清单 (仓建器) → 本体增量 → 状态机 → 沙盘
        subprocess.run([PY, 'scripts/windscada_watch_channels_build.py'], check=True, capture_output=True)
        subprocess.run([PY, '-m', 'src.ontology.populate'], check=True, capture_output=True)
        subprocess.run([PY, '-m', 'src.ontology.scenario_29'], check=True, capture_output=True)
        from . import acceptance_fsm
        acceptance_fsm.evaluate()          # 修泵登记后此处会带 repair_done 参数
        actions.append(f'数据月 {om} → {dm}: L1仓重建 + populate增量 + 沙盘刷新 + 验收状态机')
    else:
        actions.append(f'[dry] 检测到新数据月 {dm} (本体记 {om}), 未执行重跑')
    after = json.loads((ONT / 'objects.json').read_text(encoding='utf-8'))
    chg = diff_summary(before, after)
    if not dry:
        # roll 状态只在真跑时落库 — dry 落状态会把随后的真跑骗成空转 (2026-08-28 实逮自修)
        from .store import Store
        s = Store(ONT / 'objects.json')
        # F8: 滚动状态演进走 force 通道, 逻辑时钟=数据月 (旧滚动记录进 history)
        s.put(dict(id='meta/roll_state', type='Decision',
                   props=dict(name='月度滚动状态', data_month=dm, last_roll_actions=actions,
                              last_changes=chg or ['零变化'])),
              force=True, when=dm)
        s.save()
    return dict(data_month=dm, actions=actions, changes=chg or ['零变化'])


if __name__ == '__main__':
    r = roll(dry='--dry' in sys.argv)
    print(json.dumps(r, ensure_ascii=False, indent=1))