diff --git a/backend/app/services/dinsar_engine_matrix.py b/backend/app/services/dinsar_engine_matrix.py index a1e8396..20e30ed 100644 --- a/backend/app/services/dinsar_engine_matrix.py +++ b/backend/app/services/dinsar_engine_matrix.py @@ -5,8 +5,9 @@ from typing import Any, Dict, Iterable, List, Optional from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload -from ..models import DinsarTaskItemORM, ResultProductORM +from ..models import DinsarProductionRunItemORM, DinsarTaskItemORM, ResultProductORM from ..utils import normalize_satellite_family @@ -102,11 +103,32 @@ def _product_status(product: ResultProductORM) -> str: return "ready" +def _status_timestamp(value: Optional[datetime]) -> float: + return _timestamp(value) + + +def _latest_failed_attempt(attempts: Iterable[DinsarProductionRunItemORM]) -> Optional[DinsarProductionRunItemORM]: + latest: Optional[DinsarProductionRunItemORM] = None + for attempt in attempts: + if str(attempt.status or "").strip().upper() != "FAILED": + continue + if latest is None or ( + _status_timestamp(attempt.ended_at or attempt.updated_at or attempt.created_at), + attempt.id or 0, + ) > ( + _status_timestamp(latest.ended_at or latest.updated_at or latest.created_at), + latest.id or 0, + ): + latest = attempt + return latest + + def serialize_engine_result( *, engine_code: str, data_family: str, product: Optional[ResultProductORM], + failed_attempt: Optional[DinsarProductionRunItemORM] = None, allowed: bool, legacy: bool = False, ) -> Dict[str, Any]: @@ -119,6 +141,10 @@ def serialize_engine_result( status = "blocked" can_dispatch = False skip_reason = "unsupported_data_family" + elif product is None and failed_attempt is not None: + status = "failed" + can_dispatch = True + skip_reason = "production_failed" elif product is None: status = "missing" can_dispatch = True @@ -140,11 +166,18 @@ def serialize_engine_result( ), "latest_product_id": product.id if product is not None else None, "product_id": product.product_id if product is not None else None, - "run_key": product.run_key if product is not None else None, + "run_key": product.run_key if product is not None else failed_attempt.latest_run_key if failed_attempt is not None else None, "published_at": product.published_at if product is not None else None, "health_status": product.health_status if product is not None else None, "primary_asset_path": product.primary_asset_path if product is not None else None, "preview_path": product.preview_path if product is not None else None, + "production_run_id": failed_attempt.run_id if failed_attempt is not None else None, + "production_item_id": failed_attempt.id if failed_attempt is not None else None, + "production_status": failed_attempt.status if failed_attempt is not None else None, + "production_error": failed_attempt.last_error if failed_attempt is not None else None, + "latest_log_path": failed_attempt.latest_log_path if failed_attempt is not None else None, + "latest_output_dir": failed_attempt.latest_output_dir if failed_attempt is not None else None, + "attempt_ended_at": failed_attempt.ended_at if failed_attempt is not None else None, "can_dispatch": can_dispatch, "skip_reason": skip_reason, } @@ -153,6 +186,7 @@ def serialize_engine_result( def build_engine_results( *, products: Iterable[ResultProductORM], + failed_attempts: Iterable[DinsarProductionRunItemORM] = (), data_family: str = "unknown", include_legacy: bool = False, ) -> Dict[str, Dict[str, Any]]: @@ -168,6 +202,17 @@ def build_engine_results( ): target[engine] = product + failed_by_engine: Dict[str, DinsarProductionRunItemORM] = {} + attempts_by_engine: Dict[str, List[DinsarProductionRunItemORM]] = {} + for attempt in failed_attempts: + engine = normalize_dinsar_engine_code(getattr(getattr(attempt, "run", None), "engine_code", "")) + if engine: + attempts_by_engine.setdefault(engine, []).append(attempt) + for engine, attempts in attempts_by_engine.items(): + latest = _latest_failed_attempt(attempts) + if latest is not None: + failed_by_engine[engine] = latest + allowed = allowed_engines_for_data_family(data_family) matrix: Dict[str, Dict[str, Any]] = {} for engine in CURRENT_DINSAR_ENGINE_ORDER: @@ -175,6 +220,7 @@ def build_engine_results( engine_code=engine, data_family=data_family, product=latest_by_engine.get(engine), + failed_attempt=failed_by_engine.get(engine) if latest_by_engine.get(engine) is None else None, allowed=engine in allowed, ) @@ -184,6 +230,7 @@ def build_engine_results( engine_code=engine, data_family=data_family, product=product, + failed_attempt=None, allowed=False, legacy=True, ) @@ -229,6 +276,27 @@ async def build_engine_results_for_task_items( ) products = result.scalars().all() + attempt_conditions = [] + if pair_keys: + attempt_conditions.append(DinsarProductionRunItemORM.pair_key.in_(pair_keys)) + if aliases: + attempt_conditions.append(DinsarProductionRunItemORM.task_alias.in_(aliases)) + attempt_conditions.append(DinsarProductionRunItemORM.task_name.in_(aliases)) + failed_attempts: List[DinsarProductionRunItemORM] = [] + if attempt_conditions: + attempt_result = await db.execute( + select(DinsarProductionRunItemORM) + .options(selectinload(DinsarProductionRunItemORM.run)) + .where(DinsarProductionRunItemORM.status == "FAILED") + .where(or_(*attempt_conditions)) + .order_by( + DinsarProductionRunItemORM.ended_at.desc().nullslast(), + DinsarProductionRunItemORM.updated_at.desc().nullslast(), + DinsarProductionRunItemORM.id.desc(), + ) + ) + failed_attempts = attempt_result.scalars().all() + by_pair_key: Dict[str, List[ResultProductORM]] = {} by_alias: Dict[str, List[ResultProductORM]] = {} for product in products: @@ -239,15 +307,33 @@ async def build_engine_results_for_task_items( if alias: by_alias.setdefault(alias, []).append(product) + failed_by_pair_key: Dict[str, List[DinsarProductionRunItemORM]] = {} + failed_by_alias: Dict[str, List[DinsarProductionRunItemORM]] = {} + for attempt in failed_attempts: + pair_key = str(attempt.pair_key or "").strip() + if pair_key: + failed_by_pair_key.setdefault(pair_key, []).append(attempt) + for alias in {str(attempt.task_alias or "").strip(), str(attempt.task_name or "").strip()}: + if alias: + failed_by_alias.setdefault(alias, []).append(attempt) + output: Dict[int, Dict[str, Dict[str, Any]]] = {} for item in items: item_products: List[ResultProductORM] = [] + item_failed_attempts: List[DinsarProductionRunItemORM] = [] pair_key = str(item.pair_key or "").strip() alias = str(item.task_alias or item.task_name or "").strip() if pair_key: item_products.extend(by_pair_key.get(pair_key, [])) + item_failed_attempts.extend(failed_by_pair_key.get(pair_key, [])) if not item_products and alias: item_products.extend(by_alias.get(alias, [])) + if not item_failed_attempts and alias: + item_failed_attempts.extend(failed_by_alias.get(alias, [])) data_family = infer_dinsar_data_family(item.master_satellite, item.slave_satellite) - output[int(item.id)] = build_engine_results(products=item_products, data_family=data_family) + output[int(item.id)] = build_engine_results( + products=item_products, + failed_attempts=item_failed_attempts, + data_family=data_family, + ) return output diff --git a/backend/app/services/result_catalog_service.py b/backend/app/services/result_catalog_service.py index b67b587..a80348f 100644 --- a/backend/app/services/result_catalog_service.py +++ b/backend/app/services/result_catalog_service.py @@ -12,11 +12,13 @@ from typing import Any, Dict, Iterable, List, Optional, Tuple from geoalchemy2.shape import from_shape from sqlalchemy import delete, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload from shapely.geometry import Polygon, shape from ..config import settings from ..models import ( DinsarProductProfileORM, + DinsarProductionRunItemORM, DinsarTaskItemORM, PairingMetricCacheORM, PairingNetworkEdgeORM, @@ -1519,8 +1521,16 @@ class ResultCatalogService: products = result.scalars().all() groups: Dict[str, Dict[str, Any]] = {} + pair_keys_for_trace = set() + aliases_for_trace = set() for product in products: pair_key = str(product.pair_key or "").strip() or f"product:{product.id}" + if str(product.pair_key or "").strip(): + pair_keys_for_trace.add(str(product.pair_key or "").strip()) + for alias_value in (product.task_alias, product.task_name): + alias_text = str(alias_value or "").strip() + if alias_text: + aliases_for_trace.add(alias_text) group = groups.setdefault( pair_key, { @@ -1552,11 +1562,49 @@ class ResultCatalogService: if group.get(key) in (None, "") and getattr(product, key) not in (None, ""): group[key] = getattr(product, key) + failed_trace_by_pair_key: Dict[str, List[DinsarProductionRunItemORM]] = {} + failed_trace_by_alias: Dict[str, List[DinsarProductionRunItemORM]] = {} + trace_conditions = [] + if pair_keys_for_trace: + trace_conditions.append(DinsarProductionRunItemORM.pair_key.in_(sorted(pair_keys_for_trace))) + if aliases_for_trace: + trace_conditions.append(DinsarProductionRunItemORM.task_alias.in_(sorted(aliases_for_trace))) + trace_conditions.append(DinsarProductionRunItemORM.task_name.in_(sorted(aliases_for_trace))) + if trace_conditions: + trace_result = await db.execute( + select(DinsarProductionRunItemORM) + .options(selectinload(DinsarProductionRunItemORM.run)) + .where(DinsarProductionRunItemORM.status == "FAILED") + .where(or_(*trace_conditions)) + .order_by( + DinsarProductionRunItemORM.ended_at.desc().nullslast(), + DinsarProductionRunItemORM.updated_at.desc().nullslast(), + DinsarProductionRunItemORM.id.desc(), + ) + ) + for attempt in trace_result.scalars().all(): + pair_key = str(attempt.pair_key or "").strip() + if pair_key: + failed_trace_by_pair_key.setdefault(pair_key, []).append(attempt) + for alias_value in (attempt.task_alias, attempt.task_name): + alias_text = str(alias_value or "").strip() + if alias_text: + failed_trace_by_alias.setdefault(alias_text, []).append(attempt) + items: List[Dict[str, Any]] = [] for group in groups.values(): products_for_group: List[ResultProductORM] = group.pop("_products") + failed_attempts_for_group: List[DinsarProductionRunItemORM] = [] + pair_key = str(group.get("pair_key") or "").strip() + if pair_key: + failed_attempts_for_group.extend(failed_trace_by_pair_key.get(pair_key, [])) + for alias_value in (group.get("task_alias"), group.get("task_name")): + alias_text = str(alias_value or "").strip() + if alias_text and not failed_attempts_for_group: + failed_attempts_for_group.extend(failed_trace_by_alias.get(alias_text, [])) engine_results = build_engine_results( products=products_for_group, + failed_attempts=failed_attempts_for_group, data_family=group.get("data_family") or "unknown", include_legacy=include_legacy, ) diff --git a/frontend/src/components/DinsarCatalogPanel.jsx b/frontend/src/components/DinsarCatalogPanel.jsx index 1bc3e30..723e08c 100644 --- a/frontend/src/components/DinsarCatalogPanel.jsx +++ b/frontend/src/components/DinsarCatalogPanel.jsx @@ -67,6 +67,34 @@ function engineStatusTone(status) { return 'neutral'; } +function formatEngineResultStatus(status) { + const normalized = String(status || 'missing').toLowerCase(); + if (normalized === 'ready') return '已生产'; + if (normalized === 'failed') return '生产失败'; + if (normalized === 'missing') return '未生产'; + if (normalized === 'blocked') return '不适用'; + if (normalized === 'running') return '生产中'; + if (normalized === 'legacy') return '历史结果'; + return status || '未生产'; +} + +function buildEngineResultTitle(result = {}) { + const parts = [formatEngineResultStatus(result.status)]; + if (result.skip_reason === 'production_failed') { + parts.push('该引擎已有失败生产记录，不是单纯缺失结果。'); + } + if (result.production_run_id) { + parts.push(`run=${result.production_run_id}`); + } + if (result.production_error) { + parts.push(String(result.production_error).replace(/\s+/g, ' ').slice(0, 220)); + } + if (result.latest_log_path) { + parts.push(`log=${result.latest_log_path}`); + } + return parts.filter(Boolean).join('\n'); +} + function MetaField({ label, value, multiline = false }) { const displayValue = value === null || value === undefined || value === '' ? '-' : value; return ( @@ -373,8 +401,9 @@ export default function DinsarCatalogPanel({ - {engineMeta.shortLabel}: {result.status || 'missing'} + {engineMeta.shortLabel}: {formatEngineResultStatus(result.status)} ); })}