diff --git a/.env.example b/.env.example index 8ffaf9d..800d337 100644 --- a/.env.example +++ b/.env.example @@ -59,15 +59,18 @@ ALLOWED_EXPORT_DIRS= # 源数据目录 # ----------------------------------------------------------------------------- UNPACK_SOURCE_DIRS=D:\Archives -SOURCE_PRODUCT_DIRS=D:\LuTan1_Image_Pool;D:\Sentinel1_Image_Pool_ZIP +TASK_POOL_ROOT=D:\Task_Pool +DINSAR_TASK_POOL_ROOT=D:\Task_Pool\DInSAR +SBAS_TASK_POOL_ROOT=D:\Task_Pool\SBAS +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 SENTINEL1_STORAGE_DIRS=D:\Sentinel1_Image_Pool INSAR_STORAGE_DIRS=D:\LuTan1_Image_Pool MONITOR_RADAR_DIRS=D:\LuTan1_Image_Pool MONITOR_DINSAR_DIRS=D:\DInSARResult -ORBIT_SOURCE_DIRS=D:\LT1_data_lsarorbit;D:\Sentinel1_EOF_Pool +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 MONITOR_ORBIT_DIR=D:\LT1_data_lsarorbit -GF3_ARCHIVE_SOURCE_DIRS=D:\production_inputs\gf3\archives +GF3_ARCHIVE_SOURCE_DIRS=\\DESKTOP-N16HJ84\InSAR_Storage_1\GaoFen-3 GF3_ARCHIVE_EXTS=.zip,.tar,.tar.gz,.tgz GF3_UNPACK_DELETE_ARCHIVE=true GF3_LEGACY_GDAL_ENABLED=false @@ -275,7 +278,7 @@ GAMMA_SBAS_RUNTIME_ID=gamma_sbas_runtime_v1 GAMMA_SBAS_WSL_DISTRO=Ubuntu-24.04 GAMMA_SBAS_PYTHON=/home/administrator/miniconda3/envs/insar_wsl_v1/bin/python GAMMA_SBAS_ENV_SCRIPT=D:\Code\Insar_management_system_v2\deploy\wsl\profiles\gamma_env.sh -GAMMA_SBAS_WORK_ROOT=D:\production_runtime\sbas_insar_work +GAMMA_SBAS_WORK_ROOT=D:\Task_Pool\SBAS GAMMA_SBAS_PRODUCT_ROOT=D:\production_results\timeseries\sbas GAMMA_SBAS_TRIAL_ROOT=D:\production_runtime\gamma_ipta_trials GAMMA_SBAS_SCRIPT_TEMPLATE_ROOT=D:\Code\Insar_management_system_v2\backend\templates\gamma_sbas diff --git a/backend/app/config.py b/backend/app/config.py index a74376e..d470541 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -93,6 +93,14 @@ def _default_runtime_root(project_root: str) -> str: return os.path.join(normalized_root, "runtime") +def _default_task_pool_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, "Task_Pool") + return os.path.join(normalized_root, "Task_Pool") + + def _default_runtime_dir(project_root: str, *parts: str) -> str: return os.path.join(_default_runtime_root(project_root), *parts) @@ -187,6 +195,9 @@ class Settings(BaseSettings): DB_SCHEMA_RESET_CONFIRM: bool = False UNPACK_SOURCE_DIRS: str = "" + TASK_POOL_ROOT: str = "" + DINSAR_TASK_POOL_ROOT: str = "" + SBAS_TASK_POOL_ROOT: str = "" SOURCE_PRODUCT_DIRS: str = "" SENTINEL1_STORAGE_DIRS: str = "" ORBIT_SOURCE_DIRS: str = "" @@ -441,6 +452,12 @@ class Settings(BaseSettings): "WATER_RESULTS_DIR", os.path.join(backend_dir, "water_results"), ) + if not self.TASK_POOL_ROOT: + object.__setattr__(self, "TASK_POOL_ROOT", _default_task_pool_root(project_root)) + if not self.DINSAR_TASK_POOL_ROOT: + object.__setattr__(self, "DINSAR_TASK_POOL_ROOT", os.path.join(self.TASK_POOL_ROOT, "DInSAR")) + if not self.SBAS_TASK_POOL_ROOT: + object.__setattr__(self, "SBAS_TASK_POOL_ROOT", os.path.join(self.TASK_POOL_ROOT, "SBAS")) if not self.SAR_ANALYSIS_READY_ROOT: object.__setattr__( self, @@ -780,7 +797,7 @@ class Settings(BaseSettings): object.__setattr__( self, "GAMMA_SBAS_WORK_ROOT", - os.path.join(_default_runtime_root(project_root), "sbas_insar_work"), + self.SBAS_TASK_POOL_ROOT, ) if not self.GAMMA_SBAS_PRODUCT_ROOT: object.__setattr__( @@ -990,6 +1007,9 @@ class Settings(BaseSettings): os.makedirs(settings.RESULT_QUARANTINE_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) + os.makedirs(settings.DINSAR_TASK_POOL_ROOT, exist_ok=True) + os.makedirs(settings.SBAS_TASK_POOL_ROOT, exist_ok=True) for path in split_env_paths(settings.GF3_ARCHIVE_SOURCE_DIRS): os.makedirs(path, exist_ok=True) for path in split_env_paths(settings.GF3_SARSCAPE_NATIVE_DIRS): diff --git a/backend/app/copier.py b/backend/app/copier.py index 17b96a8..aa7c70d 100644 --- a/backend/app/copier.py +++ b/backend/app/copier.py @@ -2,6 +2,7 @@ import shutil import asyncio import tempfile +import tarfile import zipfile import json import hashlib @@ -43,6 +44,100 @@ def find_dinsar_source_to_copy(path: str) -> str: return path +_DINSAR_ARCHIVE_SUFFIXES = (".tar.gz", ".tgz", ".zip", ".tar") + + +def _is_supported_archive(path: str) -> bool: + lower = str(path or "").lower() + return any(lower.endswith(suffix) for suffix in _DINSAR_ARCHIVE_SUFFIXES) + + +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, dest_dir: str) -> int: + if zipfile.is_zipfile(archive_path): + extracted = 0 + with zipfile.ZipFile(archive_path) as zip_obj: + for info in zip_obj.infolist(): + rel_name = _safe_archive_member_name(info.filename, archive_path) + dest_path = os.path.abspath(os.path.join(dest_dir, rel_name)) + if not dest_path.startswith(os.path.abspath(dest_dir) + os.sep): + raise ValueError(f"Unsafe ZIP member path: {info.filename}") + if info.is_dir(): + os.makedirs(dest_path, exist_ok=True) + continue + os.makedirs(os.path.dirname(dest_path), exist_ok=True) + with zip_obj.open(info, "r") as source, open(dest_path, "wb") as target: + shutil.copyfileobj(source, target, length=1024 * 1024) + extracted += 1 + return extracted + + if tarfile.is_tarfile(archive_path): + extracted = 0 + with tarfile.open(archive_path, "r:*") as tar_obj: + for member in tar_obj: + rel_name = _safe_archive_member_name(member.name, archive_path) + dest_path = os.path.abspath(os.path.join(dest_dir, rel_name)) + if not dest_path.startswith(os.path.abspath(dest_dir) + os.sep): + raise ValueError(f"Unsafe TAR member path: {member.name}") + if member.isdir(): + os.makedirs(dest_path, exist_ok=True) + continue + if not member.isfile(): + continue + source = tar_obj.extractfile(member) + if source is None: + continue + os.makedirs(os.path.dirname(dest_path), exist_ok=True) + with source, open(dest_path, "wb") as target: + shutil.copyfileobj(source, target, length=1024 * 1024) + extracted += 1 + return extracted + + raise ValueError(f"Unsupported archive format: {archive_path}") + + +def _materialize_dinsar_source(source_path: str, dest_dir: str) -> Dict[str, Any]: + normalized = os.path.normpath(os.path.abspath(str(source_path or ""))) + if not os.path.exists(normalized): + raise FileNotFoundError(normalized) + + if os.path.isdir(normalized): + shutil.copytree(normalized, dest_dir, dirs_exist_ok=True) + return {"mode": "copy_directory", "source_path": normalized} + + if os.path.isfile(normalized) and _is_supported_archive(normalized): + os.makedirs(dest_dir, exist_ok=True) + extracted = _extract_archive_to_dir(normalized, dest_dir) + if extracted <= 0: + raise OSError(f"Archive extraction produced no files: {normalized}") + return { + "mode": "extract_archive", + "source_path": normalized, + "archive_path": normalized, + "extracted_files": extracted, + } + + if os.path.isfile(normalized): + os.makedirs(dest_dir, exist_ok=True) + target = os.path.join(dest_dir, os.path.basename(normalized)) + shutil.copy2(normalized, target) + return { + "mode": "copy_file", + "source_path": normalized, + "relative_path": os.path.basename(target), + } + + raise FileNotFoundError(normalized) + + def _resolve_orbit_dest_path( orbit_dir: str, role: str, @@ -191,6 +286,7 @@ def _build_dinsar_pair_metadata( package_format: str, include_orbit_files: bool, orbit_entries: List[Dict[str, Any]], + source_materialization: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: return { "pair_key": item.get("pair_key"), @@ -214,6 +310,7 @@ def _build_dinsar_pair_metadata( "master_orbit_file_path": item.get("master_orbit_file_path"), "slave_orbit_file_path": item.get("slave_orbit_file_path"), "orbit_files": orbit_entries, + "source_materialization": source_materialization or {}, "scene_pair_uid": item.get("scene_pair_uid") or item.get("pair_uid"), "pair_uid": item.get("pair_uid") or item.get("scene_pair_uid"), "network_run_id": item.get("network_run_id"), @@ -1094,8 +1191,26 @@ async def run_dinsar_copy_items( failed_count += 1 continue - await asyncio.to_thread(shutil.copytree, master_src_path, master_dir, dirs_exist_ok=True) - await asyncio.to_thread(shutil.copytree, slave_src_path, slave_dir, dirs_exist_ok=True) + master_materialization = await asyncio.to_thread( + _materialize_dinsar_source, + master_src_path, + master_dir, + ) + slave_materialization = await asyncio.to_thread( + _materialize_dinsar_source, + slave_src_path, + slave_dir, + ) + source_materialization = { + "master": { + **master_materialization, + "target_relative_path": "master", + }, + "slave": { + **slave_materialization, + "target_relative_path": "slave", + }, + } orbit_entries = await _copy_dinsar_orbit_files( task_id, item, @@ -1112,6 +1227,7 @@ async def run_dinsar_copy_items( "zip" if export_zip else "folder", include_orbit_files, orbit_entries, + source_materialization, ), ) if export_zip and zip_path: diff --git a/backend/app/services/root_registry_service.py b/backend/app/services/root_registry_service.py index 29f1bc9..ee4fe6b 100644 --- a/backend/app/services/root_registry_service.py +++ b/backend/app/services/root_registry_service.py @@ -458,6 +458,35 @@ def _build_root_specs_from_settings() -> List[RootSpec]: owner_engine="pyint", ) ) + specs.extend( + _iter_single_root_specs( + env_var="TASK_POOL_ROOT", + path=settings.TASK_POOL_ROOT, + root_role="task_pool_root", + display_name="Task Pool Root", + scan_mode="workspace", + ) + ) + specs.extend( + _iter_single_root_specs( + env_var="DINSAR_TASK_POOL_ROOT", + path=settings.DINSAR_TASK_POOL_ROOT, + root_role="task_pool_dinsar", + display_name="D-InSAR Task Pool", + scan_mode="workspace", + owner_engine="dinsar", + ) + ) + specs.extend( + _iter_single_root_specs( + env_var="SBAS_TASK_POOL_ROOT", + path=settings.SBAS_TASK_POOL_ROOT, + root_role="task_pool_sbas", + display_name="SBAS Task Pool", + scan_mode="workspace", + owner_engine="sbas", + ) + ) specs.extend( _iter_single_root_specs( env_var="GAMMA_SBAS_WORK_ROOT", diff --git a/docs/DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md b/docs/DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md index c352fb8..2bc82cb 100644 --- a/docs/DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md +++ b/docs/DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md @@ -210,3 +210,22 @@ Task_20250101_20250201 - Task_Pool 根路径需要明确配置项，建议新增 `DINSAR_TASK_POOL_ROOT`，默认 `D:\Task_Pool\DInSAR`。 - 中间文件清理需要先确认每个引擎的“可删目录”和“必须保留资产”清单，不能用一套规则覆盖全部。 - 前端 grouped 结果视图需要兼容旧 flat API 一段时间，避免已有页面一次性断裂。 + +## 2026-06-15 Task_Pool Materialize Update + +Current direction: + +- `TASK_POOL_ROOT` defaults to `D:\Task_Pool`. +- `DINSAR_TASK_POOL_ROOT` defaults to `D:\Task_Pool\DInSAR`. +- `SBAS_TASK_POOL_ROOT` defaults to `D:\Task_Pool\SBAS`. +- D-InSAR distribution materializes source inputs inside the task folder: + - directory sources are copied into `master/` and `slave/`; + - `S1_ZIP`, `LT1_ARCHIVE`, and other supported archives are extracted into `master/` and `slave/`; + - staged orbit files go into `orbit/`. +- Engines must consume local Task_Pool paths, not UNC source archive paths. +- `.dinsar_pair.json` records `source_materialization` so cleanup can distinguish copied directories, extracted archives, and staged files. + +Cleanup implication: + +- `master/`, `slave/`, `orbit/`, and engine `work/` folders are local materialized inputs/workspace and may be cleaned after all required results are registered. +- `publish/`, manifests, result assets, previews, and catalog metadata are preserved. diff --git a/docs/SBAS_INSAR_CURRENT_WORKFLOW.md b/docs/SBAS_INSAR_CURRENT_WORKFLOW.md index 6608cc3..6b1d222 100644 --- a/docs/SBAS_INSAR_CURRENT_WORKFLOW.md +++ b/docs/SBAS_INSAR_CURRENT_WORKFLOW.md @@ -252,3 +252,28 @@ GET /api/sbas-insar-products/{product_id}/assets/{asset_id} - 选中候选可生成 Stack Manifest； - 结果发布支持 GeoTIFF、预览图、监测点曲线和点矢量下载； - 前端生产页和结果页构建通过。 + +## 2026-06-15 SBAS Task_Pool Update + +Recommended SBAS work root: + +```text +SBAS_TASK_POOL_ROOT=D:\Task_Pool\SBAS +GAMMA_SBAS_WORK_ROOT=D:\Task_Pool\SBAS +``` + +SBAS has its own Task_Pool sub-root and must not reuse `D:\Task_Pool\DInSAR`. + +Recommended run layout: + +```text +D:\Task_Pool\SBAS\ + ├─ task_manifest.json + ├─ sbas_stack_manifest.json + ├─ sources + ├─ orbits + ├─ work + └─ publish +``` + +Source archives remain on UNC. Selected scenes and orbit files are materialized under the SBAS task directory before Gamma/LandSAR execution. Cleanup may remove `sources`, `orbits`, and `work` after result registration, but must preserve manifests, `publish`, previews, and catalog assets. diff --git a/docs/UNC_SOURCE_ARCHIVE_AND_MATERIALIZE_DESIGN_20260615.md b/docs/UNC_SOURCE_ARCHIVE_AND_MATERIALIZE_DESIGN_20260615.md index 53a61d3..8958d60 100644 --- a/docs/UNC_SOURCE_ARCHIVE_AND_MATERIALIZE_DESIGN_20260615.md +++ b/docs/UNC_SOURCE_ARCHIVE_AND_MATERIALIZE_DESIGN_20260615.md @@ -23,13 +23,15 @@ This keeps the 20 TB storage useful for long-term source management while protec - `LT1_ARCHIVE` and `GF3_ARCHIVE` extract to a local materialized directory. - Directory assets return `DIRECTORY_READY`. -Default local materialize root is: +Default source materialization is task-scoped. D-InSAR and SBAS callers should pass a Task_Pool target directory: ```text -\source_materialized\ +D:\Task_Pool\DInSAR\\master +D:\Task_Pool\DInSAR\\slave +D:\Task_Pool\SBAS\\sources\ ``` -Callers may pass `target_root` to force a D-InSAR Task_Pool or SBAS run-specific input directory. +The generic materialize endpoint still accepts `target_root` for ad hoc checks. Production callers must provide a Task_Pool destination. ## Production Boundary @@ -104,6 +106,29 @@ Recommended source archive layout: └─ S1*.EOF ``` +Recommended local Task_Pool layout: + +```text +D:\Task_Pool + ├─ DInSAR + │ └─ + │ ├─ task_manifest.json + │ ├─ .dinsar_pair.json + │ ├─ master + │ ├─ slave + │ ├─ orbit + │ ├─ work + │ └─ publish + └─ SBAS + └─ + ├─ task_manifest.json + ├─ sbas_stack_manifest.json + ├─ sources + ├─ orbits + ├─ work + └─ publish +``` + 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 @@ -114,6 +139,10 @@ The local `.env` should keep legacy local roots and UNC roots side by side durin 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 +TASK_POOL_ROOT=D:\Task_Pool +DINSAR_TASK_POOL_ROOT=D:\Task_Pool\DInSAR +SBAS_TASK_POOL_ROOT=D:\Task_Pool\SBAS +GAMMA_SBAS_WORK_ROOT=D:\Task_Pool\SBAS ``` Do not store SMB credentials in `.env`. Credentials should be stored in Windows Credential Manager for the account that runs the backend/worker service. @@ -167,7 +196,7 @@ Keep `ORBIT_POOL_ENVI` and `PYINT_ORBIT_POOL_TXT` local. Add a later sync/materi 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. +2. Task preparation materializes master/slave scenes and orbit files under `D:\Task_Pool\DInSAR\`. 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.