diff --git a/app/api/routers/database.py b/app/api/routers/database.py index e9d2e0a..0779230 100644 --- a/app/api/routers/database.py +++ b/app/api/routers/database.py @@ -1,9 +1,11 @@ +import csv import re from datetime import datetime +from pathlib import Path from typing import Optional import pandas as pd -from fastapi import APIRouter, Body, HTTPException +from fastapi import APIRouter, Body, File, Form, HTTPException, UploadFile from fastapi.responses import FileResponse from starlette.background import BackgroundTask @@ -181,6 +183,129 @@ def drop_all_tables(database_source: DatabaseSource = Body("main", embed=True)): raise HTTPException(status_code=500, detail=str(exc)) from exc +@router.post("/api/database/table/row/update") +def update_table_row( + table_name: str = Body(..., embed=True), + database_source: DatabaseSource = Body("main"), + identifier: dict = Body(...), + values: dict = Body(...), +): + if not identifier: + raise HTTPException(status_code=400, detail="缺少行定位信息") + db = _db(database_source) + try: + affected = db.update_row(table_name, identifier, values) + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) from exc + if affected == 0: + raise HTTPException(status_code=404, detail="未找到匹配的数据行，可能已被修改或删除") + return {"success": True, "message": "已更新该行", "affected_rows": affected} + + +@router.post("/api/database/table/row/delete") +def delete_table_row( + table_name: str = Body(..., embed=True), + database_source: DatabaseSource = Body("main"), + identifier: dict = Body(...), +): + if not identifier: + raise HTTPException(status_code=400, detail="缺少行定位信息") + db = _db(database_source) + try: + affected = db.delete_row(table_name, identifier) + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) from exc + if affected == 0: + raise HTTPException(status_code=404, detail="未找到匹配的数据行，可能已被删除") + return {"success": True, "message": "已删除该行", "affected_rows": affected} + + +def _table_columns(db, table_name: str) -> list[str]: + info = db.get_table_info(table_name) + return [str(column["Field"]) for column in info.get("columns", [])] + + +@router.post("/api/database/table/template") +def download_table_template( + table_name: str = Body(..., embed=True), + database_source: DatabaseSource = Body("main"), +): + db = _db(database_source) + try: + columns = _table_columns(db, table_name) + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) from exc + if not columns: + raise HTTPException(status_code=400, detail="无法获取表字段，无法生成模板") + + CACHE_DIR.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + filepath = CACHE_DIR / f"{table_name}_template_{timestamp}.csv" + try: + with filepath.open("w", encoding="utf-8-sig", newline="") as handle: + csv.writer(handle).writerow(columns) + except Exception: + remove_file_safely(filepath) + raise + return FileResponse( + path=str(filepath), + filename=f"{table_name}_模板.csv", + media_type="text/csv", + background=BackgroundTask(remove_file_safely, filepath), + ) + + +def _read_csv_header(path: Path) -> list[str]: + with path.open("r", encoding="utf-8-sig", newline="") as handle: + for row in csv.reader(handle): + return [str(cell).strip() for cell in row] + return [] + + +# Sync def so FastAPI runs it in a threadpool: file read + DB insert are blocking. +@router.post("/api/database/table/import") +def import_table_csv( + file: UploadFile = File(...), + table_name: str = Form(...), + database_source: str = Form("main"), +): + if not file.filename or not file.filename.lower().endswith(".csv"): + raise HTTPException(status_code=400, detail="仅支持 CSV 格式文件") + db = _db(database_source) + try: + columns = _table_columns(db, table_name) + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) from exc + if not columns: + raise HTTPException(status_code=400, detail="无法获取表字段") + + CACHE_DIR.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + tmp_path = CACHE_DIR / f"import_{timestamp}.csv" + tmp_path.write_bytes(file.file.read()) + try: + header = _read_csv_header(tmp_path) + if not header: + raise HTTPException(status_code=400, detail="CSV 文件为空或缺少表头") + missing = [name for name in columns if name not in header] + extra = [name for name in header if name not in columns] + if missing or extra: + parts = [] + if missing: + parts.append("缺少字段: " + ", ".join(missing)) + if extra: + parts.append("多余字段: " + ", ".join(extra)) + raise HTTPException(status_code=400, detail="CSV 字段与模板不一致，导入失败。" + "；".join(parts)) + imported = db.import_csv(str(tmp_path), table_name) + except HTTPException: + raise + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) from exc + finally: + remove_file_safely(tmp_path) + return {"success": True, "message": f"导入成功，共 {imported} 行", "imported_rows": imported} + + @router.post("/api/database/execute") def execute_sql( sql: str = Body(..., embed=True), diff --git a/app/database.py b/app/database.py index 3dac2fa..bffaac4 100644 --- a/app/database.py +++ b/app/database.py @@ -1,6 +1,7 @@ """ 数据库连接与操作模块 """ +import csv import pymysql from contextlib import contextmanager from typing import Any, Dict, List, Optional, Tuple @@ -199,6 +200,62 @@ class DatabaseManager: "total_pages": (total + page_size - 1) // page_size } + def _row_conditions(self, identifier: Dict[str, Any]) -> Tuple[str, List[Any]]: + """按整行原值构造 WHERE：NULL 用 IS NULL；调用方再加 LIMIT 1 只影响单行。""" + parts: List[str] = [] + params: List[Any] = [] + for col, val in identifier.items(): + if val is None: + parts.append(f"`{col}` IS NULL") + else: + parts.append(f"`{col}` = %s") + params.append(val) + return (" AND ".join(parts) if parts else "1 = 0"), params + + def update_row(self, table_name: str, identifier: Dict[str, Any], values: Dict[str, Any]) -> int: + """更新单行：按 identifier(原始整行) 定位、LIMIT 1，避免影响重复行。返回影响行数。""" + if not values: + return 0 + set_clause = ", ".join(f"`{col}` = %s" for col in values) + set_params = list(values.values()) + where_clause, where_params = self._row_conditions(identifier) + sql = f"UPDATE `{table_name}` SET {set_clause} WHERE {where_clause} LIMIT 1" + with self.get_connection() as conn: + with conn.cursor() as cursor: + cursor.execute(sql, set_params + where_params) + conn.commit() + return cursor.rowcount + + def delete_row(self, table_name: str, identifier: Dict[str, Any]) -> int: + """删除单行：按 identifier(原始整行) 定位、LIMIT 1。返回影响行数。""" + where_clause, where_params = self._row_conditions(identifier) + sql = f"DELETE FROM `{table_name}` WHERE {where_clause} LIMIT 1" + with self.get_connection() as conn: + with conn.cursor() as cursor: + cursor.execute(sql, where_params) + conn.commit() + return cursor.rowcount + + def import_csv(self, file_path: str, table_name: str) -> int: + """按 CSV 表头列追加导入（列须与表字段一致，由调用方校验）。返回导入行数。""" + with open(file_path, "r", encoding="utf-8-sig", newline="") as handle: + reader = csv.reader(handle) + try: + header = next(reader) + except StopIteration: + return 0 + columns = [str(name).strip() for name in header] + width = len(columns) + data: List[Tuple] = [] + for row in reader: + if not any(str(cell).strip() for cell in row): + continue + cells = list(row[:width]) + [""] * (width - len(row)) + data.append(tuple(cells)) + if not data: + return 0 + return self.bulk_insert(table_name, columns, data) + def truncate_table(self, table_name: str) -> bool: """清空表""" with self.get_connection() as conn: diff --git a/app/services/cell_data.py b/app/services/cell_data.py index 9d730fa..cec0f1a 100644 --- a/app/services/cell_data.py +++ b/app/services/cell_data.py @@ -253,37 +253,58 @@ class CellDataProcessor: 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"] + key_field = str(key_config["field"]) rows_by_key: dict[str, dict[str, str]] = {} + self._log(f"开始解压并解析 {len(local_files)} 个 ZIP 文件...") for selected, local_path in local_files: + band_label = selected.band or "未识别频段" + size_kb = local_path.stat().st_size / 1024 if local_path.exists() else 0 + self._log(f"[{band_label}] 解压 {local_path.name}（{size_kb:.0f} KB）...") + zip_parsed_before = result.parsed_rows + zip_skipped_before = result.skipped_rows with zipfile.ZipFile(local_path) as zf: sources = list(mapping["sources"]) - for info in zf.infolist(): + csv_entries = [info for info in zf.infolist() if Path(info.filename).name.lower().endswith(".csv")] + self._log(f" 压缩包内含 {len(csv_entries)} 个 CSV 文件") + for info in csv_entries: name = Path(info.filename).name - if not name.lower().endswith(".csv"): - continue matching_sources = [ source for source in sources if name.startswith(source["file_prefix"]) and (not selected.band or source["band"] == selected.band) ] if not matching_sources: + self._log(f" 跳过未匹配规则的文件: {name}") continue if not selected.band and len(matching_sources) > 1: - self._log(f"跳过无法识别频段的文件: {name}") + self._log(f" 跳过无法识别频段的文件: {name}") continue for source in matching_sources: raw = zf.read(info.filename) text = self._decode_csv(raw) reader = csv.DictReader(text.splitlines()) + added = 0 + skipped = 0 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 + skipped += 1 continue - row[str(key_config["field"])] = key + row[key_field] = key rows_by_key[key] = {column: row.get(column, "") for column in CELLINFO_COLUMNS} result.parsed_rows += 1 + added += 1 + self._log(f" 解析 {name}（频段 {source.get('band', '') or '通用'}）：有效 {added} 行，跳过 {skipped} 行") + self._log( + f"[{band_label}] {local_path.name} 解析完成：" + f"本包有效 {result.parsed_rows - zip_parsed_before} 行，跳过 {result.skipped_rows - zip_skipped_before} 行" + ) + self._log( + f"全部解析完成：累计有效 {result.parsed_rows} 行，按 {key_field} 去重后 {len(rows_by_key)} 行，" + f"累计跳过 {result.skipped_rows} 行" + ) return list(rows_by_key.values()) def _map_row(self, fields: dict[str, Any], csv_row: dict[str, str]) -> dict[str, str]: @@ -320,6 +341,7 @@ class CellDataProcessor: 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") + self._log(f"准备写入表 `{table}`（库 {mysql.dbname}@{mysql.host}:{mysql.port}），共 {len(rows)} 行") conn = pymysql.connect( host=mysql.host, port=mysql.port, @@ -330,22 +352,29 @@ class CellDataProcessor: cursorclass=pymysql.cursors.DictCursor, autocommit=False, ) + self._log("已连接 CellData 数据库") try: with conn.cursor() as cursor: self._ensure_cellinfo_table(cursor, table) + self._log(f"已确认表结构 `{table}`") cursor.execute(f"TRUNCATE TABLE `{table}`") + self._log(f"已清空表 `{table}`（TRUNCATE）") 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): + total = len(values) + for start in range(0, total, 1000): cursor.executemany( f"INSERT INTO `{table}` ({columns}) VALUES ({placeholders})", values[start:start + 1000], ) + self._log(f"写入中 {min(start + 1000, total)}/{total} 行...") conn.commit() + self._log(f"已提交，成功写入 {len(rows)} 行到 `{table}`") return len(rows) - except Exception: + except Exception as exc: conn.rollback() + self._log(f"写入失败，已回滚: {exc}") raise finally: conn.close() diff --git a/app/warehouse.py b/app/warehouse.py index 2db94a1..858583a 100644 --- a/app/warehouse.py +++ b/app/warehouse.py @@ -9,6 +9,7 @@ query_table / truncate_table / drop_table / drop_all_tables / execute_sql。 """ from __future__ import annotations +from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.config import AppConfig @@ -34,6 +35,10 @@ def _quote_value(value: str) -> str: return "'" + str(value).replace("\\", "\\\\").replace("'", "''") + "'" +def _sql_literal(value: Any) -> str: + return "NULL" if value is None else _quote_value(str(value)) + + class MetrixWarehouse: """用 Metrix 数据库 API 实现 DatabaseManager 的只读/管理子集。""" @@ -153,6 +158,41 @@ class MetrixWarehouse: self._run(drop_sql) return {"success": True, "dropped_count": len(tables), "tables": tables} + def _row_where(self, identifier: Dict[str, Any]) -> str: + parts = [ + f"{_quote_ident(col)} IS NULL" if val is None else f"{_quote_ident(col)} = {_sql_literal(val)}" + for col, val in identifier.items() + ] + return " AND ".join(parts) if parts else "1 = 0" + + def update_row(self, table_name: str, identifier: Dict[str, Any], values: Dict[str, Any]) -> int: + if not values: + return 0 + set_clause = ", ".join(f"{_quote_ident(col)} = {_sql_literal(val)}" for col, val in values.items()) + sql = f"UPDATE {_quote_ident(table_name)} SET {set_clause} WHERE {self._row_where(identifier)} LIMIT 1" + ok, result = self.execute_sql(sql) + if not ok: + raise RuntimeError(str(result)) + return int(result.get("affected_rows", 0)) if isinstance(result, dict) else 0 + + def delete_row(self, table_name: str, identifier: Dict[str, Any]) -> int: + sql = f"DELETE FROM {_quote_ident(table_name)} WHERE {self._row_where(identifier)} LIMIT 1" + ok, result = self.execute_sql(sql) + if not ok: + raise RuntimeError(str(result)) + return int(result.get("affected_rows", 0)) if isinstance(result, dict) else 0 + + def import_csv(self, file_path: str, table_name: str) -> int: + """走 Metrix 平台导入 API 追加导入到已存在的表（字段映射由平台按列名处理）。""" + job_id = self.client.import_csv( + self.conn_id, table_name, Path(file_path), mode="append", + database=self.database, create_table=False, + ) + job = self.client.wait_job(job_id) + if job.get("status") != "success": + raise RuntimeError(job.get("error_code") or job.get("status") or "导入失败") + return int(job.get("row_count") or 0) + def execute_sql(self, sql: str) -> Tuple[bool, Any]: try: res = self.client.run_script(self.conn_id, content=sql, database=self.database, run_timeout=600) diff --git a/docs/project_context.md b/docs/project_context.md index 4b9b8e2..e6be2c9 100644 --- a/docs/project_context.md +++ b/docs/project_context.md @@ -888,3 +888,20 @@ - 数据源合并：`source_type` 从 `ftp | sftp | metrix` 改为 `external | metrix`；FTP/SFTP 协议选择保留在远程数据源配置的 `RemoteData.protocol` 中，`external` 统一代表外部储存。旧值 `ftp`/`sftp` 自动规范化为 `external`。 - 全局状态：`metrixEnabled` 通过 `composables/metrixEnabled.ts` 共享响应式状态，`AppShell` 登录后和激活弹窗切换时同步更新。 - 验证：`python -m compileall -q app` 通过；`frontend` `npm run build`（vue-tsc）通过；构建产物已清理。 + +## 2026-06-26：数据管理增强（列宽可调 + 行编辑/删除 + 模板/导入）与 CellData 处理日志细化 + +数据管理（`frontend/src/components/DatabasePanel.vue` + `app/api/routers/database.py` + `app/database.py` + `app/warehouse.py`）： +- 数据表列宽可拖拽：`n-data-table` 每列加 `resizable + width`，`scroll-x` 计入新增操作列宽度。 +- 新增固定右侧「操作」列：每行「编辑 / 删除」。编辑弹窗按列字段逐项填写后保存；删除二次确认；操作后自动刷新当前页。 +- 行定位策略：以「整行原值」作为 WHERE 条件 + `LIMIT 1`（NULL 用 `IS NULL`），无主键表也能精确改/删单行、避免误伤重复行。后端 `DatabaseManager.update_row/delete_row`（参数化）与 `MetrixWarehouse.update_row/delete_row`（经 `execute_sql` 字面量 SQL）。 +- 工具栏新增「模板」「导入」（在 刷新/清空/删除 同组）：模板下载当前表字段的 CSV 表头（`POST /api/database/table/template`，FileResponse）；导入上传 CSV（`POST /api/database/table/import`，multipart，sync def 走线程池）。**导入按模板校验**：CSV 表头必须与表字段完全一致，缺字段/多字段一律 400 失败；通过后追加导入（`DatabaseManager.import_csv` 走 `bulk_insert`；Metrix 走平台 `/import` mode=append、create_table=false）。 +- 新接口均支持 `database_source`(main/cell_data) 双库；前端用 `client.upload`(multipart) 导入、`download`(POST) 下模板。 + +CellData 处理日志细化（`app/services/cell_data.py`）： +- `_parse_zip_files`（解压/解析）：逐 ZIP 打「解压 {频段}/{文件名}（KB）」+「含 N 个 CSV」，逐 CSV 打「解析 {文件名}（频段）：有效 X / 跳过 Y 行」，每包小计，末尾打「累计有效 / 去重后 / 累计跳过」。 +- `_replace_cellinfo`（上传/导入）：打「准备写入表/库」「已连接」「确认表结构」「清空表(TRUNCATE)」「写入中 written/total 行」分批进度「已提交，成功写入 N 行」，失败回滚也记录。 + +附带修复：`frontend/src/vite-env.d.ts` 增加 `declare module 'monaco-editor/esm/vs/basic-languages/mysql/mysql.js'`，修复 ScriptPanel SQL 补全的深层导入缺类型声明导致 `vue-tsc`（`npm run build`）失败（TS7016）。 + +- 验证：`python -m compileall -q app` 通过；`frontend` `npm run build`（vue-tsc）通过（DatabasePanel chunk≈124KB）。 diff --git a/frontend/src/components/DatabasePanel.vue b/frontend/src/components/DatabasePanel.vue index 3b6fc78..2ca3417 100644 --- a/frontend/src/components/DatabasePanel.vue +++ b/frontend/src/components/DatabasePanel.vue @@ -88,6 +88,8 @@ 刷新 + 模板 + 导入 清空 删除 @@ -251,12 +253,60 @@ + + +
+
+ + +
+
+ + + + +
+

+ 请先下载「模板」，按模板填写数据后导入到表 {{ selectedTable }}。 + CSV 字段必须与表字段完全一致，否则导入失败。 +

+
+ + 选择 CSV 文件 + {{ importFile?.name || '未选择文件' }} +
+ 下载模板 +
+ +