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 16:08:46 +08:00
parent 3240027d48
commit f8a5269284
6 changed files with 142 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/routers/remote.py
+26 -4
View File
@@ -38,15 +38,25 @@ async def start_remote_processing():
work_dir.mkdir(parents=True, exist_ok=True)
logs: list[str] = []
current_stage = "downloading"
def log_callback(message: str) -> None:
logs.append(message)
state.processing_tasks[task_id] = {"logs": logs.copy(), "status": "processing"}
_set_task_stage(task_id, current_stage, logs)
logger = ProcessLogger(log_file=work_dir / "log.txt", callback=log_callback)
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,
)
state.history_manager.create(work_dir, 0, record_id=task_id)
state.history_manager.update(task_id, status="processing")
state.processing_tasks[task_id] = {"logs": [], "status": "processing"}
state.processing_tasks[task_id] = {"logs": [], "status": "processing", "stage": current_stage}
state.global_task_lock.update(
{
"locked": True,
@@ -71,6 +81,16 @@ async def start_remote_processing():
}
def _set_task_stage(task_id: str, stage: str, logs: list[str], status: str = "processing") -> None:
state.processing_tasks[task_id] = {
"logs": logs.copy(),
"status": status,
"stage": stage,
}
if state.global_task_lock["task_id"] == task_id:
state.global_task_lock["stage"] = stage
def _run_remote_processing(
task_id: str,
work_dir: Path,
@@ -93,7 +113,7 @@ def _run_remote_processing(
raise RuntimeError("远程目录中未下载到任何文件")
state.history_manager.update(task_id, file_count=download_result.file_count)
state.global_task_lock["stage"] = "processing"
logger.set_stage("extracting")
processor = DataProcessor(state.config, work_dir, logger)
result = processor.process()
@@ -115,6 +135,7 @@ def _run_remote_processing(
state.processing_tasks[task_id] = {
"logs": state.history_manager.get_logs(task_id),
"status": status,
"stage": status,
}
except Exception as exc:
logger.error(f"远程自动化任务失败: {exc}")
@@ -122,6 +143,7 @@ def _run_remote_processing(
state.processing_tasks[task_id] = {
"logs": state.history_manager.get_logs(task_id),
"status": "failed",
"stage": "failed",
}
finally:
try:
app/api/routers/tasks.py
+33 -5
View File
@@ -38,7 +38,7 @@ async def get_global_task_status():
return {
"has_active": True,
"task_id": task_id,
"stage": "processing",
"stage": active_tasks[task_id].get("stage", "processing"),
"logs": active_tasks[task_id].get("logs", []),
}
@@ -88,14 +88,24 @@ async def start_processing(task_id: str = Body(..., embed=True)):
raise HTTPException(status_code=400, detail="工作目录不存在")
logs: list[str] = []
current_stage = "processing"
def log_callback(message: str) -> None:
logs.append(message)
state.processing_tasks[task_id] = {"logs": logs.copy(), "status": "processing"}
_set_task_stage(task_id, current_stage, logs)
logger = ProcessLogger(log_file=work_dir / "log.txt", callback=log_callback)
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,
)
state.history_manager.update(task_id, status="processing")
state.processing_tasks[task_id] = {"logs": [], "status": "processing"}
state.processing_tasks[task_id] = {"logs": [], "status": "processing", "stage": current_stage}
state.global_task_lock.update(
{
"locked": True,
@@ -116,7 +126,12 @@ async def get_processing_status(task_id: str = Body(..., embed=True)):
if task_id in state.processing_tasks:
task_info = state.processing_tasks[task_id]
logs = task_info.get("logs") or state.history_manager.get_logs(task_id)
return {"task_id": task_id, "status": task_info["status"], "logs": logs}
return {
"task_id": task_id,
"status": task_info["status"],
"stage": task_info.get("stage"),
"logs": logs,
}
record = state.history_manager.get(task_id)
if not record:
@@ -125,6 +140,7 @@ async def get_processing_status(task_id: str = Body(..., embed=True)):
return {
"task_id": task_id,
"status": record.status,
"stage": record.status,
"logs": state.history_manager.get_logs(task_id),
"elapsed_time": record.elapsed_time,
"error": record.error,
@@ -140,6 +156,16 @@ def _task_finished(task_id: str) -> bool:
return bool(task_info and task_info.get("status") in {"completed", "failed"})
def _set_task_stage(task_id: str, stage: str, logs: list[str], status: str = "processing") -> None:
state.processing_tasks[task_id] = {
"logs": logs.copy(),
"status": status,
"stage": stage,
}
if state.global_task_lock["task_id"] == task_id:
state.global_task_lock["stage"] = stage
def _run_processing(task_id: str, work_dir: Path, logger: ProcessLogger) -> None:
try:
processor = DataProcessor(state.config, work_dir, logger)
@@ -155,12 +181,14 @@ def _run_processing(task_id: str, work_dir: Path, logger: ProcessLogger) -> None
state.processing_tasks[task_id] = {
"logs": state.history_manager.get_logs(task_id),
"status": status,
"stage": status,
}
except Exception as exc:
state.history_manager.update(task_id, status="failed", error=str(exc))
state.processing_tasks[task_id] = {
"logs": state.history_manager.get_logs(task_id),
"status": "failed",
"stage": "failed",
}
finally:
try:
app/processor.py
+17 -1
View File
@@ -22,10 +22,16 @@ from app.database import DatabaseManager
class ProcessLogger:
"""处理日志记录器"""
def __init__(self, log_file: Optional[Path] = None, callback: Optional[Callable[[str], None]] = None):
def __init__(
self,
log_file: Optional[Path] = None,
callback: Optional[Callable[[str], None]] = None,
stage_callback: Optional[Callable[[str], None]] = None,
):
self.logs: List[str] = []
self.log_file = log_file
self.callback = callback
self.stage_callback = stage_callback
# 如果指定了日志文件，确保目录存在
if self.log_file:
self.log_file.parent.mkdir(parents=True, exist_ok=True)
@@ -62,6 +68,10 @@ class ProcessLogger:
def success(self, message: str):
self.log(message, "SUCCESS")
def set_stage(self, stage: str):
if self.stage_callback:
self.stage_callback(stage)
def get_logs(self) -> List[str]:
return self.logs.copy()
@@ -195,24 +205,30 @@ class DataProcessor:
try:
# 1. 解压 ZIP 文件
self.logger.set_stage("extracting")
self._unzip_files()
# 2. 处理 Excel 文件（并行）
self.logger.set_stage("converting")
self._process_excel_files_parallel()
# 3. 处理 CSV 文件并上传到数据库（高性能批量插入）
self.logger.set_stage("importing")
self._process_csv_files()
# 4. 执行 SQL 脚本
self.logger.set_stage("scripting")
self._execute_sql_script()
elapsed = round(time.time() - start_time, 2)
self.logger.set_stage("completed")
self.logger.success(f"处理完成！总耗时: {elapsed} 秒")
self.results["success"] = True
self.results["elapsed_time"] = elapsed
except Exception as e:
self.logger.set_stage("failed")
self.logger.error(f"处理失败: {str(e)}")
self.results["success"] = False
self.results["error"] = str(e)