From c36dd2183828a04dea9459b91ae73c04e05d28dd Mon Sep 17 00:00:00 2001 From: Nixevol Date: Thu, 25 Jun 2026 04:28:24 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=20CellData=20?= =?UTF-8?q?=E9=A2=84=E5=A4=84=E7=90=86=E4=B8=8E=E5=8D=95=E7=8B=AC=E5=88=B7?= =?UTF-8?q?=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Configure.json | 14 +- app/api/routers/cell_data.py | 103 ++++++ app/api/routers/config.py | 84 +++++ app/api/routers/remote.py | 3 + app/api/routers/tasks.py | 2 + app/config.py | 127 +++++++- app/main.py | 2 + app/services/cell_data.py | 361 ++++++++++++++++++++++ docs/project_context.md | 9 + frontend/src/components/FileWorkflow.vue | 56 +++- frontend/src/components/SettingsPanel.vue | 195 +++++++++++- frontend/src/styles.css | 79 +++++ frontend/src/types.ts | 5 + 13 files changed, 1027 insertions(+), 13 deletions(-) create mode 100644 app/api/routers/cell_data.py create mode 100644 app/services/cell_data.py diff --git a/Configure.json b/Configure.json index 3143e7a..34bfd96 100644 --- a/Configure.json +++ b/Configure.json @@ -1,5 +1,5 @@ { - "Update": "2026/06/25 02:26:49", + "Update": "2026/06/25 03:58:45", "SourceType": "sftp", "WarehouseType": "mysql", "MySQL_DBInfo": { @@ -211,7 +211,7 @@ "host": "", "port": 22, "user": "", - "remote_dir": "/CellData", + "remote_dir": "/", "passive": true, "timeout": 30, "auto_delete_source": false, @@ -224,12 +224,18 @@ "passwd": "" }, "mysql": { - "host": "localhost", + "host": "127.0.0.1", "port": 3306, "user": "root", "dbname": "celldata", "passwd": "123456" - } + }, + "scan_paths": [ + "/网优日常优化数据文档/日常性能报表/{maxyear}年/300表" + ], + "year_dir_regex": "(?P\\d{4})年", + "file_name_regex": "^Result_300_.*\\.zip$", + "file_time_regex": "(?P\\d{14})(?=\\.zip$)" }, "HistoryRetention": { "enabled": true, diff --git a/app/api/routers/cell_data.py b/app/api/routers/cell_data.py new file mode 100644 index 0000000..8491bc8 --- /dev/null +++ b/app/api/routers/cell_data.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import time +from datetime import datetime +from pathlib import Path +from threading import Thread + +from fastapi import APIRouter, Body, HTTPException + +from app import state +from app.api.routers.task_runtime import set_task_stage +from app.config import AppConfig, CACHE_DIR +from app.processor import ProcessLogger +from app.services.cell_data import refresh_cell_data + +router = APIRouter(tags=["cell-data"]) + + +@router.post("/api/cell-data/process/start") +async def start_cell_data_processing(): + if state.global_task_lock["locked"]: + raise HTTPException(status_code=409, detail="已有任务在运行，请等待当前任务完成") + + task_id = "cell_data_" + datetime.now().strftime("%Y%m%d_%H%M%S") + work_dir = CACHE_DIR / task_id + work_dir.mkdir(parents=True, exist_ok=True) + logs: list[str] = [] + current_stage = "locating" + + def log_callback(message: str) -> None: + logs.append(message) + set_task_stage(task_id, current_stage, logs) + + def stage_callback(stage: str) -> None: + nonlocal current_stage + current_stage = stage + set_task_stage(task_id, current_stage, logs) + + logger = ProcessLogger( + log_file=work_dir / "log.txt", + callback=log_callback, + stage_callback=stage_callback, + ) + app_config = state.current_config() + state.processing_tasks[task_id] = {"logs": [], "status": "processing", "stage": current_stage} + state.global_task_lock.update( + { + "locked": True, + "task_id": task_id, + "stage": current_stage, + "started_at": datetime.now().isoformat(), + } + ) + + thread = Thread(target=_run_cell_data_processing, args=(task_id, work_dir, logger, app_config), daemon=True) + thread.start() + return {"success": True, "message": "CellData 处理已启动", "task_id": task_id, "stage": current_stage} + + +@router.post("/api/cell-data/process/status") +async def get_cell_data_processing_status(task_id: str = Body(..., embed=True)): + if task_id in state.processing_tasks: + task = state.processing_tasks[task_id] + return { + "task_id": task_id, + "status": task.get("status", "processing"), + "stage": task.get("stage", "processing"), + "logs": task.get("logs", []), + "error": task.get("error"), + "result": task.get("result"), + "elapsed_time": task.get("elapsed_time"), + } + raise HTTPException(status_code=404, detail="任务不存在") + + +def _run_cell_data_processing(task_id: str, work_dir: Path, logger: ProcessLogger, app_config: AppConfig) -> None: + started = time.time() + try: + result = refresh_cell_data(app_config, work_dir, logger) + elapsed = round(time.time() - started, 2) + state.processing_tasks[task_id] = { + "logs": logger.get_logs(), + "status": "completed", + "stage": "completed", + "elapsed_time": elapsed, + "result": { + "selected_files": result.selected_files, + "parsed_rows": result.parsed_rows, + "imported_rows": result.imported_rows, + "skipped_rows": result.skipped_rows, + }, + } + except Exception as exc: + logger.error(str(exc)) + state.processing_tasks[task_id] = { + "logs": logger.get_logs(), + "status": "failed", + "stage": "failed", + "error": str(exc), + "elapsed_time": round(time.time() - started, 2), + } + finally: + state.reset_task_lock() diff --git a/app/api/routers/config.py b/app/api/routers/config.py index dca6981..4eb55e9 100644 --- a/app/api/routers/config.py +++ b/app/api/routers/config.py @@ -1,4 +1,5 @@ import json +import re from datetime import datetime from typing import Any from urllib.parse import quote @@ -11,6 +12,7 @@ from app import state from app.config import ( CellDataConfig, DataMappingsConfig, + DEFAULT_CELL_DATA_MAPPING, HistoryRetentionConfig, MetrixConfig, MySQLConfig, @@ -103,6 +105,11 @@ async def update_cell_data_remote_config(config: dict[str, Any] = Body(...)): state.config.cell_data = CellDataConfig( remote_data=RemoteDataConfig.from_dict(config), mysql=current.mysql, + scan_paths=current.scan_paths, + year_dir_regex=current.year_dir_regex, + file_name_regex=current.file_name_regex, + file_time_regex=current.file_time_regex, + mapping=current.mapping, ).normalized() state.config.save() return {"success": True, "message": "CellData 远程数据源配置已更新", "update": state.config.update} @@ -115,6 +122,11 @@ async def update_cell_data_mysql_config(config: dict[str, Any] = Body(...)): state.config.cell_data = CellDataConfig( remote_data=current.remote_data, mysql=MySQLConfig.from_dict(config, default_dbname="celldata"), + scan_paths=current.scan_paths, + year_dir_regex=current.year_dir_regex, + file_name_regex=current.file_name_regex, + file_time_regex=current.file_time_regex, + mapping=current.mapping, ).normalized() state.config.save() return {"success": True, "message": "CellData 数据库配置已更新", "update": state.config.update} @@ -136,6 +148,36 @@ async def test_cell_data_mysql_connection(config: dict[str, Any] | None = Body(N return _test_mysql_config(mysql_config) +@router.post("/api/config/cell-data/settings") +async def update_cell_data_settings(config: dict[str, Any] = Body(...)): + validation = _validate_cell_data_settings(config) + if not validation["success"]: + raise HTTPException(status_code=400, detail=validation["message"]) + state.reload_config() + current = state.config.cell_data.normalized() + state.config.cell_data = CellDataConfig( + remote_data=current.remote_data, + mysql=current.mysql, + scan_paths=config.get("scan_paths", current.scan_paths), + year_dir_regex=str(config.get("year_dir_regex", current.year_dir_regex)), + file_name_regex=str(config.get("file_name_regex", current.file_name_regex)), + file_time_regex=str(config.get("file_time_regex", current.file_time_regex)), + mapping=config.get("mapping", current.mapping), + ).normalized() + state.config.save() + return {"success": True, "message": "CellData 规则已更新", "update": state.config.update} + + +@router.post("/api/config/cell-data/settings/validate") +async def validate_cell_data_settings(config: dict[str, Any] = Body(...)): + return _validate_cell_data_settings(config) + + +@router.get("/api/config/cell-data/mapping/default") +async def get_default_cell_data_mapping(): + return DEFAULT_CELL_DATA_MAPPING + + @router.post("/api/config/history-retention") async def update_history_retention(config: dict[str, Any] = Body(...)): state.reload_config() @@ -254,3 +296,45 @@ def _test_mysql_config(config: MySQLConfig) -> dict[str, Any]: except Exception as exc: return {"success": False, "message": str(exc)} + +def _validate_cell_data_settings(config: dict[str, Any]) -> dict[str, Any]: + scan_paths = config.get("scan_paths", []) + if not isinstance(scan_paths, list) or not any(str(path).strip() for path in scan_paths): + return {"success": False, "message": "请至少配置一个扫描路径"} + + for key in ("year_dir_regex", "file_name_regex", "file_time_regex"): + try: + re.compile(str(config.get(key, ""))) + except re.error as exc: + return {"success": False, "message": f"{key} 正则无效: {exc}"} + + mapping = config.get("mapping") + if not isinstance(mapping, dict): + return {"success": False, "message": "映射规则必须是 JSON 对象"} + if not str(mapping.get("target_table", "")).strip(): + return {"success": False, "message": "映射规则缺少 target_table"} + key_config = mapping.get("key") + if not isinstance(key_config, dict) or not key_config.get("field") or not key_config.get("expr"): + return {"success": False, "message": "映射规则缺少 key.field 或 key.expr"} + sources = mapping.get("sources") + if not isinstance(sources, list) or not sources: + return {"success": False, "message": "映射规则至少需要一个 sources 项"} + + for index, source in enumerate(sources, 1): + if not isinstance(source, dict): + return {"success": False, "message": f"sources 第 {index} 项必须是对象"} + if not source.get("band") or not source.get("file_prefix"): + return {"success": False, "message": f"sources 第 {index} 项缺少 band 或 file_prefix"} + fields = source.get("fields") + if not isinstance(fields, dict) or not fields: + return {"success": False, "message": f"sources 第 {index} 项缺少 fields"} + for target, rule in fields.items(): + if not str(target).strip(): + return {"success": False, "message": f"sources 第 {index} 项存在空目标字段"} + if isinstance(rule, str) and rule.strip(): + continue + if isinstance(rule, dict) and "value" in rule: + continue + return {"success": False, "message": f"{target} 的映射规则无效"} + return {"success": True, "message": "映射规则有效"} + diff --git a/app/api/routers/remote.py b/app/api/routers/remote.py index 7b0b6b3..bb515e1 100644 --- a/app/api/routers/remote.py +++ b/app/api/routers/remote.py @@ -14,6 +14,7 @@ from app.api.routers.task_runtime import ( ) from app.config import AppConfig, CACHE_DIR, RemoteDataConfig from app.processor import DataProcessor, ProcessLogger +from app.services.cell_data import refresh_cell_data from app.services.license import LicenseError, check_processing_allowed from app.services.platform import PlatformStorageDownloader, make_source_downloader from app.services.pipeline import RESULT_TABLES, run_import_and_report @@ -150,6 +151,8 @@ def _run_remote_processing( state.history_manager.update(task_id, file_count=download_result.file_count) + refresh_cell_data(app_config, work_dir, logger) + if app_config.warehouse_type == "metrix": started = time.time() run_import_and_report(work_dir, app_config, logger) diff --git a/app/api/routers/tasks.py b/app/api/routers/tasks.py index 234db95..3921470 100644 --- a/app/api/routers/tasks.py +++ b/app/api/routers/tasks.py @@ -13,6 +13,7 @@ from app.api.routers.task_runtime import ( ) from app.config import AppConfig from app.processor import DataProcessor, ProcessLogger +from app.services.cell_data import refresh_cell_data from app.services.license import LicenseError, check_processing_allowed @@ -168,6 +169,7 @@ def _task_finished(task_id: str) -> bool: def _run_processing(task_id: str, work_dir: Path, logger: ProcessLogger, app_config: AppConfig) -> None: try: + refresh_cell_data(app_config, work_dir, logger) if app_config.warehouse_type == "metrix": import time diff --git a/app/config.py b/app/config.py index f5b50ac..0015601 100644 --- a/app/config.py +++ b/app/config.py @@ -4,6 +4,7 @@ import json import os import sys +from copy import deepcopy from datetime import datetime from dataclasses import dataclass, field from pathlib import Path @@ -75,6 +76,99 @@ SOURCE_TYPES = ("ftp", "sftp", "metrix") # Warehouse backends: direct MySQL, or Metrix database platform. WAREHOUSE_TYPES = ("mysql", "metrix") +DEFAULT_CELL_DATA_SCAN_PATHS = [ + "/网优日常优化数据文档/日常性能报表/{maxyear}年/300表", +] +DEFAULT_CELL_DATA_YEAR_DIR_REGEX = r"(?P\d{4})年" +DEFAULT_CELL_DATA_FILE_NAME_REGEX = r"^Result_300_.*\.zip$" +DEFAULT_CELL_DATA_FILE_TIME_REGEX = r"(?P\d{14})(?=\.zip$)" +DEFAULT_CELL_DATA_MAPPING: Dict[str, Any] = { + "target_table": "cellinfo", + "key": {"field": "CGI", "expr": "{PLMN}-{eNodeBID}-{CellID}"}, + "sources": [ + { + "band": "2.6G", + "file_prefix": "LTE_ITBBU_CellInfo", + "fields": { + "eNodeBID": "eNBId", + "CellID": "cellLocalId", + "PLMN": "plmn", + "基站名称": "eNBName", + "小区名称": "CellName", + "频点": "frequency", + "带宽": "bandWidth", + "制式": "radioMode", + "功率": "cpSpeRefSigPwr", + "网络": {"value": "4G"}, + }, + }, + { + "band": "2.6G", + "file_prefix": "LTE_SDR_CellInfo", + "fields": { + "eNodeBID": "eNBId", + "CellID": "cellLocalId", + "PLMN": "plmn", + "基站名称": "eNBName", + "小区名称": "CellName", + "频点": "frequency", + "带宽": "bandWidth", + "制式": "radioMode", + "功率": "cpSpeRefSigPwr", + "网络": {"value": "4G"}, + }, + }, + { + "band": "2.6G", + "file_prefix": "NR_CellInfo", + "fields": { + "eNodeBID": "gNBId", + "CellID": "cellLocalId", + "PLMN": "plmn", + "基站名称": "gNBName", + "小区名称": "CellName", + "频点": "ssbFrequency", + "带宽": "carrierBandwidth", + "制式": {"value": "2.6G"}, + "功率": "powerPerRERef", + "网络": {"value": "5G"}, + }, + }, + { + "band": "700M", + "file_prefix": "LTE_ITBBU_CellInfo", + "fields": { + "eNodeBID": "eNBId", + "CellID": "cellLocalId", + "PLMN": "plmn", + "基站名称": "eNBName", + "小区名称": "CellName", + "频点": "frequency", + "带宽": "bandWidth", + "制式": "radioMode", + "功率": "cpSpeRefSigPwr", + "网络": {"value": "4G"}, + }, + }, + { + "band": "700M", + "file_prefix": "NR_CellInfo", + "fields": { + "eNodeBID": "gNBId", + "CellID": "cellLocalId", + "PLMN": "plmn", + "基站名称": "gNBName", + "小区名称": "CellName", + "频点": "ssbFrequency", + "带宽": "carrierBandwidth", + "制式": {"value": "700M"}, + "功率": "powerPerRERef", + "网络": {"value": "5G"}, + }, + }, + ], +} + @dataclass class MetrixConfig: @@ -335,15 +429,36 @@ class CellDataConfig: default_factory=lambda: RemoteDataConfig( enabled=False, protocol="sftp", - remote_dir="/CellData", + remote_dir="/", ) ) mysql: MySQLConfig = field(default_factory=lambda: MySQLConfig(dbname="celldata")) + scan_paths: List[str] = field(default_factory=lambda: list(DEFAULT_CELL_DATA_SCAN_PATHS)) + year_dir_regex: str = DEFAULT_CELL_DATA_YEAR_DIR_REGEX + file_name_regex: str = DEFAULT_CELL_DATA_FILE_NAME_REGEX + file_time_regex: str = DEFAULT_CELL_DATA_FILE_TIME_REGEX + mapping: Dict[str, Any] = field(default_factory=lambda: deepcopy(DEFAULT_CELL_DATA_MAPPING)) def normalized(self) -> "CellDataConfig": + scan_paths = [] + seen = set() + for path in self.scan_paths or []: + normalized = str(path).replace("\\", "/").strip() + if normalized and normalized not in seen: + scan_paths.append(normalized) + seen.add(normalized) + if not scan_paths: + scan_paths = list(DEFAULT_CELL_DATA_SCAN_PATHS) + + mapping = self.mapping if isinstance(self.mapping, dict) and self.mapping else deepcopy(DEFAULT_CELL_DATA_MAPPING) return CellDataConfig( remote_data=self.remote_data.normalized(), mysql=self.mysql.normalized(), + scan_paths=scan_paths, + year_dir_regex=str(self.year_dir_regex or DEFAULT_CELL_DATA_YEAR_DIR_REGEX).strip() or DEFAULT_CELL_DATA_YEAR_DIR_REGEX, + file_name_regex=str(self.file_name_regex or DEFAULT_CELL_DATA_FILE_NAME_REGEX).strip() or DEFAULT_CELL_DATA_FILE_NAME_REGEX, + file_time_regex=str(self.file_time_regex or DEFAULT_CELL_DATA_FILE_TIME_REGEX).strip() or DEFAULT_CELL_DATA_FILE_TIME_REGEX, + mapping=mapping, ) def to_dict(self, include_password: bool = False) -> Dict[str, Any]: @@ -351,6 +466,11 @@ class CellDataConfig: return { "remote_data": normalized.remote_data.to_dict(include_password=include_password), "mysql": normalized.mysql.to_dict(include_password=include_password), + "scan_paths": normalized.scan_paths, + "year_dir_regex": normalized.year_dir_regex, + "file_name_regex": normalized.file_name_regex, + "file_time_regex": normalized.file_time_regex, + "mapping": normalized.mapping, } @classmethod @@ -362,6 +482,11 @@ class CellDataConfig: 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, + scan_paths=data.get("scan_paths", default.scan_paths) if isinstance(data.get("scan_paths", default.scan_paths), list) else default.scan_paths, + year_dir_regex=str(data.get("year_dir_regex", default.year_dir_regex)), + file_name_regex=str(data.get("file_name_regex", default.file_name_regex)), + file_time_regex=str(data.get("file_time_regex", default.file_time_regex)), + mapping=data.get("mapping", default.mapping) if isinstance(data.get("mapping", default.mapping), dict) else default.mapping, ).normalized() diff --git a/app/main.py b/app/main.py index 8374bf2..f5b2192 100644 --- a/app/main.py +++ b/app/main.py @@ -12,6 +12,7 @@ from app import state from app.api.routers import ( auth, cache, + cell_data, config, database, health, @@ -110,6 +111,7 @@ def register_routes(app: FastAPI) -> None: database.router, config.router, cache.router, + cell_data.router, script.router, ] for router in routers: diff --git a/app/services/cell_data.py b/app/services/cell_data.py new file mode 100644 index 0000000..0aecd0d --- /dev/null +++ b/app/services/cell_data.py @@ -0,0 +1,361 @@ +from __future__ import annotations + +import csv +import re +import stat +import zipfile +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any, Callable, Iterable + +import chardet +import pymysql + +from app.config import AppConfig, CellDataConfig +from app.services.remote_download import RemoteDataDownloader, RemoteFileInfo + +LogFn = Callable[[str], None] + +CELLINFO_COLUMNS = [ + "CGI", + "eNodeBID", + "CellID", + "PLMN", + "基站名称", + "小区名称", + "频点", + "带宽", + "制式", + "功率", + "网络", +] + + +@dataclass(frozen=True) +class SelectedZip: + scan_path: str + band: str + remote_file: RemoteFileInfo + timestamp: str + + +@dataclass +class CellDataResult: + selected_files: int = 0 + parsed_rows: int = 0 + imported_rows: int = 0 + skipped_rows: int = 0 + + +class CellDataProcessor: + def __init__(self, app_config: AppConfig, work_dir: Path, logger=None): + self.app_config = app_config + self.config = app_config.cell_data.normalized() + self.work_dir = work_dir + self.logger = logger + self.downloader = RemoteDataDownloader(self.config.remote_data, self._log) + self.year_dir_re = re.compile(self.config.year_dir_regex) + self.file_name_re = re.compile(self.config.file_name_regex) + self.file_time_re = re.compile(self.config.file_time_regex) + + def run(self) -> CellDataResult: + if not self.config.remote_data.enabled: + self._log("CellData 数据源未启用，跳过") + return CellDataResult() + + self._set_stage("locating") + selected = self._select_latest_zip_files() + if not selected: + raise RuntimeError("未找到可处理的 CellData ZIP 文件") + self._log(f"已选择 {len(selected)} 个 CellData ZIP 文件") + + self._set_stage("downloading") + local_files = self._download_selected(selected) + + self._set_stage("parsing") + result = CellDataResult(selected_files=len(selected)) + rows = self._parse_zip_files(local_files, result) + if not rows: + raise RuntimeError("CellData ZIP 中未解析到有效数据") + + self._set_stage("importing") + result.imported_rows = self._replace_cellinfo(rows) + self._log(f"CellData 导入完成，共 {result.imported_rows} 行") + return result + + def _select_latest_zip_files(self) -> list[SelectedZip]: + selected: list[SelectedZip] = [] + for template in self.config.scan_paths: + scan_path = self._resolve_scan_path(template) + band_dirs = [ + entry + for entry in self._list_dir(scan_path) + if entry["type"] == "directory" + ] + for band_dir in band_dirs: + band = str(band_dir["name"]) + files = [ + entry + for entry in self._list_dir(str(band_dir["path"])) + if entry["type"] == "file" and self.file_name_re.search(str(entry["name"])) + ] + latest = self._latest_file(files) + if latest: + selected.append( + SelectedZip( + scan_path=scan_path, + band=band, + remote_file=self._remote_file_info(str(latest["path"]), int(latest.get("size") or 0)), + timestamp=str(latest["timestamp"]), + ) + ) + self._log(f"{band}: {latest['name']}") + return selected + + def _resolve_scan_path(self, template: str) -> str: + path = self._replace_date_placeholders(str(template).replace("\\", "/").strip()) + if not path.startswith("/"): + path = self._join_remote_path(self.config.remote_data.remote_dir, path) + if "{maxyear}" not in path: + return path + + parts = [part for part in path.split("/") if part] + maxyear_index = next((index for index, part in enumerate(parts) if "{maxyear}" in part), -1) + if maxyear_index < 0: + return path + parent = "/" + "/".join(parts[:maxyear_index]) if maxyear_index else "/" + year_segment = parts[maxyear_index] + years: list[int] = [] + for entry in self._list_dir(parent): + if entry["type"] != "directory": + continue + match = self.year_dir_re.search(str(entry["name"])) + if match: + years.append(int(match.group("year"))) + if not years: + raise RuntimeError(f"未找到年份目录: {parent}") + max_year = str(max(years)) + parts[maxyear_index] = year_segment.replace("{maxyear}", max_year) + return "/" + "/".join(parts) + + def _replace_date_placeholders(self, path: str) -> str: + now = datetime.now() + return ( + path.replace("{yyyy}", now.strftime("%Y")) + .replace("{yyyymm}", now.strftime("%Y%m")) + .replace("{yyyymmdd}", now.strftime("%Y%m%d")) + ) + + def _latest_file(self, files: list[dict[str, Any]]) -> dict[str, Any] | None: + candidates: list[dict[str, Any]] = [] + for item in files: + match = self.file_time_re.search(str(item["name"])) + if not match: + continue + timestamp = match.group("timestamp") + candidates.append({**item, "timestamp": timestamp}) + if not candidates: + return None + return max(candidates, key=lambda item: str(item["timestamp"])) + + def _download_selected(self, selected: list[SelectedZip]) -> list[tuple[SelectedZip, Path]]: + target_dir = self.work_dir / "cell_data" + target_dir.mkdir(parents=True, exist_ok=True) + local_files: list[tuple[SelectedZip, Path]] = [] + if self.config.remote_data.protocol == "ftp": + with self.downloader._ftp_client() as ftp: # noqa: SLF001 - reuse existing connection helpers + for item in selected: + local_path = target_dir / item.band / item.remote_file.name + self.downloader._download_ftp_file(ftp, item.remote_file.path, local_path, self._download_result(), item.remote_file.size) # noqa: SLF001 + local_files.append((item, local_path)) + return local_files + + ssh = self.downloader._sftp_ssh_client() # noqa: SLF001 + try: + with ssh.open_sftp() as sftp: + for item in selected: + local_path = target_dir / item.band / item.remote_file.name + self.downloader._download_sftp_file(sftp, item.remote_file.path, local_path, self._download_result(), item.remote_file.size) # noqa: SLF001 + local_files.append((item, local_path)) + finally: + ssh.close() + return local_files + + @staticmethod + def _download_result(): + from app.services.remote_download import RemoteDownloadResult + + return RemoteDownloadResult() + + def _parse_zip_files(self, local_files: list[tuple[SelectedZip, Path]], result: CellDataResult) -> list[dict[str, str]]: + mapping = self.config.mapping + key_config = mapping["key"] + rows_by_key: dict[str, dict[str, str]] = {} + for selected, local_path in local_files: + with zipfile.ZipFile(local_path) as zf: + for source in mapping["sources"]: + if source["band"] != selected.band: + continue + for info in zf.infolist(): + name = Path(info.filename).name + if not name.lower().endswith(".csv"): + continue + if not name.startswith(source["file_prefix"]): + continue + raw = zf.read(info.filename) + text = self._decode_csv(raw) + reader = csv.DictReader(text.splitlines()) + for csv_row in reader: + row = self._map_row(source["fields"], csv_row) + key = self._render_expr(str(key_config["expr"]), row) + if not key or "--" in key: + result.skipped_rows += 1 + continue + row[str(key_config["field"])] = key + rows_by_key[key] = {column: row.get(column, "") for column in CELLINFO_COLUMNS} + result.parsed_rows += 1 + return list(rows_by_key.values()) + + def _map_row(self, fields: dict[str, Any], csv_row: dict[str, str]) -> dict[str, str]: + row: dict[str, str] = {} + for target, rule in fields.items(): + if isinstance(rule, str): + row[target] = str(csv_row.get(rule, "") or "").strip() + elif isinstance(rule, dict) and "value" in rule: + row[target] = str(rule.get("value", "") or "").strip() + else: + row[target] = "" + return row + + @staticmethod + def _render_expr(expr: str, row: dict[str, str]) -> str: + def replace(match): + return row.get(match.group(1), "") + + return re.sub(r"\{([^{}]+)\}", replace, expr).strip() + + @staticmethod + def _decode_csv(raw: bytes) -> str: + candidates = ["utf-8-sig", "utf-8", "gb18030", "gbk"] + detected = (chardet.detect(raw).get("encoding") or "").lower() + if detected and detected not in candidates: + candidates.append(detected) + for encoding in dict.fromkeys(candidates): + try: + return raw.decode(encoding).lstrip("\ufeff") + except (LookupError, UnicodeDecodeError): + pass + return raw.decode("utf-8", errors="replace").lstrip("\ufeff") + + def _replace_cellinfo(self, rows: list[dict[str, str]]) -> int: + mysql = self.config.mysql.normalized() + table = str(self.config.mapping.get("target_table") or "cellinfo") + conn = pymysql.connect( + host=mysql.host, + port=mysql.port, + user=mysql.user, + password=mysql.passwd, + database=mysql.dbname, + charset="utf8mb4", + cursorclass=pymysql.cursors.DictCursor, + autocommit=False, + ) + try: + with conn.cursor() as cursor: + self._ensure_cellinfo_table(cursor, table) + cursor.execute(f"TRUNCATE TABLE `{table}`") + placeholders = ", ".join(["%s"] * len(CELLINFO_COLUMNS)) + columns = ", ".join(f"`{column}`" for column in CELLINFO_COLUMNS) + values = [tuple(row.get(column, "") for column in CELLINFO_COLUMNS) for row in rows] + for start in range(0, len(values), 1000): + cursor.executemany( + f"INSERT INTO `{table}` ({columns}) VALUES ({placeholders})", + values[start:start + 1000], + ) + conn.commit() + return len(rows) + except Exception: + conn.rollback() + raise + finally: + conn.close() + + @staticmethod + def _ensure_cellinfo_table(cursor, table: str) -> None: + cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{table}` ( + `CGI` varchar(120) DEFAULT NULL, + `eNodeBID` int DEFAULT NULL, + `CellID` int DEFAULT NULL, + `PLMN` varchar(100) DEFAULT NULL, + `基站名称` varchar(200) DEFAULT NULL, + `小区名称` varchar(200) DEFAULT NULL, + `频点` varchar(50) DEFAULT NULL, + `带宽` varchar(20) DEFAULT NULL, + `制式` varchar(50) DEFAULT NULL, + `功率` varchar(100) DEFAULT NULL, + `网络` varchar(20) DEFAULT NULL, + KEY `CGI` (`CGI`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 + """ + ) + + def _list_dir(self, remote_path: str) -> list[dict[str, Any]]: + if self.config.remote_data.protocol == "ftp": + with self.downloader._ftp_client() as ftp: # noqa: SLF001 + return [ + { + "name": name, + "type": "directory" if entry_type == "dir" else "file", + "size": size, + "path": self._join_remote_path(remote_path, name), + } + for name, entry_type, size in self.downloader._list_ftp_entries(ftp, remote_path) # noqa: SLF001 + if name not in {".", ".."} + ] + + ssh = self.downloader._sftp_ssh_client() # noqa: SLF001 + try: + with ssh.open_sftp() as sftp: + return [ + { + "name": item.filename, + "type": "directory" if stat.S_ISDIR(item.st_mode) else "file", + "size": int(getattr(item, "st_size", 0) or 0), + "path": self._join_remote_path(remote_path, item.filename), + } + for item in sftp.listdir_attr(remote_path) + if item.filename not in {".", ".."} + ] + finally: + ssh.close() + + def _remote_file_info(self, remote_path: str, size: int) -> RemoteFileInfo: + return self.downloader._remote_file_info(remote_path, size) # noqa: SLF001 + + @staticmethod + def _join_remote_path(parent: str, child: str) -> str: + parent = parent.replace("\\", "/").rstrip("/") + child = child.replace("\\", "/").strip("/") + if not parent: + return child + if parent == "/": + return f"/{child}" + return f"{parent}/{child}" + + def _set_stage(self, stage: str) -> None: + if hasattr(self.logger, "set_stage"): + self.logger.set_stage(stage) + + def _log(self, message: str) -> None: + if self.logger: + if hasattr(self.logger, "info"): + self.logger.info(message) + else: + self.logger(message) + + +def refresh_cell_data(app_config: AppConfig, work_dir: Path, logger=None) -> CellDataResult: + return CellDataProcessor(app_config, work_dir, logger).run() diff --git a/docs/project_context.md b/docs/project_context.md index 2cfb112..53ebe2e 100644 --- a/docs/project_context.md +++ b/docs/project_context.md @@ -744,3 +744,12 @@ - 700M 映射： - `LTE_ITBBU_CellInfo`：`eNodeBID<-eNBId`，`CellID<-cellLocalId`，`PLMN<-plmn`，`基站名称<-eNBName`，`小区名称<-CellName`，`频点<-frequency`，`带宽<-bandWidth`，`制式<-radioMode`，`功率<-cpSpeRefSigPwr`，`网络="4G"`。 - `NR_CellInfo`：`eNodeBID<-gNBId`，`CellID<-cellLocalId`，`PLMN<-plmn`，`基站名称<-gNBName`，`小区名称<-CellName`，`频点<-ssbFrequency`，`带宽<-carrierBandwidth`，`制式="700M"`，`功率<-powerPerRERef`，`网络="5G"`。 + +## 2026-06-25：实现 CellData 预处理与单独刷新入口 + +- 配置：`CellData` 新增 `scan_paths`、`year_dir_regex`、`file_name_regex`、`file_time_regex` 与 `mapping`。默认扫描路径为 `/网优日常优化数据文档/日常性能报表/{maxyear}年/300表`；默认 ZIP 过滤为 `Result_300_*.zip`；默认映射写入 `cellinfo` 并生成 `CGI={PLMN}-{eNodeBID}-{CellID}`。 +- 设置页：CellData 数据源卡片新增扫描路径列表、路径说明弹窗、三个高级正则输入、映射 JSON 编辑框，以及“校验 JSON / 恢复默认映射 / 保存规则”操作。说明弹窗包含路径模板、占位符、正则和多目录示例，避免在表单页堆长文案。 +- 后端：新增 `app/services/cell_data.py`，负责路径模板解析（含 `{maxyear}`/`{yyyy}`/`{yyyymm}`/`{yyyymmdd}`）、按扫描目录下一级子目录选择最新 ZIP、解析目标 CSV、执行 JSON 映射、清空并批量写入 `celldata.cellinfo`。CSV 解码优先 UTF-8/UTF-8-SIG，回退 GB18030/GBK。 +- 接口：新增 `app/api/routers/cell_data.py`，提供 `POST /api/cell-data/process/start` 与 `/status`，数据处理页新增独立 CellData 卡片，可只刷新 CellData，不跑容量处理。 +- 接入：本地上传、远程手动和自动调度入口都会在容量处理前调用 CellData 预处理；CellData 单独处理和容量处理共用现有全局任务锁，避免并发写库。 +- 验证：`python -m compileall -q app` 通过；`frontend` `npm run build` 通过；远程定位可从最新年份 `2026年/300表` 选出 `2.6G` 与 `700M` 各自最新 `Result_300` ZIP；用 SFTP MCP 读取的小样本 ZIP 验证解析和入库，`celldata.cellinfo` 写入 1 行且中文字段正常（样本 `CGI=460-00-12683845-1`）。 diff --git a/frontend/src/components/FileWorkflow.vue b/frontend/src/components/FileWorkflow.vue index 7b0377d..68a986a 100644 --- a/frontend/src/components/FileWorkflow.vue +++ b/frontend/src/components/FileWorkflow.vue @@ -59,6 +59,22 @@ /> +
+
+

CellData

+

刷新 CellData 数据。

+
+ + 刷新 CellData + +
+

已选择文件

@@ -226,6 +242,7 @@ const files = ref([]); const uploadProgress = ref(0); const working = ref(false); const remoteStarting = ref(false); +const cellDataStarting = ref(false); const isDragging = ref(false); const taskStatus = ref(null); const activeTask = ref(null); @@ -236,17 +253,19 @@ const activationCode = ref(''); const activationLoading = ref(false); const activationKeyLabel = ref('2026/12/30'); const licenseErrorMessage = ref('当前数据日期已超过授权到期日期，请输入激活码延长 30 天。'); -const taskMode = ref<'local' | 'remote' | 'unknown'>('unknown'); +const taskMode = ref<'local' | 'remote' | 'cell_data' | 'unknown'>('unknown'); let activationRetry: (() => Promise) | null = null; let timer: number | undefined; const stageLabels: Record = { license: '授权校验中...', + locating: '定位数据中...', downloading: '远程下载中...', uploading: '上传文件中...', processing: '处理中...', extracting: '解压数据中...', converting: '转换 Excel 中...', + parsing: '解析数据中...', importing: '上传数据中...', scripting: '运行脚本中...', completed: '处理完成', @@ -542,6 +561,34 @@ async function startRemoteProcessing() { } } +async function startCellDataProcessing() { + if (working.value || remoteStarting.value || cellDataStarting.value) return; + + cellDataStarting.value = true; + try { + const result = await apiPost('/api/cell-data/process/start'); + message.success(result.message || 'CellData 处理已启动'); + activeTask.value = { + has_active: true, + task_id: result.task_id, + stage: result.stage || 'locating', + started_at: new Date().toISOString() + }; + taskStatus.value = { + task_id: result.task_id, + status: 'processing', + stage: result.stage || 'locating', + logs: ['CellData 任务已提交，等待处理日志...'] + }; + taskMode.value = 'cell_data'; + startPolling(result.task_id); + } catch (error) { + message.error(error instanceof Error ? error.message : 'CellData 处理启动失败'); + } finally { + cellDataStarting.value = false; + } +} + async function startUploadedTask(taskId: string) { await apiPost('/api/process/start', { task_id: taskId }); taskMode.value = 'local'; @@ -566,7 +613,7 @@ async function checkActiveTask() { try { activeTask.value = await apiGet('/api/task/status'); if (activeTask.value.has_active && activeTask.value.task_id) { - taskMode.value = 'unknown'; + taskMode.value = activeTask.value.task_id.startsWith('cell_data_') ? 'cell_data' : 'unknown'; startPolling(activeTask.value.task_id); } } catch (error) { @@ -589,7 +636,10 @@ function stopPolling() { async function poll(taskId: string) { try { - taskStatus.value = await apiPost('/api/process/status', { task_id: taskId }); + const statusUrl = taskMode.value === 'cell_data' + ? '/api/cell-data/process/status' + : '/api/process/status'; + taskStatus.value = await apiPost(statusUrl, { task_id: taskId }); if (activeTask.value?.has_active && taskStatus.value.stage) { activeTask.value = { ...activeTask.value, stage: taskStatus.value.stage }; } diff --git a/frontend/src/components/SettingsPanel.vue b/frontend/src/components/SettingsPanel.vue index c8094eb..66e6c04 100644 --- a/frontend/src/components/SettingsPanel.vue +++ b/frontend/src/components/SettingsPanel.vue @@ -291,8 +291,64 @@ - + + +
+
+ 支持路径模板和多目录组合 + + + +
+
+
+ + + + +
+
+ + + 添加 + +
+ + + + + + + + + + + + + + + + + + + + + + + 校验 JSON + 恢复默认映射 + 保存规则 + @@ -611,13 +667,47 @@ + + + +
+

路径模板

+

每行填写一个扫描目录。系统会在目录下继续识别 700M、2.6G 等子目录。

+ /网优日常优化数据文档/日常性能报表/{maxyear}年/300表 +
+
+

占位符

+

{maxyear}：扫描同级年份目录并取最大年份。

+

{yyyy}：当前年份。

+

{yyyymm}：当前年月。

+

{yyyymmdd}：当前日期。

+
+
+

高级正则

+

年份目录正则：识别年份目录，默认匹配 2026年。

+

文件名正则：筛选 ZIP 文件，默认匹配 Result_300_ 开头。

+

时间戳正则：从 ZIP 文件名末尾提取文件时间。

+
+
+

多目录示例

+ /网优日常优化数据文档/日常性能报表/{maxyear}年/300表 + /网优日常优化数据文档/日常性能报表/{maxyear}年/cellinfo +
+ +