feat: 实现干扰指标最新小时入库

main.py

@@ -14,7 +14,7 @@ import re
 import shutil
 import sys
 import time
-from typing import Protocol
+from typing import Callable, Protocol
 from urllib.error import HTTPError, URLError
 from urllib.parse import quote, urlencode
 from urllib.request import Request, urlopen
@@ -138,6 +138,9 @@ SUMMARY_HEADER = (
     "source_path",
 )
 
+DEFAULT_DB_TABLE = "interference_hourly_summary"
+DB_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9_]+$")
+
 
 class ProcessingError(RuntimeError):
     pass
@@ -157,6 +160,134 @@ class Source(Protocol):
     def download(self, path: str) -> bytes: ...
 
 
+class SummaryStore(Protocol):
+    def replace_latest(self, rows: list[dict[str, str]]) -> dict[str, object]: ...
+
+
+class MySQLSummaryStore:
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        user: str,
+        password: str,
+        database: str,
+        table: str = DEFAULT_DB_TABLE,
+        connect_factory: Callable[[], object] | None = None,
+    ) -> None:
+        if not DB_IDENTIFIER_RE.fullmatch(table):
+            raise ProcessingError(f"Invalid database table name: {table!r}")
+        self.host = host
+        self.port = port
+        self.user = user
+        self.password = password
+        self.database = database
+        self.table = table
+        self.connect_factory = connect_factory
+
+    def replace_latest(self, rows: list[dict[str, str]]) -> dict[str, object]:
+        if not rows:
+            raise ProcessingError("Refusing to replace database data with an empty batch")
+        metric_times = {datetime.strptime(row["hour_start"], "%Y-%m-%d %H:%M:%S") for row in rows}
+        if len(metric_times) != 1:
+            raise ProcessingError("Database batch must contain exactly one metric hour")
+        metric_time = next(iter(metric_times))
+        table = f"`{self.table}`"
+        connection = self._connect()
+        cursor = None
+        try:
+            cursor = connection.cursor()
+            cursor.execute(
+                f"""
+                CREATE TABLE IF NOT EXISTS {table} (
+                    metric_time DATETIME NOT NULL COMMENT '指标开始时间',
+                    source_type VARCHAR(64) NOT NULL,
+                    cgi VARCHAR(128) NOT NULL,
+                    cell_name VARCHAR(255) NOT NULL,
+                    interference_dbm DECIMAL(10,3) NOT NULL,
+                    source_path VARCHAR(1024) NOT NULL,
+                    PRIMARY KEY (metric_time, source_type, cgi)
+                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
+                """
+            )
+            cursor.execute(f"SELECT MAX(metric_time) FROM {table}")
+            latest_row = cursor.fetchone()
+            latest_time = latest_row[0] if latest_row else None
+            if isinstance(latest_time, str):
+                latest_time = datetime.fromisoformat(latest_time)
+            if latest_time is not None and latest_time > metric_time:
+                raise ProcessingError(
+                    f"Database already contains newer metric time {latest_time:%Y-%m-%d %H:%M:%S}; "
+                    f"refusing to replace it with {metric_time:%Y-%m-%d %H:%M:%S}"
+                )
+
+            cursor.execute(f"DELETE FROM {table} WHERE metric_time = %s", (metric_time,))
+            refreshed_rows = cursor.rowcount
+            cursor.executemany(
+                f"""
+                INSERT INTO {table}
+                    (metric_time, source_type, cgi, cell_name, interference_dbm, source_path)
+                VALUES (%s, %s, %s, %s, %s, %s)
+                """,
+                [
+                    (
+                        metric_time,
+                        row["source_type"],
+                        row["cgi"],
+                        row["cell_name"],
+                        row["interference_dbm"],
+                        row["source_path"],
+                    )
+                    for row in rows
+                ],
+            )
+            cursor.execute(f"DELETE FROM {table} WHERE metric_time <> %s", (metric_time,))
+            old_rows_deleted = cursor.rowcount
+            connection.commit()
+        except ProcessingError:
+            connection.rollback()
+            raise
+        except Exception as exc:
+            connection.rollback()
+            raise ProcessingError(f"MySQL write failed: {exc}") from exc
+        finally:
+            if cursor is not None:
+                cursor.close()
+            connection.close()
+
+        return {
+            "enabled": True,
+            "table": self.table,
+            "metric_time": metric_time.strftime("%Y-%m-%d %H:%M:%S"),
+            "inserted_rows": len(rows),
+            "refreshed_rows": refreshed_rows,
+            "old_rows_deleted": old_rows_deleted,
+        }
+
+    def _connect(self) -> object:
+        if self.connect_factory is not None:
+            return self.connect_factory()
+        try:
+            import pymysql
+        except ImportError as exc:
+            raise ProcessingError("PyMySQL is required for database output") from exc
+        try:
+            return pymysql.connect(
+                host=self.host,
+                port=self.port,
+                user=self.user,
+                password=self.password,
+                database=self.database,
+                charset="utf8mb4",
+                autocommit=False,
+                connect_timeout=15,
+                read_timeout=60,
+                write_timeout=60,
+            )
+        except Exception as exc:
+            raise ProcessingError(f"MySQL connection failed: {exc}") from exc
+
+
 class ApiSource:
     def __init__(self, base_url: str, token: str, storage_id: str, root: str) -> None:
         if not token:
