diff --git a/backend/app/routers/__init__.py b/backend/app/routers/__init__.py index 998fe54..ad4c795 100644 --- a/backend/app/routers/__init__.py +++ b/backend/app/routers/__init__.py @@ -18,6 +18,7 @@ from . import ( license, logs, monitor, + ops_maintenance, orbit, pairing, ps_products, @@ -48,6 +49,7 @@ def include_all_routers(router: APIRouter) -> None: router.include_router(tools.router) router.include_router(unpack.router) router.include_router(monitor.router) + router.include_router(ops_maintenance.router) router.include_router(orbit.router) router.include_router(assets.router) router.include_router(root_registry.router) diff --git a/backend/app/routers/ops_maintenance.py b/backend/app/routers/ops_maintenance.py new file mode 100644 index 0000000..001a28d --- /dev/null +++ b/backend/app/routers/ops_maintenance.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from pydantic import BaseModel +from sqlalchemy.ext.asyncio import AsyncSession + +from ..database import get_db +from ..models import AuthUserORM +from ..services.ops_maintenance_service import ops_maintenance_service +from .dependencies import _add_operation_audit_log, _get_current_user, _require_admin + + +router = APIRouter(prefix="/ops-maintenance", tags=["ops-maintenance"]) + + +class CleanupRequest(BaseModel): + confirm: bool = False + delete_task_records: bool = True + delete_logs: bool = True + delete_production_records: bool = True + delete_result_products: bool = True + delete_production_dirs: bool = True + delete_task_pool_dir: bool = True + + +@router.get("/tasks") +async def list_maintenance_tasks( + task_type: Optional[str] = Query(default=None), + status: Optional[str] = Query(default=None), + limit: int = Query(default=50, ge=1, le=200), + offset: int = Query(default=0, ge=0), + _user: AuthUserORM = Depends(_get_current_user), + db: AsyncSession = Depends(get_db), +): + return await ops_maintenance_service.list_tasks( + db, + task_type=task_type, + status=status, + limit=limit, + offset=offset, + ) + + +@router.get("/tasks/{task_id}/diagnosis") +async def diagnose_maintenance_task( + task_id: str, + _user: AuthUserORM = Depends(_get_current_user), + db: AsyncSession = Depends(get_db), +): + diagnosis = await ops_maintenance_service.diagnose_task(db, task_id) + if diagnosis is None: + raise HTTPException(status_code=404, detail="Task not found.") + return diagnosis + + +@router.post("/tasks/{task_id}/cleanup-preview") +async def preview_maintenance_cleanup( + task_id: str, + _user: AuthUserORM = Depends(_get_current_user), + db: AsyncSession = Depends(get_db), +): + preview = await ops_maintenance_service.cleanup_preview(db, task_id) + if preview is None: + raise HTTPException(status_code=404, detail="Task not found.") + return preview + + +@router.post("/tasks/{task_id}/cleanup") +async def cleanup_maintenance_task( + task_id: str, + request_body: CleanupRequest, + request: Request, + admin_user: AuthUserORM = Depends(_require_admin), + db: AsyncSession = Depends(get_db), +): + if not request_body.confirm: + raise HTTPException(status_code=400, detail="Cleanup confirmation is required.") + try: + result = await ops_maintenance_service.cleanup_task( + db, + task_id, + options=request_body.model_dump(), + ) + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + if result is None: + raise HTTPException(status_code=404, detail="Task not found.") + + await _add_operation_audit_log( + db, + request, + action="ops_task_cleanup", + resource=f"ops-maintenance/tasks/{task_id}", + detail={ + "task_id": task_id, + "options": request_body.model_dump(), + "result": result, + }, + user=admin_user, + ) + await db.commit() + return result diff --git a/backend/app/services/ops_maintenance_service.py b/backend/app/services/ops_maintenance_service.py new file mode 100644 index 0000000..acfb1a0 --- /dev/null +++ b/backend/app/services/ops_maintenance_service.py @@ -0,0 +1,695 @@ +from __future__ import annotations + +import os +import shutil +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + +from sqlalchemy import delete, func, or_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..config import settings +from ..models import ( + DinsarProductionExecutionORM, + DinsarProductionRunItemORM, + DinsarProductionRunORM, + DinsarResultORM, + ResultProductORM, + SystemJobORM, + SystemTaskORM, + TaskLogORM, + WorkflowArtifactORM, + WorkflowRunORM, + WorkflowStepORM, +) +from .dinsar_production_service import dinsar_production_service + + +TERMINAL_TASK_STATUSES = {"COMPLETED", "FAILED", "PARTIAL_SUCCESS", "CANCELLED"} +MAINTENANCE_LIST_STATUSES = {"FAILED", "PARTIAL_SUCCESS", "CANCELLED", "PENDING", "RUNNING"} +ACTIVE_TASK_STATUSES = {"PENDING", "RUNNING"} +ACTIVE_JOB_STATUSES = {"READY", "RETRY", "RUNNING"} +DINSAR_TASK_TYPES = {"LANDSAR_RUN", "LANDSAR_CLUSTER_RUN", "PYINT_RUN", "IDL_RUN_DINSAR"} +SUPPORTED_CLEANUP_TASK_TYPES = DINSAR_TASK_TYPES | {"COPY_DATA"} +DEFAULT_TASK_TYPES = SUPPORTED_CLEANUP_TASK_TYPES | {"PAIRING_CACHE_REBUILD"} + + +def _utcnow_naive() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def _dt(value: Any) -> Optional[str]: + if value is None: + return None + if hasattr(value, "isoformat"): + return value.isoformat() + return str(value) + + +def _norm_status(value: Any) -> str: + return str(value or "").strip().upper() + + +def _compact(value: Any, limit: int = 220) -> Optional[str]: + if value is None: + return None + text = str(value).replace("\r", " ").replace("\n", " ").strip() + if len(text) <= limit: + return text + return text[:limit] + "..." + + +def _normalize_path_text(value: Any) -> str: + return os.path.normpath(str(value or "").strip().strip('"').strip("'")) + + +def _path_exists(path: str) -> bool: + return bool(path) and os.path.exists(path) + + +def _safe_count(rows: Iterable[Any]) -> int: + return len(list(rows)) + + +class OpsMaintenanceService: + async def list_tasks( + self, + db: AsyncSession, + *, + task_type: Optional[str] = None, + status: Optional[str] = None, + limit: int = 50, + offset: int = 0, + ) -> Dict[str, Any]: + safe_limit = max(1, min(int(limit or 50), 200)) + safe_offset = max(0, int(offset or 0)) + task_types = [task_type.strip().upper()] if task_type else sorted(DEFAULT_TASK_TYPES) + status_filter = status.strip().upper() if status else "" + + conditions = [SystemTaskORM.task_type.in_(task_types)] + if status_filter: + conditions.append(SystemTaskORM.status == status_filter) + else: + conditions.append(SystemTaskORM.status.in_(sorted(MAINTENANCE_LIST_STATUSES))) + + result = await db.execute( + select(SystemTaskORM) + .where(*conditions) + .order_by(SystemTaskORM.updated_at.desc(), SystemTaskORM.id.desc()) + .offset(safe_offset) + .limit(safe_limit) + ) + tasks = list(result.scalars().all()) + items = [await self._build_task_summary(db, task) for task in tasks] + abnormal_items = [ + item + for item in items + if item.get("issue_level") in {"warning", "danger"} + or _norm_status(item.get("status")) in {"FAILED", "PARTIAL_SUCCESS", "CANCELLED"} + ] + return { + "items": abnormal_items, + "limit": safe_limit, + "offset": safe_offset, + "returned": len(abnormal_items), + } + + async def diagnose_task(self, db: AsyncSession, task_id: str) -> Optional[Dict[str, Any]]: + task = await self._get_task(db, task_id) + if task is None: + return None + jobs = await self._get_jobs(db, task.task_id) + run = await self._get_production_run_for_task(db, task, jobs) + item_counts: Dict[str, int] = {} + execution_counts: Dict[str, int] = {} + disk_paths: List[Dict[str, Any]] = [] + products: List[Dict[str, Any]] = [] + + if run is not None: + item_counts = await self._status_counts(db, DinsarProductionRunItemORM, run.run_id) + execution_counts = await self._status_counts(db, DinsarProductionExecutionORM, run.run_id) + disk_paths.extend(await self._collect_run_disk_paths(db, run)) + products = await self._collect_result_products(db, run) + related_tasks = await self._related_copy_tasks_for_run(db, run) if run is not None else [] + + copy_dest = self._copy_task_dest_dir(task) + if copy_dest: + disk_paths.append(self._path_payload(copy_dest, "task_pool")) + + recent_logs = await self._recent_logs(db, task.task_id) + findings, cleanup_supported, cleanup_blockers = self._diagnose_findings( + task=task, + jobs=jobs, + run=run, + item_counts=item_counts, + execution_counts=execution_counts, + disk_paths=disk_paths, + ) + return { + "task": self._task_payload(task), + "jobs": [self._job_payload(job) for job in jobs], + "production_run": self._run_payload(run) if run else None, + "production_item_counts": item_counts, + "production_execution_counts": execution_counts, + "result_products": products, + "related_tasks": [self._task_payload(item) for item in related_tasks], + "recent_logs": recent_logs, + "disk_paths": disk_paths, + "diagnosis": { + "summary": findings[0] if findings else "未发现明显异常。", + "findings": findings, + "cleanup_supported": cleanup_supported, + "cleanup_blockers": cleanup_blockers, + }, + } + + async def cleanup_preview(self, db: AsyncSession, task_id: str) -> Optional[Dict[str, Any]]: + diagnosis = await self.diagnose_task(db, task_id) + if diagnosis is None: + return None + + task = diagnosis["task"] + task_type = _norm_status(task.get("task_type")) + task_status = _norm_status(task.get("status")) + cleanup_supported = bool(diagnosis["diagnosis"].get("cleanup_supported")) + blockers = list(diagnosis["diagnosis"].get("cleanup_blockers") or []) + if task_type not in SUPPORTED_CLEANUP_TASK_TYPES: + blockers.append(f"第一版暂不支持清理任务类型 {task_type}。") + if task_status in ACTIVE_TASK_STATUSES: + blockers.append("任务仍处于活动状态,不能清理。") + + db_counts = await self._cleanup_db_counts(db, diagnosis) + disk_deletes = self._cleanup_disk_targets(diagnosis) + blocked = bool(blockers) or not cleanup_supported + return { + "task_id": task_id, + "blocked": blocked, + "blockers": blockers, + "cleanup_supported": cleanup_supported and not blocked, + "database_deletes": db_counts, + "disk_deletes": disk_deletes, + } + + async def cleanup_task( + self, + db: AsyncSession, + task_id: str, + *, + options: Dict[str, bool], + ) -> Optional[Dict[str, Any]]: + preview = await self.cleanup_preview(db, task_id) + if preview is None: + return None + if preview.get("blocked"): + raise ValueError("; ".join(preview.get("blockers") or ["清理被阻止。"])) + + deleted_db: Dict[str, int] = {} + task = await self._get_task(db, task_id) + if task is None: + return None + jobs = await self._get_jobs(db, task_id) + run = await self._get_production_run_for_task(db, task, jobs) + + delete_logs = bool(options.get("delete_logs", True)) + delete_task_records = bool(options.get("delete_task_records", True)) + delete_production_records = bool(options.get("delete_production_records", True)) + delete_result_products = bool(options.get("delete_result_products", True)) + delete_task_pool_dir = bool(options.get("delete_task_pool_dir", True)) + related_copy_tasks = await self._related_copy_tasks_for_run(db, run) if run is not None else [] + + if delete_result_products and run is not None: + products = await self._result_product_orms_for_run(db, run) + product_ids = [item.product_id for item in products] + compat_ids = await self._compat_ids_for_products(db, product_ids) + if product_ids: + result = await db.execute(delete(ResultProductORM).where(ResultProductORM.product_id.in_(product_ids))) + deleted_db["result_products"] = int(result.rowcount or 0) + if compat_ids: + result = await db.execute(delete(DinsarResultORM).where(DinsarResultORM.id.in_(compat_ids))) + deleted_db["dinsar_results"] = int(result.rowcount or 0) + + if delete_production_records and run is not None: + deleted_run = await dinsar_production_service.delete_run_record(run.run_id, db=db) + deleted_db["dinsar_production_runs"] = 1 if deleted_run else 0 + deleted_db["dinsar_production_run_items"] = int(preview["database_deletes"].get("dinsar_production_run_items", 0)) + deleted_db["dinsar_production_executions"] = int(preview["database_deletes"].get("dinsar_production_executions", 0)) + deleted_db["system_jobs"] = int(preview["database_deletes"].get("system_jobs", 0)) + deleted_db["system_tasks"] = 1 + deleted_db["task_logs"] = int(preview["database_deletes"].get("task_logs", 0)) + if delete_task_pool_dir: + related_deleted = await self._delete_related_tasks(db, related_copy_tasks) + for key, value in related_deleted.items(): + deleted_db[key] = int(deleted_db.get(key, 0)) + int(value or 0) + else: + if delete_logs: + result = await db.execute(delete(TaskLogORM).where(TaskLogORM.task_id == task_id)) + deleted_db["task_logs"] = int(result.rowcount or 0) + if delete_task_records: + result = await db.execute(delete(SystemJobORM).where(SystemJobORM.task_id == task_id)) + deleted_db["system_jobs"] = int(result.rowcount or 0) + result = await db.execute(delete(SystemTaskORM).where(SystemTaskORM.task_id == task_id)) + deleted_db["system_tasks"] = int(result.rowcount or 0) + await db.commit() + + disk_result = await self._delete_disk_targets(preview, options) + return { + "task_id": task_id, + "deleted_database": deleted_db, + "deleted_disk": disk_result, + } + + async def _delete_related_tasks( + self, + db: AsyncSession, + tasks: List[SystemTaskORM], + ) -> Dict[str, int]: + task_ids = [task.task_id for task in tasks if task.task_id] + if not task_ids: + return {} + result = await db.execute(delete(TaskLogORM).where(TaskLogORM.task_id.in_(task_ids))) + logs = int(result.rowcount or 0) + result = await db.execute(delete(SystemJobORM).where(SystemJobORM.task_id.in_(task_ids))) + jobs = int(result.rowcount or 0) + result = await db.execute(delete(SystemTaskORM).where(SystemTaskORM.task_id.in_(task_ids))) + task_count = int(result.rowcount or 0) + await db.commit() + return { + "related_task_logs": logs, + "related_system_jobs": jobs, + "related_system_tasks": task_count, + } + + async def _get_task(self, db: AsyncSession, task_id: str) -> Optional[SystemTaskORM]: + result = await db.execute(select(SystemTaskORM).where(SystemTaskORM.task_id == str(task_id or "").strip())) + return result.scalar_one_or_none() + + async def _get_jobs(self, db: AsyncSession, task_id: str) -> List[SystemJobORM]: + result = await db.execute( + select(SystemJobORM) + .where(SystemJobORM.task_id == str(task_id or "").strip()) + .order_by(SystemJobORM.updated_at.desc(), SystemJobORM.id.desc()) + ) + return list(result.scalars().all()) + + async def _get_production_run_for_task( + self, + db: AsyncSession, + task: SystemTaskORM, + jobs: List[SystemJobORM], + ) -> Optional[DinsarProductionRunORM]: + candidates: List[str] = [] + params = task.params or {} + if isinstance(params, dict): + for key in ("production_run_id", "run_id"): + if params.get(key): + candidates.append(str(params[key])) + for job in jobs: + payload = job.payload or {} + if isinstance(payload, dict): + for key in ("production_run_id", "run_id", "dinsar_run_id"): + if payload.get(key): + candidates.append(str(payload[key])) + if candidates: + result = await db.execute(select(DinsarProductionRunORM).where(DinsarProductionRunORM.run_id.in_(candidates))) + run = result.scalar_one_or_none() + if run is not None: + return run + result = await db.execute(select(DinsarProductionRunORM).where(DinsarProductionRunORM.task_id == task.task_id)) + return result.scalar_one_or_none() + + async def _build_task_summary(self, db: AsyncSession, task: SystemTaskORM) -> Dict[str, Any]: + jobs = await self._get_jobs(db, task.task_id) + run = await self._get_production_run_for_task(db, task, jobs) + item_counts: Dict[str, int] = {} + execution_counts: Dict[str, int] = {} + if run is not None: + item_counts = await self._status_counts(db, DinsarProductionRunItemORM, run.run_id) + execution_counts = await self._status_counts(db, DinsarProductionExecutionORM, run.run_id) + findings, cleanup_supported, blockers = self._diagnose_findings( + task=task, + jobs=jobs, + run=run, + item_counts=item_counts, + execution_counts=execution_counts, + disk_paths=[], + ) + log_count = await self._count(db, select(func.count()).select_from(TaskLogORM).where(TaskLogORM.task_id == task.task_id)) + return { + **self._task_payload(task), + "run_id": run.run_id if run else None, + "job_id": jobs[0].job_id if jobs else None, + "job_status": jobs[0].status if jobs else None, + "issue_level": "danger" if _norm_status(task.status) == "FAILED" else ("warning" if findings else "info"), + "issue_summary": findings[0] if findings else _compact(task.message, 160), + "cleanup_supported": cleanup_supported, + "cleanup_blocked_reason": "; ".join(blockers) if blockers else "", + "counts": { + "jobs": len(jobs), + "logs": log_count, + "run_items": sum(item_counts.values()), + "executions": sum(execution_counts.values()), + "completed_items": item_counts.get("COMPLETED", 0), + "failed_items": item_counts.get("FAILED", 0), + "pending_items": item_counts.get("PENDING", 0), + "running_items": item_counts.get("RUNNING", 0), + }, + } + + async def _status_counts(self, db: AsyncSession, model: Any, run_id: str) -> Dict[str, int]: + result = await db.execute( + select(model.status, func.count()) + .where(model.run_id == run_id) + .group_by(model.status) + ) + return {str(status or "UNKNOWN").upper(): int(count or 0) for status, count in result.all()} + + async def _recent_logs(self, db: AsyncSession, task_id: str) -> List[Dict[str, Any]]: + result = await db.execute( + select(TaskLogORM) + .where(TaskLogORM.task_id == task_id) + .order_by(TaskLogORM.timestamp.desc(), TaskLogORM.id.desc()) + .limit(30) + ) + return [ + { + "id": item.id, + "level": item.log_level, + "message": _compact(item.message, 500), + "timestamp": _dt(item.timestamp), + } + for item in result.scalars().all() + ] + + async def _collect_run_disk_paths(self, db: AsyncSession, run: DinsarProductionRunORM) -> List[Dict[str, Any]]: + paths: Dict[str, Dict[str, Any]] = {} + if run.source_root: + paths[_normalize_path_text(run.source_root)] = self._path_payload(run.source_root, "task_pool") + result = await db.execute(select(DinsarProductionExecutionORM).where(DinsarProductionExecutionORM.run_id == run.run_id)) + for execution in result.scalars().all(): + if execution.output_dir: + publish_dir = self._publish_package_dir(execution.output_dir) + paths[publish_dir] = self._path_payload(publish_dir, "production_result") + log_path = dinsar_production_service.read_run_log(run.run_id, max_bytes=1).get("path") + if log_path: + paths[_normalize_path_text(log_path)] = self._path_payload(log_path, "run_log") + return list(paths.values()) + + async def _collect_result_products(self, db: AsyncSession, run: DinsarProductionRunORM) -> List[Dict[str, Any]]: + products = await self._result_product_orms_for_run(db, run) + return [ + { + "product_id": item.product_id, + "display_name": item.display_name, + "status": item.status, + "health_status": item.health_status, + "publish_dir": item.publish_dir, + "manifest_path": item.manifest_path, + } + for item in products + ] + + async def _result_product_orms_for_run(self, db: AsyncSession, run: DinsarProductionRunORM) -> List[ResultProductORM]: + result = await db.execute(select(DinsarProductionExecutionORM).where(DinsarProductionExecutionORM.run_id == run.run_id)) + dirs = [self._publish_package_dir(item.output_dir) for item in result.scalars().all() if item.output_dir] + clauses = [] + for path in dirs: + clauses.append(ResultProductORM.publish_dir == path) + clauses.append(ResultProductORM.native_output_dir.like(path + "%")) + clauses.append(ResultProductORM.manifest_path.like(path + "%")) + clauses.append(ResultProductORM.primary_asset_path.like(path + "%")) + if not clauses: + return [] + products = await db.execute(select(ResultProductORM).where(or_(*clauses))) + by_id: Dict[str, ResultProductORM] = {} + for product in products.scalars().all(): + by_id[product.product_id] = product + return list(by_id.values()) + + async def _compat_ids_for_products(self, db: AsyncSession, product_ids: List[str]) -> List[int]: + if not product_ids: + return [] + result = await db.execute(select(DinsarResultORM).where(DinsarResultORM.compat_product_id.in_(product_ids))) + return [int(item.id) for item in result.scalars().all()] + + def _publish_package_dir(self, output_dir: str) -> str: + normalized = _normalize_path_text(output_dir) + marker = os.sep + "runs" + os.sep + if marker.lower() in normalized.lower(): + lower = normalized.lower() + index = lower.index(marker.lower()) + return normalized[:index] + return normalized + + def _copy_task_dest_dir(self, task: SystemTaskORM) -> str: + params = task.params if isinstance(task.params, dict) else {} + return _normalize_path_text(params.get("dest_dir")) if params.get("dest_dir") else "" + + async def _related_copy_tasks_for_run( + self, + db: AsyncSession, + run: Optional[DinsarProductionRunORM], + ) -> List[SystemTaskORM]: + source_root = _normalize_path_text(run.source_root if run is not None else "") + if not source_root: + return [] + result = await db.execute( + select(SystemTaskORM) + .where(SystemTaskORM.task_type == "COPY_DATA") + .order_by(SystemTaskORM.updated_at.desc(), SystemTaskORM.id.desc()) + .limit(1000) + ) + tasks = [] + for task in result.scalars().all(): + if _normalize_path_text(self._copy_task_dest_dir(task)).lower() == source_root.lower(): + tasks.append(task) + return tasks + + def _diagnose_findings( + self, + *, + task: SystemTaskORM, + jobs: List[SystemJobORM], + run: Optional[DinsarProductionRunORM], + item_counts: Dict[str, int], + execution_counts: Dict[str, int], + disk_paths: List[Dict[str, Any]], + ) -> tuple[List[str], bool, List[str]]: + findings: List[str] = [] + blockers: List[str] = [] + status = _norm_status(task.status) + if status == "FAILED": + findings.append("任务已失败,需要人工确认后清理。") + elif status == "PARTIAL_SUCCESS": + findings.append("任务部分成功,清理前请确认保留策略。") + elif status == "CANCELLED": + findings.append("任务已取消,可按需清理残留记录和目录。") + elif status in ACTIVE_TASK_STATUSES: + blockers.append("任务仍处于活动状态。") + + active_jobs = [job for job in jobs if _norm_status(job.status) in ACTIVE_JOB_STATUSES] + stale_cutoff = _utcnow_naive() - timedelta(seconds=int(getattr(settings, "JOB_WORKER_STALE_RUNNING_SECONDS", 7200))) + stale_jobs = [ + job for job in active_jobs + if job.heartbeat_at is not None and job.heartbeat_at < stale_cutoff + ] + if stale_jobs: + findings.append(f"发现 {len(stale_jobs)} 个心跳超时 job。") + if active_jobs and not stale_jobs: + blockers.append("仍存在活动 job。") + + residual_running = item_counts.get("RUNNING", 0) + execution_counts.get("RUNNING", 0) + if run is not None and residual_running: + findings.append(f"生产 run 中仍有 {residual_running} 个 RUNNING 残留。") + if run is not None and item_counts.get("PENDING", 0): + findings.append(f"生产 run 中仍有 {item_counts.get('PENDING', 0)} 个 PENDING 项未执行。") + + missing_paths = [item for item in disk_paths if item.get("path") and not item.get("exists")] + if missing_paths: + findings.append(f"有 {len(missing_paths)} 个登记路径已不存在。") + + cleanup_supported = _norm_status(task.task_type) in SUPPORTED_CLEANUP_TASK_TYPES and not blockers + return findings, cleanup_supported, blockers + + async def _cleanup_db_counts(self, db: AsyncSession, diagnosis: Dict[str, Any]) -> Dict[str, int]: + task_id = diagnosis["task"]["task_id"] + run = diagnosis.get("production_run") or {} + run_id = run.get("run_id") + workflow_run_id = run.get("workflow_run_id") + counts = { + "system_tasks": await self._count(db, select(func.count()).select_from(SystemTaskORM).where(SystemTaskORM.task_id == task_id)), + "system_jobs": await self._count(db, select(func.count()).select_from(SystemJobORM).where(SystemJobORM.task_id == task_id)), + "task_logs": await self._count(db, select(func.count()).select_from(TaskLogORM).where(TaskLogORM.task_id == task_id)), + "related_system_tasks": 0, + "related_system_jobs": 0, + "related_task_logs": 0, + "dinsar_production_runs": 0, + "dinsar_production_run_items": 0, + "dinsar_production_executions": 0, + "result_products": len(diagnosis.get("result_products") or []), + "dinsar_results": 0, + "workflow_runs": 0, + "workflow_steps": 0, + "workflow_artifacts": 0, + } + if run_id: + counts["dinsar_production_runs"] = await self._count(db, select(func.count()).select_from(DinsarProductionRunORM).where(DinsarProductionRunORM.run_id == run_id)) + counts["dinsar_production_run_items"] = await self._count(db, select(func.count()).select_from(DinsarProductionRunItemORM).where(DinsarProductionRunItemORM.run_id == run_id)) + counts["dinsar_production_executions"] = await self._count(db, select(func.count()).select_from(DinsarProductionExecutionORM).where(DinsarProductionExecutionORM.run_id == run_id)) + run_obj = await self._get_run_by_id(db, run_id) + related_tasks = await self._related_copy_tasks_for_run(db, run_obj) + related_task_ids = [item.task_id for item in related_tasks if item.task_id] + counts["related_system_tasks"] = len(related_task_ids) + if related_task_ids: + counts["related_system_jobs"] = await self._count(db, select(func.count()).select_from(SystemJobORM).where(SystemJobORM.task_id.in_(related_task_ids))) + counts["related_task_logs"] = await self._count(db, select(func.count()).select_from(TaskLogORM).where(TaskLogORM.task_id.in_(related_task_ids))) + if workflow_run_id: + counts["workflow_runs"] = await self._count(db, select(func.count()).select_from(WorkflowRunORM).where(WorkflowRunORM.run_id == workflow_run_id)) + counts["workflow_steps"] = await self._count(db, select(func.count()).select_from(WorkflowStepORM).where(WorkflowStepORM.run_id == workflow_run_id)) + counts["workflow_artifacts"] = await self._count(db, select(func.count()).select_from(WorkflowArtifactORM).where(WorkflowArtifactORM.run_id == workflow_run_id)) + return counts + + async def _get_run_by_id(self, db: AsyncSession, run_id: str) -> Optional[DinsarProductionRunORM]: + result = await db.execute(select(DinsarProductionRunORM).where(DinsarProductionRunORM.run_id == run_id)) + return result.scalar_one_or_none() + + async def _count(self, db: AsyncSession, stmt: Any) -> int: + return int((await db.execute(stmt)).scalar_one() or 0) + + def _cleanup_disk_targets(self, diagnosis: Dict[str, Any]) -> List[Dict[str, Any]]: + targets: Dict[str, Dict[str, Any]] = {} + for item in diagnosis.get("disk_paths") or []: + path = _normalize_path_text(item.get("path")) + if not path: + continue + payload = self._path_payload(path, item.get("kind") or "unknown") + payload["allowed"] = self._is_allowed_delete_path(path) + targets[path.lower()] = payload + return list(targets.values()) + + def _path_payload(self, path: str, kind: str) -> Dict[str, Any]: + normalized = _normalize_path_text(path) + exists = _path_exists(normalized) + return { + "path": normalized, + "kind": kind, + "exists": exists, + "allowed": self._is_allowed_delete_path(normalized), + } + + def _is_allowed_delete_path(self, path: str) -> bool: + normalized = _normalize_path_text(path) + if not normalized: + return False + roots = [ + settings.DINSAR_TASK_POOL_ROOT, + settings.DINSAR_PRODUCT_DIR, + os.path.join(settings.PROJECT_ROOT, "backend", "runtime", "dinsar_production"), + ] + full = os.path.abspath(normalized) + for root in roots: + root_text = _normalize_path_text(root) + if not root_text: + continue + root_full = os.path.abspath(root_text) + if full == root_full: + return False + try: + if os.path.commonpath([full, root_full]) == root_full: + return True + except ValueError: + continue + return False + + async def _delete_disk_targets(self, preview: Dict[str, Any], options: Dict[str, bool]) -> Dict[str, Any]: + delete_production_dirs = bool(options.get("delete_production_dirs", True)) + delete_task_pool_dir = bool(options.get("delete_task_pool_dir", True)) + deleted: List[str] = [] + missing: List[str] = [] + skipped: List[str] = [] + failed: List[Dict[str, str]] = [] + for item in preview.get("disk_deletes") or []: + kind = item.get("kind") + path = _normalize_path_text(item.get("path")) + if not item.get("allowed"): + skipped.append(path) + continue + if kind == "task_pool" and not delete_task_pool_dir: + skipped.append(path) + continue + if kind == "production_result" and not delete_production_dirs: + skipped.append(path) + continue + if not _path_exists(path): + missing.append(path) + continue + try: + if os.path.isdir(path): + shutil.rmtree(path) + else: + Path(path).unlink() + deleted.append(path) + except OSError as exc: + failed.append({"path": path, "error": str(exc)}) + return { + "deleted": deleted, + "missing": missing, + "skipped": skipped, + "failed": failed, + } + + def _task_payload(self, task: SystemTaskORM) -> Dict[str, Any]: + return { + "id": task.id, + "task_id": task.task_id, + "task_type": task.task_type, + "task_name": task.task_name, + "status": task.status, + "progress": task.progress, + "message": _compact(task.message), + "params": task.params, + "created_at": _dt(task.created_at), + "updated_at": _dt(task.updated_at), + "started_at": _dt(task.started_at), + "ended_at": _dt(task.ended_at), + } + + def _job_payload(self, job: SystemJobORM) -> Dict[str, Any]: + return { + "job_id": job.job_id, + "job_type": job.job_type, + "status": job.status, + "attempts": job.attempts, + "max_attempts": job.max_attempts, + "locked_by": job.locked_by, + "locked_at": _dt(job.locked_at), + "heartbeat_at": _dt(job.heartbeat_at), + "started_at": _dt(job.started_at), + "finished_at": _dt(job.finished_at), + "last_error": _compact(job.last_error), + } + + def _run_payload(self, run: DinsarProductionRunORM) -> Dict[str, Any]: + return { + "run_id": run.run_id, + "task_id": run.task_id, + "workflow_run_id": run.workflow_run_id, + "engine_code": run.engine_code, + "profile_code": run.profile_code, + "mode": run.mode, + "source_root": run.source_root, + "publish_root_dir": run.publish_root_dir, + "status": run.status, + "total_items": run.total_items, + "completed_items": run.completed_items, + "failed_items": run.failed_items, + "skipped_items": run.skipped_items, + "latest_message": _compact(run.latest_message), + "created_at": _dt(run.created_at), + "updated_at": _dt(run.updated_at), + "started_at": _dt(run.started_at), + "ended_at": _dt(run.ended_at), + } + + +ops_maintenance_service = OpsMaintenanceService() diff --git a/frontend/src/App.css b/frontend/src/App.css index a725a16..8fbe323 100644 --- a/frontend/src/App.css +++ b/frontend/src/App.css @@ -3109,6 +3109,269 @@ input[type="checkbox"] { cursor: not-allowed; } +.health-action-button.danger { + border-color: #b91c1c; + background: #b91c1c; +} + +.ops-task-panel { + display: grid; + gap: 12px; + padding: 12px; + border: 1px solid var(--color-border); + border-radius: 8px; + background: #fff; +} + +.ops-task-header, +.ops-task-detail-header, +.ops-task-actions { + display: flex; + align-items: flex-start; + justify-content: space-between; + gap: 12px; +} + +.ops-task-header h4, +.ops-task-detail-header h4 { + margin: 0; + color: var(--color-text-primary); + font-size: 0.95em; +} + +.ops-task-header p, +.ops-task-detail-header p { + margin: 4px 0 0; + color: var(--color-text-secondary); + font-size: 0.82em; + line-height: 1.5; +} + +.ops-task-summary { + display: grid; + grid-template-columns: repeat(3, minmax(0, 1fr)); + gap: 8px; +} + +.ops-task-summary > div, +.ops-task-preview-cell { + display: flex; + align-items: center; + justify-content: space-between; + gap: 8px; + min-width: 0; + padding: 8px 10px; + border: 1px solid #e2e8f0; + border-radius: 6px; + background: #f8fafc; +} + +.ops-task-summary span, +.ops-task-preview-cell span { + color: #64748b; + font-size: 0.78em; +} + +.ops-task-summary strong, +.ops-task-preview-cell strong { + color: #0f172a; + font-size: 0.9em; +} + +.ops-task-filters { + display: flex; + align-items: flex-end; + gap: 8px; + flex-wrap: wrap; +} + +.ops-task-filters label { + display: grid; + gap: 4px; + color: #475569; + font-size: 0.78em; + font-weight: 700; +} + +.ops-task-filters select { + min-width: 150px; + padding: 7px 9px; + border: 1px solid #cbd5e1; + border-radius: 6px; + background: #fff; + color: #0f172a; +} + +.ops-task-table-wrap { + overflow: auto; + border: 1px solid #e2e8f0; + border-radius: 8px; +} + +.ops-task-table { + width: 100%; + min-width: 880px; + border-collapse: collapse; + font-size: 0.82em; +} + +.ops-task-table th, +.ops-task-table td { + padding: 9px 10px; + border-bottom: 1px solid #e2e8f0; + text-align: left; + vertical-align: top; +} + +.ops-task-table th { + color: #475569; + background: #f8fafc; + font-weight: 800; +} + +.ops-task-table tr.selected td { + background: #eff6ff; +} + +.ops-task-table tr:last-child td { + border-bottom: 0; +} + +.ops-task-name { + color: #0f172a; + font-weight: 800; +} + +.ops-task-muted { + color: #64748b; + font-size: 0.82em; + line-height: 1.5; +} + +.ops-task-empty { + color: #64748b; + text-align: center !important; +} + +.ops-task-badge { + display: inline-flex; + align-items: center; + min-height: 22px; + padding: 2px 8px; + border-radius: 999px; + font-size: 0.76em; + font-weight: 800; + white-space: nowrap; +} + +.ops-task-badge.tone-danger { + background: #fee2e2; + color: #991b1b; +} + +.ops-task-badge.tone-warn { + background: #fef3c7; + color: #92400e; +} + +.ops-task-badge.tone-ok { + background: #dcfce7; + color: #166534; +} + +.ops-task-badge.tone-info { + background: #dbeafe; + color: #1d4ed8; +} + +.ops-task-badge.tone-neutral { + background: #e2e8f0; + color: #334155; +} + +.ops-task-detail, +.ops-task-preview { + display: grid; + gap: 12px; + padding: 12px; + border: 1px solid #dbe3ef; + border-radius: 8px; + background: #fbfdff; +} + +.ops-task-diagnosis { + display: grid; + gap: 6px; +} + +.ops-task-two-col { + display: grid; + grid-template-columns: minmax(0, 1fr) minmax(0, 1fr); + gap: 12px; +} + +.ops-task-section-title { + margin-bottom: 6px; + color: #334155; + font-size: 0.82em; + font-weight: 800; +} + +.ops-task-preview-grid { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); + gap: 6px; +} + +.ops-task-path-list, +.ops-task-log-list { + display: grid; + gap: 6px; +} + +.ops-task-path { + display: grid; + gap: 4px; + padding: 8px; + border: 1px solid #e2e8f0; + border-radius: 6px; + background: #fff; +} + +.ops-task-path.blocked { + border-color: #fecaca; + background: #fff7f7; +} + +.ops-task-path span, +.ops-task-path em, +.ops-task-log-list span { + color: #64748b; + font-size: 0.76em; + font-style: normal; +} + +.ops-task-path code { + color: #0f172a; + font-size: 0.78em; + white-space: normal; + overflow-wrap: anywhere; +} + +.ops-task-log-list > div { + padding: 8px; + border: 1px solid #e2e8f0; + border-radius: 6px; + background: #fff; +} + +.ops-task-log-list p { + margin: 3px 0 0; + color: #334155; + font-size: 0.8em; + line-height: 1.45; +} + .log-management-panel { display: grid; gap: 10px; @@ -3460,6 +3723,23 @@ input[type="checkbox"] { .health-grid { grid-template-columns: 1fr; } + + .ops-task-header, + .ops-task-detail-header, + .ops-task-actions { + align-items: stretch; + flex-direction: column; + } + + .ops-task-summary, + .ops-task-two-col { + grid-template-columns: 1fr; + } + + .ops-task-filters label, + .ops-task-filters select { + width: 100%; + } } /* Top Status Bar */ diff --git a/frontend/src/HealthCheckPanel.jsx b/frontend/src/HealthCheckPanel.jsx index 146d315..cf9d48c 100644 --- a/frontend/src/HealthCheckPanel.jsx +++ b/frontend/src/HealthCheckPanel.jsx @@ -7,6 +7,7 @@ import { syncWaterScenesFromDisk } from './api/water'; import { listEngines, runWslCheck } from './api/dinsarProduction'; import { getOrbitStatus, syncOrbitPools } from './api/orbit'; import LogManagementPanel from './LogManagementPanel'; +import OpsTaskMaintenancePanel from './OpsTaskMaintenancePanel'; const toNumber = (value) => { const parsed = Number(value); @@ -1507,6 +1508,7 @@ const HealthCheckPanel = ({ currentUser }) => { )} {/* 日志管理 */} + diff --git a/frontend/src/OpsTaskMaintenancePanel.jsx b/frontend/src/OpsTaskMaintenancePanel.jsx new file mode 100644 index 0000000..a3f2f35 --- /dev/null +++ b/frontend/src/OpsTaskMaintenancePanel.jsx @@ -0,0 +1,368 @@ +import React, { useCallback, useEffect, useMemo, useState } from 'react'; +import { + cleanupMaintenanceTask, + getMaintenanceTaskDiagnosis, + listMaintenanceTasks, + previewMaintenanceCleanup, +} from './api/opsMaintenance'; + +const TASK_TYPES = [ + ['', '全部类型'], + ['LANDSAR_RUN', 'LandSAR D-InSAR'], + ['LANDSAR_CLUSTER_RUN', 'LandSAR 集群'], + ['PYINT_RUN', 'PyINT/Gamma'], + ['IDL_RUN_DINSAR', 'SARscape D-InSAR'], + ['COPY_DATA', '数据准备'], + ['PAIRING_CACHE_REBUILD', '配对缓存'], +]; + +const STATUSES = [ + ['', '全部状态'], + ['FAILED', '失败'], + ['PARTIAL_SUCCESS', '部分成功'], + ['CANCELLED', '已取消'], + ['RUNNING', '运行中'], + ['PENDING', '等待中'], +]; + +const formatDate = (value) => { + if (!value) return '-'; + const date = new Date(value); + if (Number.isNaN(date.getTime())) return value; + return date.toLocaleString('zh-CN', { hour12: false }); +}; + +const formatCount = (value) => Number(value || 0).toLocaleString('zh-CN'); + +const toneForStatus = (status) => { + const value = String(status || '').toUpperCase(); + if (value === 'FAILED') return 'danger'; + if (value === 'PARTIAL_SUCCESS' || value === 'CANCELLED') return 'warn'; + if (value === 'COMPLETED') return 'ok'; + if (value === 'RUNNING' || value === 'PENDING') return 'info'; + return 'neutral'; +}; + +const StatusBadge = ({ status }) => ( + {status || '-'} +); + +const CountLine = ({ counts = {} }) => ( + + 成功 {formatCount(counts.completed_items)} / 失败 {formatCount(counts.failed_items)} / 等待 {formatCount(counts.pending_items)} / 运行 {formatCount(counts.running_items)} + +); + +const DatabasePreview = ({ counts = {} }) => { + const entries = Object.entries(counts).filter(([, value]) => Number(value || 0) > 0); + if (!entries.length) return
没有将删除的数据库记录。
; + return ( +
+ {entries.map(([key, value]) => ( +
+ {key} + {formatCount(value)} +
+ ))} +
+ ); +}; + +const DiskPreview = ({ paths = [] }) => { + if (!paths.length) return
没有将删除的磁盘路径。
; + return ( +
+ {paths.map(item => ( +
+ {item.kind || 'path'} + {item.path} + {item.exists ? '存在' : '不存在'} / {item.allowed ? '允许' : '禁止'} +
+ ))} +
+ ); +}; + +const OpsTaskMaintenancePanel = ({ isAdmin }) => { + const [filters, setFilters] = useState({ task_type: '', status: '' }); + const [tasks, setTasks] = useState([]); + const [loading, setLoading] = useState(false); + const [error, setError] = useState(''); + const [selectedTaskId, setSelectedTaskId] = useState(''); + const [diagnosis, setDiagnosis] = useState(null); + const [diagnosisLoading, setDiagnosisLoading] = useState(false); + const [preview, setPreview] = useState(null); + const [previewLoading, setPreviewLoading] = useState(false); + const [cleanupLoading, setCleanupLoading] = useState(false); + const [cleanupResult, setCleanupResult] = useState(null); + const [panelMessage, setPanelMessage] = useState(null); + + const loadTasks = useCallback(async () => { + setLoading(true); + setError(''); + try { + const data = await listMaintenanceTasks({ + task_type: filters.task_type || undefined, + status: filters.status || undefined, + limit: 80, + }); + setTasks(Array.isArray(data?.items) ? data.items : []); + } catch (err) { + setError(err.response?.data?.detail || err.message || '任务维护列表加载失败'); + } finally { + setLoading(false); + } + }, [filters]); + + useEffect(() => { + loadTasks(); + }, [loadTasks]); + + const selectedTask = useMemo( + () => tasks.find(item => item.task_id === selectedTaskId) || null, + [tasks, selectedTaskId], + ); + + const loadDiagnosis = useCallback(async (taskId) => { + setSelectedTaskId(taskId); + setDiagnosis(null); + setPreview(null); + setCleanupResult(null); + setPanelMessage(null); + setDiagnosisLoading(true); + try { + const data = await getMaintenanceTaskDiagnosis(taskId); + setDiagnosis(data); + } catch (err) { + setCleanupResult({ ok: false, message: err.response?.data?.detail || err.message || '诊断加载失败' }); + } finally { + setDiagnosisLoading(false); + } + }, []); + + const loadPreview = useCallback(async () => { + if (!selectedTaskId) return; + setPreviewLoading(true); + setCleanupResult(null); + try { + const data = await previewMaintenanceCleanup(selectedTaskId); + setPreview(data); + } catch (err) { + setCleanupResult({ ok: false, message: err.response?.data?.detail || err.message || '清理预览失败' }); + } finally { + setPreviewLoading(false); + } + }, [selectedTaskId]); + + const executeCleanup = useCallback(async () => { + if (!selectedTaskId || !preview || preview.blocked || !isAdmin) return; + setCleanupLoading(true); + setCleanupResult(null); + try { + const data = await cleanupMaintenanceTask(selectedTaskId, { + confirm: true, + delete_task_records: true, + delete_logs: true, + delete_production_records: true, + delete_result_products: true, + delete_production_dirs: true, + delete_task_pool_dir: true, + }); + setCleanupResult({ + ok: true, + message: `清理完成:数据库 ${Object.values(data.deleted_database || {}).reduce((sum, value) => sum + Number(value || 0), 0)} 条,目录 ${(data.deleted_disk?.deleted || []).length} 个。`, + }); + setPanelMessage({ + ok: true, + message: `清理完成:数据库 ${Object.values(data.deleted_database || {}).reduce((sum, value) => sum + Number(value || 0), 0)} 条,目录 ${(data.deleted_disk?.deleted || []).length} 个。`, + }); + setPreview(null); + setDiagnosis(null); + setSelectedTaskId(''); + await loadTasks(); + } catch (err) { + setCleanupResult({ ok: false, message: err.response?.data?.detail || err.message || '清理失败' }); + setPanelMessage({ ok: false, message: err.response?.data?.detail || err.message || '清理失败' }); + } finally { + setCleanupLoading(false); + } + }, [isAdmin, loadTasks, preview, selectedTaskId]); + + const abnormalCount = tasks.length; + const cleanableCount = tasks.filter(item => item.cleanup_supported).length; + const blockedCount = tasks.filter(item => !item.cleanup_supported).length; + + return ( +
+
+
+

任务维护

+

查看失败、部分成功、取消和残留运行的任务;清理后请回到业务页面重新提交。

+
+ +
+ +
+
异常任务{formatCount(abnormalCount)}
+
可清理{formatCount(cleanableCount)}
+
需复核{formatCount(blockedCount)}
+
+ +
+ + +
+ + {error &&
{error}
} + {panelMessage && ( +
+ {panelMessage.message} +
+ )} + +
+ + + + + + + + + + + + + {tasks.map(task => ( + + + + + + + + + ))} + {!loading && !tasks.length && ( + + + + )} + +
状态任务统计问题摘要更新时间操作
+
{task.task_name || task.task_id}
+
{task.task_type} / {task.task_id}
+
{task.issue_summary || '-'}{formatDate(task.updated_at)} + +
当前没有匹配的异常任务。
+
+ + {selectedTaskId && ( +
+
+
+

{selectedTask?.task_name || selectedTaskId}

+

{selectedTaskId}

+
+ +
+ + {diagnosisLoading ? ( +
正在加载诊断...
+ ) : diagnosis ? ( + <> +
+ {(diagnosis.diagnosis?.findings || []).map(item => ( +
{item}
+ ))} + {!(diagnosis.diagnosis?.findings || []).length && ( +
未发现明显异常。
+ )} + {(diagnosis.diagnosis?.cleanup_blockers || []).map(item => ( +
{item}
+ ))} +
+ +
+
+
生产统计
+ [`execution_${key}`, value])), + }} /> +
+
+
最近日志
+
+ {(diagnosis.recent_logs || []).slice(0, 8).map(log => ( +
+ [{formatDate(log.timestamp)}] {log.level} +

{log.message}

+
+ ))} + {!(diagnosis.recent_logs || []).length &&
无日志。
} +
+
+
+ + ) : null} + + {preview && ( +
+
清理预览
+ {preview.blocked && ( +
+ {preview.blockers?.join(';') || '当前任务不允许清理。'} +
+ )} +
+
+
数据库记录
+ +
+
+
磁盘路径
+ +
+
+
+ {!isAdmin && 仅管理员可执行清理。} + +
+
+ )} + + {cleanupResult && ( +
+ {cleanupResult.message} +
+ )} +
+ )} +
+ ); +}; + +export default OpsTaskMaintenancePanel; diff --git a/frontend/src/api/opsMaintenance.js b/frontend/src/api/opsMaintenance.js new file mode 100644 index 0000000..94ea568 --- /dev/null +++ b/frontend/src/api/opsMaintenance.js @@ -0,0 +1,13 @@ +import apiClient from './client'; + +export const listMaintenanceTasks = (params = {}) => + apiClient.get('/ops-maintenance/tasks', { params }).then(r => r.data); + +export const getMaintenanceTaskDiagnosis = (taskId) => + apiClient.get(`/ops-maintenance/tasks/${encodeURIComponent(taskId)}/diagnosis`).then(r => r.data); + +export const previewMaintenanceCleanup = (taskId) => + apiClient.post(`/ops-maintenance/tasks/${encodeURIComponent(taskId)}/cleanup-preview`).then(r => r.data); + +export const cleanupMaintenanceTask = (taskId, payload) => + apiClient.post(`/ops-maintenance/tasks/${encodeURIComponent(taskId)}/cleanup`, payload).then(r => r.data);