diff --git a/app/api/routers/cell_data.py b/app/api/routers/cell_data.py index d595e02..ffaddec 100644 --- a/app/api/routers/cell_data.py +++ b/app/api/routers/cell_data.py @@ -9,9 +9,9 @@ from fastapi import APIRouter, Body, File, HTTPException, UploadFile from app import state from app.api.routers.task_runtime import set_task_stage -from app.config import AppConfig, CACHE_DIR +from app.config import AppConfig, CACHE_DIR, CELLDATA_SCRIPT from app.processor import ProcessLogger -from app.services.cell_data import CellDataProcessor, refresh_cell_data +from app.services.cell_data import CellDataProcessor, execute_celldata_script, refresh_cell_data from app.utils.files import safe_relative_path router = APIRouter(tags=["cell-data"]) @@ -139,6 +139,7 @@ def _run_cell_data_processing(task_id: str, work_dir: Path, logger: ProcessLogge started = time.time() try: result = refresh_cell_data(app_config, work_dir, logger) + execute_celldata_script(CELLDATA_SCRIPT, app_config, logger) elapsed = round(time.time() - started, 2) state.processing_tasks[task_id] = { "logs": logger.get_logs(), @@ -169,6 +170,7 @@ def _run_uploaded_cell_data_processing(task_id: str, upload_dir: Path, work_dir: started = time.time() try: result = CellDataProcessor(app_config, work_dir, logger).run_local(upload_dir) + execute_celldata_script(CELLDATA_SCRIPT, app_config, logger) elapsed = round(time.time() - started, 2) state.processing_tasks[task_id] = { "logs": logger.get_logs(), diff --git a/app/api/routers/remote.py b/app/api/routers/remote.py index 8a1c71c..e086c6a 100644 --- a/app/api/routers/remote.py +++ b/app/api/routers/remote.py @@ -14,7 +14,8 @@ 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.config import CELLDATA_SCRIPT +from app.services.cell_data import copy_celldata_tables_to_capacity, execute_celldata_script, 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 @@ -221,6 +222,8 @@ def _try_refresh_cell_data(app_config: AppConfig, work_dir: Path, logger: Proces f"CellData 更新完成：{result.imported_rows} 行" f"（解析 {result.parsed_rows}，跳过 {result.skipped_rows}）" ) + execute_celldata_script(CELLDATA_SCRIPT, app_config, logger) + copy_celldata_tables_to_capacity(app_config, logger) except Exception as exc: logger.warning(f"CellData 更新失败，继续容量处理: {exc}") diff --git a/app/api/routers/script.py b/app/api/routers/script.py index ee4a454..cc96b3c 100644 --- a/app/api/routers/script.py +++ b/app/api/routers/script.py @@ -4,41 +4,62 @@ from datetime import datetime from pathlib import Path from threading import Thread -from fastapi import APIRouter, Body, HTTPException +from fastapi import APIRouter, Body, HTTPException, Query from app import state from app.api.routers.task_runtime import set_task_stage -from app.config import AppConfig, CACHE_DIR, SQL_SCRIPT +from app.config import AppConfig, CACHE_DIR, CELLDATA_SCRIPT, SQL_SCRIPT from app.processor import DataProcessor, ProcessLogger router = APIRouter(tags=["script"]) +SCRIPT_PATHS = { + "report": SQL_SCRIPT, + "celldata": CELLDATA_SCRIPT, +} +SCRIPT_LABELS = { + "report": "报表脚本", + "celldata": "CellData 脚本", +} + + +def _resolve_script_path(script_type: str) -> Path: + path = SCRIPT_PATHS.get(script_type) + if path is None: + raise HTTPException(status_code=400, detail=f"不支持的脚本类型: {script_type}") + return path + @router.get("/api/script/content") -async def get_script_content(): +async def get_script_content(type: str = Query("report")): + script_path = _resolve_script_path(type) + label = SCRIPT_LABELS.get(type, type) try: - if not SQL_SCRIPT.exists(): + if not script_path.exists(): return { "success": True, - "content": "# SQL 脚本文件不存在，请在此编写脚本\n", + "content": f"# {label}文件不存在，请在此编写脚本\n", "modified": None, - "path": str(SQL_SCRIPT), + "path": str(script_path), + "script_type": type, } - modified = datetime.fromtimestamp(SQL_SCRIPT.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S") + modified = datetime.fromtimestamp(script_path.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S") return { "success": True, - "content": SQL_SCRIPT.read_text(encoding="utf-8"), + "content": script_path.read_text(encoding="utf-8"), "modified": modified, - "path": str(SQL_SCRIPT), + "path": str(script_path), + "script_type": type, } except Exception as exc: return {"success": False, "error": str(exc)} @router.post("/api/script/execute") -async def execute_script(): +async def execute_script(script_type: str = Body("report", embed=True)): + script_path = _resolve_script_path(script_type) if state.global_task_lock["locked"]: raise HTTPException(status_code=409, detail="已有任务在运行，请等待完成") @@ -62,29 +83,48 @@ async def execute_script(): set_task_stage(task_id, "processing", logs) app_config = state.current_config() - thread = Thread(target=_run_script, args=(task_id, logger, logs, app_config), daemon=True) + thread = Thread( + target=_run_script, + args=(task_id, logger, logs, app_config, script_path, script_type), + daemon=True, + ) thread.start() - return {"success": True, "message": "脚本执行任务已启动", "task_id": task_id} + label = SCRIPT_LABELS.get(script_type, script_type) + return {"success": True, "message": f"{label}执行任务已启动", "task_id": task_id} @router.post("/api/script/save") -async def save_script_content(content: str = Body(..., embed=True)): +async def save_script_content( + content: str = Body(...), + script_type: str = Body("report"), +): + script_path = _resolve_script_path(script_type) try: - if SQL_SCRIPT.exists(): - shutil.copy(SQL_SCRIPT, SQL_SCRIPT.with_suffix(".sql.bak")) + if script_path.exists(): + shutil.copy(script_path, script_path.with_suffix(".sql.bak")) - SQL_SCRIPT.write_text(content, encoding="utf-8") - modified = datetime.fromtimestamp(SQL_SCRIPT.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S") + script_path.write_text(content, encoding="utf-8") + modified = datetime.fromtimestamp(script_path.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S") return {"success": True, "message": "脚本保存成功", "modified": modified} except Exception as exc: return {"success": False, "error": str(exc)} -def _run_script(task_id: str, logger: ProcessLogger, logs: list[str], app_config: AppConfig) -> None: +def _run_script( + task_id: str, + logger: ProcessLogger, + logs: list[str], + app_config: AppConfig, + script_path: Path, + script_type: str, +) -> None: temp_work_dir: Path | None = None + label = SCRIPT_LABELS.get(script_type, script_type) try: - logger.info("开始执行 SQL 脚本...") - if app_config.warehouse_type == "metrix": + logger.info(f"开始执行{label}...") + if script_type == "celldata": + _run_celldata_script(script_path, app_config, logger) + elif app_config.warehouse_type == "metrix": from app.services.pipeline import run_report_sql run_report_sql(app_config, logger) @@ -93,12 +133,18 @@ def _run_script(task_id: str, logger: ProcessLogger, logs: list[str], app_config temp_work_dir.mkdir(parents=True, exist_ok=True) processor = DataProcessor(app_config, temp_work_dir, logger) processor._execute_sql_script() - logger.success("SQL 脚本执行完成") + logger.success(f"{label}执行完成") set_task_stage(task_id, "completed", logs, status="completed") except Exception as exc: - logger.error(f"SQL 脚本执行失败: {exc}") + logger.error(f"{label}执行失败: {exc}") set_task_stage(task_id, "failed", logs, status="failed") finally: if temp_work_dir and temp_work_dir.exists(): shutil.rmtree(temp_work_dir, ignore_errors=True) state.reset_task_lock() + + +def _run_celldata_script(script_path: Path, app_config: AppConfig, logger: ProcessLogger) -> None: + from app.services.cell_data import execute_celldata_script + + execute_celldata_script(script_path, app_config, logger) diff --git a/app/api/routers/tasks.py b/app/api/routers/tasks.py index 528a02a..87a707e 100644 --- a/app/api/routers/tasks.py +++ b/app/api/routers/tasks.py @@ -13,7 +13,8 @@ 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.config import CELLDATA_SCRIPT +from app.services.cell_data import copy_celldata_tables_to_capacity, execute_celldata_script, refresh_cell_data from app.services.license import LicenseError, check_processing_allowed @@ -227,5 +228,7 @@ def _try_refresh_cell_data(app_config: AppConfig, work_dir: Path, logger: Proces f"CellData 更新完成：{result.imported_rows} 行" f"（解析 {result.parsed_rows}，跳过 {result.skipped_rows}）" ) + execute_celldata_script(CELLDATA_SCRIPT, app_config, logger) + copy_celldata_tables_to_capacity(app_config, logger) except Exception as exc: logger.warning(f"CellData 更新失败，继续容量处理: {exc}") diff --git a/app/config.py b/app/config.py index 0ae520f..242ce4e 100644 --- a/app/config.py +++ b/app/config.py @@ -24,6 +24,7 @@ BASE_DIR = _resolve_base_dir() CACHE_DIR = BASE_DIR / "cache" CONFIG_FILE = BASE_DIR / "Configure.json" SQL_SCRIPT = BASE_DIR / "ReportScript.sql" +CELLDATA_SCRIPT = BASE_DIR / "CellDataScript.sql" @dataclass diff --git a/app/services/cell_data.py b/app/services/cell_data.py index 90110c4..9d730fa 100644 --- a/app/services/cell_data.py +++ b/app/services/cell_data.py @@ -428,3 +428,153 @@ class CellDataProcessor: def refresh_cell_data(app_config: AppConfig, work_dir: Path, logger=None) -> CellDataResult: return CellDataProcessor(app_config, work_dir, logger).run() + + +def execute_celldata_script(script_path: Path, app_config: AppConfig, logger=None) -> None: + if not script_path.exists(): + if logger: + logger.info("CellData 脚本文件不存在，跳过") + return + sql_text = script_path.read_text(encoding="utf-8").strip() + if not sql_text: + if logger: + logger.info("CellData 脚本为空，跳过") + return + + from app.processor import DataProcessor + + statements = DataProcessor.parse_sql_script(sql_text) + if not statements: + if logger: + logger.info("CellData 脚本中没有有效语句，跳过") + return + + mysql = app_config.cell_data.mysql.normalized() + 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: + for i, sql in enumerate(statements, 1): + preview = sql[:80].replace("\n", " ") + if logger: + logger.info(f"CellData SQL ({i}/{len(statements)}): {preview}...") + cursor.execute(sql) + affected = cursor.rowcount if cursor.rowcount >= 0 else 0 + if affected > 0 and logger: + logger.info(f"完成，影响 {affected} 行") + conn.commit() + if logger: + logger.success(f"CellData 脚本执行完成，共 {len(statements)} 条语句") + except Exception: + conn.rollback() + raise + finally: + conn.close() + + +def copy_celldata_tables_to_capacity(app_config: AppConfig, logger=None) -> int: + cd_mysql = app_config.cell_data.mysql.normalized() + cap_mysql = app_config.mysql.normalized() + + if ( + cd_mysql.host == cap_mysql.host + and cd_mysql.port == cap_mysql.port + and cd_mysql.dbname == cap_mysql.dbname + ): + if logger: + logger.info("CellData 与容量数据库相同，跳过表复制") + return 0 + + cd_conn = pymysql.connect( + host=cd_mysql.host, port=cd_mysql.port, + user=cd_mysql.user, password=cd_mysql.passwd, + database=cd_mysql.dbname, charset="utf8mb4", + cursorclass=pymysql.cursors.DictCursor, + ) + cap_conn = pymysql.connect( + host=cap_mysql.host, port=cap_mysql.port, + user=cap_mysql.user, password=cap_mysql.passwd, + database=cap_mysql.dbname, charset="utf8mb4", + cursorclass=pymysql.cursors.DictCursor, + local_infile=True, autocommit=False, + ) + try: + tables = _list_tables(cd_conn) + if not tables: + if logger: + logger.info("CellData 数据库中没有表，跳过复制") + return 0 + + copied = 0 + for table in tables: + rows = _copy_one_table(cd_conn, cap_conn, table, logger) + if rows >= 0: + copied += 1 + cap_conn.commit() + if logger: + logger.success(f"已将 {copied} 张表从 CellData 复制到容量数据库") + return copied + except Exception: + cap_conn.rollback() + raise + finally: + cd_conn.close() + cap_conn.close() + + +def _list_tables(conn) -> list[str]: + with conn.cursor() as cur: + cur.execute("SHOW TABLES") + return [list(row.values())[0] for row in cur.fetchall()] + + +def _copy_one_table(src_conn, dst_conn, table: str, logger=None) -> int: + with src_conn.cursor() as cur: + cur.execute(f"SHOW CREATE TABLE `{table}`") + row = cur.fetchone() + create_sql = row.get("Create Table") or list(row.values())[1] + + with src_conn.cursor() as cur: + cur.execute(f"SELECT COUNT(*) AS `cnt` FROM `{table}`") + count = cur.fetchone()["cnt"] + + with dst_conn.cursor() as cur: + cur.execute(f"DROP TABLE IF EXISTS `{table}`") + cur.execute(create_sql) + + if count == 0: + if logger: + logger.info(f"复制表 {table}: 0 行（空表）") + return 0 + + with src_conn.cursor() as src_cur: + src_cur.execute(f"SELECT * FROM `{table}`") + columns = [desc[0] for desc in src_cur.description] + col_list = ", ".join(f"`{c}`" for c in columns) + placeholders = ", ".join(["%s"] * len(columns)) + insert_sql = f"INSERT INTO `{table}` ({col_list}) VALUES ({placeholders})" + + batch: list[tuple] = [] + inserted = 0 + with dst_conn.cursor() as dst_cur: + for row in src_cur: + batch.append(tuple(row.values())) + if len(batch) >= 5000: + dst_cur.executemany(insert_sql, batch) + inserted += len(batch) + batch.clear() + if batch: + dst_cur.executemany(insert_sql, batch) + inserted += len(batch) + + if logger: + logger.info(f"复制表 {table}: {inserted} 行") + return inserted diff --git a/docs/project_context.md b/docs/project_context.md index 62ed75f..d36fbe2 100644 --- a/docs/project_context.md +++ b/docs/project_context.md @@ -868,3 +868,14 @@ - 修复：`processor.py::_convert_datetime_column` 在 ISO8601 解析前先用正则剥离时区后缀（`+08:00`、`-05:30`、`Z`）和 `T` 分隔符，再按朴素日期时间解析，保留原始壁钟时间。 - SQL 清理：`ReportScript.sql` 删除 4G/5G 的 `DATE_ADD(INTERVAL 8 HOUR)` 补偿语句，因为数据导入阶段已正确保留本地时间，不再需要 SQL 层面修正。 - 验证：`python -m compileall -q app` 通过。 + +## 2026-06-25：CellData 脚本编辑与跨库表复制 + +- 新增 `CellDataScript.sql` 脚本路径（`config.py::CELLDATA_SCRIPT = BASE_DIR / "CellDataScript.sql"`），与 `ReportScript.sql` 并列，用于 CellData 数据导入后执行的 SQL。 +- 脚本 API（`script.py`）三个接口（读取/保存/执行）新增 `script_type` 参数（`report` / `celldata`），默认 `report` 保持向后兼容；CellData 脚本执行在 CellData 数据库上下文中运行。 +- 新增 `cell_data.py::execute_celldata_script()`：在 CellData MySQL 库上执行 CellDataScript.sql，复用 `DataProcessor.parse_sql_script` 解析。 +- 新增 `cell_data.py::copy_celldata_tables_to_capacity()`：列出 CellData 库所有表，逐表 `SHOW CREATE TABLE` → 目标库建表 → 分批 `SELECT * / INSERT INTO` 复制数据；两库相同时跳过。 +- CellData 独立处理（远程 / 本地上传）完成后自动执行 CellData 脚本（不复制表）。 +- 容量处理集成（`tasks.py` / `remote.py` 的 `_try_refresh_cell_data`）：CellData 数据更新 → 执行 CellData 脚本 → 复制表到容量库 → 后续 ReportScript.sql 可直接引用 CellData 表。 +- 前端脚本编辑页（`ScriptPanel.vue`）编辑器上方新增 Naive UI segment tabs 切换「报表脚本」和「CellData 脚本」；切换时检查未保存状态、加载对应脚本内容；保存和运行按钮跟随当前脚本类型。 +- 验证：`python -m compileall -q app` 通过；`frontend` `npm run build`（vue-tsc）通过；构建产物已清理。 diff --git a/frontend/src/components/ScriptPanel.vue b/frontend/src/components/ScriptPanel.vue index 6445c02..4e2331e 100644 --- a/frontend/src/components/ScriptPanel.vue +++ b/frontend/src/components/ScriptPanel.vue @@ -1,6 +1,17 @@