Metrix/CapacityReport
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 实时重载配置文件

Browse Source
This commit is contained in:
nixevol
2026-05-19 17:20:46 +08:00
parent 59b14b4938
commit db543395ba
9 changed files with 61 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/routers/config.py
+9 -3
View File
@@ -15,12 +15,12 @@ router = APIRouter(tags=["config"])
@router.get("/api/config")
async def get_config():
return state.config.to_dict()
return state.current_config().to_dict()
@router.get("/api/config/full")
async def get_config_full():
return state.config.to_dict_full()
return state.current_config().to_dict_full()
@router.post("/api/config/mysql")
@@ -31,6 +31,7 @@ async def update_mysql_config(
passwd: str = Body(...),
dbname: str = Body(...),
):
state.reload_config()
state.config.mysql.host = host
state.config.mysql.port = port
state.config.mysql.user = user
@@ -42,6 +43,7 @@ async def update_mysql_config(
@router.post("/api/config/remote")
async def update_remote_config(config: dict[str, Any] = Body(...)):
state.reload_config()
state.config.remote_data = RemoteDataConfig.from_dict(config)
state.config.save()
return {"success": True, "message": "远程数据配置已更新", "update": state.config.update}
@@ -49,6 +51,7 @@ async def update_remote_config(config: dict[str, Any] = Body(...)):
@router.post("/api/config/history-retention")
async def update_history_retention(config: dict[str, Any] = Body(...)):
state.reload_config()
state.config.history_retention = HistoryRetentionConfig.from_dict(config)
state.config.save()
return {"success": True, "message": "处理历史保留配置已更新", "update": state.config.update}
@@ -56,6 +59,7 @@ async def update_history_retention(config: dict[str, Any] = Body(...)):
@router.post("/api/config/sheet-filter")
async def update_sheet_filter(filters: list[str] = Body(...)):
state.reload_config()
state.config.sheet_filter = filters
state.config.save()
return {"success": True, "message": "Sheet 过滤规则已更新", "update": state.config.update}
@@ -63,6 +67,7 @@ async def update_sheet_filter(filters: list[str] = Body(...)):
@router.post("/api/config/extract-fields")
async def update_extract_fields(fields: list[dict[str, Any]] = Body(...)):
state.reload_config()
state.config.extract_fields = fields
state.config.save()
return {"success": True, "message": "字段映射配置已更新", "update": state.config.update}
@@ -72,7 +77,7 @@ async def update_extract_fields(fields: list[dict[str, Any]] = Body(...)):
async def download_config():
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"Configure_{timestamp}.json"
content = json.dumps(state.config.to_file_dict(), ensure_ascii=False, indent=2)
content = json.dumps(state.current_config().to_file_dict(), ensure_ascii=False, indent=2)
return Response(
content=content,
media_type="application/json",
@@ -90,6 +95,7 @@ async def upload_config(file: UploadFile = File(...)):
if not isinstance(data, dict):
raise ValueError("配置文件格式错误：必须是 JSON 对象")
state.reload_config()
_apply_config_data(data)
state.config.save()
return {"success": True, "message": "配置文件上传成功", "update": state.config.update}
app/api/routers/database.py
+15 -11
View File
@@ -53,16 +53,20 @@ def _dataframe_from_table(db: DatabaseManager, table_name: str) -> pd.DataFrame:
return pd.DataFrame(result["data"], columns=columns)
def _db() -> DatabaseManager:
return DatabaseManager(state.current_config())
@router.post("/api/database/test")
async def test_database():
db = DatabaseManager(state.config)
db = _db()
success, message = db.test_connection()
return {"success": success, "message": message}
@router.get("/api/database/info")
async def get_database_info():
db = DatabaseManager(state.config)
db = _db()
try:
return {"success": True, **db.get_server_info()}
except Exception as exc:
@@ -72,7 +76,7 @@ async def get_database_info():
@router.get("/api/database/tables")
@router.post("/api/database/tables")
async def get_tables():
db = DatabaseManager(state.config)
db = _db()
try:
return {"tables": db.get_tables()}
except Exception as exc:
@@ -81,7 +85,7 @@ async def get_tables():
@router.post("/api/database/table/info")
async def get_table_info(table_name: str = Body(..., embed=True)):
db = DatabaseManager(state.config)
db = _db()
try:
return db.get_table_info(table_name)
except Exception as exc:
@@ -96,7 +100,7 @@ async def query_table_data(
order_by: Optional[str] = Body(None),
order_dir: str = Body("ASC"),
):
db = DatabaseManager(state.config)
db = _db()
try:
return db.query_table(table_name, page, page_size, order_by=order_by, order_dir=order_dir)
except Exception as exc:
@@ -112,7 +116,7 @@ async def query_table_with_filter(
order_by: Optional[str] = Body(None),
order_dir: str = Body("ASC"),
):
db = DatabaseManager(state.config)
db = _db()
try:
return db.query_table(
table_name,
@@ -128,7 +132,7 @@ async def query_table_with_filter(
@router.post("/api/database/table/truncate")
async def truncate_table(table_name: str = Body(..., embed=True)):
db = DatabaseManager(state.config)
db = _db()
try:
db.truncate_table(table_name)
return {"success": True, "message": f"表 {table_name} 已清空"}
@@ -138,7 +142,7 @@ async def truncate_table(table_name: str = Body(..., embed=True)):
@router.post("/api/database/table/drop")
async def drop_table(table_name: str = Body(..., embed=True)):
db = DatabaseManager(state.config)
db = _db()
try:
db.drop_table(table_name)
return {"success": True, "message": f"表 {table_name} 已删除"}
@@ -148,7 +152,7 @@ async def drop_table(table_name: str = Body(..., embed=True)):
@router.post("/api/database/table/drop-all")
async def drop_all_tables():
db = DatabaseManager(state.config)
db = _db()
try:
result = db.drop_all_tables()
return {
@@ -163,7 +167,7 @@ async def drop_all_tables():
@router.post("/api/database/execute")
async def execute_sql(sql: str = Body(..., embed=True)):
db = DatabaseManager(state.config)
db = _db()
try:
success, result = db.execute_sql(sql)
if success:
@@ -190,7 +194,7 @@ async def download_table(
if file_format == "csv" and len(requested_tables) != 1:
raise HTTPException(status_code=400, detail="CSV 每次只能导出一张表")
db = DatabaseManager(state.config)
db = _db()
try:
available_tables = set(db.get_tables())
missing_tables = [name for name in requested_tables if name not in available_tables]
app/api/routers/health.py
+1 -1
View File
@@ -18,7 +18,7 @@ async def health_check():
}
try:
db_manager = DatabaseManager(state.config)
db_manager = DatabaseManager(state.current_config())
server_info = db_manager.get_server_info()
checks["database"] = {
"status": "ok",
app/api/routers/remote.py
+7 -5
View File
@@ -6,7 +6,7 @@ from typing import Any
from fastapi import APIRouter, Body, HTTPException
from app import state
from app.config import CACHE_DIR, RemoteDataConfig
from app.config import AppConfig, CACHE_DIR, RemoteDataConfig
from app.processor import DataProcessor, ProcessLogger
from app.services.remote_download import RemoteDataDownloader
@@ -16,7 +16,7 @@ router = APIRouter(tags=["remote"])
@router.post("/api/remote/test")
async def test_remote_connection(config: dict[str, Any] | None = Body(None)):
remote_config = RemoteDataConfig.from_dict(config) if config else state.config.remote_data
remote_config = RemoteDataConfig.from_dict(config) if config else state.current_config().remote_data
try:
RemoteDataDownloader(remote_config).test_connection()
return {"success": True, "message": "远程服务器连接成功"}
@@ -29,7 +29,8 @@ async def start_remote_processing():
if state.global_task_lock["locked"]:
raise HTTPException(status_code=409, detail="已有任务在运行，请等待当前任务完成")
remote_config = state.config.remote_data.normalized()
app_config = state.current_config()
remote_config = app_config.remote_data.normalized()
if not remote_config.enabled:
raise HTTPException(status_code=400, detail="请先启用远程数据配置")
@@ -68,7 +69,7 @@ async def start_remote_processing():
thread = Thread(
target=_run_remote_processing,
args=(task_id, work_dir, remote_config, logger),
args=(task_id, work_dir, app_config, remote_config, logger),
daemon=True,
)
thread.start()
@@ -94,6 +95,7 @@ def _set_task_stage(task_id: str, stage: str, logs: list[str], status: str = "pr
def _run_remote_processing(
task_id: str,
work_dir: Path,
app_config: AppConfig,
remote_config: RemoteDataConfig,
logger: ProcessLogger,
) -> None:
@@ -115,7 +117,7 @@ def _run_remote_processing(
state.history_manager.update(task_id, file_count=download_result.file_count)
logger.set_stage("extracting")
processor = DataProcessor(state.config, work_dir, logger)
processor = DataProcessor(app_config, work_dir, logger)
result = processor.process()
status = "completed" if result.get("success") else "failed"
if status == "completed" and remote_config.auto_delete_source:
app/api/routers/script.py
+5 -5
View File
@@ -7,7 +7,7 @@ from threading import Thread
from fastapi import APIRouter, Body, HTTPException
from app import state
from app.config import CACHE_DIR, SQL_SCRIPT
from app.config import AppConfig, CACHE_DIR, SQL_SCRIPT
from app.processor import DataProcessor, ProcessLogger
@@ -59,8 +59,9 @@ async def execute_script():
logger = ProcessLogger(log_file=None, callback=log_callback)
state.processing_tasks[task_id] = {"logs": [], "status": "processing"}
app_config = state.current_config()
thread = Thread(target=_run_script, args=(task_id, logger, logs), daemon=True)
thread = Thread(target=_run_script, args=(task_id, logger, logs, app_config), daemon=True)
thread.start()
return {"success": True, "message": "脚本执行任务已启动", "task_id": task_id}
@@ -78,14 +79,14 @@ async def save_script_content(content: str = Body(..., embed=True)):
return {"success": False, "error": str(exc)}
def _run_script(task_id: str, logger: ProcessLogger, logs: list[str]) -> None:
def _run_script(task_id: str, logger: ProcessLogger, logs: list[str], app_config: AppConfig) -> None:
temp_work_dir: Path | None = None
try:
logger.info("开始执行 SQL 脚本...")
temp_work_dir = CACHE_DIR / task_id
temp_work_dir.mkdir(parents=True, exist_ok=True)
processor = DataProcessor(state.config, temp_work_dir, logger)
processor = DataProcessor(app_config, temp_work_dir, logger)
processor._execute_sql_script()
logger.success("SQL 脚本执行完成")
state.processing_tasks[task_id] = {"logs": logs.copy(), "status": "completed"}
@@ -96,4 +97,3 @@ def _run_script(task_id: str, logger: ProcessLogger, logs: list[str]) -> None:
if temp_work_dir and temp_work_dir.exists():
shutil.rmtree(temp_work_dir, ignore_errors=True)
state.reset_task_lock()
app/api/routers/tasks.py
+5 -3
View File
@@ -6,6 +6,7 @@ from typing import Any
from fastapi import APIRouter, Body, HTTPException
from app import state
from app.config import AppConfig
from app.processor import DataProcessor, ProcessLogger
@@ -104,6 +105,7 @@ async def start_processing(task_id: str = Body(..., embed=True)):
callback=log_callback,
stage_callback=stage_callback,
)
app_config = state.current_config()
state.history_manager.update(task_id, status="processing")
state.processing_tasks[task_id] = {"logs": [], "status": "processing", "stage": current_stage}
state.global_task_lock.update(
@@ -115,7 +117,7 @@ async def start_processing(task_id: str = Body(..., embed=True)):
}
)
thread = Thread(target=_run_processing, args=(task_id, work_dir, logger), daemon=True)
thread = Thread(target=_run_processing, args=(task_id, work_dir, logger, app_config), daemon=True)
thread.start()
return {"success": True, "message": "处理任务已启动", "task_id": task_id}
@@ -166,9 +168,9 @@ def _set_task_stage(task_id: str, stage: str, logs: list[str], status: str = "pr
state.global_task_lock["stage"] = stage
def _run_processing(task_id: str, work_dir: Path, logger: ProcessLogger) -> None:
def _run_processing(task_id: str, work_dir: Path, logger: ProcessLogger, app_config: AppConfig) -> None:
try:
processor = DataProcessor(state.config, work_dir, logger)
processor = DataProcessor(app_config, work_dir, logger)
result = processor.process()
status = "completed" if result.get("success") else "failed"
state.history_manager.update(