From 6004655b978ec023eda57fc67c83a71f39264933 Mon Sep 17 00:00:00 2001 From: Harmon Date: Thu, 16 Apr 2026 23:07:10 +0800 Subject: [PATCH] Fix D-InSAR task log access after completion --- backend/app/routers/tasks_runtime.py | 29 ++++- .../app/services/dinsar_production_service.py | 56 +++++++++ backend/app/services/job_handlers.py | 38 +++++- backend/app/services/task_service.py | 97 ++++++++++++--- frontend/src/DinsarProductionPanel.jsx | 107 ++++++++++++---- frontend/src/DinsarProductsPanel.jsx | 116 ++++++++++++++---- frontend/src/api/tasks.js | 9 ++ 7 files changed, 379 insertions(+), 73 deletions(-) diff --git a/backend/app/routers/tasks_runtime.py b/backend/app/routers/tasks_runtime.py index 353a8f3..a0086f4 100644 --- a/backend/app/routers/tasks_runtime.py +++ b/backend/app/routers/tasks_runtime.py @@ -4,7 +4,7 @@ import asyncio import json from typing import List, Optional -from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel @@ -38,6 +38,15 @@ def _new_session(): return database.AsyncSessionLocal() +def _split_csv_param(raw: Optional[str]) -> List[str]: + values: List[str] = [] + for chunk in str(raw or "").split(","): + text = chunk.strip() + if text and text not in values: + values.append(text) + return values + + @router.get("/tasks/active", response_model=List[TaskInfo]) async def get_active_tasks(limit: int = TASK_ACTIVE_DEFAULT_LIMIT, offset: int = 0): safe_limit = min(TASK_ACTIVE_MAX_LIMIT, max(1, int(limit or TASK_ACTIVE_DEFAULT_LIMIT))) @@ -46,6 +55,24 @@ async def get_active_tasks(limit: int = TASK_ACTIVE_DEFAULT_LIMIT, offset: int = return [TaskInfo.model_validate(task) for task in orm_tasks] +@router.get("/tasks/recent", response_model=List[TaskInfo]) +async def get_recent_tasks( + task_types: Optional[str] = Query(None, description="Comma-separated task types."), + statuses: Optional[str] = Query(None, description="Comma-separated task statuses."), + limit: int = TASK_ACTIVE_DEFAULT_LIMIT, + offset: int = 0, +): + safe_limit = min(TASK_ACTIVE_MAX_LIMIT, max(1, int(limit or TASK_ACTIVE_DEFAULT_LIMIT))) + safe_offset = min(TASK_QUERY_MAX_OFFSET, max(0, int(offset or 0))) + orm_tasks = await task_service.list_tasks( + task_types=_split_csv_param(task_types), + statuses=_split_csv_param(statuses), + limit=safe_limit, + offset=safe_offset, + ) + return [TaskInfo.model_validate(task) for task in orm_tasks] + + @router.get("/tasks/active/stream") async def stream_active_tasks(request: Request): token = request.cookies.get(SESSION_COOKIE_NAME) diff --git a/backend/app/services/dinsar_production_service.py b/backend/app/services/dinsar_production_service.py index 158f6ab..e0bff80 100644 --- a/backend/app/services/dinsar_production_service.py +++ b/backend/app/services/dinsar_production_service.py @@ -16,6 +16,7 @@ from ..models import ( DinsarProductionExecutionORM, DinsarProductionRunItemORM, DinsarProductionRunORM, + SystemTaskORM, ) from .envi_service import RUNTIME_DIR, _collect_task_folders, _resolve_dinsar_pair_identity, _to_local_path from .task_service import task_service @@ -206,6 +207,41 @@ def _public_run_status(value: str) -> str: class DinsarProductionService: + async def reconcile_run_with_task( + self, + run: DinsarProductionRunORM, + task: Optional[SystemTaskORM], + *, + db: AsyncSession, + ) -> bool: + if task is None: + return False + + run_status = str(run.status or "").strip().upper() + task_status = str(task.status or "").strip().upper() + if run_status in TERMINAL_RUN_STATUSES: + return False + if task_status not in {"COMPLETED", "FAILED", "CANCELLED"}: + return False + + if task_status == "COMPLETED": + next_status = RUN_STATUS_COMPLETED + elif task_status == "CANCELLED": + next_status = RUN_STATUS_CANCELLED + run.cancel_requested = True + else: + next_status = RUN_STATUS_FAILED + + summary_payload = dict(run.summary_json or {}) + summary_payload["reconciled_from_task_status"] = task_status + latest_message = str(task.message or "").strip() or f"Reconciled from task status {task_status}" + run.status = next_status + run.summary_json = summary_payload + run.latest_message = latest_message + run.ended_at = run.ended_at or _utcnow() + await self.refresh_run_counters(run, db=db, latest_message=latest_message) + return True + async def create_run( self, *, @@ -374,6 +410,26 @@ class DinsarProductionService: ) result = await db.execute(stmt) runs = result.scalars().all() + pending_reconcile = [ + run + for run in runs + if run.task_id and str(run.status or "").strip().upper() not in TERMINAL_RUN_STATUSES + ] + if pending_reconcile: + task_ids = [run.task_id for run in pending_reconcile if run.task_id] + task_result = await db.execute( + select(SystemTaskORM).where(SystemTaskORM.task_id.in_(task_ids)) + ) + tasks_by_id = {task.task_id: task for task in task_result.scalars().all()} + changed = False + for run in pending_reconcile: + changed = await self.reconcile_run_with_task( + run, + tasks_by_id.get(run.task_id), + db=db, + ) or changed + if changed: + await db.commit() return { "runs": [ { diff --git a/backend/app/services/job_handlers.py b/backend/app/services/job_handlers.py index d84e1f4..535e508 100644 --- a/backend/app/services/job_handlers.py +++ b/backend/app/services/job_handlers.py @@ -1823,7 +1823,43 @@ async def _handle_idl_run_dinsar(job: SystemJobORM) -> None: payload = job.payload or {} production_run_id = str(payload.get("production_run_id") or "").strip() if production_run_id: - await _run_dinsar_production_controller(job) + try: + await _run_dinsar_production_controller(job) + except Exception as exc: + latest_message = f"D-InSAR production controller failed: {exc}" + try: + async with AsyncSessionLocal() as db: + run = await dinsar_production_service.get_run(production_run_id, db) + if run is not None and str(run.status or "").strip().upper() not in {"COMPLETED", "FAILED", "CANCELLED"}: + summary_payload = dict(run.summary_json or {}) + summary_payload["controller_error"] = str(exc) + await dinsar_production_service.finalize_run( + run, + db=db, + status="FAILED", + summary_payload=summary_payload, + latest_message=latest_message, + ) + dinsar_production_service.append_run_log( + run.run_id, + f"[controller-failed] {exc}", + ) + except Exception: + pass + + try: + current_task = await task_service.get_task(job.task_id) + if current_task and current_task.status not in {"COMPLETED", "FAILED", "CANCELLED"}: + await task_service.add_log(job.task_id, "ERROR", latest_message) + await task_service.update_task( + job.task_id, + status="FAILED", + progress=100, + message=latest_message, + ) + except Exception: + pass + raise return mode = payload.get("mode", "metatask") diff --git a/backend/app/services/task_service.py b/backend/app/services/task_service.py index 64fcadd..114de4f 100644 --- a/backend/app/services/task_service.py +++ b/backend/app/services/task_service.py @@ -65,6 +65,19 @@ def _clamp_pagination(limit: int, offset: int, *, default_limit: int, max_limit: return safe_limit, safe_offset +def _normalize_string_list(values: Optional[List[str]], *, uppercase: bool = False) -> List[str]: + normalized: List[str] = [] + for raw in values or []: + text = str(raw or "").strip() + if not text: + continue + if uppercase: + text = text.upper() + if text not in normalized: + normalized.append(text) + return normalized + + def _task_type_lock_key(task_type: str) -> int: normalized = (task_type or "").strip().lower().encode("utf-8") digest = hashlib.sha256(normalized).digest() @@ -110,6 +123,27 @@ class TaskService: def __init__(self): pass + async def _expire_zombie_tasks(self, db: AsyncSession) -> None: + timeout_threshold = datetime.now() - timedelta(minutes=TASK_TIMEOUT_MINUTES) + result = await db.execute( + select(SystemTaskORM).where( + and_( + SystemTaskORM.status == "RUNNING", + SystemTaskORM.updated_at < timeout_threshold + ) + ) + ) + zombie_tasks = result.scalars().all() + + for task in zombie_tasks: + task.status = "FAILED" + task.message = "系统检测超时: 任务被认为已失效 (心跳超时)" + log = TaskLogORM(task_id=task.task_id, log_level="WARNING", message="任务因超时被自动标记为失败") + db.add(log) + + if zombie_tasks: + await db.commit() + async def create_task( self, task_type: str, @@ -269,26 +303,7 @@ class TaskService: max_limit=TASK_ACTIVE_MAX_LIMIT, ) - # 1. 查找僵尸任务并标记为失败 - timeout_threshold = datetime.now() - timedelta(minutes=TASK_TIMEOUT_MINUTES) - result = await db.execute( - select(SystemTaskORM).where( - and_( - SystemTaskORM.status == "RUNNING", - SystemTaskORM.updated_at < timeout_threshold - ) - ) - ) - zombie_tasks = result.scalars().all() - - for task in zombie_tasks: - task.status = "FAILED" - task.message = "系统检测超时: 任务被认为已失效 (心跳超时)" - log = TaskLogORM(task_id=task.task_id, log_level="WARNING", message="任务因超时被自动标记为失败") - db.add(log) - - if zombie_tasks: - await db.commit() + await self._expire_zombie_tasks(db) # 2. 获取活跃任务 active_result = await db.execute( @@ -304,6 +319,48 @@ class TaskService: if gen_db: await db.close() + async def list_tasks( + self, + task_types: Optional[List[str]] = None, + statuses: Optional[List[str]] = None, + limit: int = TASK_ACTIVE_DEFAULT_LIMIT, + offset: int = 0, + db: Optional[AsyncSession] = None, + ) -> List[SystemTaskORM]: + gen_db = db is None + if gen_db: + db = get_db_session() + + try: + safe_limit, safe_offset = _clamp_pagination( + limit, + offset, + default_limit=TASK_ACTIVE_DEFAULT_LIMIT, + max_limit=TASK_ACTIVE_MAX_LIMIT, + ) + await self._expire_zombie_tasks(db) + + normalized_task_types = _normalize_string_list(task_types) + normalized_statuses = _normalize_string_list(statuses, uppercase=True) + + stmt = select(SystemTaskORM) + if normalized_task_types: + stmt = stmt.where(SystemTaskORM.task_type.in_(normalized_task_types)) + if normalized_statuses: + stmt = stmt.where(SystemTaskORM.status.in_(normalized_statuses)) + + stmt = ( + stmt + .order_by(SystemTaskORM.created_at.desc(), SystemTaskORM.id.desc()) + .offset(safe_offset) + .limit(safe_limit) + ) + result = await db.execute(stmt) + return result.scalars().all() + finally: + if gen_db: + await db.close() + async def get_task(self, task_id: str, db: Optional[AsyncSession] = None) -> Optional[SystemTaskORM]: gen_db = db is None if gen_db: diff --git a/frontend/src/DinsarProductionPanel.jsx b/frontend/src/DinsarProductionPanel.jsx index 8f464d9..68312f7 100644 --- a/frontend/src/DinsarProductionPanel.jsx +++ b/frontend/src/DinsarProductionPanel.jsx @@ -253,6 +253,20 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) const currentProfiles = currentEngineObj?.profiles || EMPTY_ARRAY; const currentProfileObj = currentProfiles.find(profile => profile.code === selectedProfile) || null; const currentParamSchema = currentProfileObj?.params_schema || EMPTY_OBJECT; + const latestRunWithTask = runs.find(run => run?.task_id) || null; + const monitoredTask = activeTask || ( + latestRunWithTask + ? { + task_id: latestRunWithTask.task_id, + task_type: latestRunWithTask.engine === 'isce2' ? 'ISCE2_RUN' : 'IDL_RUN_DINSAR', + status: latestRunWithTask.raw_status || latestRunWithTask.status, + progress: latestRunWithTask.raw_status === 'COMPLETED' || latestRunWithTask.status === 'success' ? 100 : null, + message: latestRunWithTask.message || '最近一次任务', + } + : null + ); + const logTaskId = monitoredTask?.task_id || ''; + const showingRecentTask = !activeTask && !!monitoredTask; const loadEngines = useCallback(async () => { setEnginesLoading(true); @@ -306,7 +320,7 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) }, []); const handleDeleteTaskLog = useCallback(async logId => { - const taskId = activeTask?.task_id; + const taskId = logTaskId; if (!taskId || !logId || taskLogActionLoading) return; if (!window.confirm('确定要删除这条任务日志吗？')) return; @@ -322,12 +336,12 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) setTaskLogDeletingId(null); setTaskLogActionLoading(false); } - }, [activeTask?.task_id, loadTaskLogs, taskLogActionLoading]); + }, [logTaskId, loadTaskLogs, taskLogActionLoading]); const handleClearTaskLogs = useCallback(async () => { - const taskId = activeTask?.task_id; + const taskId = logTaskId; if (!taskId || taskLogActionLoading || taskLogs.length === 0) return; - if (!window.confirm('确定要清空当前任务的全部日志吗？')) return; + if (!window.confirm(`确定要清空任务 ${taskId} 的全部日志吗？`)) return; setTaskLogActionLoading(true); try { @@ -339,7 +353,7 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) } finally { setTaskLogActionLoading(false); } - }, [activeTask?.task_id, loadTaskLogs, taskLogActionLoading, taskLogs.length]); + }, [logTaskId, loadTaskLogs, taskLogActionLoading, taskLogs.length]); useEffect(() => { loadEngines(); @@ -348,17 +362,20 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) }, [loadActiveTask, loadEngines, loadRuns]); useEffect(() => { - const timer = setInterval(loadActiveTask, 5000); + const timer = setInterval(() => { + loadRuns(); + loadActiveTask(); + }, 5000); return () => clearInterval(timer); - }, [loadActiveTask]); + }, [loadActiveTask, loadRuns]); useEffect(() => { - const taskId = activeTask?.task_id || ''; + const taskId = logTaskId; loadTaskLogs(taskId); if (!taskId) return undefined; const timer = setInterval(() => loadTaskLogs(taskId), 5000); return () => clearInterval(timer); - }, [activeTask?.task_id, loadTaskLogs]); + }, [logTaskId, loadTaskLogs]); useEffect(() => { if (currentProfiles.length > 0) { @@ -672,31 +689,65 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) - {activeTask && ( -
-
当前任务
-
- {activeTask.task_id} - {formatTaskType(activeTask.task_type)} - {formatStatus(activeTask.status)} - {activeTask.message} + {monitoredTask && ( +
+
+ {showingRecentTask ? '最近一次任务' : '当前任务'}
- {activeTask.progress != null && ( +
+ {monitoredTask.task_id} - {formatTaskType(monitoredTask.task_type)} - {formatStatus(monitoredTask.status)} - {monitoredTask.message} +
+ {monitoredTask.progress != null && (
-
+
-
{activeTask.progress}%
+
{monitoredTask.progress}%
)} -
+
-
任务日志
+
+ {showingRecentTask ? '最近一次任务日志' : '当前任务日志'} +
{!readOnly && (
- {!activeTask ? ( + {!monitoredTask ? (
当前没有正在执行的产物处理任务。
) : ( -
-
当前任务
-
- {activeTask.task_id} - {formatTaskType(activeTask.task_type)} - {formatStatus(activeTask.status)} - {activeTask.message} +
+
+ {showingRecentTask ? '最近一次任务' : '当前任务'}
- {activeTask.progress != null && ( +
+ {monitoredTask.task_id} - {formatTaskType(monitoredTask.task_type)} - {formatStatus(monitoredTask.status)} - {monitoredTask.message} +
+ {monitoredTask.progress != null && (
-
+
-
{activeTask.progress}%
+
{monitoredTask.progress}%
)} -
+
-
任务日志
+
+ {showingRecentTask ? '最近一次任务日志' : '当前任务日志'} +
{!readOnly && (