diff --git a/backend/app/routers/assets.py b/backend/app/routers/assets.py index 84a9742..5cfce0d 100644 --- a/backend/app/routers/assets.py +++ b/backend/app/routers/assets.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime from typing import Any, Dict, List, Optional from fastapi import APIRouter, Depends, HTTPException, Query @@ -39,6 +40,11 @@ class S1BatchUnpackRequest(BaseModel): scan_before_unpack: bool = True +class SourceMaterializeRequest(BaseModel): + target_root: Optional[str] = None + overwrite: bool = False + + @router.get("/inventory/status") async def get_asset_inventory_status( current_user: AuthUserORM = Depends(_get_current_user), @@ -199,6 +205,35 @@ async def unpack_sentinel1_source_asset( return {"message": "Sentinel-1 unpack task queued", "task_id": task_id, "job_id": job_id} +@router.post("/sources/{asset_id}/materialize") +async def materialize_source_asset( + asset_id: int, + request: Optional[SourceMaterializeRequest] = None, + admin_user: AuthUserORM = Depends(_require_admin), + db: AsyncSession = Depends(get_db), +): + _ = admin_user + asset = await db.get(SourceProductAssetORM, asset_id) + if asset is None: + raise HTTPException(status_code=404, detail="Source product asset not found.") + request_data = request or SourceMaterializeRequest() + try: + result = asset_inventory_service.materialize_source_asset( + asset, + target_root=request_data.target_root, + overwrite=bool(request_data.overwrite), + ) + except Exception as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + metadata = dict(asset.metadata_json or {}) + metadata["last_materialized_dir"] = result.get("safe_dir") or result.get("target_dir") + metadata["last_materialized_at"] = datetime.utcnow().isoformat() + metadata["last_materialized_status"] = result.get("status") + asset.metadata_json = metadata + await db.commit() + return {"message": "Source asset materialized", "result": result} + + @router.post("/inventory/unpack-sentinel1", status_code=202) async def run_sentinel1_unpack_batch( request: Optional[S1BatchUnpackRequest] = None, diff --git a/backend/app/services/asset_inventory_service.py b/backend/app/services/asset_inventory_service.py index 9cb9634..649ca1e 100644 --- a/backend/app/services/asset_inventory_service.py +++ b/backend/app/services/asset_inventory_service.py @@ -5,8 +5,10 @@ import hashlib import os import re import shutil +import tarfile import zipfile from datetime import datetime, timedelta +from pathlib import PurePosixPath from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple from geoalchemy2.shape import from_shape @@ -30,6 +32,7 @@ from ..models import ( from ..utils import ( find_xml_file, normalize_satellite_family, + parse_gf3_l2_dirname, parse_lt1_radar_filename, parse_xml_metadata, ) @@ -68,6 +71,8 @@ _LT1_ORBIT_RE = re.compile( r"^(?PLT1[A-Z]?)_GpsData_GAS_C_(?P\d{8})\.txt$", re.IGNORECASE, ) +_LT1_ARCHIVE_EXTS = (".tar.gz", ".tgz", ".zip", ".tar") +_GF3_ARCHIVE_EXTS = (".tar.gz", ".tgz", ".zip", ".tar") def _parse_bool(value: Any, default: bool = False) -> bool: @@ -190,13 +195,167 @@ def _asset_uid(prefix: str, path: str) -> str: def _strip_known_suffix(name: str) -> str: lower = name.lower() + for suffix in (".tar.gz", ".tgz"): + if lower.endswith(suffix): + return name[: -len(suffix)] if lower.endswith(".zip"): return name[:-4] + if lower.endswith(".tar"): + return name[:-4] if lower.endswith(".safe"): return name[:-5] return name +def _has_archive_suffix(name: str, suffixes: Sequence[str]) -> bool: + lower = str(name or "").lower() + return any(lower.endswith(suffix) for suffix in suffixes) + + +def _archive_member_base_name(member_name: str) -> str: + text = str(member_name or "").replace("\\", "/").strip("/") + return PurePosixPath(text).name + + +def _archive_member_scene_name(member_name: str, fallback: str) -> str: + parts = [part for part in str(member_name or "").replace("\\", "/").split("/") if part] + for part in parts: + stem = _strip_known_suffix(part) + if parse_lt1_radar_filename(stem) or parse_gf3_l2_dirname(stem): + return stem + return fallback + + +def _archive_read_first_matching(path: str, predicate: Callable[[str], bool]) -> Tuple[Optional[str], Optional[bytes], List[str]]: + members: List[str] = [] + if zipfile.is_zipfile(path): + with zipfile.ZipFile(path) as archive: + for info in archive.infolist(): + name = info.filename + if info.is_dir(): + continue + members.append(name) + if predicate(name): + return name, archive.read(info), members + return None, None, members + + if tarfile.is_tarfile(path): + with tarfile.open(path, "r:*") as archive: + for member in archive: + name = member.name + if not member.isfile(): + continue + members.append(name) + if predicate(name): + source = archive.extractfile(member) + if source is None: + continue + with source: + return name, source.read(), members + return None, None, members + + return None, None, members + + +def _archive_list_matching(path: str, predicate: Callable[[str], bool], *, limit: int = 20) -> List[str]: + matches: List[str] = [] + + def _visit(name: str) -> None: + if predicate(name): + matches.append(name) + + if zipfile.is_zipfile(path): + with zipfile.ZipFile(path) as archive: + for info in archive.infolist(): + if info.is_dir(): + continue + _visit(info.filename) + if len(matches) >= limit: + break + return matches + + if tarfile.is_tarfile(path): + with tarfile.open(path, "r:*") as archive: + for member in archive: + if not member.isfile(): + continue + _visit(member.name) + if len(matches) >= limit: + break + return matches + + +def _safe_archive_member_name(member_name: str, archive_path: str) -> str: + name = str(member_name or "").replace("\\", "/").strip("/") + if not name or name.startswith("../") or "/../" in f"/{name}/": + raise ValueError(f"Unsafe archive member path in {archive_path}: {member_name}") + if os.path.isabs(name) or os.path.splitdrive(name)[0]: + raise ValueError(f"Unsafe archive member path in {archive_path}: {member_name}") + return name + + +def _extract_archive_to_dir(archive_path: str, target_dir: str, *, overwrite: bool = False) -> Dict[str, Any]: + archive = _normalize_path(archive_path) + target = _normalize_path(target_dir) + if not os.path.isfile(archive): + raise FileNotFoundError(f"Archive does not exist: {archive}") + if os.path.exists(target): + if not overwrite: + return {"status": "EXISTS", "archive_path": archive, "target_dir": target, "extracted": False, "member_count": None} + shutil.rmtree(target) + + tmp_dir = target + ".materialize_tmp" + if os.path.exists(tmp_dir): + shutil.rmtree(tmp_dir) + os.makedirs(tmp_dir, exist_ok=True) + extracted = 0 + try: + if zipfile.is_zipfile(archive): + with zipfile.ZipFile(archive) as zip_obj: + for info in zip_obj.infolist(): + rel_name = _safe_archive_member_name(info.filename, archive) + destination = os.path.abspath(os.path.join(tmp_dir, rel_name)) + if not destination.startswith(os.path.abspath(tmp_dir) + os.sep): + raise ValueError(f"Unsafe ZIP member path: {info.filename}") + if info.is_dir(): + os.makedirs(destination, exist_ok=True) + continue + os.makedirs(os.path.dirname(destination), exist_ok=True) + with zip_obj.open(info, "r") as source, open(destination, "wb") as target_stream: + shutil.copyfileobj(source, target_stream, length=1024 * 1024) + extracted += 1 + elif tarfile.is_tarfile(archive): + with tarfile.open(archive, "r:*") as tar_obj: + for member in tar_obj: + rel_name = _safe_archive_member_name(member.name, archive) + destination = os.path.abspath(os.path.join(tmp_dir, rel_name)) + if not destination.startswith(os.path.abspath(tmp_dir) + os.sep): + raise ValueError(f"Unsafe TAR member path: {member.name}") + if member.isdir(): + os.makedirs(destination, exist_ok=True) + continue + if not member.isfile(): + continue + source = tar_obj.extractfile(member) + if source is None: + continue + os.makedirs(os.path.dirname(destination), exist_ok=True) + with source, open(destination, "wb") as target_stream: + shutil.copyfileobj(source, target_stream, length=1024 * 1024) + extracted += 1 + else: + raise ValueError(f"Unsupported archive format: {archive}") + + if extracted <= 0: + raise OSError(f"Archive extraction produced no files: {archive}") + os.makedirs(os.path.dirname(target), exist_ok=True) + os.replace(tmp_dir, target) + return {"status": "EXTRACTED", "archive_path": archive, "target_dir": target, "extracted": True, "member_count": extracted} + finally: + if os.path.exists(tmp_dir): + shutil.rmtree(tmp_dir, ignore_errors=True) + + def _parse_datetime_token(value: Optional[str]) -> Optional[datetime]: text = str(value or "").strip() if not text: @@ -216,6 +375,105 @@ def _parse_datetime_token(value: Optional[str]) -> Optional[datetime]: return None +def _xml_text_by_local_names(root: etree._Element, names: Sequence[str]) -> Optional[str]: + name_set = {str(item).lower() for item in names} + for element in root.iter(): + local_name = etree.QName(element).localname.lower() + if local_name not in name_set: + continue + text = str(element.text or "").strip() + if text: + return text + return None + + +def _xml_text_under_local_path(root: etree._Element, parent_name: str, child_name: str) -> Optional[str]: + parent_key = parent_name.lower() + child_key = child_name.lower() + for parent in root.iter(): + if etree.QName(parent).localname.lower() != parent_key: + continue + for child in parent.iter(): + if child is parent: + continue + if etree.QName(child).localname.lower() == child_key: + text = str(child.text or "").strip() + if text: + return text + return None + + +def _xml_float(value: Optional[str]) -> Optional[float]: + try: + if value is None or str(value).strip() == "": + return None + return float(str(value).strip()) + except (TypeError, ValueError): + return None + + +def _parse_radar_xml_metadata_bytes(data: bytes) -> Tuple[Optional[List[Tuple[float, float]]], Dict[str, Any]]: + parser = _xml_parser() + root = etree.fromstring(data, parser=parser) + + corners: List[Tuple[float, float]] = [] + for element in root.iter(): + if etree.QName(element).localname.lower() != "scenecornercoord": + continue + lon = _xml_float(_xml_text_under_local_path(element, "sceneCornerCoord", "lon") or _xml_text_by_local_names(element, ["lon"])) + lat = _xml_float(_xml_text_under_local_path(element, "sceneCornerCoord", "lat") or _xml_text_by_local_names(element, ["lat"])) + if lon is not None and lat is not None: + corners.append((lon, lat)) + + coverage_polygon: Optional[List[Tuple[float, float]]] = None + if len(corners) >= 4: + coverage_polygon = corners[:4] + if coverage_polygon[0] != coverage_polygon[-1]: + coverage_polygon.append(coverage_polygon[0]) + + start_time = ( + _xml_text_under_local_path(root, "start", "timeUTC") + or _xml_text_by_local_names(root, ["startTime", "start_time", "beginPosition"]) + ) + stop_time = ( + _xml_text_under_local_path(root, "stop", "timeUTC") + or _xml_text_by_local_names(root, ["stopTime", "stop_time", "endPosition"]) + ) + center_lon = _xml_float(_xml_text_under_local_path(root, "sceneCenterCoord", "lon")) + center_lat = _xml_float(_xml_text_under_local_path(root, "sceneCenterCoord", "lat")) + metadata = { + "orbit_direction": (_xml_text_by_local_names(root, ["pass", "orbitDirection"]) or "").upper() or None, + "imaging_mode": _xml_text_under_local_path(root, "acquisitionInfo", "imagingMode") + or _xml_text_under_local_path(root, "orderInfo", "imagingMode") + or _xml_text_by_local_names(root, ["imagingMode"]), + "polarization": _xml_text_under_local_path(root, "acquisitionInfo", "polarisationMode") + or _xml_text_under_local_path(root, "polarisationList", "polLayer") + or _xml_text_under_local_path(root, "polList", "polLayer") + or _xml_text_by_local_names(root, ["polarisationMode", "polarization", "polarisation", "polLayer"]), + "receiving_station": _xml_text_under_local_path(root, "generationInfo", "receivingStation") + or _xml_text_by_local_names(root, ["receivingStation"]), + "satellite_mode": _xml_text_by_local_names(root, ["satelliteMode"]), + "orbit_circle": _xml_text_under_local_path(root, "missionInfo", "absOrbit") + or _xml_text_by_local_names(root, ["absOrbit", "absoluteOrbit"]), + "relative_orbit": _xml_text_under_local_path(root, "missionInfo", "relOrbit") + or _xml_text_by_local_names(root, ["relOrbit", "relativeOrbit"]), + "scene_center_lon": center_lon, + "scene_center_lat": center_lat, + "acquisition_time_utc": start_time, + "acquisition_stop_time_utc": stop_time, + "product_type": _xml_text_under_local_path(root, "imageDataInfo", "imageDataType") + or _xml_text_under_local_path(root, "orderInfo", "productVariant") + or _xml_text_by_local_names(root, ["productType", "imageDataType", "productVariant"]), + "image_data_format": _xml_text_under_local_path(root, "imageDataInfo", "imageDataFormat") + or _xml_text_by_local_names(root, ["imageDataFormat"]), + "product_level": _xml_text_by_local_names(root, ["productLevel", "itemName"]), + "product_unique_id": _xml_text_by_local_names(root, ["logicalProductID", "sceneID", "productID"]), + "look_direction": (_xml_text_under_local_path(root, "acquisitionInfo", "lookDirection") or "").upper() or None, + "coverage_polygon": coverage_polygon, + } + return coverage_polygon, {key: value for key, value in metadata.items() if value not in (None, "", [])} + + def _date_start_stop(date_yyyymmdd: str) -> Tuple[Optional[datetime], Optional[datetime]]: try: start = datetime.strptime(date_yyyymmdd, "%Y%m%d") @@ -416,6 +674,62 @@ def _parse_s1_safe_manifest(path: str) -> Dict[str, Any]: } +def _parse_lt1_archive_metadata(path: str) -> Dict[str, Any]: + archive_stem = _strip_known_suffix(os.path.basename(path)) + xml_member, xml_data, members = _archive_read_first_matching( + path, + lambda name: _archive_member_base_name(name).lower().endswith(".meta.xml"), + ) + tiff_members = _archive_list_matching( + path, + lambda name: _archive_member_base_name(name).lower().endswith((".tiff", ".tif")), + limit=8, + ) + if not xml_member or not xml_data: + return { + "archive_parse_status": "MISSING_XML", + "archive_member_count_scanned": len(members), + "contained_tiff_members": tiff_members, + } + coverage_polygon, xml_meta = _parse_radar_xml_metadata_bytes(xml_data) + return { + "archive_parse_status": "OK", + "archive_xml_member": xml_member, + "archive_scene_name": _archive_member_scene_name(xml_member, archive_stem), + "contained_tiff_members": tiff_members, + "coverage_polygon": coverage_polygon, + **xml_meta, + } + + +def _parse_gf3_archive_metadata(path: str) -> Dict[str, Any]: + archive_stem = _strip_known_suffix(os.path.basename(path)) + xml_member, xml_data, members = _archive_read_first_matching( + path, + lambda name: _archive_member_base_name(name).lower().endswith(".xml"), + ) + quicklooks = _archive_list_matching( + path, + lambda name: _archive_member_base_name(name).lower().endswith((".jpg", ".jpeg", ".png", ".bmp", "_ql.tif", "_ql.tiff")), + limit=8, + ) + if not xml_member or not xml_data: + return { + "archive_parse_status": "MISSING_XML", + "archive_member_count_scanned": len(members), + "quicklook_members": quicklooks, + } + coverage_polygon, xml_meta = _parse_radar_xml_metadata_bytes(xml_data) + return { + "archive_parse_status": "OK", + "archive_xml_member": xml_member, + "archive_scene_name": _archive_member_scene_name(xml_member, archive_stem), + "quicklook_members": quicklooks, + "coverage_polygon": coverage_polygon, + **xml_meta, + } + + def _parse_s1_eof_header(path: str) -> Dict[str, Any]: try: root = etree.parse(path, parser=_xml_parser()).getroot() @@ -443,6 +757,7 @@ def _parse_source_entry(path: str, root: ManagedRootORM) -> Optional[Dict[str, A lower_name = name.lower() stat = _stat_path(path) now = _utcnow() + name_stem = _strip_known_suffix(name) if lower_name.endswith(".zip") and name.upper().startswith("S1"): name_meta = _parse_s1_source_name(name) @@ -459,6 +774,55 @@ def _parse_source_entry(path: str, root: ManagedRootORM) -> Optional[Dict[str, A manifest_meta = {"manifest_parse_status": "FAILED", "manifest_parse_error": str(exc)} return _build_s1_source_asset(path, root, name_meta, manifest_meta, stat, parse_status, parse_error, now) + if _has_archive_suffix(name, _LT1_ARCHIVE_EXTS) and name_stem.upper().startswith("LT1"): + parsed = parse_lt1_radar_filename(name_stem) + if not parsed: + return None + parse_status = "OK" + parse_error = None + archive_meta: Dict[str, Any] = {} + try: + archive_meta = _parse_lt1_archive_metadata(path) + if archive_meta.get("archive_parse_status") != "OK": + parse_status = "PARTIAL" + parse_error = str(archive_meta.get("archive_parse_status") or "archive metadata incomplete") + except Exception as exc: + parse_status = "PARTIAL" + parse_error = str(exc) + archive_meta = {"archive_parse_status": "FAILED", "archive_parse_error": str(exc)} + return _build_lt1_source_asset( + path, + root, + parsed, + archive_meta, + archive_meta.get("coverage_polygon"), + stat, + now, + source_format="LT1_ARCHIVE", + archive_path=path, + parser_name="lt1_archive_metadata", + parse_status=parse_status, + parse_error=parse_error, + ) + + if _has_archive_suffix(name, _GF3_ARCHIVE_EXTS) and name_stem.upper().startswith("GF3"): + parsed = parse_gf3_l2_dirname(name_stem) + if not parsed: + return None + parse_status = "OK" + parse_error = None + archive_meta = {} + try: + archive_meta = _parse_gf3_archive_metadata(path) + if archive_meta.get("archive_parse_status") != "OK": + parse_status = "PARTIAL" + parse_error = str(archive_meta.get("archive_parse_status") or "archive metadata incomplete") + except Exception as exc: + parse_status = "PARTIAL" + parse_error = str(exc) + archive_meta = {"archive_parse_status": "FAILED", "archive_parse_error": str(exc)} + return _build_gf3_archive_asset(path, root, parsed, archive_meta, stat, parse_status, parse_error, now) + if lower_name.endswith(".safe") and os.path.isdir(path) and name.upper().startswith("S1"): name_meta = _parse_s1_source_name(name) if not name_meta: @@ -569,6 +933,12 @@ def _build_lt1_source_asset( coverage_polygon: Optional[List[Tuple[float, float]]], stat: Dict[str, Optional[float]], now: datetime, + *, + source_format: str = "LT1_DIR", + archive_path: Optional[str] = None, + parser_name: str = "lt1_source_directory", + parse_status: str = "OK", + parse_error: Optional[str] = None, ) -> Dict[str, Any]: metadata = dict(parsed) metadata.update({key: value for key, value in xml_meta.items() if value not in (None, "")}) @@ -582,35 +952,99 @@ def _build_lt1_source_asset( return { "asset_uid": _asset_uid("source", path), - "logical_product_uid": os.path.basename(path), + "logical_product_uid": _strip_known_suffix(os.path.basename(path)), "satellite_family": normalize_satellite_family(satellite), "satellite": satellite, - "source_format": "LT1_DIR", - "product_type": parsed.get("product_type"), - "product_level": parsed.get("product_level"), - "imaging_mode": parsed.get("imaging_mode"), - "polarization": parsed.get("polarization"), - "absolute_orbit": parsed.get("orbit_circle"), - "relative_orbit": None, + "source_format": source_format, + "product_type": xml_meta.get("product_type") or parsed.get("product_type"), + "product_level": xml_meta.get("product_level") or parsed.get("product_level"), + "imaging_mode": xml_meta.get("imaging_mode") or parsed.get("imaging_mode"), + "polarization": xml_meta.get("polarization") or parsed.get("polarization"), + "absolute_orbit": xml_meta.get("orbit_circle") or parsed.get("orbit_circle"), + "relative_orbit": xml_meta.get("relative_orbit"), "orbit_direction": xml_meta.get("orbit_direction") or parsed.get("orbit_direction"), - "acquisition_start_time_utc": None, - "acquisition_stop_time_utc": None, + "acquisition_start_time_utc": _parse_datetime_token(xml_meta.get("acquisition_time_utc")), + "acquisition_stop_time_utc": _parse_datetime_token(xml_meta.get("acquisition_stop_time_utc")), "imaging_date": imaging_date, "root_ref_id": root.id, "root_path": root.path, "file_path": path, - "archive_path": None, + "archive_path": archive_path, "path_kind": _path_kind(path), "file_name": os.path.basename(path), - "file_stem": os.path.basename(path), - "file_ext": "", + "file_stem": _strip_known_suffix(os.path.basename(path)), + "file_ext": os.path.splitext(path)[1].lower(), "size_bytes": stat.get("size_bytes"), "mtime_epoch": stat.get("mtime_epoch"), "checksum_status": "NOT_COMPUTED", - "parser_name": "lt1_source_directory", + "parser_name": parser_name, "parser_version": PARSER_VERSION, - "parse_status": "OK", - "parse_error": None, + "parse_status": parse_status, + "parse_error": parse_error, + "parsed_at": now, + "metadata_json": _json_safe(metadata), + "is_active": True, + "missing_since": None, + "updated_at": now, + } + + +def _build_gf3_archive_asset( + path: str, + root: ManagedRootORM, + parsed: Dict[str, Any], + archive_meta: Dict[str, Any], + stat: Dict[str, Optional[float]], + parse_status: str, + parse_error: Optional[str], + now: datetime, +) -> Dict[str, Any]: + coverage_polygon = archive_meta.get("coverage_polygon") + metadata = dict(parsed) + metadata.update({key: value for key, value in archive_meta.items() if value not in (None, "")}) + metadata["coverage_polygon"] = coverage_polygon + metadata["coverage_bbox"] = _bbox_from_polygon(coverage_polygon) + centroid_lon, centroid_lat = _centroid_from_polygon(coverage_polygon) + metadata["scene_center_lon"] = parsed.get("scene_center_lon") if parsed.get("scene_center_lon") is not None else centroid_lon + metadata["scene_center_lat"] = parsed.get("scene_center_lat") if parsed.get("scene_center_lat") is not None else centroid_lat + start_time = _parse_datetime_token(archive_meta.get("acquisition_time_utc")) + stop_time = _parse_datetime_token(archive_meta.get("acquisition_stop_time_utc")) + imaging_date = parsed.get("imaging_date") + if not imaging_date and start_time: + imaging_date = start_time.strftime("%Y%m%d") + stem = _strip_known_suffix(os.path.basename(path)) + + return { + "asset_uid": _asset_uid("source", path), + "logical_product_uid": archive_meta.get("product_unique_id") or stem, + "satellite_family": "GF3", + "satellite": "GF3", + "source_format": "GF3_ARCHIVE", + "product_type": archive_meta.get("product_type") or parsed.get("product_type") or "L1A", + "product_level": archive_meta.get("product_level") or parsed.get("product_level") or "L1A", + "imaging_mode": archive_meta.get("imaging_mode") or parsed.get("imaging_mode"), + "polarization": archive_meta.get("polarization") or parsed.get("polarization"), + "absolute_orbit": archive_meta.get("orbit_circle") or parsed.get("orbit_circle"), + "relative_orbit": archive_meta.get("relative_orbit"), + "orbit_direction": archive_meta.get("orbit_direction") or parsed.get("orbit_direction"), + "acquisition_start_time_utc": start_time, + "acquisition_stop_time_utc": stop_time, + "imaging_date": imaging_date, + "root_ref_id": root.id, + "root_path": root.path, + "file_path": path, + "archive_path": path, + "path_kind": _path_kind(path), + "file_name": os.path.basename(path), + "file_stem": stem, + "file_ext": os.path.splitext(path)[1].lower(), + "size_bytes": stat.get("size_bytes"), + "mtime_epoch": stat.get("mtime_epoch"), + "checksum_status": "NOT_COMPUTED", + "parser_name": "gf3_archive_metadata", + "parser_version": PARSER_VERSION, + "parse_status": parse_status, + "parse_error": parse_error, "parsed_at": now, "metadata_json": _json_safe(metadata), "is_active": True, @@ -732,6 +1166,9 @@ def _iter_source_candidates(root_path: str) -> Iterable[str]: yield _normalize_path(entry.path) continue stack.append(entry.path) + elif entry.is_file(follow_symlinks=False): + if entry.name.upper().startswith("S1") and entry.name.lower().endswith(".zip"): + yield _normalize_path(entry.path) except OSError: continue except OSError: @@ -749,8 +1186,17 @@ def _iter_s1_zip_candidates(root_path: str) -> Iterable[str]: if entry.is_dir(follow_symlinks=False): stack.append(entry.path) elif entry.is_file(follow_symlinks=False): - if entry.name.upper().startswith("S1") and entry.name.lower().endswith(".zip"): + name_upper = entry.name.upper() + stem_upper = _strip_known_suffix(entry.name).upper() + if name_upper.startswith("S1") and entry.name.lower().endswith(".zip"): yield _normalize_path(entry.path) + continue + if stem_upper.startswith("LT1") and _has_archive_suffix(entry.name, _LT1_ARCHIVE_EXTS): + yield _normalize_path(entry.path) + continue + if stem_upper.startswith("GF3") and _has_archive_suffix(entry.name, _GF3_ARCHIVE_EXTS): + yield _normalize_path(entry.path) + continue except OSError: continue except OSError: @@ -860,6 +1306,13 @@ def _insar_source_ready(row: Dict[str, Any], coverage_polygon: Optional[List[Tup return True, None +def _image_data_format_for_source(row: Dict[str, Any]) -> str: + source_format = str(row.get("source_format") or "").upper() + if source_format in {"S1_ZIP", "LT1_ARCHIVE", "GF3_ARCHIVE"}: + return "ARCHIVE" + return "DIRECTORY" + + class AssetInventoryService: async def _progress(self, task_id: Optional[str], message: str, progress: int) -> None: if not task_id: @@ -876,7 +1329,12 @@ class AssetInventoryService: type_set = {str(item or "").strip().lower() for item in (inventory_types or []) if str(item or "").strip()} roles: List[str] = [] if not type_set or "source_product" in type_set or "source" in type_set: - roles.append("source_product_pool") + roles.extend( + [ + "source_product_pool", + "source_pool_gf3_archive", + ] + ) if not type_set or "orbit_asset" in type_set or "orbit" in type_set: roles.append("orbit_asset_pool") stmt = ( @@ -921,7 +1379,7 @@ class AssetInventoryService: for index, root in enumerate(roots, start=1): progress = 5 + int((index - 1) / max(1, total_roots) * 75) await self._progress(task_id, f"Scanning {root.display_name}: {root.path}", progress) - if root.root_role == "source_product_pool": + if root.root_role in {"source_product_pool", "source_pool_gf3_archive"}: result = await self.scan_source_root(db, root) totals["source_roots"] += 1 totals["source_assets"] += int(result.get("asset_count") or 0) @@ -1453,6 +1911,39 @@ class AssetInventoryService: "member_count": len(names), } + def materialize_source_asset( + self, + asset: SourceProductAssetORM, + *, + target_root: Optional[str] = None, + overwrite: bool = False, + ) -> Dict[str, Any]: + source_format = str(asset.source_format or "").upper() + source_path = _normalize_path(str(asset.archive_path or asset.file_path or "")) + if not source_path: + raise ValueError("Source asset path is empty.") + if source_format == "S1_ZIP": + return self.unpack_sentinel1_archive(source_path, target_root=target_root, overwrite=overwrite) + if source_format not in {"LT1_ARCHIVE", "GF3_ARCHIVE"}: + if os.path.isdir(source_path): + return { + "status": "DIRECTORY_READY", + "source_path": source_path, + "target_dir": source_path, + "extracted": False, + "source_format": source_format, + } + raise ValueError(f"Source format is not materializable from archive: {source_format}") + + requested_root = _normalize_path(target_root or "") + if not requested_root: + requested_root = _normalize_path(os.path.join(settings.PYINT_WORK_ROOT, "source_materialized", source_format.lower())) + scene_name = _strip_known_suffix(os.path.basename(source_path)) + target_dir = os.path.join(requested_root, scene_name) + result = _extract_archive_to_dir(source_path, target_dir, overwrite=overwrite) + result["source_format"] = source_format + return result + async def run_sentinel1_unpack_task(self, task_id: str, payload: Optional[Dict[str, Any]] = None) -> None: payload = payload if isinstance(payload, dict) else {} asset_id = payload.get("asset_id") @@ -1831,7 +2322,7 @@ class AssetInventoryService: "product_type": row.get("product_type"), "source_product_token": metadata.get("filename_class_token") or metadata.get("source_product_token"), "image_data_type": "COMPLEX", - "image_data_format": "ZIP" if row.get("source_format") == "S1_ZIP" else "DIRECTORY", + "image_data_format": _image_data_format_for_source(row), "product_variant": metadata.get("product_variant"), "product_level": row.get("product_level"), "product_unique_id": row.get("logical_product_uid"), @@ -2149,7 +2640,6 @@ class AssetInventoryService: await db.execute( select(func.count(SourceProductAssetORM.id)).where( SourceProductAssetORM.is_active == True, # noqa: E712 - SourceProductAssetORM.source_format != "S1_ZIP", ) ) ).scalar_one() @@ -2202,7 +2692,6 @@ class AssetInventoryService: filters = [] if not include_inactive: filters.append(SourceProductAssetORM.is_active == True) # noqa: E712 - filters.append(SourceProductAssetORM.source_format != "S1_ZIP") if satellite_family: filters.append(SourceProductAssetORM.satellite_family == satellite_family.upper()) if satellite: diff --git a/docs/INDEX.md b/docs/INDEX.md index 4e13a26..2ec865f 100644 --- a/docs/INDEX.md +++ b/docs/INDEX.md @@ -30,6 +30,8 @@ - [DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md](DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md) D-InSAR 保留 ENVI/SARscape、LandSAR、Gamma/PyINT 三引擎，退出 ISCE2，统一 Task_Pool、结果聚合和中间文件清理的当前设计。 +- [UNC_SOURCE_ARCHIVE_AND_MATERIALIZE_DESIGN_20260615.md](UNC_SOURCE_ARCHIVE_AND_MATERIALIZE_DESIGN_20260615.md) + UNC/SMB 源压缩包管理、包内 XML/manifest 资产化、本地 materialize 和 D-InSAR/SBAS 生产边界。 - [DINSAR_PRODUCTION_CORES_OVERVIEW.md](DINSAR_PRODUCTION_CORES_OVERVIEW.md) 旧版 ENVI/SARscape、ISCE2、Gamma/PyINT D-InSAR 生产核心说明。ISCE2 相关内容仅作历史背景。 diff --git a/docs/UNC_SOURCE_ARCHIVE_AND_MATERIALIZE_DESIGN_20260615.md b/docs/UNC_SOURCE_ARCHIVE_AND_MATERIALIZE_DESIGN_20260615.md new file mode 100644 index 0000000..53a61d3 --- /dev/null +++ b/docs/UNC_SOURCE_ARCHIVE_AND_MATERIALIZE_DESIGN_20260615.md @@ -0,0 +1,278 @@ +# UNC Source Archive and Local Materialize Design + +## Decision + +UNC/SMB storage is treated as the source archive pool. Production engines should not use UNC paths as their working input. D-InSAR, SBAS, Gamma/PyINT, LandSAR, and SARscape should consume local materialized task inputs. + +This keeps the 20 TB storage useful for long-term source management while protecting production from SMB disconnects, credential scope, WSL path conversion, and external engine UNC compatibility. + +## Current Implementation + +- Source asset inventory now recognizes archive assets: + - `S1_ZIP` + - `LT1_ARCHIVE` + - `GF3_ARCHIVE` +- Sentinel-1 ZIP manifest parsing already reads `manifest.safe` directly from the ZIP. +- LT-1 archive parsing reads `*.meta.xml` directly from `.zip`, `.tar`, `.tar.gz`, or `.tgz` and records contained TIFF members. +- GF3 archive parsing reads the first XML member directly from `.zip`, `.tar`, `.tar.gz`, or `.tgz` and records quicklook-like members when present. +- `GF3_ARCHIVE_SOURCE_DIRS` roots are included in asset inventory scans as source pools. +- Source asset listing and inventory counts now include archive assets instead of hiding `S1_ZIP`. +- A generic source materialize endpoint exists: + - `POST /api/assets/sources/{asset_id}/materialize` + - `S1_ZIP` uses the existing Sentinel-1 SAFE unpacker. + - `LT1_ARCHIVE` and `GF3_ARCHIVE` extract to a local materialized directory. + - Directory assets return `DIRECTORY_READY`. + +Default local materialize root is: + +```text +\source_materialized\ +``` + +Callers may pass `target_root` to force a D-InSAR Task_Pool or SBAS run-specific input directory. + +## Production Boundary + +D-InSAR and SBAS should store source asset references in task/run manifests, then materialize selected inputs into the run directory before engine execution. + +Required next integration points: + +- D-InSAR Task_Pool publishing: + - store `source_product_asset_id`, `archive_path`, `source_format`; + - materialize master/slave archive assets into the task directory before engine dispatch. +- Gamma/PyINT: + - always consume local materialized paths because WSL conversion rejects or cannot reliably map UNC paths. +- LandSAR and ENVI/SARscape: + - prefer local materialized paths even when Windows can see UNC, to avoid external engine path and credential issues. +- SBAS: + - stack discovery can use archive metadata; + - selected scenes must be materialized into the SBAS `RAW`/input structure before Gamma commands such as `par_LT1_SLC`. + +## GF3 Management + +GF3 has two asset layers: + +- `GF3_ARCHIVE`: original source archive, suitable for UNC source management and migration tracking. +- GF3 SARscape standardized L2: production result/analysis-ready layer, used for map footprint, preview, radar data management, and water extraction. + +Do not replace standardized L2 management with raw archive management. Archive assets should link migration and production status; previews and water extraction should continue to consume standardized L2/analysis-ready products. + +## Migration Guidance + +1. Register UNC roots first and scan inventory. +2. Verify archive asset counts and parse status. +3. Keep existing local standardized results and D-InSAR/SBAS products in place. +4. Move source archives to UNC and update root configuration. +5. Only after inventory and materialize tests pass, switch D-InSAR/SBAS publishing to archive asset references. + +Production safety rule: if a run cannot materialize every selected source asset locally, the run must fail before invoking the engine. + +## Recommended UNC Layout + +The current deployment uses two SMB shares: + +```text +\\DESKTOP-N16HJ84\InSAR_Storage_1 +\\DESKTOP-N16HJ84\InSAR_Storage_2 +``` + +Recommended source archive layout: + +```text +\\DESKTOP-N16HJ84\InSAR_Storage_1 + └─ GaoFen-3 + ├─ 20260513 + │ └─ GF3_*.tar.gz + └─ 20260514 + +\\DESKTOP-N16HJ84\InSAR_Storage_2 + ├─ LuTan-1 + │ └─ Archive + │ ├─ 20260513 + │ │ └─ LT1*.tar.gz / LT1*.tgz / LT1*.zip / LT1*.tar + │ └─ 20260514 + ├─ Sentinel-1 + │ └─ Archive + │ ├─ 20260513 + │ │ └─ S1*.zip + │ └─ 20260514 + └─ Orbit + ├─ LuTan-1 + │ ├─ LT1A_GpsData_GAS_C_YYYYMMDD.txt + │ └─ LT1B_GpsData_GAS_C_YYYYMMDD.txt + └─ Sentinel-1 + └─ S1*.EOF +``` + +Date folders are optional for the scanner because source and orbit inventory recurse through configured roots. They are recommended for operator readability and migration checks. + +## Current Local Configuration Example + +The local `.env` should keep legacy local roots and UNC roots side by side during migration: + +```text +SOURCE_PRODUCT_DIRS=D:\LuTan1_Image_Pool;D:\Sentinel1_Image_Pool_ZIP;\\DESKTOP-N16HJ84\InSAR_Storage_2\LuTan-1\Archive;\\DESKTOP-N16HJ84\InSAR_Storage_2\Sentinel-1\Archive +ORBIT_SOURCE_DIRS=D:\LT1_data_lsarorbit;D:\Sentinel1_EOF_Pool;\\DESKTOP-N16HJ84\InSAR_Storage_2\Orbit\LuTan-1;\\DESKTOP-N16HJ84\InSAR_Storage_2\Orbit\Sentinel-1 +GF3_ARCHIVE_SOURCE_DIRS=\\DESKTOP-N16HJ84\InSAR_Storage_1\GaoFen-3 +``` + +Do not store SMB credentials in `.env`. Credentials should be stored in Windows Credential Manager for the account that runs the backend/worker service. + +## Orbit Pool Contract + +There are two different orbit concepts: + +- `ORBIT_SOURCE_DIRS`: source inventory roots. These can be UNC and may be date-organized or flat. +- `ORBIT_POOL_ENVI` / `PYINT_ORBIT_POOL_TXT`: local production orbit pools. These should remain local disk paths. + +LT-1 local production orbit pool should support both flat and satellite-split layouts: + +```text +D:\orbit_pools\envi + ├─ LT1A + │ └─ LT1A_GpsData_GAS_C_YYYYMMDD.txt + ├─ LT1B + │ └─ LT1B_GpsData_GAS_C_YYYYMMDD.txt + └─ converted + └─ envi +``` + +The `LT1A` and `LT1B` names are satellite names, not product levels. ENVI/Gamma/PyINT/SBAS should use local orbit files copied or synchronized from `ORBIT_SOURCE_DIRS`; they should not be required to read UNC directly. + +Sentinel-1 EOF files can be indexed from UNC. Gamma/PyINT/SBAS execution should stage required EOF files locally with the selected scenes. + +## Migration Phases + +### Phase 1: Source archive migration + +Move or copy source archives only: + +- LT-1 compressed scenes to `\\DESKTOP-N16HJ84\InSAR_Storage_2\LuTan-1\Archive\\`. +- Sentinel-1 ZIP scenes to `\\DESKTOP-N16HJ84\InSAR_Storage_2\Sentinel-1\Archive\\`. +- GF3 raw archives to `\\DESKTOP-N16HJ84\InSAR_Storage_1\GaoFen-3\\`. + +Keep current local unpacked scene directories in place until D-InSAR and SBAS archive materialization have been tested. + +### Phase 2: Orbit source migration + +Copy orbit source files to UNC: + +- LT-1 TXT files to `\\DESKTOP-N16HJ84\InSAR_Storage_2\Orbit\LuTan-1\`. +- Sentinel-1 EOF files to `\\DESKTOP-N16HJ84\InSAR_Storage_2\Orbit\Sentinel-1\`. + +Keep `ORBIT_POOL_ENVI` and `PYINT_ORBIT_POOL_TXT` local. Add a later sync/materialize step to populate local orbit pools from the indexed UNC source assets. + +### Phase 3: Production cutover + +After inventory scan verifies UNC assets: + +1. D-InSAR Task_Pool stores source asset IDs and archive paths. +2. Task preparation materializes master/slave scenes and orbit files locally. +3. Engines run only against local Task_Pool paths. +4. Results register normally. +5. Local materialized inputs and intermediate products are eligible for cleanup after result registration. + +### Phase 4: Retire old local source pools + +Only after repeated D-InSAR/SBAS runs succeed from archive materialization: + +- remove old local source roots from `SOURCE_PRODUCT_DIRS`; +- keep local work/result roots; +- keep standardized GF3 L2 products unless explicitly migrated and revalidated. + +## Local Cleanup Design + +After source archives are managed on UNC and production results are registered as assets, local disk can be treated as a cache/work area. Cleanup should be explicit and asset-aware. + +### Keep Classes + +Cleanup must never delete: + +- configured UNC source archive roots; +- local or UNC orbit source roots; +- registered D-InSAR result assets; +- registered SBAS result assets; +- registered GF3 standardized L2 assets; +- `SAR_ANALYSIS_READY_ROOT` products and water extraction result assets; +- current pointers, manifests, previews, and catalog metadata needed to open results. + +### Cleanup Classes + +Cleanup may delete only these local classes after verification: + +- materialized source inputs under `source_materialized`; +- D-InSAR Task_Pool copied inputs after every required engine run is registered; +- D-InSAR engine intermediate folders not listed in the result manifest; +- Gamma/PyINT temporary project work directories after result registration; +- SBAS `RAW`, `SLC`, `RSLC`, `MLI`, `DIFF`, `DIFF1`, script logs, and temporary staging after SBAS product registration; +- GF3 SARscape native intermediates only after standardized L2 registration and optional native-retention policy allows cleanup. + +### Safety Contract + +Every cleanup operation should run in two phases: + +1. `preview`: enumerate candidate paths, classify each path, show size, last modified time, owning task/run/product, and keep/delete reason. +2. `execute`: delete only candidates from a persisted preview token or exact candidate list. + +Deletion must require: + +- path is inside an approved local work root; +- path is not inside any configured source archive root; +- path is not inside a result publish root unless the exact file is classified as intermediate; +- associated result or standardized asset is registered; +- candidate is older than a configurable minimum age; +- no active task references the path. + +### Proposed API + +```text +POST /api/maintenance/cleanup/preview +POST /api/maintenance/cleanup/execute +``` + +Preview request fields: + +```json +{ + "scope": "dinsar|sbas|gf3|materialized|all", + "root_ids": [], + "older_than_hours": 24, + "require_registered_result": true, + "include_task_pool_inputs": false +} +``` + +Preview response should include: + +```json +{ + "preview_id": "...", + "total_bytes": 0, + "candidates": [ + { + "path": "D:\\production_runtime\\...", + "class": "materialized_source", + "owner": "task/run/product id", + "size_bytes": 0, + "eligible": true, + "reason": "registered_result_exists" + } + ], + "blocked": [] +} +``` + +### Recommended Defaults + +- `materialized`: delete after 24 hours if no active task references it. +- `dinsar`: delete engine intermediates after result registration; keep Task_Pool inputs until all selected engines are complete or user opts in. +- `sbas`: delete heavy Gamma working directories after SBAS catalog registration and product assets exist. +- `gf3`: keep standardized L2; clean SARscape native only when `GF3_SARSCAPE_CLEAN_AFTER_SUCCESS=true` and standardized registration is confirmed. + +### Implementation Order + +1. Add read-only cleanup preview service. +2. Add path classification and approved-root checks. +3. Add execute endpoint with preview token. +4. Add frontend maintenance panel. +5. Wire D-InSAR/SBAS/GF3 run pages to show cleanup eligibility after successful registration. diff --git a/frontend/src/api/assets.js b/frontend/src/api/assets.js index 2fa9b4e..882229d 100644 --- a/frontend/src/api/assets.js +++ b/frontend/src/api/assets.js @@ -18,5 +18,8 @@ export const listAssetIssues = (params = {}) => export const unpackSentinel1Source = (assetId, payload = {}) => apiClient.post(`/assets/sources/${assetId}/unpack-sentinel1`, payload).then(r => r.data); +export const materializeSourceAsset = (assetId, payload = {}) => + apiClient.post(`/assets/sources/${assetId}/materialize`, payload).then(r => r.data); + export const unpackSentinel1Batch = (payload = {}) => apiClient.post('/assets/inventory/unpack-sentinel1', payload).then(r => r.data);