Compressor/backend/indicator_features/audit_wave_metadata.py

#!/usr/bin/env python3
"""Read-only inventory of indicator-diagram files; does not fetch sample points."""
from __future__ import annotations

import argparse
import csv
import sys
from pathlib import Path

import pymysql

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))

from app.config import settings  # noqa: E402


def connect():
    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",
        cursorclass=pymysql.cursors.DictCursor,
        connect_timeout=8,
        read_timeout=120,
    )


def main() -> None:
    parser = argparse.ArgumentParser(description="只读盘点功图文件元数据，不读取采样点")
    parser.add_argument("--units", nargs="+", default=["7", "8", "9"])
    parser.add_argument("--out", default="cache/indicator/元数据盘点.csv")
    args = parser.parse_args()

    unit_like = [f"{unit}号机组%" for unit in args.units]
    connection = connect()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT
                    CASE
                      WHEN point_name LIKE '7号机组%%' THEN '7'
                      WHEN point_name LIKE '8号机组%%' THEN '8'
                      WHEN point_name LIKE '9号机组%%' THEN '9'
                    END AS unit,
                    DATE(sample_time) AS day,
                    COUNT(*) AS files,
                    SUM(CASE WHEN rpm > 0 THEN 1 ELSE 0 END) AS running_files,
                    MIN(sample_time) AS first_time,
                    MAX(sample_time) AS last_time
                FROM wave_file
                WHERE measurement_type = '压力'
                  AND (point_name LIKE %s OR point_name LIKE %s OR point_name LIKE %s)
                GROUP BY unit, DATE(sample_time)
                ORDER BY unit, day
                """,
                tuple(unit_like),
            )
            daily = list(cursor.fetchall())

            cursor.execute(
                """
                SELECT
                    CASE
                      WHEN point_name LIKE '7号机组%%' THEN '7'
                      WHEN point_name LIKE '8号机组%%' THEN '8'
                      WHEN point_name LIKE '9号机组%%' THEN '9'
                    END AS unit,
                    COUNT(*) AS files,
                    SUM(CASE WHEN rpm > 0 THEN 1 ELSE 0 END) AS running_files,
                    COUNT(DISTINCT point_name) AS points,
                    MIN(sample_time) AS first_time,
                    MAX(sample_time) AS last_time
                FROM wave_file
                WHERE measurement_type = '压力'
                  AND (point_name LIKE %s OR point_name LIKE %s OR point_name LIKE %s)
                GROUP BY unit
                ORDER BY unit
                """,
                tuple(unit_like),
            )
            totals = list(cursor.fetchall())

            cursor.execute(
                """
                SELECT measurement_type, COUNT(*) AS files
                FROM wave_file
                WHERE point_name LIKE %s OR point_name LIKE %s OR point_name LIKE %s
                GROUP BY measurement_type
                ORDER BY measurement_type
                """,
                tuple(unit_like),
            )
            types = list(cursor.fetchall())
    finally:
        connection.close()

    selected = {str(row["unit"]) for row in totals}
    totals = [row for row in totals if str(row["unit"]) in selected]
    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    with out.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=["unit", "day", "files", "running_files", "first_time", "last_time"],
        )
        writer.writeheader()
        writer.writerows(daily)

    print("功图文件元数据盘点完成（没有读取 wave_sample_one）")
    print(f"按天明细：{out}，共 {len(daily):,} 个有记录的机组-日期")
    print("\n按机组汇总：")
    for row in totals:
        print(
            f"  {row['unit']}号：文件 {row['files']:,}，运行文件 {row['running_files']:,}，"
            f"测点 {row['points']}，时间 {row['first_time']} ~ {row['last_time']}"
        )
    print("\n测量类型汇总：")
    for row in types:
        print(f"  {row['measurement_type']}: {row['files']:,}")


if __name__ == "__main__":
    main()