# -*- coding: utf-8 -*-
"""数据层: 离线导出 → 本系统视图. 只读 M5 产物 (各窗 index/spectra, 模型产物), 不重算诊断.
窗自动发现: config.windows (w0127 特殊路径) + m5/windows/*/index.parquet (新导入的窗自动进界面; 排除 exclude_windows)."""
import os, json, hashlib
import numpy as np
import pandas as pd
from .config import KEY_SCALARS, SENSORS

# 验证/测试窗不进生产集 (名字含 test / _reimport)。★2026-09-20 补 `_superseded`:
# 重算按"当前数据"替换同名窗时, 旧窗留档为 `<窗>_superseded_<时分>` —— 它是留档, 不该参与生产集。
EXCLUDE_DEFAULT = ('_reimport', '_superseded', 'test')


def windows(cfg):
    """{window: dict(index=Path, spectra_dir=Path|None)} 按名排序."""
    out = {}
    for w, p in cfg['windows'].items():
        if p.exists():
            d = p.parent if p.name == 'index.parquet' else None
            out[w] = dict(index=p, spectra_dir=(d / 'spectra') if d and (d / 'spectra').exists() else None)
    root = cfg['m5'] / 'windows'
    if root.exists():
        for d in sorted(root.iterdir()):
            if (d / 'index.parquet').exists() and d.name not in out and not any(x in d.name for x in cfg.get('exclude_windows', EXCLUDE_DEFAULT)):
                out[d.name] = dict(index=d / 'index.parquet', spectra_dir=(d / 'spectra') if (d / 'spectra').exists() else None)
    sel = os.environ.get('M5_WINDOWS')   # 与六层链脚本同一开关: 设了则只取该子集 (冲击轴/融合与 oem_scan/model_run 窗集合一致)
    if sel:
        keep = set(sel.split(','))
        out = {k: v for k, v in out.items() if k in keep}
    return dict(sorted(out.items()))


_SPAN_CACHE: dict = {}


def window_spans(cfg):
    """{窗: dict(rows, t_min, t_max, turbines, start_late)} —— 每个窗**真实覆盖的日期区间**。

    为什么要有它（用户令 2026-09-19 实逮）: 窗名 `w0316` 只是"窗起始日 03-16"的代号, 页面上到处只有
    这个名字与"证据窗末 2026-04-21", **看不出这个窗从哪天到哪天** —— 用户拿到的振动导出是
    `CMS_RuDong_CGN_202603-04`（2026-03 至 2026-04）, 于是他看到逐台页只有 4 月的谱, 就以为
    "3 月的数据没被呈现"。实测源件本身最早只到 **2026-03-16 17:03**（逐台起始不等: 最晚 03-30）,
    也就是说数据没错、是**呈现上没把覆盖区间说出来**。这里把区间算出来, 供报告/页面如实写明。

    只读每个窗索引的 `trigger_time` + `turbine` 两列（百万行级也只几秒）, 进程内缓存。
    """
    key = tuple(sorted(windows(cfg)))
    if _SPAN_CACHE.get('key') == key:
        return _SPAN_CACHE['val']
    out = {}
    for w, info in windows(cfg).items():
        try:
            d = pd.read_parquet(info['index'], columns=['trigger_time', 'turbine'])
            t = pd.to_datetime(d['trigger_time'], errors='coerce')
            per_first = t.groupby(d['turbine']).min()
            out[w] = dict(rows=int(len(d)), t_min=str(t.min()), t_max=str(t.max()),
                          turbines=int(d['turbine'].nunique()),
                          first_min=str(per_first.min()), first_max=str(per_first.max()),
                          start_late=bool(str(per_first.min())[:10] != str(per_first.max())[:10]))
        except Exception as e:
            out[w] = dict(rows=0, t_min='', t_max='', turbines=0, err=f'{type(e).__name__}: {e}')
    _SPAN_CACHE.update(key=key, val=out)
    return out


def span_text(cfg, win: str | None = None) -> str:
    """给人看的一句话: `w0316 2026-03-16 → 2026-04-21`（多窗时逗号分隔; 逐台起始不等时如实写出来）。"""
    sp = window_spans(cfg)
    ws = [win] if win else list(sp)
    parts = []
    for w in ws:
        s = sp.get(w) or {}
        if not s.get('rows'):
            continue
        rng = f"{str(s['t_min'])[:10]} → {str(s['t_max'])[:10]}"
        if s.get('start_late'):
            rng += f"（逐台起始不等: 最早 {str(s['first_min'])[:10]} · 最晚 {str(s['first_max'])[:10]}）"
        parts.append(f'{w} {rng}')
    return ' · '.join(parts)


