feat: 新增 CellData 数据源和数据库配置

2026-06-24 23:31:04 +08:00
parent 0a4012ad4e
commit c3dcb3f5b1
6 changed files with 599 additions and 82 deletions
@@ -3,18 +3,22 @@ from datetime import datetime
 from typing import Any
 from urllib.parse import quote

+import pymysql
 from fastapi import APIRouter, Body, HTTPException, UploadFile, File
 from fastapi.responses import Response

 from app import state
 from app.config import (
+    CellDataConfig,
    DataMappingsConfig,
    HistoryRetentionConfig,
    MetrixConfig,
+    MySQLConfig,
    RemoteDataConfig,
    SOURCE_TYPES,
    WAREHOUSE_TYPES,
 )
+from app.services.remote_download import RemoteDataDownloader


 router = APIRouter(tags=["config"])
@@ -92,6 +96,46 @@ async def update_data_mappings_config(config: dict[str, Any] = Body(...)):
    return {"success": True, "message": "数据目录映射已更新", "update": state.config.update}


+@router.post("/api/config/cell-data/remote")
+async def update_cell_data_remote_config(config: dict[str, Any] = Body(...)):
+    state.reload_config()
+    current = state.config.cell_data.normalized()
+    state.config.cell_data = CellDataConfig(
+        remote_data=RemoteDataConfig.from_dict(config),
+        mysql=current.mysql,
+    ).normalized()
+    state.config.save()
+    return {"success": True, "message": "CellData 远程数据源配置已更新", "update": state.config.update}
+
+
+@router.post("/api/config/cell-data/mysql")
+async def update_cell_data_mysql_config(config: dict[str, Any] = Body(...)):
+    state.reload_config()
+    current = state.config.cell_data.normalized()
+    state.config.cell_data = CellDataConfig(
+        remote_data=current.remote_data,
+        mysql=MySQLConfig.from_dict(config, default_dbname="celldata"),
+    ).normalized()
+    state.config.save()
+    return {"success": True, "message": "CellData 数据库配置已更新", "update": state.config.update}
+
+
+@router.post("/api/config/cell-data/remote/test")
+async def test_cell_data_remote_connection(config: dict[str, Any] | None = Body(None)):
+    try:
+        remote_config = RemoteDataConfig.from_dict(config) if config else state.current_config().cell_data.remote_data
+        RemoteDataDownloader(remote_config).test_connection()
+        return {"success": True, "message": "CellData 远程服务器连接成功"}
+    except Exception as exc:
+        return {"success": False, "message": f"连接失败: {exc}"}
+
+
+@router.post("/api/config/cell-data/mysql/test")
+async def test_cell_data_mysql_connection(config: dict[str, Any] | None = Body(None)):
+    mysql_config = MySQLConfig.from_dict(config, default_dbname="celldata") if config else state.current_config().cell_data.mysql
+    return _test_mysql_config(mysql_config)
+
+
@router.post("/api/config/history-retention")
 async def update_history_retention(config: dict[str, Any] = Body(...)):
    state.reload_config()
@@ -167,9 +211,7 @@ def _apply_config_data(data: dict[str, Any]) -> None:

    mysql_data = data.get("MySQL_DBInfo")
    if isinstance(mysql_data, dict):
-        for key in ("host", "port", "user", "passwd", "dbname"):
-            if key in mysql_data:
-                setattr(state.config.mysql, key, mysql_data[key])
+        state.config.mysql = MySQLConfig.from_dict(mysql_data)

    if "SheetFilter" in data:
        state.config.sheet_filter = data["SheetFilter"] if isinstance(data["SheetFilter"], list) else []
@@ -181,7 +223,34 @@ def _apply_config_data(data: dict[str, Any]) -> None:
    if isinstance(remote_data, dict):
        state.config.remote_data = RemoteDataConfig.from_dict(remote_data)

+    cell_data = data.get("CellData")
+    if isinstance(cell_data, dict):
+        state.config.cell_data = CellDataConfig.from_dict(cell_data)
+
    history_retention = data.get("HistoryRetention")
    if isinstance(history_retention, dict):
        state.config.history_retention = HistoryRetentionConfig.from_dict(history_retention)