@@ -302,7 +433,14 @@ def parse_workbook(raw_zip: bytes, source_type: str) -> tuple[tuple[str, ...], l
         workbook.close()
 
 
-def process(source: Source, output_root: Path, lookback_days: int, requested_window: str = "", plmn: str = "460-00") -> Path:
+def process(
+    source: Source,
+    output_root: Path,
+    lookback_days: int,
+    requested_window: str = "",
+    plmn: str = "460-00",
+    store: SummaryStore | None = None,
+) -> Path:
     candidates = source.candidates(lookback_days)
     window, selected, warnings_out = select_window(candidates, requested_window)
     temp_dir = output_root.resolve() / f".{window}.tmp-{os.getpid()}"
@@ -315,6 +453,7 @@ def process(source: Source, output_root: Path, lookback_days: int, requested_win
 
     summary_rows: list[dict[str, str]] = []
     manifest_files: list[dict[str, object]] = []
+    database_result: dict[str, object] = {"enabled": False}
     try:
         for source_type in EXPECTED_TYPES:
             candidate = selected[source_type]
@@ -340,6 +479,8 @@ def process(source: Source, output_root: Path, lookback_days: int, requested_win
         summary_rows.sort(key=lambda item: (EXPECTED_TYPES.index(item["source_type"]), item["cgi"], item["cell_name"]))
         summary_name = f"interference_summary_{window}.csv"
         write_dict_csv(temp_dir / summary_name, SUMMARY_HEADER, summary_rows)
+        if store is not None:
+            database_result = store.replace_latest(summary_rows)
         manifest = {
             "generated_at": datetime.now(timezone.utc).isoformat(),
             "window": window,
@@ -351,6 +492,7 @@ def process(source: Source, output_root: Path, lookback_days: int, requested_win
             "warnings": warnings_out,
             "files": manifest_files,
             "summary_csv": summary_name,
+            "database": database_result,
         }
         (temp_dir / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
         output_root.resolve().mkdir(parents=True, exist_ok=True)
@@ -365,6 +507,10 @@ def process(source: Source, output_root: Path, lookback_days: int, requested_win
     print(f"selected_window={window}")
     print(f"source_files={len(manifest_files)}")
     print(f"summary_rows={len(summary_rows)}")
+    if database_result["enabled"]:
+        print(f"database_metric_time={database_result['metric_time']}")
+        print(f"database_inserted_rows={database_result['inserted_rows']}")
+        print(f"database_old_rows_deleted={database_result['old_rows_deleted']}")
     for message in warnings_out:
         print(f"warning={message}")
     print(f"output={final_dir}")
@@ -471,15 +617,45 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--storage-id", default=os.getenv("METRIX_STORAGE_ID", DEFAULT_STORAGE_ID))
     parser.add_argument("--root", default=os.getenv("INTERFERENCE_SOURCE_ROOT", DEFAULT_ROOT))
     parser.add_argument("--plmn", default=os.getenv("INTERFERENCE_PLMN", "460-00"))
+    parser.add_argument("--no-database", action="store_true", help="Generate files without writing MySQL")
+    parser.add_argument("--db-host", default=os.getenv("INTERFERENCE_DB_HOST", ""))
+    parser.add_argument("--db-port", type=int, default=int(os.getenv("INTERFERENCE_DB_PORT", "3306")))
+    parser.add_argument("--db-user", default=os.getenv("INTERFERENCE_DB_USER", ""))
+    parser.add_argument("--db-password", default=os.getenv("INTERFERENCE_DB_PASSWORD", ""))
+    parser.add_argument("--db-name", default=os.getenv("INTERFERENCE_DB_NAME", ""))
+    parser.add_argument("--db-table", default=os.getenv("INTERFERENCE_DB_TABLE", DEFAULT_DB_TABLE))
     return parser
 
 
+def mysql_store_from_args(args: argparse.Namespace) -> MySQLSummaryStore:
+    required_values = {
+        "INTERFERENCE_DB_HOST": args.db_host,
+        "INTERFERENCE_DB_USER": args.db_user,
+        "INTERFERENCE_DB_PASSWORD": args.db_password,
+        "INTERFERENCE_DB_NAME": args.db_name,
+    }
+    missing = [name for name, value in required_values.items() if not value]
+    if missing:
+        raise ProcessingError(f"Missing database settings: {', '.join(missing)}; use --no-database for file-only runs")
+    if not 1 <= args.db_port <= 65535:
+        raise ProcessingError("db-port must be between 1 and 65535")
+    return MySQLSummaryStore(
+        host=args.db_host,
+        port=args.db_port,
+        user=args.db_user,
+        password=args.db_password,
+        database=args.db_name,
+        table=args.db_table,
+    )
+
+
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
     if args.lookback_days < 1:
         raise ProcessingError("lookback-days must be at least 1")
     source: Source = LocalSource(args.source_dir) if args.source_dir else ApiSource(args.api_base, args.api_token, args.storage_id, args.root)
-    process(source, args.output_dir, args.lookback_days, args.window, args.plmn)
+    store = None if args.no_database else mysql_store_from_args(args)
+    process(source, args.output_dir, args.lookback_days, args.window, args.plmn, store)
     return 0