def _read_index(p):
    cols = ['turbine', 'sensor_name', 'meas_name', 'trigger_time', 'rpm', 'condition_key', 'alarm_type', 'ds_size', 'scalar_value', 'overload']
    d = pd.read_parquet(p)
    return d[[c for c in cols if c in d.columns]]


def load_scalars(cfg, meas=KEY_SCALARS):
    """各窗标量 (ds_size==1) 拼接; 缓存按窗集合命名。

    ★ 2026-09-17 修: 产物被"清除产物"删掉后 `windows()` 返回空集 ⇒ 原来 `pd.concat([])` 直接抛
    `ValueError: No objects to concatenate`, **把整个 CMS 服务进程带走**(现场实测: 清产物后点"登录"
    → /detail/ 与 /cms/ 全挂)。现在空集返回**空表**(列齐、dtype 对), 页面照常起来显示"无产物";
    空结果不写缓存 (否则下次放了产物还会读到这份空缓存)。
    """
    ws = windows(cfg)
    key = hashlib.md5(('|'.join(ws) + '|' + ','.join(meas)).encode()).hexdigest()[:10]
    cache = cfg['out'] / 'cache' / f'scalars_{key}.parquet'
    if cache.exists():
        return pd.read_parquet(cache)
    parts = []
    for w, info in ws.items():
        d = _read_index(info['index'])
        d = d[(d.ds_size == 1) & d.meas_name.isin(meas) & d.sensor_name.isin(SENSORS)].copy()
        d['window'] = w
        parts.append(d)
    if not parts:
        return _empty_scalars()
    s = pd.concat(parts, ignore_index=True)
    s['trigger_time'] = pd.to_datetime(s.trigger_time)
    s['scalar_value'] = pd.to_numeric(s.scalar_value, errors='coerce')
    cache.parent.mkdir(parents=True, exist_ok=True)
    s.to_parquet(cache, index=False)
    return s


def _empty_scalars() -> pd.DataFrame:
    """无窗(无产物)时的空标量表 —— 列名与 _read_index 下的标量子集一致, 供下游直接 groupby/取值。"""
    cols = ['turbine', 'sensor_name', 'meas_name', 'trigger_time', 'rpm', 'condition_key',
            'alarm_type', 'ds_size', 'scalar_value', 'overload', 'window']
    d = pd.DataFrame({c: pd.Series(dtype='float64' if c in ('rpm', 'ds_size', 'scalar_value') else 'object')
                      for c in cols})
    d['trigger_time'] = pd.to_datetime(d['trigger_time'])
    return d


def load_alarm_counts(cfg):
    """各窗报警计数 —— 无窗时返回空表而不是 `pd.concat([])` 崩掉 (同上, 2026-09-17)。"""
    rows = []
    for w, info in windows(cfg).items():
        d = pd.read_parquet(info['index'], columns=['turbine', 'alarm_type'])
        c = d.groupby(['turbine', 'alarm_type']).size().unstack(fill_value=0)
        c['window'] = w
        rows.append(c.reset_index())
    if not rows:
        return pd.DataFrame(columns=['turbine', 'window'])
    return pd.concat(rows, ignore_index=True).fillna(0)


_META = {}


def _mtime_stamp(paths) -> str:
    """一组文件的 (路径, mtime_ns, size) 摘要 —— 用来判断"产物是否换过"。

    2026-09-16 用户令"重算后的产物可即时用于页面呈现": 本模块的 _META 是**进程内**缓存,
    长期跑着的服务 (windcms serve / 观澜组件) 在摄入新窗之后会一直用旧 meta (实测同类问题
    在 scripts/windscada_serve.py 的 _CACHE 上出现过)。把文件指纹并进缓存键即可自动失效,
    不必重启服务。"""
    h = hashlib.sha1()
    for p in paths:
        try:
            st = p.stat()
            h.update(f'{p}|{st.st_mtime_ns}|{st.st_size}\n'.encode('utf-8'))
        except OSError:
            h.update(f'{p}|MISSING\n'.encode('utf-8'))
    return h.hexdigest()[:12]


