diff --git a/.env.example b/.env.example index 73c5477..f9e9d6a 100644 --- a/.env.example +++ b/.env.example @@ -109,6 +109,12 @@ DINSAR_PRODUCT_DIR=D:\production_results\dinsar TIMESERIES_PRODUCT_DIR=D:\production_results\timeseries PSINSAR_PRODUCT_DIR=D:\production_results\timeseries RESULT_QUARANTINE_ROOT=D:\production_results\_quarantine +RESULT_DELIVERY_ROOT=D:\Result_Delivery +RESULT_DELIVERY_PUBLIC_BASE_URL=/deliveries +RESULT_DELIVERY_RETENTION_DAYS=7 +RESULT_DELIVERY_MAX_ITEMS=500 +RESULT_DELIVERY_ZIP_MAX_BYTES=21474836480 +RESULT_DELIVERY_CHECKSUM_ENABLED=true RESULT_CATALOG_AUTO_REBUILD_ON_STARTUP=true diff --git a/backend/app/config.py b/backend/app/config.py index cd87ea0..7b09f12 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -77,6 +77,14 @@ def _default_result_publish_root(project_root: str) -> str: return os.path.join(normalized_root, "production_results") +def _default_result_delivery_root(project_root: str) -> str: + normalized_root = os.path.normpath(project_root) + drive, _tail = os.path.splitdrive(normalized_root) + if drive: + return os.path.join(drive + os.sep, "Result_Delivery") + return os.path.join(normalized_root, "result_delivery") + + def _default_input_root(project_root: str) -> str: normalized_root = os.path.normpath(project_root) drive, _tail = os.path.splitdrive(normalized_root) @@ -313,6 +321,12 @@ class Settings(BaseSettings): PSINSAR_PRODUCT_DIR: str = "" RESULT_QUARANTINE_ROOT: str = "" RESULT_CATALOG_AUTO_REBUILD_ON_STARTUP: bool = True + RESULT_DELIVERY_ROOT: str = "" + RESULT_DELIVERY_PUBLIC_BASE_URL: str = "/deliveries" + RESULT_DELIVERY_RETENTION_DAYS: int = 7 + RESULT_DELIVERY_MAX_ITEMS: int = 500 + RESULT_DELIVERY_ZIP_MAX_BYTES: int = 21474836480 + RESULT_DELIVERY_CHECKSUM_ENABLED: bool = True WSL_DISTRO: str = "" WSL_SHARED_CONDA_ENV: str = "" @@ -626,6 +640,12 @@ class Settings(BaseSettings): "RESULT_QUARANTINE_ROOT", os.path.join(self.RESULT_PUBLISH_ROOT, "_quarantine"), ) + if not self.RESULT_DELIVERY_ROOT: + object.__setattr__( + self, + "RESULT_DELIVERY_ROOT", + _default_result_delivery_root(project_root), + ) if not self.ISCE2_WORK_ROOT: object.__setattr__( self, @@ -1088,6 +1108,7 @@ class Settings(BaseSettings): os.makedirs(settings.TIMESERIES_PRODUCT_DIR, exist_ok=True) os.makedirs(settings.PSINSAR_PRODUCT_DIR, exist_ok=True) os.makedirs(settings.RESULT_QUARANTINE_ROOT, exist_ok=True) + os.makedirs(settings.RESULT_DELIVERY_ROOT, exist_ok=True) os.makedirs(settings.SAR_ANALYSIS_READY_ROOT, exist_ok=True) os.makedirs(settings.SAR_ANALYSIS_WORK_ROOT, exist_ok=True) os.makedirs(settings.TASK_POOL_ROOT, exist_ok=True) diff --git a/backend/app/db_maintenance.py b/backend/app/db_maintenance.py index 2619613..64e7970 100644 --- a/backend/app/db_maintenance.py +++ b/backend/app/db_maintenance.py @@ -42,6 +42,7 @@ MIGRATION_FILES = [ "010_source_orbit_asset_inventory.sql", "011_source_metadata_documents.sql", "012_source_archive_integrity.sql", + "013_result_delivery_requests.sql", ] diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 72494b8..5f9d51a 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -10,6 +10,8 @@ from .orm import ( ResultAssetORM, ResultIssueORM, ResultCatalogStateORM, + ResultDeliveryRequestORM, + ResultDeliveryItemORM, PairingCacheStateORM, PairingDirtySceneORM, PairingMetricCacheORM, @@ -100,6 +102,7 @@ __all__ = [ "RadarDataORM", "DinsarResultORM", "HazardPointORM", "ResultProductORM", "DinsarProductProfileORM", "ResultAssetORM", "ResultIssueORM", "ResultCatalogStateORM", + "ResultDeliveryRequestORM", "ResultDeliveryItemORM", "PairingCacheStateORM", "PairingDirtySceneORM", "PairingMetricCacheORM", "PairingNetworkRunORM", "PairingNetworkEdgeORM", "TimeseriesStackPlanORM", "TimeseriesStackPlanItemORM", "TimeseriesStackPlanEdgeORM", diff --git a/backend/app/models/orm.py b/backend/app/models/orm.py index 2f173e0..325a47c 100644 --- a/backend/app/models/orm.py +++ b/backend/app/models/orm.py @@ -322,6 +322,82 @@ class ResultCatalogStateORM(Base): updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) +class ResultDeliveryRequestORM(Base): + __tablename__ = "result_delivery_requests" + + id = Column(Integer, primary_key=True, autoincrement=True) + delivery_id = Column(String(64), unique=True, index=True, nullable=False) + owner_user_id = Column(Integer, ForeignKey("auth_users.id", ondelete="SET NULL"), nullable=True, index=True) + owner_username = Column(String(64), index=True, nullable=False) + channel = Column(String(32), index=True, nullable=False) + status = Column(String(32), index=True, nullable=False, default="PENDING", server_default="PENDING") + package_mode = Column(String(16), nullable=False, default="directory", server_default="directory") + item_count = Column(Integer, nullable=False, default=0, server_default="0") + total_bytes = Column(BigInteger, nullable=False, default=0, server_default="0") + copied_bytes = Column(BigInteger, nullable=False, default=0, server_default="0") + delivery_root = Column(String, nullable=False) + delivery_dir = Column(String, nullable=False) + zip_path = Column(String, nullable=True) + manifest_path = Column(String, nullable=True) + expires_at = Column(DateTime, nullable=True, index=True) + task_id = Column(String(128), nullable=True, index=True) + job_id = Column(String(128), nullable=True, index=True) + error_message = Column(Text, nullable=True) + request_json = Column(JSON, nullable=True) + summary_json = Column(JSON, nullable=True) + created_at = Column(DateTime, server_default=func.now(), nullable=False) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False) + started_at = Column(DateTime, nullable=True) + completed_at = Column(DateTime, nullable=True) + + owner = relationship("AuthUserORM") + items = relationship( + "ResultDeliveryItemORM", + back_populates="delivery", + cascade="all, delete-orphan", + ) + + __table_args__ = ( + Index("idx_result_delivery_owner_status", "owner_user_id", "status"), + Index("idx_result_delivery_channel_status", "channel", "status"), + Index("idx_result_delivery_created", "created_at"), + ) + + +class ResultDeliveryItemORM(Base): + __tablename__ = "result_delivery_items" + + id = Column(Integer, primary_key=True, autoincrement=True) + delivery_id = Column( + String(64), + ForeignKey("result_delivery_requests.delivery_id", ondelete="CASCADE"), + index=True, + nullable=False, + ) + source_product_id = Column(Integer, ForeignKey("result_products.id", ondelete="SET NULL"), nullable=True, index=True) + source_result_id = Column(Integer, ForeignKey("dinsar_results.id", ondelete="SET NULL"), nullable=True, index=True) + source_asset_id = Column(Integer, ForeignKey("result_assets.id", ondelete="SET NULL"), nullable=True, index=True) + display_name = Column(String(255), nullable=False) + source_path = Column(String, nullable=False) + relative_path = Column(String, nullable=True) + file_size = Column(BigInteger, nullable=False, default=0, server_default="0") + checksum_sha256 = Column(String(64), nullable=True) + status = Column(String(32), index=True, nullable=False, default="PENDING", server_default="PENDING") + error_message = Column(Text, nullable=True) + created_at = Column(DateTime, server_default=func.now(), nullable=False) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False) + + delivery = relationship("ResultDeliveryRequestORM", back_populates="items") + product = relationship("ResultProductORM", foreign_keys=[source_product_id]) + compat_result = relationship("DinsarResultORM", foreign_keys=[source_result_id]) + asset = relationship("ResultAssetORM", foreign_keys=[source_asset_id]) + + __table_args__ = ( + Index("idx_result_delivery_items_delivery_status", "delivery_id", "status"), + Index("idx_result_delivery_items_product", "source_product_id"), + ) + + class PairingCacheStateORM(Base): __tablename__ = "pairing_cache_state" diff --git a/backend/app/routers/__init__.py b/backend/app/routers/__init__.py index eae33af..998fe54 100644 --- a/backend/app/routers/__init__.py +++ b/backend/app/routers/__init__.py @@ -21,6 +21,7 @@ from . import ( orbit, pairing, ps_products, + result_deliveries, sbas_insar_products, radar, root_registry, @@ -55,6 +56,7 @@ def include_all_routers(router: APIRouter) -> None: router.include_router(pairing.router) router.include_router(dinsar.router) router.include_router(dinsar_products.router) + router.include_router(result_deliveries.router) router.include_router(dinsar_production.router) router.include_router(landsar_lt1_production.router) router.include_router(sbas_insar_production.router) diff --git a/backend/app/routers/dependencies.py b/backend/app/routers/dependencies.py index df8b142..1add7ad 100644 --- a/backend/app/routers/dependencies.py +++ b/backend/app/routers/dependencies.py @@ -296,6 +296,12 @@ def _is_high_risk_write_path(path: str, method: str) -> bool: return any(normalized.startswith(prefix) for prefix in HIGH_RISK_WRITE_PATH_PREFIXES) +def _is_user_self_service_write_path(path: str, method: str) -> bool: + normalized = _normalize_request_path(path) + upper_method = (method or "").upper() + return upper_method == "POST" and normalized == "/api/result-deliveries" + + def _get_client_ip(request: Request) -> Optional[str]: direct_ip = request.client.host if request.client else None if direct_ip in _TRUSTED_PROXY_IPS: @@ -400,7 +406,11 @@ async def _require_auth(request: Request, db: AsyncSession = Depends(get_db)): await db.commit() raise HTTPException(status_code=401, detail="Authentication required.") - if (not _is_read_only_operation(path, method)) and user.role != ROLE_ADMIN: + if ( + (not _is_read_only_operation(path, method)) + and user.role != ROLE_ADMIN + and not _is_user_self_service_write_path(path, method) + ): if _is_high_risk_write_path(path, method): await add_audit_log( db, diff --git a/backend/app/routers/result_deliveries.py b/backend/app/routers/result_deliveries.py new file mode 100644 index 0000000..19e7362 --- /dev/null +++ b/backend/app/routers/result_deliveries.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +import mimetypes +import os +from typing import List, Optional + +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import FileResponse +from pydantic import BaseModel, field_validator +from sqlalchemy.ext.asyncio import AsyncSession + +from ..database import get_db +from ..models import AuthUserORM +from ..services.result_delivery_service import result_delivery_service +from .dependencies import _add_operation_audit_log, _get_current_user + + +router = APIRouter() + + +class ResultDeliveryCreateRequest(BaseModel): + channel: str + product_ids: Optional[List[int]] = None + compat_result_ids: Optional[List[int]] = None + package_mode: str = "directory" + include_checksums: Optional[bool] = None + + @field_validator("channel", mode="before") + @classmethod + def _validate_channel(cls, value): + text = str(value or "").strip().lower() + if not text: + raise ValueError("channel is required") + return text + + @field_validator("package_mode", mode="before") + @classmethod + def _validate_package_mode(cls, value): + text = str(value or "directory").strip().lower() + if text not in {"directory", "zip"}: + raise ValueError("package_mode must be directory or zip") + return text + + +@router.get("/result-deliveries/channels") +async def get_result_delivery_channels( + current_user: AuthUserORM = Depends(_get_current_user), +): + _ = current_user + return {"items": result_delivery_service.channels()} + + +@router.post("/result-deliveries", status_code=202) +async def create_result_delivery( + request: ResultDeliveryCreateRequest, + http_request: Request, + db: AsyncSession = Depends(get_db), + current_user: AuthUserORM = Depends(_get_current_user), +): + try: + delivery = await result_delivery_service.create_delivery( + db, + user=current_user, + channel=request.channel, + product_ids=request.product_ids, + compat_result_ids=request.compat_result_ids, + package_mode=request.package_mode, + include_checksums=request.include_checksums, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + await _add_operation_audit_log( + db, + request=http_request, + action="result_delivery_created", + resource="result-deliveries", + detail={ + "delivery_id": delivery.delivery_id, + "channel": delivery.channel, + "package_mode": delivery.package_mode, + "item_count": delivery.item_count, + "task_id": delivery.task_id, + "job_id": delivery.job_id, + }, + user=current_user, + ) + await db.commit() + return result_delivery_service.serialize_delivery(delivery) + + +@router.get("/result-deliveries") +async def list_result_deliveries( + mine: bool = True, + include_all: bool = False, + limit: int = 50, + offset: int = 0, + db: AsyncSession = Depends(get_db), + current_user: AuthUserORM = Depends(_get_current_user), +): + allow_all = bool(include_all) and str(current_user.role or "").lower() == "admin" + if mine: + allow_all = False + return await result_delivery_service.list_deliveries( + db, + user=current_user, + include_all=allow_all, + include_items=True, + item_limit=5, + limit=limit, + offset=offset, + ) + + +@router.get("/result-deliveries/{delivery_id}") +async def get_result_delivery( + delivery_id: str, + db: AsyncSession = Depends(get_db), + current_user: AuthUserORM = Depends(_get_current_user), +): + delivery = await result_delivery_service.get_delivery( + db, + delivery_id=delivery_id, + user=current_user, + include_items=True, + ) + if delivery is None: + raise HTTPException(status_code=404, detail="delivery not found") + return result_delivery_service.serialize_delivery(delivery, include_items=True) + + +@router.get("/result-deliveries/{delivery_id}/manifest") +async def download_result_delivery_manifest( + delivery_id: str, + db: AsyncSession = Depends(get_db), + current_user: AuthUserORM = Depends(_get_current_user), +): + try: + path = await result_delivery_service.resolve_manifest_path( + db, + delivery_id=delivery_id, + user=current_user, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return FileResponse(path, filename=os.path.basename(path), media_type="application/json") + + +@router.get("/result-deliveries/{delivery_id}/archive/download") +async def download_result_delivery_archive( + delivery_id: str, + db: AsyncSession = Depends(get_db), + current_user: AuthUserORM = Depends(_get_current_user), +): + try: + path = await result_delivery_service.resolve_archive_path( + db, + delivery_id=delivery_id, + user=current_user, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return FileResponse(path, filename=os.path.basename(path), media_type="application/zip") + + +@router.get("/result-deliveries/{delivery_id}/files/{item_id}/download") +async def download_result_delivery_item( + delivery_id: str, + item_id: int, + db: AsyncSession = Depends(get_db), + current_user: AuthUserORM = Depends(_get_current_user), +): + try: + path = await result_delivery_service.resolve_item_path( + db, + delivery_id=delivery_id, + item_id=item_id, + user=current_user, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + media_type = mimetypes.guess_type(path)[0] or "application/octet-stream" + return FileResponse(path, filename=os.path.basename(path), media_type=media_type) diff --git a/backend/app/services/job_handlers.py b/backend/app/services/job_handlers.py index aacc902..f292641 100644 --- a/backend/app/services/job_handlers.py +++ b/backend/app/services/job_handlers.py @@ -44,6 +44,7 @@ from .engine_lock_service import engine_lock_service from .envi_service import build_envi_runner_command, extract_disp_results, get_envi_runner_cwd, get_envi_runner_env from .psinsar_catalog_service import psinsar_catalog_service from .result_catalog_service import result_catalog_service +from .result_delivery_service import JOB_TYPE_RESULT_DELIVERY_BUILD, result_delivery_service from .sbas_insar_catalog_service import sbas_insar_catalog_service from .task_service import task_service from .timeseries_service import ( @@ -5160,6 +5161,30 @@ async def _handle_rebuild_dinsar_catalog_clean(job: SystemJobORM) -> None: ) +async def _handle_result_delivery_build(job: SystemJobORM) -> None: + if not job.task_id: + raise ValueError("RESULT_DELIVERY_BUILD requires task_id for progress tracking.") + payload = job.payload or {} + delivery_id = str(payload.get("delivery_id") or "").strip() + if not delivery_id: + raise ValueError("RESULT_DELIVERY_BUILD requires delivery_id payload.") + + await task_service.start_task(job.task_id, message=f"正在生成成果交付包: {delivery_id}") + await task_service.add_log(job.task_id, "INFO", f"Result delivery build started: {delivery_id}") + result = await result_delivery_service.build_delivery(delivery_id) + summary = result.get("summary") or {} + await task_service.add_log( + job.task_id, + "INFO", + ( + f"Result delivery build summary: delivery_id={delivery_id}, " + f"status={result.get('status')}, files={summary.get('file_count', 0)}, " + f"copied={summary.get('copied_files', 0)}, skipped={summary.get('skipped_files', 0)}, " + f"failed={summary.get('failed_items', 0)}, dir={result.get('delivery_dir')}" + ), + ) + + async def _handle_timeseries_prepare(job: SystemJobORM) -> None: if not job.task_id: raise ValueError("TIMESERIES_PREPARE requires task_id for progress tracking.") @@ -6206,6 +6231,7 @@ _HANDLERS = { JOB_TYPE_EXTRACT_DINSAR_PRODUCTS: _handle_extract_dinsar_products, JOB_TYPE_PUBLISH_DINSAR_PRODUCTS: _handle_publish_dinsar_products_clean, JOB_TYPE_REBUILD_DINSAR_CATALOG: _handle_rebuild_dinsar_catalog_clean, + JOB_TYPE_RESULT_DELIVERY_BUILD: _handle_result_delivery_build, JOB_TYPE_PAIRING_CACHE_REBUILD: _handle_pairing_cache_rebuild, JOB_TYPE_TIMESERIES_PREPARE: _handle_timeseries_prepare, JOB_TYPE_TIMESERIES_STACK_PREP: _handle_timeseries_stack_prep, diff --git a/backend/app/services/result_delivery_service.py b/backend/app/services/result_delivery_service.py new file mode 100644 index 0000000..bbbdd95 --- /dev/null +++ b/backend/app/services/result_delivery_service.py @@ -0,0 +1,907 @@ +from __future__ import annotations + +import hashlib +import json +import os +import re +import shutil +import zipfile +from dataclasses import dataclass +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + +from sqlalchemy import func, or_, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from .. import database +from ..config import settings +from ..models import ( + AuthUserORM, + DinsarResultORM, + ResultAssetORM, + ResultDeliveryItemORM, + ResultDeliveryRequestORM, + ResultProductORM, +) +from .dinsar_read_service import dinsar_read_service +from .job_queue_service import job_queue_service +from .task_service import task_service + + +JOB_TYPE_RESULT_DELIVERY_BUILD = "RESULT_DELIVERY_BUILD" +TASK_TYPE_RESULT_DELIVERY_BUILD = "RESULT_DELIVERY_BUILD" + +DELIVERY_STATUS_PENDING = "PENDING" +DELIVERY_STATUS_RUNNING = "RUNNING" +DELIVERY_STATUS_READY = "READY" +DELIVERY_STATUS_FAILED = "FAILED" +DELIVERY_STATUS_CANCELLED = "CANCELLED" +DELIVERY_STATUS_EXPIRED = "EXPIRED" + +ITEM_STATUS_PENDING = "PENDING" +ITEM_STATUS_COPIED = "COPIED" +ITEM_STATUS_FAILED = "FAILED" +ITEM_STATUS_SKIPPED = "SKIPPED" + +CHANNEL_DINSAR = "dinsar" +SUPPORTED_READY_CHANNELS = {CHANNEL_DINSAR} + +PACKAGE_MODE_DIRECTORY = "directory" +PACKAGE_MODE_ZIP = "zip" + +_ASSOCIATED_EXTENSIONS = ("", ".hdr", ".sml", ".xml", ".aux.xml", ".prj") +_INVALID_SEGMENT_RE = re.compile(r'[<>:"/\\|?*\x00-\x1F]+') +_SAFE_ID_RE = re.compile(r"[^0-9A-Za-z_-]+") + + +@dataclass(frozen=True) +class DeliverySource: + product: ResultProductORM + compat_row: Optional[DinsarResultORM] + display_name: str + source_path: str + source_asset_id: Optional[int] = None + + +def _new_session() -> AsyncSession: + if database.AsyncSessionLocal is None: + database.init_db() + if database.AsyncSessionLocal is None: + raise RuntimeError("Database session factory is not initialized.") + return database.AsyncSessionLocal() + + +def _utcnow() -> datetime: + return datetime.utcnow() + + +def _normalize_channel(value: str) -> str: + channel = str(value or "").strip().lower() + if not channel: + raise ValueError("channel is required") + return channel + + +def _normalize_package_mode(value: str) -> str: + mode = str(value or PACKAGE_MODE_DIRECTORY).strip().lower() + if mode not in {PACKAGE_MODE_DIRECTORY, PACKAGE_MODE_ZIP}: + raise ValueError("package_mode must be directory or zip") + return mode + + +def _normalize_int_ids(values: Optional[Iterable[Any]], *, max_count: int) -> List[int]: + normalized: List[int] = [] + seen = set() + for raw in values or []: + try: + value = int(raw) + except (TypeError, ValueError): + continue + if value <= 0 or value in seen: + continue + seen.add(value) + normalized.append(value) + if len(normalized) > max_count: + raise ValueError(f"selected item count exceeds max limit ({max_count})") + return normalized + + +def _sanitize_segment(value: Any, *, default: str = "item", max_len: int = 120) -> str: + text = str(value or "").strip() + text = _INVALID_SEGMENT_RE.sub("_", text).strip(" .") + text = _SAFE_ID_RE.sub("_", text) + return (text or default)[:max_len] + + +def _same_file_size(left: str, right: str) -> bool: + try: + return os.path.getsize(left) == os.path.getsize(right) + except OSError: + return False + + +def _file_size(path: str) -> int: + try: + return int(os.path.getsize(path)) + except OSError: + return 0 + + +def _sha256_file(path: str) -> str: + digest = hashlib.sha256() + with open(path, "rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _write_json(path: str, payload: Dict[str, Any]) -> None: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as stream: + json.dump(payload, stream, ensure_ascii=False, indent=2, default=str) + + +def _path_within(parent: str, child: str) -> bool: + parent_path = Path(parent).resolve() + child_path = Path(child).resolve() + try: + child_path.relative_to(parent_path) + return True + except ValueError: + return False + + +def _iter_associated_files(source_path: str) -> List[str]: + source = os.path.normpath(source_path) + if os.path.isdir(source): + files: List[str] = [] + for root, _dirs, names in os.walk(source): + for name in names: + files.append(os.path.join(root, name)) + return sorted(files) + if not os.path.isfile(source): + return [] + + src_dir = os.path.dirname(source) + base_name = os.path.basename(source) + files: List[str] = [] + for ext in _ASSOCIATED_EXTENSIONS: + candidate = os.path.join(src_dir, base_name + ext) if ext else source + if os.path.isfile(candidate) and candidate not in files: + files.append(candidate) + return files + + +def _delivery_root() -> str: + root = str(settings.RESULT_DELIVERY_ROOT or "").strip() + if not root: + raise ValueError("RESULT_DELIVERY_ROOT is not configured") + return os.path.normpath(os.path.abspath(root)) + + +def _owner_dir(root: str, username: str) -> str: + return os.path.join(root, _sanitize_segment(username, default="user", max_len=64)) + + +def _make_delivery_id(username: str) -> str: + stamp = _utcnow().strftime("%Y%m%d%H%M%S") + digest = hashlib.sha1(f"{username}|{stamp}|{os.urandom(8).hex()}".encode("utf-8")).hexdigest()[:10] + return f"rd_{stamp}_{digest}" + + +def _resolve_source_path(product: ResultProductORM, assets: List[ResultAssetORM]) -> tuple[str, Optional[int]]: + primary_assets = [ + asset for asset in assets + if asset.exists_flag and (asset.is_primary or str(asset.asset_role or "").lower() in {"disp", "primary_geotiff"}) + ] + primary_assets.sort(key=lambda item: (not item.is_primary, item.id)) + for asset in primary_assets: + path = str(asset.absolute_path or "").strip() + if path: + return path, asset.id + for path in (product.primary_asset_path, product.source_primary_path, product.publish_dir): + text = str(path or "").strip() + if text: + return text, None + return "", None + + +class ResultDeliveryService: + def channels(self) -> List[Dict[str, Any]]: + return [ + { + "key": CHANNEL_DINSAR, + "group": "InSAR 成果", + "label": "D-InSAR 结果", + "state": "ready", + "state_text": "可交付", + "description": "已登记 D-InSAR catalog，可创建后台交付包并下载到本地。", + }, + { + "key": "sbas", + "group": "InSAR 成果", + "label": "SBAS-InSAR 结果", + "state": "planned", + "state_text": "目录可查", + "description": "SBAS 结果 catalog 已有基础能力，本阶段暂不开放交付打包。", + }, + { + "key": "lt1_ortho", + "group": "正射成果", + "label": "LT-1 正射结果", + "state": "placeholder", + "state_text": "待接入", + "description": "陆探一正射生产将由 LandSAR 生产链注册后接入统一交付。", + }, + { + "key": "s1_ortho", + "group": "正射成果", + "label": "Sentinel-1 正射结果", + "state": "placeholder", + "state_text": "待接入", + "description": "Sentinel-1 正射生产尚未接入，当前只保留交付通道占位。", + }, + { + "key": "gf3_ortho", + "group": "正射成果", + "label": "GF3 SARscape _geo", + "state": "placeholder", + "state_text": "待接入", + "description": "GF3 外部生产成果登记后再接入统一交付。", + }, + ] + + async def _resolve_dinsar_sources( + self, + db: AsyncSession, + *, + product_ids: List[int], + compat_result_ids: List[int], + ) -> List[DeliverySource]: + sources: List[DeliverySource] = [] + seen_products = set() + + if product_ids: + result = await db.execute( + select(ResultProductORM) + .options(selectinload(ResultProductORM.assets)) + .where( + ResultProductORM.id.in_(product_ids), + ResultProductORM.catalog_name == CHANNEL_DINSAR, + ResultProductORM.status.in_(["READY", "PUBLISHED", "OK"]), + ) + .order_by(ResultProductORM.id.asc()) + ) + for product in result.scalars().unique().all(): + assets = list(product.assets or []) + source_path, source_asset_id = _resolve_source_path(product, assets) + if not source_path: + continue + seen_products.add(int(product.id)) + sources.append( + DeliverySource( + product=product, + compat_row=None, + display_name=dinsar_read_service.get_display_name(product), + source_path=source_path, + source_asset_id=source_asset_id, + ) + ) + + if compat_result_ids: + records = await dinsar_read_service.list_compat_records_by_ids( + db, + compat_result_ids=compat_result_ids, + ) + product_lookup_ids = [record.product.id for record in records if record.product is not None] + assets_by_product: Dict[int, List[ResultAssetORM]] = {} + if product_lookup_ids: + asset_result = await db.execute( + select(ResultAssetORM) + .where(ResultAssetORM.product_ref_id.in_(product_lookup_ids)) + .order_by(ResultAssetORM.id.asc()) + ) + for asset in asset_result.scalars().all(): + assets_by_product.setdefault(int(asset.product_ref_id), []).append(asset) + + for record in records: + product = record.product + if int(product.id) in seen_products: + continue + assets = assets_by_product.get(int(product.id), []) + source_path, source_asset_id = _resolve_source_path(product, assets) + if not source_path and record.compat_row is not None: + source_path = str(record.compat_row.file_path or "").strip() + if not source_path: + continue + seen_products.add(int(product.id)) + sources.append( + DeliverySource( + product=product, + compat_row=record.compat_row, + display_name=record.display_name, + source_path=source_path, + source_asset_id=source_asset_id, + ) + ) + + return sources + + async def create_delivery( + self, + db: AsyncSession, + *, + user: AuthUserORM, + channel: str, + product_ids: Optional[List[int]] = None, + compat_result_ids: Optional[List[int]] = None, + package_mode: str = PACKAGE_MODE_DIRECTORY, + include_checksums: Optional[bool] = None, + ) -> ResultDeliveryRequestORM: + channel = _normalize_channel(channel) + if channel not in SUPPORTED_READY_CHANNELS: + raise ValueError(f"{channel} delivery is not connected yet") + + package_mode = _normalize_package_mode(package_mode) + max_items = max(1, int(settings.RESULT_DELIVERY_MAX_ITEMS or 500)) + normalized_product_ids = _normalize_int_ids(product_ids, max_count=max_items) + normalized_compat_ids = _normalize_int_ids(compat_result_ids, max_count=max_items) + if not normalized_product_ids and not normalized_compat_ids: + raise ValueError("select at least one result") + + if len(normalized_product_ids) + len(normalized_compat_ids) > max_items: + raise ValueError(f"selected item count exceeds max limit ({max_items})") + + if channel == CHANNEL_DINSAR: + sources = await self._resolve_dinsar_sources( + db, + product_ids=normalized_product_ids, + compat_result_ids=normalized_compat_ids, + ) + else: + sources = [] + + if not sources: + raise ValueError("selected results do not have deliverable files") + if len(sources) > max_items: + raise ValueError(f"resolved result count exceeds max limit ({max_items})") + + delivery_root = _delivery_root() + delivery_id = _make_delivery_id(str(user.username or "user")) + delivery_dir = os.path.join(_owner_dir(delivery_root, str(user.username or "user")), delivery_id) + expires_at = _utcnow() + timedelta(days=max(1, int(settings.RESULT_DELIVERY_RETENTION_DAYS or 7))) + request_json = { + "channel": channel, + "product_ids": normalized_product_ids, + "compat_result_ids": normalized_compat_ids, + "package_mode": package_mode, + "include_checksums": ( + bool(settings.RESULT_DELIVERY_CHECKSUM_ENABLED) + if include_checksums is None + else bool(include_checksums) + ), + } + delivery = ResultDeliveryRequestORM( + delivery_id=delivery_id, + owner_user_id=user.id, + owner_username=str(user.username or "unknown"), + channel=channel, + status=DELIVERY_STATUS_PENDING, + package_mode=package_mode, + item_count=len(sources), + total_bytes=0, + copied_bytes=0, + delivery_root=delivery_root, + delivery_dir=delivery_dir, + zip_path=(f"{delivery_dir}.zip" if package_mode == PACKAGE_MODE_ZIP else None), + expires_at=expires_at, + request_json=request_json, + summary_json={ + "source_count": len(sources), + "created_message": "Delivery package queued.", + }, + ) + db.add(delivery) + await db.flush() + + task_type = f"{TASK_TYPE_RESULT_DELIVERY_BUILD}_{delivery_id}" + task_id = await task_service.create_task( + task_type, + f"成果交付包生成: {delivery_id}", + params={ + "delivery_id": delivery_id, + "channel": channel, + "item_count": len(sources), + "package_mode": package_mode, + }, + db=db, + ) + job_id = await job_queue_service.create_job( + JOB_TYPE_RESULT_DELIVERY_BUILD, + payload={"delivery_id": delivery_id}, + task_id=task_id, + max_attempts=1, + db=db, + ) + delivery.task_id = task_id + delivery.job_id = job_id + await db.flush() + return delivery + + async def list_deliveries( + self, + db: AsyncSession, + *, + user: AuthUserORM, + include_all: bool = False, + include_items: bool = False, + item_limit: Optional[int] = None, + limit: int = 50, + offset: int = 0, + ) -> Dict[str, Any]: + safe_limit = min(200, max(1, int(limit or 50))) + safe_offset = max(0, int(offset or 0)) + stmt = select(ResultDeliveryRequestORM) + if include_items: + stmt = stmt.options(selectinload(ResultDeliveryRequestORM.items)) + count_stmt = select(func.count(ResultDeliveryRequestORM.id)) + if not include_all: + owner_filter = ResultDeliveryRequestORM.owner_user_id == user.id + stmt = stmt.where(owner_filter) + count_stmt = count_stmt.where(owner_filter) + stmt = ( + stmt.order_by(ResultDeliveryRequestORM.created_at.desc(), ResultDeliveryRequestORM.id.desc()) + .offset(safe_offset) + .limit(safe_limit) + ) + result = await db.execute(stmt) + total_result = await db.execute(count_stmt) + deliveries = result.scalars().unique().all() if include_items else result.scalars().all() + items = [ + self.serialize_delivery(item, include_items=include_items, item_limit=item_limit) + for item in deliveries + ] + total = int(total_result.scalar_one() or 0) + return { + "items": items, + "total": total, + "limit": safe_limit, + "offset": safe_offset, + "has_more": safe_offset + len(items) < total, + } + + async def get_delivery( + self, + db: AsyncSession, + *, + delivery_id: str, + user: AuthUserORM, + include_items: bool = True, + ) -> Optional[ResultDeliveryRequestORM]: + stmt = select(ResultDeliveryRequestORM).where( + ResultDeliveryRequestORM.delivery_id == str(delivery_id or "").strip() + ) + if include_items: + stmt = stmt.options(selectinload(ResultDeliveryRequestORM.items)) + if str(user.role or "").lower() != "admin": + stmt = stmt.where(ResultDeliveryRequestORM.owner_user_id == user.id) + result = await db.execute(stmt) + return result.scalars().unique().one_or_none() + + def serialize_delivery( + self, + delivery: ResultDeliveryRequestORM, + *, + include_items: bool = False, + item_limit: Optional[int] = None, + ) -> Dict[str, Any]: + items = list(delivery.items or []) if include_items else [] + if item_limit is not None: + items = items[: max(0, int(item_limit))] + summary = delivery.summary_json if isinstance(delivery.summary_json, dict) else {} + payload = { + "id": delivery.id, + "delivery_id": delivery.delivery_id, + "owner_user_id": delivery.owner_user_id, + "owner_username": delivery.owner_username, + "channel": delivery.channel, + "status": delivery.status, + "package_mode": delivery.package_mode, + "item_count": delivery.item_count, + "total_bytes": delivery.total_bytes, + "copied_bytes": delivery.copied_bytes, + "delivery_dir": delivery.delivery_dir, + "zip_path": delivery.zip_path, + "manifest_path": delivery.manifest_path, + "expires_at": delivery.expires_at, + "task_id": delivery.task_id, + "job_id": delivery.job_id, + "error_message": delivery.error_message, + "summary": summary, + "created_at": delivery.created_at, + "updated_at": delivery.updated_at, + "started_at": delivery.started_at, + "completed_at": delivery.completed_at, + "download_urls": { + "manifest": f"/api/result-deliveries/{delivery.delivery_id}/manifest", + "archive": ( + f"/api/result-deliveries/{delivery.delivery_id}/archive/download" + if delivery.zip_path + else None + ), + }, + } + if include_items: + payload["items"] = [self.serialize_item(item, delivery_id=delivery.delivery_id) for item in items] + return payload + + def serialize_item(self, item: ResultDeliveryItemORM, *, delivery_id: str) -> Dict[str, Any]: + return { + "id": item.id, + "delivery_id": item.delivery_id, + "source_product_id": item.source_product_id, + "source_result_id": item.source_result_id, + "source_asset_id": item.source_asset_id, + "display_name": item.display_name, + "relative_path": item.relative_path, + "file_size": item.file_size, + "checksum_sha256": item.checksum_sha256, + "status": item.status, + "error_message": item.error_message, + "download_url": ( + f"/api/result-deliveries/{delivery_id}/files/{item.id}/download" + if item.status == ITEM_STATUS_COPIED + else None + ), + } + + async def build_delivery(self, delivery_id: str) -> Dict[str, Any]: + async with _new_session() as db: + result = await db.execute( + select(ResultDeliveryRequestORM).where(ResultDeliveryRequestORM.delivery_id == delivery_id) + ) + delivery = result.scalar_one_or_none() + if delivery is None: + raise ValueError(f"delivery not found: {delivery_id}") + if delivery.status in {DELIVERY_STATUS_READY, DELIVERY_STATUS_CANCELLED, DELIVERY_STATUS_EXPIRED}: + return self.serialize_delivery(delivery) + + delivery.status = DELIVERY_STATUS_RUNNING + delivery.started_at = _utcnow() + delivery.error_message = None + await db.commit() + + try: + summary = await self._build_delivery_files(delivery_id) + except Exception as exc: + async with _new_session() as db: + result = await db.execute( + select(ResultDeliveryRequestORM).where(ResultDeliveryRequestORM.delivery_id == delivery_id) + ) + delivery = result.scalar_one_or_none() + if delivery is not None: + delivery.status = DELIVERY_STATUS_FAILED + delivery.error_message = str(exc) + delivery.completed_at = _utcnow() + delivery.summary_json = { + **(delivery.summary_json if isinstance(delivery.summary_json, dict) else {}), + "error": str(exc), + } + await db.commit() + if delivery.task_id: + await task_service.update_task(delivery.task_id, status="FAILED", progress=100, message=str(exc)) + raise + + async with _new_session() as db: + result = await db.execute( + select(ResultDeliveryRequestORM) + .options(selectinload(ResultDeliveryRequestORM.items)) + .where(ResultDeliveryRequestORM.delivery_id == delivery_id) + ) + delivery = result.scalars().unique().one_or_none() + if delivery is None: + raise ValueError(f"delivery not found: {delivery_id}") + available_files = int(summary.get("copied_files", 0) or 0) + int(summary.get("skipped_files", 0) or 0) + delivery.status = DELIVERY_STATUS_READY if available_files > 0 else DELIVERY_STATUS_FAILED + delivery.completed_at = _utcnow() + delivery.error_message = summary.get("error_message") + delivery.summary_json = summary + await db.commit() + if delivery.task_id: + if delivery.status == DELIVERY_STATUS_READY: + await task_service.update_task( + delivery.task_id, + status="COMPLETED", + progress=100, + message=( + f"成果交付包已生成: files={summary.get('copied_files', 0)}, " + f"bytes={summary.get('copied_bytes', 0)}" + ), + ) + else: + await task_service.update_task( + delivery.task_id, + status="FAILED", + progress=100, + message=summary.get("error_message") or "成果交付包生成失败", + ) + return self.serialize_delivery(delivery, include_items=True) + + async def _build_delivery_files(self, delivery_id: str) -> Dict[str, Any]: + async with _new_session() as db: + result = await db.execute( + select(ResultDeliveryRequestORM).where(ResultDeliveryRequestORM.delivery_id == delivery_id) + ) + delivery = result.scalar_one() + request_json = delivery.request_json if isinstance(delivery.request_json, dict) else {} + if delivery.channel == CHANNEL_DINSAR: + sources = await self._resolve_dinsar_sources( + db, + product_ids=_normalize_int_ids(request_json.get("product_ids"), max_count=int(settings.RESULT_DELIVERY_MAX_ITEMS or 500)), + compat_result_ids=_normalize_int_ids(request_json.get("compat_result_ids"), max_count=int(settings.RESULT_DELIVERY_MAX_ITEMS or 500)), + ) + else: + sources = [] + task_id = delivery.task_id + + if not sources: + raise ValueError("delivery has no deliverable sources") + + include_checksums = bool( + request_json.get("include_checksums") + if "include_checksums" in request_json + else settings.RESULT_DELIVERY_CHECKSUM_ENABLED + ) + delivery_dir = os.path.normpath(os.path.abspath(delivery.delivery_dir)) + root = os.path.normpath(os.path.abspath(delivery.delivery_root)) + if not _path_within(root, delivery_dir): + raise ValueError("delivery directory is outside RESULT_DELIVERY_ROOT") + + os.makedirs(delivery_dir, exist_ok=True) + copied_files = 0 + skipped_files = 0 + failed_items = 0 + copied_bytes = 0 + total_bytes = 0 + checksum_lines: List[str] = [] + item_payloads: List[Dict[str, Any]] = [] + manifest_items: List[Dict[str, Any]] = [] + + total_sources = len(sources) + for source_index, source in enumerate(sources, start=1): + folder = _sanitize_segment( + source.product.task_alias + or source.product.task_name + or source.display_name + or source.product.product_id, + default=f"product_{source_index}", + ) + source_files = _iter_associated_files(source.source_path) + if not source_files: + failed_items += 1 + item_payloads.append( + { + "delivery_id": delivery_id, + "source_product_id": source.product.id, + "source_result_id": source.compat_row.id if source.compat_row else None, + "source_asset_id": source.source_asset_id, + "display_name": source.display_name, + "source_path": source.source_path, + "relative_path": None, + "file_size": 0, + "checksum_sha256": None, + "status": ITEM_STATUS_FAILED, + "error_message": "source file not found", + } + ) + continue + + for file_index, source_file in enumerate(source_files, start=1): + if os.path.isdir(source.source_path): + relative_under_source = os.path.relpath(source_file, source.source_path) + relative_path = os.path.join(folder, relative_under_source) + else: + relative_path = os.path.join(folder, os.path.basename(source_file)) + target_path = os.path.join(delivery_dir, relative_path) + size = _file_size(source_file) + total_bytes += size + status = ITEM_STATUS_COPIED + error_message = None + checksum = None + try: + os.makedirs(os.path.dirname(target_path), exist_ok=True) + if os.path.isfile(target_path) and _same_file_size(source_file, target_path): + skipped_files += 1 + else: + shutil.copy2(source_file, target_path) + copied_files += 1 + copied_bytes += size + if include_checksums and os.path.isfile(target_path): + checksum = _sha256_file(target_path) + checksum_lines.append(f"{checksum} {relative_path.replace(os.sep, '/')}") + except OSError as exc: + status = ITEM_STATUS_FAILED + error_message = str(exc) + failed_items += 1 + + if task_id and (copied_files + skipped_files + failed_items) % 10 == 0: + progress = min(95, int(((source_index - 1) / max(1, total_sources)) * 100)) + await task_service.update_task( + task_id, + progress=progress, + message=f"成果交付包生成中: {source_index}/{total_sources}", + ) + + item_payload = { + "delivery_id": delivery_id, + "source_product_id": source.product.id, + "source_result_id": source.compat_row.id if source.compat_row else None, + "source_asset_id": source.source_asset_id if file_index == 1 else None, + "display_name": source.display_name, + "source_path": source_file, + "relative_path": relative_path, + "file_size": size, + "checksum_sha256": checksum, + "status": status, + "error_message": error_message, + } + item_payloads.append(item_payload) + manifest_items.append( + { + "display_name": source.display_name, + "product_id": source.product.product_id, + "product_ref_id": source.product.id, + "source_result_id": source.compat_row.id if source.compat_row else None, + "relative_path": relative_path.replace(os.sep, "/"), + "file_size": size, + "checksum_sha256": checksum, + "status": status, + "error_message": error_message, + } + ) + + checksums_path = os.path.join(delivery_dir, "checksums.sha256") + if include_checksums: + with open(checksums_path, "w", encoding="utf-8") as stream: + stream.write("\n".join(checksum_lines)) + if checksum_lines: + stream.write("\n") + + manifest_path = os.path.join(delivery_dir, "manifest.json") + summary = { + "delivery_id": delivery_id, + "channel": delivery.channel, + "package_mode": delivery.package_mode, + "source_count": total_sources, + "file_count": len(item_payloads), + "copied_files": copied_files, + "skipped_files": skipped_files, + "failed_items": failed_items, + "total_bytes": total_bytes, + "copied_bytes": copied_bytes, + "include_checksums": include_checksums, + "manifest_path": manifest_path, + "checksums_path": checksums_path if include_checksums else None, + "created_at": _utcnow().isoformat(timespec="seconds"), + } + if failed_items: + summary["error_message"] = f"{failed_items} files failed during delivery build" + _write_json( + manifest_path, + { + "schema_version": "insar.result-delivery/v1", + "summary": summary, + "items": manifest_items, + }, + ) + + zip_path = None + if delivery.package_mode == PACKAGE_MODE_ZIP: + if total_bytes > int(settings.RESULT_DELIVERY_ZIP_MAX_BYTES or 0): + raise ValueError( + f"delivery size exceeds zip limit: {total_bytes} > {settings.RESULT_DELIVERY_ZIP_MAX_BYTES}" + ) + zip_path = str(delivery.zip_path or f"{delivery_dir}.zip") + with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True) as archive: + for root_dir, _dirs, files in os.walk(delivery_dir): + for name in files: + path = os.path.join(root_dir, name) + archive.write(path, os.path.relpath(path, delivery_dir)) + + async with _new_session() as db: + result = await db.execute( + select(ResultDeliveryRequestORM).where(ResultDeliveryRequestORM.delivery_id == delivery_id) + ) + delivery_row = result.scalar_one() + await db.execute( + ResultDeliveryItemORM.__table__.delete().where( + ResultDeliveryItemORM.delivery_id == delivery_id + ) + ) + for payload in item_payloads: + db.add(ResultDeliveryItemORM(**payload)) + delivery_row.total_bytes = total_bytes + delivery_row.copied_bytes = copied_bytes + delivery_row.item_count = len(item_payloads) + delivery_row.manifest_path = manifest_path + delivery_row.zip_path = zip_path or delivery_row.zip_path + await db.commit() + + return summary + + async def resolve_manifest_path( + self, + db: AsyncSession, + *, + delivery_id: str, + user: AuthUserORM, + ) -> str: + delivery = await self.get_delivery(db, delivery_id=delivery_id, user=user, include_items=False) + if delivery is None: + raise ValueError("delivery not found") + if delivery.status != DELIVERY_STATUS_READY: + raise ValueError("delivery is not ready") + path = str(delivery.manifest_path or "").strip() + if not path or not os.path.isfile(path): + raise ValueError("manifest file is missing") + if not _path_within(delivery.delivery_root, path): + raise ValueError("manifest path is outside delivery root") + return path + + async def resolve_archive_path( + self, + db: AsyncSession, + *, + delivery_id: str, + user: AuthUserORM, + ) -> str: + delivery = await self.get_delivery(db, delivery_id=delivery_id, user=user, include_items=False) + if delivery is None: + raise ValueError("delivery not found") + if delivery.status != DELIVERY_STATUS_READY: + raise ValueError("delivery is not ready") + path = str(delivery.zip_path or "").strip() + if not path or not os.path.isfile(path): + raise ValueError("zip archive is not available") + if not _path_within(delivery.delivery_root, path): + raise ValueError("zip path is outside delivery root") + return path + + async def resolve_item_path( + self, + db: AsyncSession, + *, + delivery_id: str, + item_id: int, + user: AuthUserORM, + ) -> str: + delivery = await self.get_delivery(db, delivery_id=delivery_id, user=user, include_items=False) + if delivery is None: + raise ValueError("delivery not found") + if delivery.status != DELIVERY_STATUS_READY: + raise ValueError("delivery is not ready") + result = await db.execute( + select(ResultDeliveryItemORM).where( + ResultDeliveryItemORM.id == int(item_id), + ResultDeliveryItemORM.delivery_id == delivery_id, + ) + ) + item = result.scalar_one_or_none() + if item is None: + raise ValueError("delivery item not found") + if item.status != ITEM_STATUS_COPIED: + raise ValueError("delivery item is not available") + relative_path = str(item.relative_path or "").strip() + if not relative_path: + raise ValueError("delivery item path is missing") + path = os.path.normpath(os.path.abspath(os.path.join(delivery.delivery_dir, relative_path))) + if not os.path.isfile(path): + raise ValueError("delivery item file is missing") + if not _path_within(delivery.delivery_dir, path): + raise ValueError("delivery item path is outside delivery directory") + return path + + +result_delivery_service = ResultDeliveryService() diff --git a/backend/migrations/013_result_delivery_requests.sql b/backend/migrations/013_result_delivery_requests.sql new file mode 100644 index 0000000..d49338f --- /dev/null +++ b/backend/migrations/013_result_delivery_requests.sql @@ -0,0 +1,94 @@ +CREATE TABLE IF NOT EXISTS result_delivery_requests ( + id SERIAL PRIMARY KEY, + delivery_id VARCHAR(64) NOT NULL UNIQUE, + owner_user_id INTEGER NULL REFERENCES auth_users(id) ON DELETE SET NULL, + owner_username VARCHAR(64) NOT NULL, + channel VARCHAR(32) NOT NULL, + status VARCHAR(32) NOT NULL DEFAULT 'PENDING', + package_mode VARCHAR(16) NOT NULL DEFAULT 'directory', + item_count INTEGER NOT NULL DEFAULT 0, + total_bytes BIGINT NOT NULL DEFAULT 0, + copied_bytes BIGINT NOT NULL DEFAULT 0, + delivery_root TEXT NOT NULL, + delivery_dir TEXT NOT NULL, + zip_path TEXT NULL, + manifest_path TEXT NULL, + expires_at TIMESTAMP NULL, + task_id VARCHAR(128) NULL, + job_id VARCHAR(128) NULL, + error_message TEXT NULL, + request_json JSON NULL, + summary_json JSON NULL, + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP NOT NULL DEFAULT NOW(), + started_at TIMESTAMP NULL, + completed_at TIMESTAMP NULL +); + +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS owner_user_id INTEGER NULL REFERENCES auth_users(id) ON DELETE SET NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS owner_username VARCHAR(64) NOT NULL DEFAULT 'unknown'; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS channel VARCHAR(32) NOT NULL DEFAULT 'dinsar'; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS status VARCHAR(32) NOT NULL DEFAULT 'PENDING'; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS package_mode VARCHAR(16) NOT NULL DEFAULT 'directory'; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS item_count INTEGER NOT NULL DEFAULT 0; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS total_bytes BIGINT NOT NULL DEFAULT 0; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS copied_bytes BIGINT NOT NULL DEFAULT 0; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS delivery_root TEXT NOT NULL DEFAULT ''; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS delivery_dir TEXT NOT NULL DEFAULT ''; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS zip_path TEXT NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS manifest_path TEXT NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS task_id VARCHAR(128) NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS job_id VARCHAR(128) NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS error_message TEXT NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS request_json JSON NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS summary_json JSON NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS created_at TIMESTAMP NOT NULL DEFAULT NOW(); +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS updated_at TIMESTAMP NOT NULL DEFAULT NOW(); +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS started_at TIMESTAMP NULL; +ALTER TABLE result_delivery_requests ADD COLUMN IF NOT EXISTS completed_at TIMESTAMP NULL; + +CREATE TABLE IF NOT EXISTS result_delivery_items ( + id SERIAL PRIMARY KEY, + delivery_id VARCHAR(64) NOT NULL REFERENCES result_delivery_requests(delivery_id) ON DELETE CASCADE, + source_product_id INTEGER NULL REFERENCES result_products(id) ON DELETE SET NULL, + source_result_id INTEGER NULL REFERENCES dinsar_results(id) ON DELETE SET NULL, + source_asset_id INTEGER NULL REFERENCES result_assets(id) ON DELETE SET NULL, + display_name VARCHAR(255) NOT NULL, + source_path TEXT NOT NULL, + relative_path TEXT NULL, + file_size BIGINT NOT NULL DEFAULT 0, + checksum_sha256 VARCHAR(64) NULL, + status VARCHAR(32) NOT NULL DEFAULT 'PENDING', + error_message TEXT NULL, + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); + +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS source_product_id INTEGER NULL REFERENCES result_products(id) ON DELETE SET NULL; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS source_result_id INTEGER NULL REFERENCES dinsar_results(id) ON DELETE SET NULL; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS source_asset_id INTEGER NULL REFERENCES result_assets(id) ON DELETE SET NULL; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS display_name VARCHAR(255) NOT NULL DEFAULT 'result'; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS source_path TEXT NOT NULL DEFAULT ''; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS relative_path TEXT NULL; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS file_size BIGINT NOT NULL DEFAULT 0; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS checksum_sha256 VARCHAR(64) NULL; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS status VARCHAR(32) NOT NULL DEFAULT 'PENDING'; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS error_message TEXT NULL; +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS created_at TIMESTAMP NOT NULL DEFAULT NOW(); +ALTER TABLE result_delivery_items ADD COLUMN IF NOT EXISTS updated_at TIMESTAMP NOT NULL DEFAULT NOW(); + +CREATE INDEX IF NOT EXISTS idx_result_delivery_owner_status ON result_delivery_requests(owner_user_id, status); +CREATE INDEX IF NOT EXISTS idx_result_delivery_channel_status ON result_delivery_requests(channel, status); +CREATE INDEX IF NOT EXISTS idx_result_delivery_created ON result_delivery_requests(created_at); +CREATE INDEX IF NOT EXISTS ix_result_delivery_requests_delivery_id ON result_delivery_requests(delivery_id); +CREATE INDEX IF NOT EXISTS ix_result_delivery_requests_owner_user_id ON result_delivery_requests(owner_user_id); +CREATE INDEX IF NOT EXISTS ix_result_delivery_requests_expires_at ON result_delivery_requests(expires_at); +CREATE INDEX IF NOT EXISTS ix_result_delivery_requests_task_id ON result_delivery_requests(task_id); +CREATE INDEX IF NOT EXISTS ix_result_delivery_requests_job_id ON result_delivery_requests(job_id); + +CREATE INDEX IF NOT EXISTS idx_result_delivery_items_delivery_status ON result_delivery_items(delivery_id, status); +CREATE INDEX IF NOT EXISTS idx_result_delivery_items_product ON result_delivery_items(source_product_id); +CREATE INDEX IF NOT EXISTS ix_result_delivery_items_delivery_id ON result_delivery_items(delivery_id); +CREATE INDEX IF NOT EXISTS ix_result_delivery_items_source_result_id ON result_delivery_items(source_result_id); +CREATE INDEX IF NOT EXISTS ix_result_delivery_items_source_asset_id ON result_delivery_items(source_asset_id); diff --git a/docs/INDEX.md b/docs/INDEX.md index e214c05..1a6eec3 100644 --- a/docs/INDEX.md +++ b/docs/INDEX.md @@ -33,6 +33,8 @@ 缁熶竴缁撴灉鐩綍銆佹爣鍑嗕骇鍝佸寘銆乧atalog 涓庡寮曟搸缁撴灉鍏卞瓨绾﹀畾銆? - [RESULT_EXTRACTION_ACCESS_CONTROL_AUDIT_20260630.md](RESULT_EXTRACTION_ACCESS_CONTROL_AUDIT_20260630.md) Result extraction and access-control audit: current D-InSAR export/registration boundaries, placeholder channels, admin/viewer limitations, and recommended exporter/operator/admin permission model. +- [RESULT_DELIVERY_DOWNLOAD_DESIGN_20260630.md](RESULT_DELIVERY_DOWNLOAD_DESIGN_20260630.md) + Result delivery/download design: asynchronous delivery packages, user-owned downloads, temporary delivery root, DB maintenance, and placeholder boundaries for SBAS/LT-1/Sentinel-1/GF3 ortho. - [DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md](DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md) D-InSAR 淇濈暀 ENVI/SARscape銆丩andSAR銆丟amma/PyINT 涓夊紩鎿庯紝閫€鍑?ISCE2锛岀粺涓€ Task_Pool銆佺粨鏋滆仛鍚堝拰涓棿鏂囦欢娓呯悊鐨勫綋鍓嶈璁°€? - [LANDSAR_DEM_PREPARATION_CONTRACT_20260618.md](LANDSAR_DEM_PREPARATION_CONTRACT_20260618.md) diff --git a/docs/RESULT_DELIVERY_DOWNLOAD_DESIGN_20260630.md b/docs/RESULT_DELIVERY_DOWNLOAD_DESIGN_20260630.md new file mode 100644 index 0000000..5ac5bd3 --- /dev/null +++ b/docs/RESULT_DELIVERY_DOWNLOAD_DESIGN_20260630.md @@ -0,0 +1,309 @@ +# 成果交付与本地下载设计 + +日期：2026-06-30 + +## 背景 + +当前“结果提取”页面同时承载了两个不同概念： + +1. 生产结果入库：从 LandSAR/ENVI/Gamma 等生产目录提取结果并登记到 catalog。 +2. 成果交付导出：用户从已登记 catalog 中选择成果，下载或复制到本地使用。 + +现有 `POST /api/dinsar-results/export` 是同步文件复制接口，并且要求 admin。这个模型不适合普通用户下载大体量成果：请求容易超时，目标路径由用户输入也不利于审计和权限控制。 + +## 目标 + +- 所有登录用户都可以申请成果交付。 +- D-InSAR 已登记成果先接入真实交付下载。 +- SBAS、LT-1 正射、Sentinel-1 正射、GF3 正射在页面和接口中保留清晰占位，不暴露假执行能力。 +- 大文件交付改为后台任务，不再由 HTTP 请求同步复制。 +- 用户最终可以把成果下载到本地；服务器交付区只是临时缓存。 +- 每次交付可审计、可过期清理、可校验。 +- 数据库变更必须接入现有自维护机制。 + +## 非目标 + +- 本阶段不实现 Sentinel-1 正射生产。 +- 本阶段不实现 SBAS 成果交付打包，只显示目录已接入但交付未接入。 +- 本阶段不做跨节点对象存储或外部网盘。 +- 本阶段不把普通用户开放到任意服务器路径写入。 + +## 权限模型 + +当前系统只有 `admin` 和 `viewer`。本阶段不强制新增角色，采用能力约定： + +- 登录用户：可查看已授权 catalog，可创建自己的成果交付任务，可查看和下载自己的交付包。 +- admin：除普通用户能力外，可查看所有交付任务，可配置交付根目录，可清理或取消交付任务。 + +后续如果拆角色，建议增加： + +- `exporter`：可创建成果交付任务。 +- `operator`：可做生产、入库和成果交付。 +- `admin`：用户、系统配置和全局清理。 + +## 交付模式 + +### 1. 目录交付 + +默认模式。后台将选中的结果文件复制到： + +```text +{RESULT_DELIVERY_ROOT}/{username}/{delivery_id}/ +``` + +目录内包含： + +- `manifest.json` +- `checksums.sha256` +- 结果文件或结果子目录 + +适合几十 GB 到 TB 级数据。用户可以通过共享目录或逐文件 HTTP 下载到本地。 + +### 2. 压缩包交付 + +可选模式。只允许低于阈值的交付包生成 zip： + +```text +{RESULT_DELIVERY_ROOT}/{username}/{delivery_id}.zip +``` + +阈值由环境变量控制，例如 `RESULT_DELIVERY_ZIP_MAX_BYTES`。超过阈值时，接口返回明确错误，要求使用目录交付或逐文件下载。 + +### 3. HTTP 下载 + +下载不由 FastAPI 直接流式传大文件。推荐 Nginx 静态服务交付区，并支持 Range 断点续传。 + +第一版接口可以返回文件下载 URL，由后端验证交付归属后通过 `FileResponse` 交付；后续切到 Nginx `X-Accel-Redirect` 或专门静态路径。 + +## 数据模型 + +新增两张表： + +### `result_delivery_requests` + +- `delivery_id`：业务 ID。 +- `owner_user_id` / `owner_username`：申请人。 +- `channel`：`dinsar`、`sbas`、`lt1_ortho`、`s1_ortho`、`gf3_ortho`。 +- `status`：`PENDING`、`RUNNING`、`READY`、`FAILED`、`CANCELLED`、`EXPIRED`。 +- `package_mode`：`directory`、`zip`。 +- `item_count`、`total_bytes`、`copied_bytes`。 +- `delivery_root`、`delivery_dir`、`zip_path`、`manifest_path`。 +- `expires_at`。 +- `task_id`、`job_id`。 +- `error_message`。 +- `request_json`、`summary_json`。 + +### `result_delivery_items` + +- `delivery_id`。 +- `source_product_id` / `source_result_id`。 +- `display_name`。 +- `source_path`。 +- `relative_path`。 +- `file_size`。 +- `checksum_sha256`。 +- `status`：`PENDING`、`COPIED`、`FAILED`、`SKIPPED`。 +- `error_message`。 + +## 数据库自维护 + +新增 migration： + +```text +backend/migrations/013_result_delivery_requests.sql +``` + +并加入 `backend/app/db_maintenance.py` 的维护文件列表。迁移必须幂等： + +- `CREATE TABLE IF NOT EXISTS` +- `ALTER TABLE ... ADD COLUMN IF NOT EXISTS` +- `CREATE INDEX IF NOT EXISTS` + +ORM 同步也要定义对应模型，避免启动时 metadata 检查缺表。 + +## API 设计 + +### 获取通道能力 + +```http +GET /api/result-deliveries/channels +``` + +返回： + +- `dinsar`: `ready` +- `sbas`: `planned` +- `lt1_ortho`: `planned` +- `s1_ortho`: `placeholder` +- `gf3_ortho`: `placeholder` + +### 创建交付任务 + +```http +POST /api/result-deliveries +``` + +请求： + +```json +{ + "channel": "dinsar", + "product_ids": ["..."], + "compat_result_ids": [1, 2, 3], + "package_mode": "directory", + "include_manifest": true, + "include_checksums": true +} +``` + +约束： + +- 普通用户只创建自己的任务。 +- `channel != dinsar` 时第一版返回 409 或 422，提示“通道尚未接入交付”。 +- 每次最大数量由 `RESULT_DELIVERY_MAX_ITEMS` 控制。 +- 不允许传入任意服务器输出路径。 + +### 列出交付任务 + +```http +GET /api/result-deliveries?mine=true +``` + +普通用户只能看到自己的任务，admin 可查看全部。 + +### 查看交付详情 + +```http +GET /api/result-deliveries/{delivery_id} +``` + +返回交付状态、文件清单、下载 URL、过期时间。 + +### 下载文件 + +```http +GET /api/result-deliveries/{delivery_id}/files/{item_id}/download +GET /api/result-deliveries/{delivery_id}/archive/download +GET /api/result-deliveries/{delivery_id}/manifest +``` + +第一版由后端验证权限后返回文件。后续可迁移到 Nginx token 或 `X-Accel-Redirect`。 + +## 后台任务 + +新增 job type： + +```text +RESULT_DELIVERY_BUILD +``` + +处理流程： + +1. 将 delivery 标记为 `RUNNING`。 +2. 解析 D-InSAR catalog 中的文件路径。 +3. 复制到交付目录。 +4. 生成 `manifest.json`。 +5. 可选计算 checksum。 +6. 可选生成 zip。 +7. 更新状态为 `READY` 或 `FAILED`。 + +任务日志应写清： + +- 总项目数。 +- 已复制数量。 +- 总大小。 +- 失败项和原因。 +- 交付目录。 +- 过期时间。 + +## 存储与清理 + +环境变量建议： + +```text +RESULT_DELIVERY_ROOT=D:\Result_Delivery +RESULT_DELIVERY_PUBLIC_BASE_URL=/deliveries +RESULT_DELIVERY_RETENTION_DAYS=7 +RESULT_DELIVERY_MAX_ITEMS=500 +RESULT_DELIVERY_ZIP_MAX_BYTES=21474836480 +RESULT_DELIVERY_CHECKSUM_ENABLED=true +``` + +清理策略： + +- `expires_at < now` 的 `READY/FAILED/CANCELLED` 交付包可清理。 +- 清理后状态改为 `EXPIRED`，保留数据库审计记录。 +- 正式成果 catalog 原文件绝不能被清理任务删除。 + +## 前端设计 + +结果提取页面改名语义： + +- “生产结果入库”：保留现有 D-InSAR 入库入口，admin 可用。 +- “成果交付下载”：所有登录用户可用。 + +页面结构： + +- 通道栏：D-InSAR 可用，SBAS/LT-1/Sentinel-1/GF3 明确显示“未接入交付”。 +- 成果选择区：复用现有 D-InSAR catalog 列表。 +- 交付选项：目录交付 / 压缩包交付。 +- 我的交付包：状态、大小、文件数、过期时间、下载入口。 + +交互约束： + +- 不再让普通用户输入服务器路径。 +- 创建后显示任务 ID 和交付 ID。 +- 对大文件提示“建议使用逐文件下载或共享目录复制”。 +- 下载入口只在 `READY` 状态显示。 + +## 与用户管理联动 + +第一版： + +- `viewer` 也可以创建自己的成果交付任务。 +- 前端不再用 `readOnly` 禁用成果交付下载。 +- 生产结果入库、删除、系统配置仍要求 admin。 + +后续： + +- 增加 `exporter/operator` 角色后，前端用户管理页需要增加角色选项。 +- 后端新增能力级依赖，例如 `require_capability("result.delivery.create")`。 + +## 风险与防护 + +- 大文件复制拖慢生产盘：限制 worker 并发，交付任务可单独限流。 +- 用户重复申请导致空间膨胀：按用户限制未过期交付包数量和总大小。 +- 任意路径写入风险：只允许系统配置的交付根目录。 +- HTTP 下载超时：优先 Nginx Range，后端只负责授权。 +- catalog 文件被移动：任务记录 item 失败，不影响其他文件。 + +## 实施阶段 + +### 阶段 1 + +- 文档落地。 +- 数据表和自维护 migration。 +- D-InSAR 目录交付后台任务。 +- 我的交付包列表和详情。 + +### 阶段 2 + +- zip 打包阈值和下载。 +- manifest/checksum 下载。 +- Nginx 静态交付路径或 `X-Accel-Redirect`。 + +### 阶段 3 + +- SBAS 交付接入。 +- LT-1 正射、Sentinel-1 正射、GF3 正射在生产 catalog 完成后接入。 +- exporter/operator 角色拆分。 + +## 验收标准 + +- 普通登录用户能创建 D-InSAR 成果交付任务。 +- HTTP 请求只排队任务，不再同步复制大文件。 +- 交付完成后用户能下载到本地。 +- 普通用户不能指定任意服务器目录。 +- admin 能看到所有交付任务。 +- 数据库重启自维护能创建交付相关表和索引。 +- Sentinel-1 正射通道显示占位，不可误点击执行。 diff --git a/frontend/src/App.css b/frontend/src/App.css index 023b75f..d184dc5 100644 --- a/frontend/src/App.css +++ b/frontend/src/App.css @@ -7025,7 +7025,7 @@ input[type="checkbox"] { .result-extraction-controls { display: grid; - grid-template-columns: minmax(180px, 0.75fr) minmax(260px, 1.4fr) auto; + grid-template-columns: minmax(180px, 0.9fr) minmax(150px, 0.45fr) minmax(160px, auto) auto; gap: 12px; padding: 14px 16px 10px; } @@ -7042,7 +7042,8 @@ input[type="checkbox"] { color: var(--color-text-muted); } -.result-extraction-field input { +.result-extraction-field input, +.result-extraction-field select { width: 100%; min-width: 0; box-sizing: border-box; @@ -7053,6 +7054,21 @@ input[type="checkbox"] { color: var(--color-text-primary); } +.result-extraction-checkbox { + display: inline-flex; + align-items: end; + gap: 8px; + min-height: 55px; + color: var(--color-text-secondary); + font-size: 12px; + font-weight: 800; +} + +.result-extraction-checkbox input { + width: 16px; + height: 16px; +} + .result-extraction-action-stack { display: flex; align-items: end; @@ -7119,6 +7135,95 @@ input[type="checkbox"] { align-content: start; } +.result-extraction-workspace-split { + flex: 1 1 auto; + min-height: 0; + display: grid; + grid-template-columns: minmax(0, 1.25fr) minmax(330px, 0.75fr); + border-top: 1px solid var(--color-border); +} + +.result-extraction-result-column { + min-width: 0; + min-height: 0; + display: flex; + flex-direction: column; +} + +.result-delivery-panel { + min-width: 0; + min-height: 0; + display: flex; + flex-direction: column; + border-left: 1px solid var(--color-border); + background: #fbfdff; +} + +.result-delivery-list { + flex: 1 1 auto; + min-height: 0; + overflow-y: auto; + scrollbar-gutter: stable; +} + +.result-delivery-row { + display: grid; + grid-template-columns: minmax(0, 1fr) auto; + gap: 10px; + align-items: start; + padding: 11px 16px; + border-bottom: 1px solid var(--color-border); +} + +.result-delivery-row-main { + min-width: 0; + display: grid; + gap: 5px; +} + +.result-delivery-row-main strong { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + font-size: 13px; + color: var(--color-text-primary); +} + +.result-delivery-row-main span, +.result-delivery-row-main em { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + font-size: 12px; + color: var(--color-text-muted); +} + +.result-delivery-row-main em { + color: #b91c1c; + font-style: normal; +} + +.result-delivery-downloads { + display: flex; + flex-wrap: wrap; + gap: 6px; +} + +.result-delivery-downloads a { + max-width: 150px; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + padding: 4px 7px; + border: 1px solid rgba(37, 99, 235, 0.2); + border-radius: 6px; + background: #eff6ff; + color: #1d4ed8; + font-size: 11px; + font-weight: 800; + text-decoration: none; +} + .result-extraction-result-row { display: grid; grid-template-columns: auto minmax(0, 1fr) auto auto; @@ -7269,6 +7374,16 @@ input[type="checkbox"] { grid-template-columns: minmax(220px, 280px) minmax(0, 1fr); } + .result-extraction-workspace-split { + grid-template-columns: 1fr; + } + + .result-delivery-panel { + border-left: 0; + border-top: 1px solid var(--color-border); + min-height: 260px; + } + .dinsar-filter-layout { grid-template-columns: repeat(2, minmax(0, 1fr)); } @@ -7357,6 +7472,11 @@ input[type="checkbox"] { .result-extraction-action-stack { align-items: stretch; } + + .result-extraction-checkbox { + align-items: center; + min-height: auto; + } } @media (max-width: 640px) { diff --git a/frontend/src/ResultExtractionPanel.jsx b/frontend/src/ResultExtractionPanel.jsx index ba19399..9369428 100644 --- a/frontend/src/ResultExtractionPanel.jsx +++ b/frontend/src/ResultExtractionPanel.jsx @@ -1,10 +1,16 @@ import { useEffect, useMemo, useState } from 'react'; -import { exportDinsarResults, getDinsarResults } from './api/dinsar'; +import { getDinsarResults } from './api/dinsar'; import { listSbasInsarProducts } from './api/sbasInsarProducts'; +import { + createResultDelivery, + getResultDeliveryArchiveUrl, + getResultDeliveryDownloadUrl, + getResultDeliveryManifestUrl, + listResultDeliveries, +} from './api/resultDeliveries'; import { getDinsarEngineMeta } from './utils/dinsarEngines'; -const DEFAULT_TARGET_DIR = String.raw`D:\Result_Export\DInSAR`; const PAGE_SIZE = 100; const PRODUCT_CHANNELS = [ @@ -13,8 +19,8 @@ const PRODUCT_CHANNELS = [ group: 'InSAR 成果', label: 'D-InSAR 结果', state: 'ready', - stateText: '可提取', - description: '从已登记的 D-InSAR 成果中选择位移结果，复制到服务器指定交付目录。', + stateText: '可交付', + description: '从已登记的 D-InSAR catalog 中选择成果，后台生成受控交付包并下载到本地。', }, { key: 'sbas', @@ -22,7 +28,7 @@ const PRODUCT_CHANNELS = [ label: 'SBAS-InSAR 结果', state: 'planned', stateText: '目录可查', - description: '成果目录和预览已接入，统一提取接口待补齐。', + description: 'SBAS 成果目录已接入，本阶段只展示目录状态，交付打包后续接入。', }, { key: 'lt1_ortho', @@ -30,7 +36,7 @@ const PRODUCT_CHANNELS = [ label: 'LT-1 正射结果', state: 'placeholder', stateText: '待接入', - description: '陆探一正射生产结果后续接入标准成果目录，并开放提取。', + description: '陆探一正射生产将由 LandSAR 生产链注册到统一成果目录后开放交付。', }, { key: 's1_ortho', @@ -38,7 +44,7 @@ const PRODUCT_CHANNELS = [ label: 'Sentinel-1 正射结果', state: 'placeholder', stateText: '待接入', - description: 'Sentinel-1 正射生产占位，后续登记后统一提取。', + description: 'Sentinel-1 正射生产尚未接入，当前只保留交付通道占位。', }, { key: 'gf3_ortho', @@ -46,7 +52,7 @@ const PRODUCT_CHANNELS = [ label: 'GF3 SARscape _geo', state: 'placeholder', stateText: '待接入', - description: 'GF3 外部生产后的 _geo 二进制和 WebP 已按本机登记思路设计，统一导出接口待接入。', + description: 'GF3 外部生产成果登记后再接入统一交付。', }, ]; @@ -56,6 +62,19 @@ function formatNumber(value) { return new Intl.NumberFormat('zh-CN').format(number); } +function formatBytes(value) { + const bytes = Number(value); + if (!Number.isFinite(bytes) || bytes <= 0) return '0 B'; + const units = ['B', 'KB', 'MB', 'GB', 'TB']; + let size = bytes; + let index = 0; + while (size >= 1024 && index < units.length - 1) { + size /= 1024; + index += 1; + } + return `${size.toFixed(index === 0 ? 0 : 1)} ${units[index]}`; +} + function normalizeItems(payload) { return Array.isArray(payload?.items) ? payload.items : []; } @@ -83,33 +102,71 @@ function stateClass(state) { return 'pending'; } +function statusText(status) { + const value = String(status || '').toUpperCase(); + return { + PENDING: '排队中', + RUNNING: '生成中', + READY: '可下载', + FAILED: '失败', + CANCELLED: '已取消', + EXPIRED: '已过期', + }[value] || value || '-'; +} + +function statusClass(status) { + const value = String(status || '').toUpperCase(); + if (value === 'READY') return 'ready'; + if (value === 'RUNNING' || value === 'PENDING') return 'planned'; + return 'pending'; +} + export default function ResultExtractionPanel({ readOnly = false }) { const [activeChannel, setActiveChannel] = useState('dinsar'); const [dinsarPayload, setDinsarPayload] = useState({ items: [], total: 0 }); const [sbasPayload, setSbasPayload] = useState({ items: [], total: 0 }); + const [deliveriesPayload, setDeliveriesPayload] = useState({ items: [], total: 0 }); const [loading, setLoading] = useState(true); + const [deliveryLoading, setDeliveryLoading] = useState(false); const [error, setError] = useState(''); const [query, setQuery] = useState(''); - const [targetDir, setTargetDir] = useState(DEFAULT_TARGET_DIR); const [selectedIds, setSelectedIds] = useState(() => new Set()); - const [exporting, setExporting] = useState(false); - const [exportError, setExportError] = useState(''); - const [exportResult, setExportResult] = useState(null); + const [packageMode, setPackageMode] = useState('directory'); + const [includeChecksums, setIncludeChecksums] = useState(true); + const [creating, setCreating] = useState(false); + const [createError, setCreateError] = useState(''); + const [createResult, setCreateResult] = useState(null); const selectedChannel = PRODUCT_CHANNELS.find(channel => channel.key === activeChannel) || PRODUCT_CHANNELS[0]; + const loadDeliveries = async () => { + setDeliveryLoading(true); + try { + const payload = await listResultDeliveries({ mine: true, limit: 20, offset: 0 }); + const items = normalizeItems(payload); + setDeliveriesPayload({ ...payload, items, total: extractTotal(payload, items.length) }); + } catch (err) { + setCreateError(err?.response?.data?.detail || err.message || '交付包列表加载失败'); + } finally { + setDeliveryLoading(false); + } + }; + const loadCatalogs = async () => { setLoading(true); setError(''); try { - const [dinsarData, sbasData] = await Promise.all([ + const [dinsarData, sbasData, deliveryData] = await Promise.all([ getDinsarResults({ limit: PAGE_SIZE, offset: 0 }), listSbasInsarProducts({ limit: 30, offset: 0 }), + listResultDeliveries({ mine: true, limit: 20, offset: 0 }), ]); const dinsarItems = normalizeItems(dinsarData); setDinsarPayload({ ...dinsarData, items: dinsarItems, total: extractTotal(dinsarData, dinsarItems.length) }); const sbasItems = normalizeItems(sbasData); setSbasPayload({ ...sbasData, items: sbasItems, total: extractTotal(sbasData, sbasItems.length) }); + const deliveryItems = normalizeItems(deliveryData); + setDeliveriesPayload({ ...deliveryData, items: deliveryItems, total: extractTotal(deliveryData, deliveryItems.length) }); setSelectedIds(new Set(dinsarItems.map(item => item.id).filter(id => id !== undefined && id !== null))); } catch (err) { setError(err?.response?.data?.detail || err.message || '结果目录加载失败'); @@ -122,6 +179,18 @@ export default function ResultExtractionPanel({ readOnly = false }) { loadCatalogs(); }, []); + useEffect(() => { + const hasActiveDelivery = (deliveriesPayload.items || []).some(item => { + const status = String(item.status || '').toUpperCase(); + return status === 'PENDING' || status === 'RUNNING'; + }); + if (!hasActiveDelivery) return undefined; + const timer = window.setInterval(() => { + loadDeliveries(); + }, 5000); + return () => window.clearInterval(timer); + }, [deliveriesPayload.items]); + const filteredDinsar = useMemo(() => { const value = query.trim().toLowerCase(); const items = dinsarPayload.items || []; @@ -147,21 +216,22 @@ export default function ResultExtractionPanel({ readOnly = false }) { const selectedCountInView = filteredIds.filter(id => selectedIds.has(id)).length; const allVisibleSelected = filteredIds.length > 0 && selectedCountInView === filteredIds.length; + const latestDelivery = deliveriesPayload.items?.[0] || null; const orthoPlaceholderCount = PRODUCT_CHANNELS.filter(channel => channel.group === '正射成果').length; const currentCatalogTotal = Number(dinsarPayload.total || 0) + Number(sbasPayload.total || 0); const metrics = [ { - label: 'D-InSAR 可提取', + label: 'D-InSAR 可交付', value: dinsarPayload.total, note: `当前载入 ${filteredDinsar.length}/${dinsarPayload.items.length} 条`, tone: 'primary', }, { - label: 'SBAS 目录', - value: sbasPayload.total, - note: '统一提取接口待接入', + label: '我的交付包', + value: deliveriesPayload.total, + note: latestDelivery ? `最近状态：${statusText(latestDelivery.status)}` : '暂无交付记录', tone: 'neutral', }, { @@ -202,38 +272,95 @@ export default function ResultExtractionPanel({ readOnly = false }) { }); }; - const handleExport = async () => { - const dir = targetDir.trim(); - if (!dir) { - setExportError('请输入服务器目标目录。'); - return; - } + const handleCreateDelivery = async () => { const ids = [...selectedIds].filter(id => filteredIds.includes(id)); if (ids.length === 0) { - setExportError('请至少选择一条 D-InSAR 结果。'); + setCreateError('请至少选择一条 D-InSAR 结果。'); return; } - setExporting(true); - setExportError(''); - setExportResult(null); + setCreating(true); + setCreateError(''); + setCreateResult(null); try { - const response = await exportDinsarResults(ids, dir); - setExportResult(response); + const response = await createResultDelivery({ + channel: 'dinsar', + compat_result_ids: ids, + package_mode: packageMode, + include_checksums: includeChecksums, + }); + setCreateResult(response); + await loadDeliveries(); } catch (err) { - setExportError(err?.response?.data?.detail || err.message || 'D-InSAR 结果提取失败'); + setCreateError(err?.response?.data?.detail || err.message || '成果交付任务创建失败'); } finally { - setExporting(false); + setCreating(false); } }; + const renderDeliveryDownloads = (delivery) => { + if (String(delivery.status || '').toUpperCase() !== 'READY') { + return null; + } + const items = Array.isArray(delivery.items) ? delivery.items : []; + return ( +
+ manifest + {delivery.zip_path && ( + zip + )} + {items.slice(0, 3).map(item => ( + + {item.relative_path?.split(/[\\/]/).pop() || `文件 ${item.id}`} + + ))} +
+ ); + }; + + const renderDeliveryList = () => ( +
+
+ 我的交付包 + {deliveryLoading ? '刷新中' : `${deliveriesPayload.items.length}/${deliveriesPayload.total}`} +
+
+ {deliveriesPayload.items.length === 0 ? ( +
还没有创建过成果交付包。
+ ) : ( + deliveriesPayload.items.map(delivery => ( +
+
+ {delivery.delivery_id} + + {delivery.channel} + {' · '} + {delivery.package_mode} + {' · '} + {formatNumber(delivery.item_count)} 文件 + {' · '} + {formatBytes(delivery.copied_bytes || delivery.total_bytes)} + + {delivery.error_message && {delivery.error_message}} + {renderDeliveryDownloads(delivery)} +
+ + {statusText(delivery.status)} + +
+ )) + )} +
+
+ ); + const renderDinsarWorkspace = () => (
- D-InSAR 交付提取 - 选择已登记结果并复制到服务器目录 + D-InSAR 成果交付 + 选择已登记结果并生成下载包
-
@@ -246,83 +373,94 @@ export default function ResultExtractionPanel({ readOnly = false }) { value={query} onChange={event => setQuery(event.target.value)} placeholder="任务名、日期、pair_key、引擎" - disabled={loading || exporting} + disabled={loading || creating} /> -
); @@ -338,15 +476,15 @@ export default function ResultExtractionPanel({ readOnly = false }) {
登记入口 - {selectedChannel.key === 'sbas' ? 'SBAS-InSAR 成果目录' : '生产管理成果登记'} + {selectedChannel.key === 'sbas' ? 'SBAS-InSAR 成果目录' : '生产结果 catalog'}
- 提取接口 - 待实现 + 交付接口 + {selectedChannel.key === 'dinsar' ? '已接入' : '待实现'}
- 交付目录 - 服务器固定/指定路径 + 用户权限 + 登录用户自助申请
{selectedChannel.key === 'sbas' && ( @@ -375,14 +513,14 @@ export default function ResultExtractionPanel({ readOnly = false }) { 成果交付出口 结果提取工作台

- 将三类正射生产成果、D-InSAR 成果和 SBAS-InSAR 成果集中管理。当前 D-InSAR 已接入真实提取， - 其余链路先保留清晰占位，避免把未完成流程误当成可执行功能。 + 将 D-InSAR 成果交付下载和后续正射成果交付统一管理。当前 D-InSAR 支持后台交付包， + SBAS、LT-1 正射、Sentinel-1 正射和 GF3 正射先保留清晰占位。

D-InSAR {formatNumber(dinsarPayload.total)} SBAS {formatNumber(sbasPayload.total)} - {readOnly ? '只读账号' : '可执行账号'} + {readOnly ? '自助交付' : '管理员'}
diff --git a/frontend/src/api/index.js b/frontend/src/api/index.js index fe52c78..5bd81bf 100644 --- a/frontend/src/api/index.js +++ b/frontend/src/api/index.js @@ -18,3 +18,4 @@ export * as statsApi from './stats'; export * as timeseriesProductionApi from './timeseriesProduction'; export * as psinsarProductsApi from './psinsarProducts'; export * as landsarLt1ProductionApi from './landsarLt1Production'; +export * as resultDeliveriesApi from './resultDeliveries'; diff --git a/frontend/src/api/resultDeliveries.js b/frontend/src/api/resultDeliveries.js new file mode 100644 index 0000000..3cb7357 --- /dev/null +++ b/frontend/src/api/resultDeliveries.js @@ -0,0 +1,22 @@ +import apiClient from './client'; + +export const getResultDeliveryChannels = () => + apiClient.get('/result-deliveries/channels').then(r => r.data); + +export const createResultDelivery = payload => + apiClient.post('/result-deliveries', payload).then(r => r.data); + +export const listResultDeliveries = (params = {}) => + apiClient.get('/result-deliveries', { params }).then(r => r.data); + +export const getResultDelivery = deliveryId => + apiClient.get(`/result-deliveries/${encodeURIComponent(deliveryId)}`).then(r => r.data); + +export const getResultDeliveryDownloadUrl = (deliveryId, itemId) => + `/api/result-deliveries/${encodeURIComponent(deliveryId)}/files/${encodeURIComponent(itemId)}/download`; + +export const getResultDeliveryManifestUrl = deliveryId => + `/api/result-deliveries/${encodeURIComponent(deliveryId)}/manifest`; + +export const getResultDeliveryArchiveUrl = deliveryId => + `/api/result-deliveries/${encodeURIComponent(deliveryId)}/archive/download`; diff --git a/frontend/src/config/taskUiPolicies.js b/frontend/src/config/taskUiPolicies.js index 3e8c478..531ef82 100644 --- a/frontend/src/config/taskUiPolicies.js +++ b/frontend/src/config/taskUiPolicies.js @@ -25,6 +25,7 @@ const TASK_UI_POLICIES = { ISCE2_RUN: { label: 'D-InSAR 历史任务', featureScope: 'dinsar_production' }, PYINT_RUN: { label: 'PyINT D-InSAR 生产', featureScope: 'dinsar_production' }, LANDSAR_RUN: { label: 'LandSAR D-InSAR 生产', featureScope: 'dinsar_production' }, + RESULT_DELIVERY_BUILD: { label: '成果交付包生成', featureScope: 'result_extraction' }, SBAS_GAMMA_WORKFLOW: { label: 'Gamma SBAS 工作流', featureScope: 'sbas_insar' }, SBAS_LANDSAR_WORKFLOW: { label: 'LandSAR SBAS 工作流', featureScope: 'sbas_insar' }, SBAS_COREGISTRATION: { label: 'SBAS 配准', featureScope: 'sbas_insar' }, @@ -42,6 +43,7 @@ const PREFIX_POLICIES = [ { prefix: 'FLOOD_WATER_EXTRACTION_', label: '洪涝水体提取', featureScope: 'flood' }, { prefix: 'FLOOD_DETECTION_', label: '洪涝检测', featureScope: 'flood' }, { prefix: 'GF3_PROCESS_', label: 'GF3 场景处理', featureScope: 'water' }, + { prefix: 'RESULT_DELIVERY_BUILD_', label: '成果交付包生成', featureScope: 'result_extraction' }, ]; export function getTaskUiPolicy(taskType) {