feat: 数据管理列宽可调/行编辑删除/模板导入,CellData 处理日志细化

- 数据表列宽可拖拽 + 新增「操作」列(编辑/删除该行)
- 工具栏新增「模板」「导入」:CSV 按模板字段校验,不一致则导入失败
- 后端 DatabaseManager/MetrixWarehouse 增加行更新/删除/CSV 导入 + 对应路由
- CellData 解压/解析/写库日志细化(逐文件、分批进度)
- 修复 ScriptPanel monaco mysql 深层导入缺类型声明导致的 vue-tsc 构建失败
This commit is contained in:
2026-06-26 05:32:06 +08:00
parent 639bae694d
commit f0bbc77763
7 changed files with 543 additions and 15 deletions
+126 -1
View File
@@ -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),
+57
View File
@@ -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:
+36 -7
View File
@@ -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()
+40
View File
@@ -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)