def spectra_meta(cfg):
    """合并谱元数据: 合并库 (cfg.spectra_dir, 覆盖 w0127) + 各窗 spectra 库; 加列 store (npz 所在目录).

    缓存键 = 产物目录 + **各窗 index/spectra_meta 的指纹** → 重算/新摄入后自动失效 (见 _mtime_stamp)。"""
    base = cfg['spectra_dir']
    wins = windows(cfg)
    stamp_files = [base / 'spectra_meta.parquet']
    for _w, info in wins.items():
        stamp_files.append(info['index'])
        if info['spectra_dir']:
            sd = info['spectra_dir']
            stamp_files.append(sd.parent / 'spectra_meta.parquet')
            stamp_files.append(sd / 'spectra_meta.parquet')
    key = f'{cfg["out"]}|{_mtime_stamp(stamp_files)}'
    if key in _META:
        return _META[key]
    parts = []
    if (base / 'spectra_meta.parquet').exists():
        m = pd.read_parquet(base / 'spectra_meta.parquet')
        m['store'] = str(base)
        m['window'] = 'w0127'
        parts.append(m)
    for w, info in wins.items():
        sd = info['spectra_dir']
        if not sd:
            continue
        mp = sd.parent / 'spectra_meta.parquet' if (sd.parent / 'spectra_meta.parquet').exists() else sd / 'spectra_meta.parquet'
        if mp.exists():
            m = pd.read_parquet(mp)
            m['store'] = str(sd)
            m['window'] = w
            parts.append(m)
    M = pd.concat(parts, ignore_index=True) if parts else pd.DataFrame(columns=['turbine', 'sensor', 'meas_name', 'trigger_time', 'store'])
    M['trigger_time'] = M['trigger_time'].astype(str)
    # 只留最近 4 个指纹键, 免得长期运行的服务把每代 meta 都攒在内存里
    if len(_META) > 4:
        _META.clear()
    _META[key] = M
    return M


def spectrum(cfg, turbine, sensor, meas_name, which='latest'):
    """取一条谱 (最新或指定 trigger_time) → (x_hz, y, meta)."""
    m = spectra_meta(cfg)
    q = m[(m.turbine == turbine) & (m.sensor == sensor) & (m.meas_name == meas_name)]
    if q.empty:
        return None
    r = q.sort_values('trigger_time').iloc[-1] if which == 'latest' else q[q.trigger_time == str(which)].iloc[0]
    shard = str(r.shard)
    npz = np.load(f"{r.store}/{shard if shard.endswith('.npz') else shard + '.npz'}")
    v = npz['values'][int(r.shard_row)]
    x = float(r.x_offset) + np.arange(len(v)) * float(r.x_delta)
    return x, v.astype(float), r.to_dict()


def load_model(cfg):
    l6 = pd.read_parquet(cfg['model_l6']) if cfg['model_l6'].exists() else pd.DataFrame()
    if len(l6) and '台' in l6.columns:
        # 归一化台号: 源文件补零 ('03#') 而消费端全用 f'{n}#' ('3#') — 不归一化则个位数台 L4 线整体查空
        # (2026-08-24 实逮: 03# 发电机 候选 16.15× 线因此丢失, 台被标优秀)
        l6['台'] = l6['台'].astype(str).str.replace(r'^0(\d#)$', r'\1', regex=True)
    fusion = pd.read_csv(cfg['fusion'], dtype=str) if cfg['fusion'].exists() else pd.DataFrame()
    summ = json.load(open(cfg['model_summary'], encoding='utf-8')) if cfg['model_summary'].exists() else {}
    return l6, fusion, summ


def load_findings(cfg):
    return json.load(open(cfg['findings'], encoding='utf-8')) if cfg['findings'].exists() else {}


def findings_for_turbine(findings, tid):
    n = int(tid[3:])
    toks = [tid, f'{n}#', f'{n:02d}#']
    out = []
    for k, v in findings.items():
        s = json.dumps(v, ensure_ascii=False)
        if any(t in s for t in toks):
            out.append(k)
    return out


def load_masks(cfg):
    p = cfg.get('mask_thresholds')
    if not p or not p.exists():
        return {}
    m = pd.read_csv(p, sep='\t', dtype=str, usecols=['state', 'measurement_key', 'condition_key', 'mask_time', 'value_median', 'yellow_scalar', 'red_scalar'])
    m = m[m.state == 'Active'].copy()
    for c in ['value_median', 'yellow_scalar', 'red_scalar']:
        m[c] = pd.to_numeric(m[c], errors='coerce')
    m = m.sort_values('mask_time').drop_duplicates(['measurement_key', 'condition_key'], keep='last')
    return {(r.measurement_key, r.condition_key): dict(median=r.value_median, yellow=r.yellow_scalar, red=r.red_scalar, mask_time=r.mask_time) for r in m.itertuples()}


def mask_key(farm_prefix, turbine, sensor, meas):
    return f'{farm_prefix}/{turbine}/{sensor}/{meas}'