refactor: 提取任务运行通用逻辑

This commit is contained in:
2026-06-01 18:24:39 +08:00
parent 8ba562e63e
commit c061b66169
4 changed files with 58 additions and 60 deletions
+10 -30
View File
@@ -6,9 +6,14 @@ from typing import Any, Callable, Iterable
from fastapi import APIRouter, Body, HTTPException
from app import state
from app.api.routers.task_runtime import (
apply_history_retention_safely,
log_license_check,
set_task_stage,
)
from app.config import AppConfig, CACHE_DIR, RemoteDataConfig
from app.processor import DataProcessor, ProcessLogger
from app.services.license import LicenseError, LicenseInfo, check_processing_allowed
from app.services.license import LicenseError, check_processing_allowed
from app.services.remote_download import RemoteDataDownloader
@@ -67,12 +72,12 @@ def start_remote_processing_job(
def log_callback(message: str) -> None:
logs.append(message)
_set_task_stage(task_id, current_stage, logs)
set_task_stage(task_id, current_stage, logs)
def stage_callback(stage: str) -> None:
nonlocal current_stage
current_stage = stage
_set_task_stage(task_id, current_stage, logs)
set_task_stage(task_id, current_stage, logs)
logger = ProcessLogger(
log_file=work_dir / "log.txt",
@@ -106,16 +111,6 @@ def start_remote_processing_job(
}
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,
@@ -144,7 +139,7 @@ def _run_remote_processing(
state.history_manager.update(task_id, file_count=download_result.file_count)
logger.set_stage("license")
_log_license_check(logger, check_processing_allowed(work_dir))
log_license_check(logger, check_processing_allowed(work_dir))
processor = DataProcessor(app_config, work_dir, logger)
result = processor.process()
@@ -189,10 +184,7 @@ def _run_remote_processing(
"error_detail": error_detail,
}
finally:
try:
state.apply_history_retention()
except Exception as exc:
print(f"自动清理处理历史失败: {exc}")
apply_history_retention_safely()
state.reset_task_lock()
if on_finish:
on_finish(task_id, final_status)
@@ -205,15 +197,3 @@ def _format_bytes(size: int) -> str:
return f"{value:.1f} {unit}"
value /= 1024
return f"{value:.1f} TB"
def _log_license_check(logger: ProcessLogger, info: LicenseInfo) -> None:
if info.current_date:
logger.info(
f"授权校验通过,数据日期: {info.current_date.isoformat()},"
f"到期日期: {info.expires_on.isoformat()}"
)
elif info.zip_count:
logger.warning("未从 ZIP 文件名识别到日期,已跳过授权日期比对")
else:
logger.warning("未找到 ZIP 文件,已跳过授权日期比对")
+32
View File
@@ -0,0 +1,32 @@
from typing import Any
from app import state
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 log_license_check(logger: Any, info: Any) -> None:
if info.current_date:
logger.info(
f"授权校验通过,数据日期: {info.current_date.isoformat()},"
f"到期日期: {info.expires_on.isoformat()}"
)
elif info.zip_count:
logger.warning("未从 ZIP 文件名识别到日期,已跳过授权日期比对")
else:
logger.warning("未找到 ZIP 文件,已跳过授权日期比对")
def apply_history_retention_safely() -> None:
try:
state.apply_history_retention()
except Exception as exc:
print(f"自动清理处理历史失败: {exc}")
+10 -30
View File
@@ -6,9 +6,14 @@ from typing import Any
from fastapi import APIRouter, Body, HTTPException
from app import state
from app.api.routers.task_runtime import (
apply_history_retention_safely,
log_license_check,
set_task_stage,
)
from app.config import AppConfig
from app.processor import DataProcessor, ProcessLogger
from app.services.license import LicenseError, LicenseInfo, check_processing_allowed
from app.services.license import LicenseError, check_processing_allowed
router = APIRouter(tags=["tasks"])
@@ -94,12 +99,12 @@ async def start_processing(task_id: str = Body(..., embed=True)):
def log_callback(message: str) -> None:
logs.append(message)
_set_task_stage(task_id, current_stage, logs)
set_task_stage(task_id, current_stage, logs)
def stage_callback(stage: str) -> None:
nonlocal current_stage
current_stage = stage
_set_task_stage(task_id, current_stage, logs)
set_task_stage(task_id, current_stage, logs)
logger = ProcessLogger(
log_file=work_dir / "log.txt",
@@ -161,20 +166,10 @@ 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, app_config: AppConfig) -> None:
try:
logger.set_stage("license")
_log_license_check(logger, check_processing_allowed(work_dir))
log_license_check(logger, check_processing_allowed(work_dir))
processor = DataProcessor(app_config, work_dir, logger)
result = processor.process()
@@ -205,20 +200,5 @@ def _run_processing(task_id: str, work_dir: Path, logger: ProcessLogger, app_con
"error_detail": error_detail,
}
finally:
try:
state.apply_history_retention()
except Exception as exc:
print(f"自动清理处理历史失败: {exc}")
apply_history_retention_safely()
state.reset_task_lock()
def _log_license_check(logger: ProcessLogger, info: LicenseInfo) -> None:
if info.current_date:
logger.info(
f"授权校验通过,数据日期: {info.current_date.isoformat()},"
f"到期日期: {info.expires_on.isoformat()}"
)
elif info.zip_count:
logger.warning("未从 ZIP 文件名识别到日期,已跳过授权日期比对")
else:
logger.warning("未找到 ZIP 文件,已跳过授权日期比对")
+6
View File
@@ -626,3 +626,9 @@
- `自动调度` is now a separate settings tab. Scheduler controls are disabled unless `启用远程自动化` is on, and turning off remote automation automatically turns off scheduler enablement in the current form state and saved payload.
- `debug.bat` starts the Python backend on `127.0.0.1:9081` and the Vite frontend on `127.0.0.1:5173` for source-level debugging without requiring `frontend/dist`.
- Verification performed: `cmd /c debug.bat` started the debug backend/frontend, `cd frontend && npm run build` passed, and browser inspection confirmed the remote connection fields align in one row with no horizontal overflow. Debug processes and `frontend/dist/` were removed after verification.
## 2026-06-01: Task runtime cleanup
- `app/api/routers/task_runtime.py` centralizes shared task-stage updates, processing license log output, and safe history-retention cleanup for manual processing and remote processing routes.
- `app/api/routers/tasks.py` and `app/api/routers/remote.py` now reuse the shared helpers instead of carrying duplicate `_set_task_stage` and `_log_license_check` implementations.
- Verification performed: `.venv\Scripts\python.exe -m compileall app`, `npm run build`, and `cargo check --manifest-path src-tauri\Cargo.toml` with a temporary sidecar placeholder all passed. Generated build output, Python caches, and temporary Tauri sidecar files were removed after verification.