Compressor/backend/openclose/sync_device_status.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
开机停机表合并 + 数据表 device_status 全量覆盖脚本。

职责（脚本内不含 DDL，建表/加列请先执行 schema_device_status.sql）：
  1. 解析 开机停机表/{7,8,9}号机.xls，将“按天逐条”记录合并成连续的开/关机时间周期；
  2. 重建 machine_run_status 表（DELETE 全部 + 重插，可重复执行）；
  3. 全量覆盖 pks_long_sample.device_status（import_batch_id 30/31/32 匹配 7/8/9 号机）；
  4. 全量覆盖 wave_file.device_status（按 device_part/point_name 文本识别机组后按时间匹配）。

用法：
  python sync_device_status.py                 # 全流程
  python sync_device_status.py --dry-run        # 只解析Excel并打印合并段，不写库
  python sync_device_status.py --smoke          # 冒烟：pks仅 batch=30 且 2025-04-01~04-05，wave同窗
  python sync_device_status.py --skip-pks --skip-wave   # 只重建开机停机表
"""
from __future__ import annotations

import argparse
import importlib.util
import math
import sys
from datetime import datetime, timedelta
from pathlib import Path

import xlrd
import pymysql
from pymysql.constants import CLIENT

ROOT = Path(__file__).resolve().parent
PROJECT_ROOT = ROOT.parents[1] if ROOT.name == "openclose" else ROOT
EXCEL_DIR = PROJECT_ROOT / "开机停机表"
STATUS_TABLE = "machine_run_status"

# 机组 -> (import_batch_id)
UNIT_BATCH = {7: 30, 8: 31, 9: 32}
UNIT_FILES = {u: f"{u}号机.xls" for u in (7, 8, 9)}

CHUNK_DAYS = 15          # pks 清零分块天数
PKS_GLOBAL_MIN = datetime(2025, 4, 1, 0, 0, 0)
# 上界取次日 00:00，保证把末行 2026-05-01 23:59:55 也纳入 [min,max) 更新范围
PKS_GLOBAL_MAX = datetime(2026, 5, 2, 0, 0, 0)


def load_settings():
    spec = importlib.util.spec_from_file_location("app_config", PROJECT_ROOT / "backend" / "app" / "config.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod.settings


def connect(settings):
    return pymysql.connect(
        host=settings.db_host, port=settings.db_port, user=settings.db_user,
        password=settings.db_password, database=settings.db_name, charset="utf8mb4",
        autocommit=False, client_flag=CLIENT.FOUND_ROWS,
        cursorclass=pymysql.cursors.DictCursor,
    )


# ---------------------------------------------------------------- Excel 解析
def parse_xls(path: Path):
    """返回该机 (unit_no, [记录])，记录为 dict；同时收集问题警告。"""
    wb = xlrd.open_workbook(str(path))
    sh = wb.sheet_by_index(0)
    rows = [sh.row_values(r) for r in range(sh.nrows)]
    header = [str(x).strip() for x in rows[0]]
    idx = {name: i for i, name in enumerate(header)}
    unit_no = int(path.stem.replace("号机", ""))
    records = []
    problems = []

    for r_i, raw in enumerate(rows[1:], start=2):
        if not raw or not any(str(c).strip() for c in raw):
            continue
        status_raw = str(raw[idx["设备状态"]]).strip()
        if status_raw not in ("开机", "关机"):
            problems.append(f"行{r_i}：无法识别状态 {status_raw!r}，已跳过")
            continue
        try:
            base = xlrd.xldate_as_datetime(raw[idx["时间"]], wb.datemode)
        except Exception:
            problems.append(f"行{r_i}：时间无法解析 {raw[idx['时间']]!r}，已跳过")
            continue
        tp_raw = raw[idx["时间点"]] if idx["时间点"] is not None else 1.0
        try:
            tp = float(tp_raw)
        except (TypeError, ValueError):
            tp = 1.0
            problems.append(f"行{r_i}：时间点无法解析 {tp_raw!r}，按 24:00 处理")
        if tp >= 1.0:
            dt = base + timedelta(days=1)
        else:
            dt = base + timedelta(seconds=int(round(tp * 86400)))
        start_cat = str(raw[idx["启动类别"]]).strip() if idx["启动类别"] is not None else ""
        stop_cat = str(raw[idx["停机类别"]]).strip() if idx["停机类别"] is not None else ""
        if start_cat == "nan":
            start_cat = ""
        if stop_cat == "nan":
            stop_cat = ""
        try:
            cum = float(raw[idx["累计运行时间"]])
        except (TypeError, ValueError):
            cum = None
        records.append({
            "unit": unit_no,
            "dt": dt,
            "status": 1 if status_raw == "开机" else 0,
            "start_cat": start_cat,
            "stop_cat": stop_cat,
            "cum": cum,
        })

    records.sort(key=lambda r: r["dt"])
    return unit_no, records, problems


def build_periods(records):
    """把逐条记录合并为连续状态周期。

    状态只在“相邻记录状态发生变化”处切段；状态相同则继续合并。
    最后一个周期结束于 Excel 末行时刻（状态结束时间截到末行）。
    """
    periods = []
    changes = []          # 状态发生变化的记录（含首条），作为每段的开启点
    for rec in records:
        if not changes or rec["status"] != changes[-1]["status"]:
            changes.append(rec)
    if not changes:
        return periods
    last_time = records[-1]["dt"]
    for i, chg in enumerate(changes):
        start = chg["dt"]
        end = changes[i + 1]["dt"] if i + 1 < len(changes) else last_time
        if end <= start:
            continue
        status = chg["status"]
        cat = chg["start_cat"] if status == 1 else chg["stop_cat"]
        cum_vals = [r["cum"] for r in records if start <= r["dt"] < end and r["cum"] is not None]
        cum = max(cum_vals) if cum_vals else None
        run_hours = round((end - start).total_seconds() / 3600.0, 2) if status == 1 else 0.0
        periods.append({
            "unit": chg["unit"],
            "import_batch_id": UNIT_BATCH[chg["unit"]],
            "status": status,
            "status_start": start,
            "status_end": end,
            "start_category": cat or None,
            "run_hours": run_hours,
            "cum_run_hours": round(cum, 2) if cum is not None else None,
        })
    return periods


def validate_periods(periods):
    warns = []
    per_unit = {}
    for p in periods:
        per_unit.setdefault(p["unit"], []).append(p)
    for u, ps in sorted(per_unit.items()):
        on_days = sum((p["status_end"] - p["status_start"]).total_seconds() for p in ps if p["status"] == 1) / 86400.0
        off_days = sum((p["status_end"] - p["status_start"]).total_seconds() for p in ps if p["status"] == 0) / 86400.0
        prev = None
        for p in ps:
            if prev and p["status"] == prev["status"]:
                warns.append(f"{u}号机：相邻两段状态未交替 {prev['status_start']}->{p['status_start']}")
            if prev and p["status_start"] < prev["status_end"]:
                warns.append(f"{u}号机：时间段重叠 {prev['status_start']}..{prev['status_end']} 与 {p['status_start']}")
            prev = p
        print(f"  {u}号机：{len(ps)} 段，开机 {on_days:.1f} 天 / 关机 {off_days:.1f} 天，"
              f"区间 {ps[0]['status_start']:%Y-%m-%d %H:%M} ~ {ps[-1]['status_end']:%Y-%m-%d %H:%M}")
    return warns


# ---------------------------------------------------------------- DB 更新
def rebuild_status_table(conn, periods):
    cur = conn.cursor()
    cur.execute(f"DELETE FROM {STATUS_TABLE}")
    sql = (f"INSERT INTO {STATUS_TABLE} "
           f"(unit_no, import_batch_id, status, status_start, status_end, "
           f"start_category, run_hours, cum_run_hours) "
           f"VALUES (%(unit)s, %(import_batch_id)s, %(status)s, %(status_start)s, "
           f"%(status_end)s, %(start_category)s, %(run_hours)s, %(cum_run_hours)s)")
    data = [
        {
            "unit": p["unit"],
            "import_batch_id": p["import_batch_id"],
            "status": p["status"],
            "status_start": p["status_start"],
            "status_end": p["status_end"],
            "start_category": p["start_category"],
            "run_hours": p["run_hours"],
            "cum_run_hours": p["cum_run_hours"],
        }
        for p in periods
    ]
    cur.executemany(sql, data)
    conn.commit()
    print(f"machine_run_status 重建完成，共 {cur.rowcount} 条（已先清空）")


def fetch_periods(conn):
    cur = conn.cursor()
    cur.execute(f"SELECT unit_no, status, status_start, status_end FROM {STATUS_TABLE} "
                f"ORDER BY unit_no, status_start")
    per = {}
    for r in cur.fetchall():
        per.setdefault(r["unit_no"], []).append(r)
    return per


def _clip(s, e, lo, hi):
    return max(s, lo), min(e, hi)


def _grid_count(a, b):
    """区间 [a,b) 内落在 5 秒采样网格上的点数（网格=epoch秒可被5整除）。"""
    a_s, b_s = int(a.timestamp()), int(b.timestamp())
    k0 = math.ceil(a_s / 5)
    k1 = math.floor((b_s - 1) / 5)
    return max(0, k1 - k0 + 1)


def update_pks(conn, per, lo=None, hi=None, only_batch=None, chunk_days=CHUNK_DAYS):
    cur = conn.cursor()
    lo = lo or PKS_GLOBAL_MIN
    hi = hi or PKS_GLOBAL_MAX
    for unit in sorted(UNIT_BATCH):
        batch = UNIT_BATCH[unit]
        if only_batch is not None and batch != only_batch:
            continue
        on_periods = [p for p in per.get(unit, []) if p["status"] == 1]
        t = lo
        n_clear = 0
        while t < hi:
            c1 = min(t + timedelta(days=chunk_days), hi)
            cur.execute("UPDATE pks_long_sample SET device_status=0 "
                        "WHERE import_batch_id=%s AND sample_time>=%s AND sample_time<%s",
                        (batch, t, c1))
            n_clear += cur.rowcount
            conn.commit()
            t = c1
        expected = 0
        for p in on_periods:
            s, e = _clip(p["status_start"], p["status_end"], lo, hi)
            if e <= s:
                continue
            expected += _grid_count(s, e)
            cur.execute("UPDATE pks_long_sample SET device_status=1 "
                        "WHERE import_batch_id=%s AND sample_time>=%s AND sample_time<%s",
                        (batch, s, e))
            conn.commit()
        cur.execute("SELECT COUNT(*) n FROM pks_long_sample WHERE import_batch_id=%s AND device_status=1",
                    (batch,))
        actual = cur.fetchone()["n"]
        print(f"  pks batch{batch}({unit}号机)：清零 {n_clear} 行，置1后 {actual} 行（理论 {expected}，差 {actual - expected}）")


def _unit_expr():
    """由文本识别机组（7/8/9号机）；识别不到返回 NULL。"""
    p = "CONCAT(IFNULL(device_part,''),' ',IFNULL(point_name,''),' ',IFNULL(device_code,''))"
    return (
        "CASE "
        f"WHEN {p} LIKE '%9号机组%' THEN 9 "
        f"WHEN {p} LIKE '%8号机组%' THEN 8 "
        f"WHEN {p} LIKE '%7号机组%' THEN 7 "
        "ELSE NULL END"
    )


def update_wave(conn, per, lo=None, hi=None):
    cur = conn.cursor()
    if lo and hi:
        wnd = "sample_time>=%s AND sample_time<%s"
        args = (lo, hi)
        cur.execute("UPDATE wave_file SET device_status=0 WHERE " + wnd, args)
    else:
        args = ()
        cur.execute("UPDATE wave_file SET device_status=0")
    print(f"  wave_file 已清零 {cur.rowcount} 行")
    conn.commit()

    # 注：SQL 内含字面 %(LIKE)，不能用 %s 参数格式化，窗口条件直接内联
    expr = _unit_expr()
    sql_set = (
        "UPDATE wave_file w "
        "JOIN machine_run_status m ON m.status=1 "
        "  AND w.sample_time>=m.status_start AND w.sample_time<m.status_end "
        f"  AND ({expr})=m.unit_no "
        "SET w.device_status=1 "
    )
    if lo and hi:
        sql_set += (f"WHERE w.sample_time>='{lo:%Y-%m-%d %H:%M:%S}' "
                    f"AND w.sample_time<'{hi:%Y-%m-%d %H:%M:%S}' ")
    cur.execute(sql_set)
    conn.commit()
    print(f"  wave_file 置1完成 {cur.rowcount} 行")


# ---------------------------------------------------------------- main
def main():
    ap = argparse.ArgumentParser(description="开机停机表合并 + device_status 全量覆盖")
    ap.add_argument("--dry-run", action="store_true", help="只解析Excel并打印，不写库")
    ap.add_argument("--smoke", action="store_true", help="冒烟：pks仅batch=30，窗口2025-04-01~04-05")
    ap.add_argument("--skip-status-table", action="store_true", help="跳过重建 machine_run_status")
    ap.add_argument("--skip-pks", action="store_true", help="跳过 pks_long_sample 更新")
    ap.add_argument("--skip-wave", action="store_true", help="跳过 wave_file 更新")
    args = ap.parse_args()

    all_periods = []
    all_problems = []
    for u in (7, 8, 9):
        f = EXCEL_DIR / UNIT_FILES[u]
        if not f.exists():
            print(f"[跳过] 缺少 {f.name}")
            continue
        unit, records, problems = parse_xls(f)
        all_problems += [f"{unit}号机：{x}" for x in problems]
        periods = build_periods(records)
        all_periods += periods
        print(f"{unit}号机.xls：{len(records)} 条原始记录 -> {len(periods)} 个合并周期")
    for p in all_problems:
        print("  [警告]", p)
    if not all_periods:
        print("没有可用的合并周期，退出")
        sys.exit(1)

    print("---- 合并结果预览（每台 前2/后2 段）----")
    for p in all_periods[:2]:
        print("  ", p)
    print("   ...")
    for p in all_periods[-2:]:
        print("  ", p)

    print("---- 连续性/交替性校验 ----")
    for w in validate_periods(all_periods):
        print("  [警告]", w)

    if args.dry_run:
        print("[dry-run] 结束，未写库")
        return

    settings = load_settings()
    conn = connect(settings)
    try:
        if not args.skip_status_table:
            rebuild_status_table(conn, all_periods)
        per = fetch_periods(conn)

        smoke_lo = smoke_hi = None
        if args.smoke:
            smoke_lo = datetime(2025, 4, 29)
            smoke_hi = datetime(2025, 5, 4)
            print("---- 冒烟模式：pks batch=30，窗口 2025-04-29 ~ 2025-05-04 ----")

        if not args.skip_pks:
            print("---- 更新 pks_long_sample.device_status ----")
            update_pks(conn, per, lo=smoke_lo, hi=smoke_hi, only_batch=30 if args.smoke else None)

        if not args.skip_wave:
            print("---- 更新 wave_file.device_status ----")
            update_wave(conn, per, lo=smoke_lo, hi=smoke_hi)
    finally:
        conn.close()
    print("完成")


if __name__ == "__main__":
    main()