diff --git a/backend/app/routers/dinsar_production.py b/backend/app/routers/dinsar_production.py index 274b922..f3589be 100644 --- a/backend/app/routers/dinsar_production.py +++ b/backend/app/routers/dinsar_production.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio +import re from typing import Any, Dict, Literal, Optional from fastapi import APIRouter, Depends, HTTPException @@ -18,6 +19,9 @@ from ..services.task_service import task_service router = APIRouter(prefix="/dinsar-production", tags=["dinsar-production"]) +_RUN_ID_RE = re.compile(r"^[\w\-]{4,128}$") +_LOG_MAX_BYTES = 200 * 1024 + DINSAR_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env( "DINSAR_PRODUCTION_JOB_MAX_ATTEMPTS", 1, @@ -351,7 +355,74 @@ async def submit_run( @router.get("/runs") -async def list_runs(limit: int = 20): +async def list_runs(limit: int = 20, offset: int = 0): async with _new_session() as db: - result = await dinsar_production_service.list_runs(db, limit=limit) + result = await dinsar_production_service.list_runs(db, limit=limit, offset=offset) return result + + +@router.get("/runs/{run_id}/log") +async def get_run_log( + run_id: str, + current_user: AuthUserORM = Depends(_get_current_user), +): + _ = current_user + normalized_run_id = str(run_id or "").strip() + if not _RUN_ID_RE.match(normalized_run_id): + raise HTTPException(status_code=400, detail="Invalid run_id format.") + try: + return await asyncio.to_thread( + dinsar_production_service.read_run_log, + normalized_run_id, + max_bytes=_LOG_MAX_BYTES, + ) + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail="Run log file not found.") from exc + + +@router.delete("/runs/{run_id}/log") +async def delete_run_log( + run_id: str, + current_user: AuthUserORM = Depends(_require_admin), +): + _ = current_user + normalized_run_id = str(run_id or "").strip() + if not _RUN_ID_RE.match(normalized_run_id): + raise HTTPException(status_code=400, detail="Invalid run_id format.") + deleted = await asyncio.to_thread( + dinsar_production_service.delete_run_log, + normalized_run_id, + ) + if not deleted: + raise HTTPException(status_code=404, detail="Run log file not found.") + return { + "run_id": normalized_run_id, + "deleted": True, + } + + +@router.delete("/runs/{run_id}") +async def delete_run_record( + run_id: str, + current_user: AuthUserORM = Depends(_require_admin), +): + _ = current_user + normalized_run_id = str(run_id or "").strip() + if not _RUN_ID_RE.match(normalized_run_id): + raise HTTPException(status_code=400, detail="Invalid run_id format.") + + async with _new_session() as db: + try: + result = await dinsar_production_service.delete_run_record( + normalized_run_id, + db=db, + ) + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + if result is None: + raise HTTPException(status_code=404, detail="Production run not found.") + return { + **result, + "deleted": True, + "products_deleted": False, + } diff --git a/backend/app/routers/tasks_runtime.py b/backend/app/routers/tasks_runtime.py index a0086f4..b0551a7 100644 --- a/backend/app/routers/tasks_runtime.py +++ b/backend/app/routers/tasks_runtime.py @@ -181,6 +181,26 @@ async def clear_task_logs( } +@router.delete("/tasks/{task_id}") +async def delete_task_record( + task_id: str, + admin_user: AuthUserORM = Depends(_require_admin), +): + task = await task_service.get_task(task_id) + if not task: + raise HTTPException(status_code=404, detail="Task not found.") + if str(task.status or "").upper() in {"PENDING", "RUNNING"}: + raise HTTPException(status_code=409, detail="Cannot delete a pending or running task.") + + deleted = await task_service.delete_task_record(task_id) + if not deleted: + raise HTTPException(status_code=404, detail="Task not found.") + return { + "task_id": task_id, + "deleted": True, + } + + @router.post("/tasks/{task_id}/force-cancel") async def force_cancel_task( task_id: str, diff --git a/backend/app/services/dinsar_production_service.py b/backend/app/services/dinsar_production_service.py index 8335123..c301802 100644 --- a/backend/app/services/dinsar_production_service.py +++ b/backend/app/services/dinsar_production_service.py @@ -9,7 +9,7 @@ import uuid from datetime import datetime from typing import Any, Dict, List, Optional -from sqlalchemy import func, select +from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from .. import database @@ -18,7 +18,12 @@ from ..models import ( DinsarProductionExecutionORM, DinsarProductionRunItemORM, DinsarProductionRunORM, + SystemJobORM, SystemTaskORM, + TaskLogORM, + WorkflowArtifactORM, + WorkflowRunORM, + WorkflowStepORM, ) from .envi_service import RUNTIME_DIR, _collect_task_folders, _resolve_dinsar_pair_identity, _to_local_path from .task_service import task_service @@ -1000,6 +1005,81 @@ class DinsarProductionService: def append_run_log(self, run_id: str, message: str) -> str: return _append_run_log_sync(run_id, message) + def read_run_log(self, run_id: str, *, max_bytes: int = 200 * 1024) -> Dict[str, Any]: + log_path = _run_log_path(run_id) + if not os.path.isfile(log_path): + raise FileNotFoundError(log_path) + size_bytes = os.path.getsize(log_path) + truncated = size_bytes > max_bytes + with open(log_path, "r", encoding="utf-8", errors="replace") as fp: + if truncated: + fp.seek(size_bytes - max_bytes) + content = "...[日志已截断，仅显示末尾]...\n" + fp.read() + else: + content = fp.read() + return { + "run_id": run_id, + "content": content, + "size_bytes": size_bytes, + "truncated": truncated, + "log_path": log_path, + } + + def delete_run_log(self, run_id: str) -> bool: + log_path = _run_log_path(run_id) + if not os.path.isfile(log_path): + return False + os.unlink(log_path) + return True + + async def delete_run_record( + self, + run_id: str, + *, + db: AsyncSession, + ) -> Optional[Dict[str, Any]]: + run = await self.get_run(run_id, db) + if run is None: + return None + if str(run.status or "").strip().upper() not in TERMINAL_RUN_STATUSES: + raise ValueError("Cannot delete a pending or running production run.") + + task_id = str(run.task_id or "").strip() + workflow_run_id = str(run.workflow_run_id or "").strip() + if task_id: + task_result = await db.execute( + select(SystemTaskORM).where(SystemTaskORM.task_id == task_id) + ) + task = task_result.scalar_one_or_none() + if task is not None and str(task.status or "").strip().upper() in {"PENDING", "RUNNING"}: + raise ValueError("Cannot delete a production run with a pending or running task.") + + deleted = { + "run_id": run.run_id, + "task_id": task_id or None, + "workflow_run_id": workflow_run_id or None, + } + await db.execute(delete(DinsarProductionExecutionORM).where(DinsarProductionExecutionORM.run_id == run.run_id)) + await db.execute(delete(DinsarProductionRunItemORM).where(DinsarProductionRunItemORM.run_id == run.run_id)) + await db.execute(delete(DinsarProductionRunORM).where(DinsarProductionRunORM.run_id == run.run_id)) + + if task_id: + await db.execute(delete(TaskLogORM).where(TaskLogORM.task_id == task_id)) + await db.execute(delete(SystemJobORM).where(SystemJobORM.task_id == task_id)) + await db.execute(delete(SystemTaskORM).where(SystemTaskORM.task_id == task_id)) + if workflow_run_id: + await db.execute(delete(SystemJobORM).where(SystemJobORM.workflow_run_id == workflow_run_id)) + await db.execute(delete(WorkflowArtifactORM).where(WorkflowArtifactORM.run_id == workflow_run_id)) + await db.execute(delete(WorkflowStepORM).where(WorkflowStepORM.run_id == workflow_run_id)) + await db.execute(delete(WorkflowRunORM).where(WorkflowRunORM.run_id == workflow_run_id)) + + await db.commit() + try: + deleted["log_deleted"] = await asyncio.to_thread(self.delete_run_log, run_id) + except OSError: + deleted["log_deleted"] = False + return deleted + def build_execution_manifest( self, *, diff --git a/backend/app/services/task_service.py b/backend/app/services/task_service.py index 6e66da2..8fc70f8 100644 --- a/backend/app/services/task_service.py +++ b/backend/app/services/task_service.py @@ -479,5 +479,32 @@ class TaskService: if gen_db: await db.close() + async def delete_task_record( + self, + task_id: str, + db: Optional[AsyncSession] = None, + ) -> bool: + gen_db = db is None + if gen_db: + db = get_db_session() + + try: + result = await db.execute( + select(SystemTaskORM).where(SystemTaskORM.task_id == task_id) + ) + task = result.scalar_one_or_none() + if task is None: + return False + await db.execute(delete(TaskLogORM).where(TaskLogORM.task_id == task_id)) + await db.delete(task) + await db.commit() + return True + except Exception: + await db.rollback() + raise + finally: + if gen_db: + await db.close() + task_service = TaskService() diff --git a/frontend/src/DinsarProductionPanel.jsx b/frontend/src/DinsarProductionPanel.jsx index 80dfa68..2ac7220 100644 --- a/frontend/src/DinsarProductionPanel.jsx +++ b/frontend/src/DinsarProductionPanel.jsx @@ -1,8 +1,7 @@ import React, { useCallback, useEffect, useState } from 'react'; -import { listEngines, listRuns, previewPyintInputAssets, submitRun } from './api/dinsarProduction'; -import { getJobLog } from './api/idl'; -import { clearTaskLogs, deleteTaskLog, getActiveTasks, getRecentTasks, getTaskLogs } from './api/tasks'; +import { deleteRunLog, deleteRunRecord, getRunLog, listEngines, listRuns, previewPyintInputAssets, submitRun } from './api/dinsarProduction'; +import { clearTaskLogs, deleteTaskLog, deleteTaskRecord, getActiveTasks, getRecentTasks, getTaskLogs } from './api/tasks'; const card = { background: '#fff', @@ -14,6 +13,10 @@ const card = { const EMPTY_ARRAY = []; const EMPTY_OBJECT = {}; +const RUN_HISTORY_PAGE_SIZE = 200; +const TASK_HISTORY_PAGE_SIZE = 500; +const TASK_LOG_PAGE_SIZE = 1000; +const TERMINAL_STATUS_VALUES = new Set(['COMPLETED', 'FAILED', 'CANCELLED', 'CANCELED', 'success', 'failed', 'cancelled', 'canceled']); const ENGINE_STATUS_COLOR = { ok: '#22c55e', @@ -120,6 +123,11 @@ function taskStatusToRunStatus(status) { return 'pending'; } +function isTerminalRunRow(run) { + return TERMINAL_STATUS_VALUES.has(String(run?.raw_status || '').trim()) + || TERMINAL_STATUS_VALUES.has(String(run?.status || '').trim()); +} + function epochSeconds(value) { if (value == null || value === '') return null; if (typeof value === 'number') return value; @@ -166,7 +174,7 @@ function inferTaskPaths(task) { }; } -function mergeRunRows(productionRuns, recentTasks, limit = 20) { +function mergeRunRows(productionRuns, recentTasks, limit = null) { const rows = (productionRuns || []).map(run => ({ ...run, record_type: run?.record_type || 'run', @@ -190,9 +198,21 @@ function mergeRunRows(productionRuns, recentTasks, limit = 20) { representedTaskIds.add(taskId); }); - return rows - .sort((a, b) => Number(b?.started_at || 0) - Number(a?.started_at || 0)) - .slice(0, limit); + const sortedRows = rows.sort((a, b) => Number(b?.started_at || 0) - Number(a?.started_at || 0)); + return limit == null ? sortedRows : sortedRows.slice(0, limit); +} + +async function fetchAllTaskLogs(taskId) { + const allLogs = []; + let offset = 0; + while (true) { + const data = await getTaskLogs(taskId, TASK_LOG_PAGE_SIZE, offset); + const pageLogs = data?.logs || []; + allLogs.push(...pageLogs); + if (pageLogs.length < TASK_LOG_PAGE_SIZE) break; + offset += pageLogs.length; + } + return allLogs; } function formatTaskLogContent(taskId, logs) { @@ -535,7 +555,15 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) const [runs, setRuns] = useState([]); const [runsLoading, setRunsLoading] = useState(false); - const [logModal, setLogModal] = useState({ open: false, runId: '', content: '', loading: false }); + const [logModal, setLogModal] = useState({ + open: false, + runId: '', + taskId: '', + source: 'run', + content: '', + loading: false, + }); + const [runLogDeletingId, setRunLogDeletingId] = useState(''); const [activeTask, setActiveTask] = useState(null); const [recentTask, setRecentTask] = useState(null); const [taskLogs, setTaskLogs] = useState([]); @@ -591,12 +619,36 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) const silent = !!options.silent; if (!silent) setRunsLoading(true); try { - const [runData, taskData] = await Promise.all([ - listRuns(20), - getRecentTasks(['ISCE2_RUN', 'PYINT_RUN', 'IDL_RUN_DINSAR'], [], 20, 0), + const loadProductionRuns = async () => { + const allRuns = []; + let offset = 0; + while (true) { + const data = await listRuns(RUN_HISTORY_PAGE_SIZE, offset); + const pageRuns = data?.runs || []; + allRuns.push(...pageRuns); + const total = Number(data?.total || 0); + if (pageRuns.length < RUN_HISTORY_PAGE_SIZE || allRuns.length >= total) break; + offset += pageRuns.length; + } + return allRuns; + }; + const loadRecentTasks = async () => { + const allTasks = []; + let offset = 0; + while (true) { + const data = await getRecentTasks(['ISCE2_RUN', 'PYINT_RUN', 'IDL_RUN_DINSAR'], [], TASK_HISTORY_PAGE_SIZE, offset); + const pageTasks = Array.isArray(data) ? data : (data?.tasks || []); + allTasks.push(...pageTasks); + if (pageTasks.length < TASK_HISTORY_PAGE_SIZE) break; + offset += pageTasks.length; + } + return allTasks; + }; + const [productionRuns, recentTasks] = await Promise.all([ + loadProductionRuns(), + loadRecentTasks(), ]); - const recentTasks = Array.isArray(taskData) ? taskData : (taskData?.tasks || []); - const nextRuns = mergeRunRows(runData.runs || [], recentTasks, 20); + const nextRuns = mergeRunRows(productionRuns, recentTasks); setRuns(nextRuns); return nextRuns; } catch { @@ -641,8 +693,8 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) } if (!silent) setTaskLogsLoading(true); try { - const data = await getTaskLogs(taskId, 120, 0); - setTaskLogs(data?.logs || []); + const logs = await fetchAllTaskLogs(taskId); + setTaskLogs(logs); } catch { setTaskLogs([]); } finally { @@ -830,25 +882,112 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) const handleViewLog = async run => { const runId = typeof run === 'string' ? run : (run?.run_id || run?.task_id || ''); - setLogModal({ open: true, runId, content: '', loading: true }); + const source = typeof run !== 'string' && run?.log_source === 'task' ? 'task' : 'run'; + const taskId = typeof run !== 'string' ? (run?.task_id || '') : ''; + setLogModal({ open: true, runId, taskId, source, content: '', loading: true }); try { - if (typeof run !== 'string' && run?.log_source === 'task' && run?.task_id) { - const data = await getTaskLogs(run.task_id, 200, 0); + if (source === 'task' && taskId) { + const logs = await fetchAllTaskLogs(taskId); setLogModal({ open: true, runId, - content: formatTaskLogContent(run.task_id, data?.logs || []), + taskId, + source, + content: formatTaskLogContent(taskId, logs), loading: false, }); return; } - const data = await getJobLog(runId); - setLogModal({ open: true, runId, content: data.content || '', loading: false }); + const data = await getRunLog(runId); + setLogModal({ open: true, runId, taskId: '', source, content: data.content || '', loading: false }); } catch { - setLogModal({ open: true, runId, content: '日志加载失败。', loading: false }); + setLogModal({ open: true, runId, taskId, source, content: '日志加载失败，可能日志文件不存在或已被删除。', loading: false }); } }; + const handleDeleteRunRowLog = useCallback(async run => { + if (readOnly || runLogDeletingId) return; + const source = run?.log_source === 'task' ? 'task' : 'run'; + const runId = run?.run_id || run?.task_id || ''; + const taskId = run?.task_id || ''; + const label = source === 'task' ? `任务 ${taskId}` : `运行 ${runId}`; + if (!runId) return; + if (!window.confirm(`确定要删除${label}的日志吗？运行记录和产物不会删除。`)) return; + + setRunLogDeletingId(runId); + try { + if (source === 'task') { + await clearTaskLogs(taskId || runId); + if (logTaskId === (taskId || runId)) { + await loadTaskLogs(taskId || runId); + } + } else { + await deleteRunLog(runId); + } + if (logModal.open && logModal.runId === runId) { + setLogModal(current => ({ + ...current, + content: '日志已删除。', + loading: false, + })); + } + setSubmitError(false); + setSubmitMsg('日志已删除。'); + await refreshMonitor({ silent: true }); + } catch (err) { + setSubmitError(true); + setSubmitMsg(`删除日志失败：${err?.response?.data?.detail || err.message}`); + } finally { + setRunLogDeletingId(''); + } + }, [loadTaskLogs, logModal.open, logModal.runId, logTaskId, readOnly, refreshMonitor, runLogDeletingId]); + + const handleDeleteOpenLog = useCallback(async () => { + if (!logModal.open || readOnly || runLogDeletingId) return; + await handleDeleteRunRowLog({ + run_id: logModal.runId, + task_id: logModal.taskId, + log_source: logModal.source, + }); + }, [handleDeleteRunRowLog, logModal, readOnly, runLogDeletingId]); + + const handleDeleteRunHistory = useCallback(async run => { + if (readOnly || runLogDeletingId) return; + const isTaskRecord = run?.log_source === 'task'; + const recordId = isTaskRecord ? (run?.task_id || run?.run_id || '') : (run?.run_id || ''); + const taskId = run?.task_id || ''; + if (!recordId) return; + if (!isTerminalRunRow(run)) { + setSubmitError(true); + setSubmitMsg('运行中记录不能删除。'); + return; + } + if (!window.confirm(`确定要删除运行记录 ${recordId} 吗？只删除记录和日志，不删除产物目录。`)) return; + + setRunLogDeletingId(recordId); + try { + if (isTaskRecord) { + await deleteTaskRecord(recordId); + } else { + await deleteRunRecord(recordId); + } + if (logTaskId === (taskId || recordId)) { + setTaskLogs([]); + } + if (logModal.open && logModal.runId === recordId) { + setLogModal({ open: false, runId: '', taskId: '', source: 'run', content: '', loading: false }); + } + setSubmitError(false); + setSubmitMsg('运行记录已删除，产物目录未删除。'); + await refreshMonitor({ silent: true }); + } catch (err) { + setSubmitError(true); + setSubmitMsg(`删除运行记录失败：${err?.response?.data?.detail || err.message}`); + } finally { + setRunLogDeletingId(''); + } + }, [logModal.open, logModal.runId, logTaskId, readOnly, refreshMonitor, runLogDeletingId]); + const isSubmitDisabled = readOnly || submitting || !currentEngineObj?.available || pyintPreviewBlocksSubmit; return ( @@ -871,30 +1010,53 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued }) color: '#e2e8f0', borderRadius: 10, padding: 24, - width: 720, - maxHeight: '80vh', + width: 'min(960px, 92vw)', + maxHeight: '82vh', display: 'flex', flexDirection: 'column', }} >
运行日志 - {logModal.runId} - +
+ {!readOnly && ( + + )} + + 
               {logModal.loading ? '加载中...' : logModal.content || '（日志为空）'}
@@ -1560,63 +1722,115 @@ export default function DinsarProductionPanel({ readOnly = false, onJobQueued })
           
         )}
 
-        
最近 20 条运行记录

+        
运行记录（已加载 {runs.length} 条）

         {runsLoading ? (
           
加载中...

         ) : runs.length === 0 ? (
           
暂无记录。

         ) : (
-          
-            
-              
-                {['运行ID', '引擎', '状态', '时间', '路径', '操作'].map(header => (
-                  
-                ))}
-              
-            
-            
-              {runs.map(run => (
-                
-                  
-                  
-                  
-                  
-                  
-                  
+                {runs.map(run => {
+                  const recordId = run.log_source === 'task' ? (run.task_id || run.run_id) : run.run_id;
+                  const canDeleteRecord = isTerminalRunRow(run);
+                  return (
+                    
+                      
+                      
+                      
+                      
+                      
+                      
+                    
+                  );
+                })}
+              
+            

-                    {header}
-                  
{run.run_id}{formatEngineLabel(run.engine)}
-                    {formatStatus(run.status)}
-                  
-                    {run.started_at ? new Date(run.started_at * 1000).toLocaleString() : '-'}
-                  
-                    
-                  
-                    
{run.run_id}{formatEngineLabel(run.engine)}
+                        {formatStatus(run.status)}
+                      
+                        {run.started_at ? new Date(run.started_at * 1000).toLocaleString() : '-'}
+                      
+                        
+                      
+                        
+                        {!readOnly && (
+                          
+                        )}
+                        {!readOnly && (
+                          
+                        )}
+                      

+          
         )}
       
     
diff --git a/frontend/src/api/dinsarProduction.js b/frontend/src/api/dinsarProduction.js
index 7055e7c..d31b086 100644
--- a/frontend/src/api/dinsarProduction.js
+++ b/frontend/src/api/dinsarProduction.js
@@ -16,8 +16,19 @@ export const submitRun = (payload) =>
   apiClient.post('/dinsar-production/run', payload).then(r => r.data);
 
 // 运行历史
-export const listRuns = (limit = 20) =>
-  apiClient.get(`/dinsar-production/runs?limit=${encodeURIComponent(limit)}`).then(r => r.data);
+export const listRuns = (limit = 20, offset = 0) =>
+  apiClient.get(
+    `/dinsar-production/runs?limit=${encodeURIComponent(limit)}&offset=${encodeURIComponent(offset)}`,
+  ).then(r => r.data);
+
+export const getRunLog = (runId) =>
+  apiClient.get(`/dinsar-production/runs/${encodeURIComponent(runId)}/log`).then(r => r.data);
+
+export const deleteRunLog = (runId) =>
+  apiClient.delete(`/dinsar-production/runs/${encodeURIComponent(runId)}/log`).then(r => r.data);
+
+export const deleteRunRecord = (runId) =>
+  apiClient.delete(`/dinsar-production/runs/${encodeURIComponent(runId)}`).then(r => r.data);
 
 export const previewPyintInputAssets = (payload) =>
   apiClient.post('/dinsar-production/engines/pyint/preview-input-assets', payload).then(r => r.data);
diff --git a/frontend/src/api/tasks.js b/frontend/src/api/tasks.js
index a6dc5fa..bc8c9fe 100644
--- a/frontend/src/api/tasks.js
+++ b/frontend/src/api/tasks.js
@@ -17,3 +17,5 @@ export const deleteTaskLog = (taskId, logId) =>
   apiClient.delete(`/tasks/${taskId}/logs/${encodeURIComponent(logId)}`).then(r => r.data);
 export const clearTaskLogs = (taskId) =>
   apiClient.delete(`/tasks/${taskId}/logs`).then(r => r.data);
+export const deleteTaskRecord = (taskId) =>
+  apiClient.delete(`/tasks/${taskId}`).then(r => r.data);