+
+def _test_mysql_config(config: MySQLConfig) -> dict[str, Any]:
+    normalized = config.normalized()
+    try:
+        conn = pymysql.connect(
+            host=normalized.host,
+            port=normalized.port,
+            user=normalized.user,
+            password=normalized.passwd,
+            database=normalized.dbname,
+            charset="utf8mb4",
+            cursorclass=pymysql.cursors.DictCursor,
+            connect_timeout=10,
+        )
+        try:
+            with conn.cursor() as cursor:
+                cursor.execute("SELECT 1")
+        finally:
+            conn.close()
+        return {"success": True, "message": "连接成功"}
+    except Exception as exc:
+        return {"success": False, "message": str(exc)}
+
@@ -33,6 +33,42 @@ class MySQLConfig:
    passwd: str = ""
    dbname: str = "CapacityReport"

+    def normalized(self) -> "MySQLConfig":
+        try:
+            port = int(self.port or 3306)
+        except (TypeError, ValueError):
+            port = 3306
+        return MySQLConfig(
+            host=str(self.host or "").strip() or "localhost",
+            port=port,
+            user=str(self.user or "").strip(),
+            passwd=self.passwd or "",
+            dbname=str(self.dbname or "").strip(),
+        )
+
+    def to_dict(self, include_password: bool = False) -> Dict[str, Any]:
+        normalized = self.normalized()
+        data = {
+            "host": normalized.host,
+            "port": normalized.port,
+            "user": normalized.user,
+            "dbname": normalized.dbname,
+        }
+        if include_password:
+            data["passwd"] = normalized.passwd
+        return data
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any] | None, default_dbname: str = "CapacityReport") -> "MySQLConfig":
+        data = data or {}
+        return cls(
+            host=str(data.get("host", "localhost")),
+            port=data.get("port", 3306),
+            user=str(data.get("user", "root")),
+            passwd=str(data.get("passwd", "")),
+            dbname=str(data.get("dbname", default_dbname)),
+        ).normalized()
+

 # Source backends: direct FTP/SFTP, or Metrix storage platform.
 SOURCE_TYPES = ("ftp", "sftp", "metrix")
@@ -292,6 +328,43 @@ class RemoteDataConfig:
        ).normalized()


+@dataclass
+class CellDataConfig:
+    """CellData side data source and database connection used by later processing steps."""
+    remote_data: RemoteDataConfig = field(
+        default_factory=lambda: RemoteDataConfig(
+            enabled=False,
+            protocol="sftp",
+            remote_dir="/CellData",
+        )
+    )
+    mysql: MySQLConfig = field(default_factory=lambda: MySQLConfig(dbname="celldata"))
+
+    def normalized(self) -> "CellDataConfig":
+        return CellDataConfig(
+            remote_data=self.remote_data.normalized(),
+            mysql=self.mysql.normalized(),
+        )
+
+    def to_dict(self, include_password: bool = False) -> Dict[str, Any]:
+        normalized = self.normalized()
+        return {
+            "remote_data": normalized.remote_data.to_dict(include_password=include_password),
+            "mysql": normalized.mysql.to_dict(include_password=include_password),
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any] | None) -> "CellDataConfig":
+        data = data or {}
+        default = cls()
+        remote_data = data.get("RemoteData") or data.get("remote_data")
+        mysql_data = data.get("MySQL_DBInfo") or data.get("mysql")
+        return cls(
+            remote_data=RemoteDataConfig.from_dict(remote_data) if isinstance(remote_data, dict) else default.remote_data,
+            mysql=MySQLConfig.from_dict(mysql_data, default_dbname="celldata") if isinstance(mysql_data, dict) else default.mysql,
+        ).normalized()
+
+
@dataclass
 class HistoryRetentionConfig:
    enabled: bool = False
@@ -346,6 +419,7 @@ class AppConfig:
    metrix: MetrixConfig = field(default_factory=MetrixConfig)
    data_mappings: DataMappingsConfig = field(default_factory=DataMappingsConfig)
    remote_data: RemoteDataConfig = field(default_factory=RemoteDataConfig)
+    cell_data: CellDataConfig = field(default_factory=CellDataConfig)
    history_retention: HistoryRetentionConfig = field(default_factory=HistoryRetentionConfig)
    sheet_filter: List[str] = field(default_factory=list)
    extract_fields: List[Dict[str, Any]] = field(default_factory=list)
@@ -359,17 +433,11 @@ class AppConfig:
        with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
            data = json.load(f)

-        mysql_data = data.get("MySQL_DBInfo", {})
-        mysql_config = MySQLConfig(
-            host=mysql_data.get("host", "localhost"),
-            port=mysql_data.get("port", 3306),
-            user=mysql_data.get("user", "root"),
-            passwd=mysql_data.get("passwd", ""),
-            dbname=mysql_data.get("dbname", "CapacityReport")
-        )
+        mysql_config = MySQLConfig.from_dict(data.get("MySQL_DBInfo"))
        remote_config = RemoteDataConfig.from_dict(data.get("RemoteData"))
        metrix_config = MetrixConfig.from_dict(data.get("Metrix"))
        data_mappings = DataMappingsConfig.from_dict(data.get("DataMappings"))
+        cell_data = CellDataConfig.from_dict(data.get("CellData"))
        history_retention = HistoryRetentionConfig.from_dict(data.get("HistoryRetention"))

        return cls(
@@ -380,6 +448,7 @@ class AppConfig:
            metrix=metrix_config,
            data_mappings=data_mappings,
            remote_data=remote_config,
+            cell_data=cell_data,
            history_retention=history_retention,
            sheet_filter=data.get("SheetFilter", []),
            extract_fields=data.get("ExtractField", [])
@@ -399,16 +468,11 @@ class AppConfig:
            "Update": self.update,
            "SourceType": self.source_type,
            "WarehouseType": self.warehouse_type,
-            "MySQL_DBInfo": {
-                "host": self.mysql.host,
-                "port": self.mysql.port,
-                "user": self.mysql.user,
-                "passwd": self.mysql.passwd,
-                "dbname": self.mysql.dbname
-            },
+            "MySQL_DBInfo": self.mysql.normalized().to_dict(include_password=True),
            "Metrix": self.metrix.normalized().to_dict(include_token=True),
            "DataMappings": self.data_mappings.normalized().to_dict(),
            "RemoteData": self.remote_data.normalized().to_dict(include_password=True),
+            "CellData": self.cell_data.normalized().to_dict(include_password=True),
            "HistoryRetention": self.history_retention.normalized().to_dict(),
            "SheetFilter": self.sheet_filter,
            "ExtractField": self.extract_fields
@@ -420,15 +484,11 @@ class AppConfig:
            "update": self.update,
            "source_type": self.source_type,
            "warehouse_type": self.warehouse_type,
-            "mysql": {
-                "host": self.mysql.host,
-                "port": self.mysql.port,
-                "user": self.mysql.user,
-                "dbname": self.mysql.dbname
-            },
+            "mysql": self.mysql.normalized().to_dict(),
            "metrix": self.metrix.normalized().to_dict(),
            "data_mappings": self.data_mappings.normalized().to_dict(),
            "remote_data": self.remote_data.normalized().to_dict(),
+            "cell_data": self.cell_data.normalized().to_dict(),
            "history_retention": self.history_retention.normalized().to_dict(),
            "sheet_filter": self.sheet_filter,
            "extract_fields": self.extract_fields
@@ -440,16 +500,11 @@ class AppConfig:
            "update": self.update,
            "source_type": self.source_type,
            "warehouse_type": self.warehouse_type,
-            "mysql": {
-                "host": self.mysql.host,
-                "port": self.mysql.port,
-                "user": self.mysql.user,
-                "passwd": self.mysql.passwd,
-                "dbname": self.mysql.dbname
-            },
+            "mysql": self.mysql.normalized().to_dict(include_password=True),
            "metrix": self.metrix.normalized().to_dict(include_token=True),
            "data_mappings": self.data_mappings.normalized().to_dict(),
            "remote_data": self.remote_data.normalized().to_dict(include_password=True),
+            "cell_data": self.cell_data.normalized().to_dict(include_password=True),
            "history_retention": self.history_retention.normalized().to_dict(),
            "sheet_filter": self.sheet_filter,
            "extract_fields": self.extract_fields