diff --git a/backend/app/copier.py b/backend/app/copier.py index 225eb1d..17b96a8 100644 --- a/backend/app/copier.py +++ b/backend/app/copier.py @@ -3,6 +3,8 @@ import shutil import asyncio import tempfile import zipfile +import json +import hashlib from datetime import datetime from typing import List, Tuple, Optional, Dict, Any @@ -158,6 +160,724 @@ def _zip_task_directory(task_dir: str, zip_path: str) -> None: pass +def _directory_has_entries(path: str) -> bool: + if not os.path.isdir(path): + return False + try: + with os.scandir(path) as entries: + return any(True for _ in entries) + except OSError: + return False + + +def _is_existing_dinsar_folder_complete(task_dir: str) -> bool: + return ( + _directory_has_entries(os.path.join(task_dir, "master")) + and _directory_has_entries(os.path.join(task_dir, "slave")) + ) + + +def _is_existing_dinsar_zip_complete(zip_path: str) -> bool: + try: + return os.path.isfile(zip_path) and os.path.getsize(zip_path) > 0 + except OSError: + return False + + +def _build_dinsar_pair_metadata( + item: Dict[str, Any], + task_name: str, + task_alias: str, + package_format: str, + include_orbit_files: bool, + orbit_entries: List[Dict[str, Any]], +) -> Dict[str, Any]: + return { + "pair_key": item.get("pair_key"), + "task_name": task_name, + "task_alias": task_alias, + "master_path": item.get("master_path"), + "slave_path": item.get("slave_path"), + "master_satellite": item.get("master_satellite"), + "slave_satellite": item.get("slave_satellite"), + "master_imaging_date": item.get("master_imaging_date"), + "slave_imaging_date": item.get("slave_imaging_date"), + "master_imaging_mode": item.get("master_imaging_mode"), + "slave_imaging_mode": item.get("slave_imaging_mode"), + "master_polarization": item.get("master_polarization"), + "slave_polarization": item.get("slave_polarization"), + "time_baseline_days": item.get("time_baseline_days"), + "spatial_baseline_meters": item.get("spatial_baseline_meters"), + "scene_center_distance_meters": item.get("scene_center_distance_meters"), + "package_format": package_format, + "include_orbit_files": bool(include_orbit_files), + "master_orbit_file_path": item.get("master_orbit_file_path"), + "slave_orbit_file_path": item.get("slave_orbit_file_path"), + "orbit_files": orbit_entries, + "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"), + "network_edge_id": item.get("network_edge_id"), + "policy_version": item.get("policy_version"), + "selection_strategy": item.get("selection_strategy"), + "copied_at": datetime.utcnow().isoformat(timespec="seconds") + "Z", + } + + +def _safe_bundle_entry_name(source_path: str, prefix: str) -> str: + base_name = os.path.basename(os.path.normpath(str(source_path or ""))) or "item" + digest = hashlib.sha1(os.path.normcase(os.path.abspath(source_path)).encode("utf-8")).hexdigest()[:10] + return f"{prefix}_{digest}_{base_name}" + + +def _bundle_relative_scene_path(source_path: str) -> str: + normalized = os.path.normpath(os.path.abspath(str(source_path))) + return os.path.join("data", _safe_bundle_entry_name(normalized, "scene")).replace(os.sep, "/") + + +def _normalize_bundle_relative_path(path: Any) -> str: + return str(path or "").strip().replace("\\", "/") + + +def _source_bundle_id_number(value: Any, prefix: str) -> int: + text = str(value or "") + marker = f"{prefix}_" + if not text.startswith(marker): + return 0 + try: + return int(text[len(marker):]) + except ValueError: + return 0 + + +def _next_source_bundle_id(records: List[Dict[str, Any]], id_key: str, prefix: str) -> int: + max_number = 0 + for record in records: + max_number = max(max_number, _source_bundle_id_number(record.get(id_key), prefix)) + return max(max_number, len(records)) + 1 + + +def _source_bundle_pair_keys_from_item(item: Dict[str, Any]) -> List[str]: + keys: List[str] = [] + + def add(label: str, value: Any) -> None: + text = str(value or "").strip() + if text: + keys.append(f"{label}:{text}") + + add("uid", item.get("scene_pair_uid") or item.get("pair_uid")) + add("pair_key", item.get("pair_key")) + network_run_id = str(item.get("network_run_id") or "").strip() + network_edge_id = str(item.get("network_edge_id") or "").strip() + if network_run_id and network_edge_id: + keys.append(f"network:{network_run_id}:{network_edge_id}") + + master_path = item.get("master_path") + slave_path = item.get("slave_path") + if master_path and slave_path: + master_abs = os.path.normcase(os.path.normpath(os.path.abspath(str(master_path)))) + slave_abs = os.path.normcase(os.path.normpath(os.path.abspath(str(slave_path)))) + keys.append(f"source_paths:{master_abs}|{slave_abs}") + keys.append( + "bundle_paths:" + f"{_bundle_relative_scene_path(str(master_path))}|" + f"{_bundle_relative_scene_path(str(slave_path))}" + ) + return keys + + +def _source_bundle_pair_keys_from_pair(pair: Dict[str, Any]) -> List[str]: + keys: List[str] = [] + + identity_key = str(pair.get("identity_key") or "").strip() + if identity_key: + keys.append(identity_key) + + def add(label: str, value: Any) -> None: + text = str(value or "").strip() + if text: + keys.append(f"{label}:{text}") + + add("uid", pair.get("scene_pair_uid") or pair.get("pair_uid")) + add("pair_key", pair.get("pair_key")) + network_run_id = str(pair.get("network_run_id") or "").strip() + network_edge_id = str(pair.get("network_edge_id") or "").strip() + if network_run_id and network_edge_id: + keys.append(f"network:{network_run_id}:{network_edge_id}") + + master_source_path = pair.get("master_source_path") + slave_source_path = pair.get("slave_source_path") + if master_source_path and slave_source_path: + master_abs = os.path.normcase(os.path.normpath(os.path.abspath(str(master_source_path)))) + slave_abs = os.path.normcase(os.path.normpath(os.path.abspath(str(slave_source_path)))) + keys.append(f"source_paths:{master_abs}|{slave_abs}") + + master_data = _normalize_bundle_relative_path(pair.get("master_data")) + slave_data = _normalize_bundle_relative_path(pair.get("slave_data")) + if master_data and slave_data: + keys.append(f"bundle_paths:{master_data}|{slave_data}") + + return keys + + +def _read_json_file(path: str) -> Dict[str, Any]: + if not os.path.isfile(path): + return {} + with open(path, "r", encoding="utf-8-sig") as handle: + payload = json.load(handle) + if not isinstance(payload, dict): + return {} + return payload + + +def _copy_source_into_bundle(source_path: str, dest_path: str, skip_existing: bool) -> str: + if skip_existing and os.path.exists(dest_path): + return "skipped" + if os.path.exists(dest_path): + if os.path.isdir(dest_path): + shutil.rmtree(dest_path) + else: + os.remove(dest_path) + parent_dir = os.path.dirname(dest_path) + os.makedirs(parent_dir, exist_ok=True) + if os.path.isdir(source_path): + shutil.copytree(source_path, dest_path) + else: + shutil.copy2(source_path, dest_path) + return "copied" + + +def _write_json_file(path: str, payload: Dict[str, Any]) -> None: + os.makedirs(os.path.dirname(path), exist_ok=True) + temp_path = f"{path}.tmp" + try: + with open(temp_path, "w", encoding="utf-8") as handle: + json.dump(payload, handle, ensure_ascii=False, indent=2) + os.replace(temp_path, path) + finally: + if os.path.exists(temp_path): + try: + os.remove(temp_path) + except OSError: + pass + + +async def run_dinsar_source_bundle_items( + task_id: str, + items: List[Dict[str, Any]], + dest_dir: str, + *, + include_orbit_files: bool = True, + skip_existing: bool = True, + max_items: Optional[int] = None, +) -> None: + try: + if max_items is not None: + try: + max_items = int(max_items) + except (TypeError, ValueError): + max_items = None + if max_items is not None and max_items <= 0: + max_items = None + + await task_service.start_task(task_id, message="Starting D-InSAR source bundle export...") + await _log_and_update(task_id, f"D-InSAR source bundle export started. Dest: {dest_dir}") + await _log_and_update( + task_id, + ( + "D-InSAR source bundle options: " + f"include_orbit_files={include_orbit_files}, " + f"skip_existing={skip_existing}, " + f"max_items={max_items if max_items is not None else 'unlimited'}" + ), + ) + + await asyncio.to_thread(os.makedirs, dest_dir, exist_ok=True) + data_dir = os.path.join(dest_dir, "data") + orbit_dir = os.path.join(dest_dir, "orbit") + await asyncio.to_thread(os.makedirs, data_dir, exist_ok=True) + if include_orbit_files: + await asyncio.to_thread(os.makedirs, orbit_dir, exist_ok=True) + + pairs_path = os.path.join(dest_dir, "pairs.json") + manifest_path = os.path.join(dest_dir, "manifest.json") + existing_pairs_payload = await asyncio.to_thread(_read_json_file, pairs_path) + existing_manifest_payload = await asyncio.to_thread(_read_json_file, manifest_path) + + raw_existing_pairs = existing_pairs_payload.get("pairs") + existing_pairs: List[Dict[str, Any]] = [ + dict(pair) + for pair in raw_existing_pairs + if isinstance(pair, dict) + ] if isinstance(raw_existing_pairs, list) else [] + + raw_existing_scenes = existing_manifest_payload.get("scenes") + existing_scene_list: List[Dict[str, Any]] = [ + dict(scene) + for scene in raw_existing_scenes + if isinstance(scene, dict) + ] if isinstance(raw_existing_scenes, list) else [] + + raw_existing_orbits = existing_manifest_payload.get("orbits") + existing_orbit_list: List[Dict[str, Any]] = [ + dict(orbit) + for orbit in raw_existing_orbits + if isinstance(orbit, dict) + ] if isinstance(raw_existing_orbits, list) else [] + + existing_pair_keys = { + key + for pair in existing_pairs + for key in _source_bundle_pair_keys_from_pair(pair) + } + if existing_pairs: + await _log_and_update(task_id, f"Existing source bundle pairs found: {len(existing_pairs)}") + + candidate_tasks: List[Dict[str, Any]] = [] + for item in items: + task_name = item.get("task_name") or item.get("task_alias") or "task" + task_alias = item.get("task_alias") or task_name + master_path = item.get("master_path") + slave_path = item.get("slave_path") + if not master_path or not slave_path: + continue + candidate_tasks.append( + { + **item, + "task_name": task_name, + "task_alias": task_alias, + "master_path": master_path, + "slave_path": slave_path, + } + ) + + skipped_existing_pair_count = 0 + exportable_tasks: List[Dict[str, Any]] = [] + for item in candidate_tasks: + pair_keys = _source_bundle_pair_keys_from_item(item) + item["identity_key"] = pair_keys[0] if pair_keys else None + if existing_pair_keys and any(key in existing_pair_keys for key in pair_keys): + skipped_existing_pair_count += 1 + continue + exportable_tasks.append(item) + + if max_items is not None: + deferred_count = max(0, len(exportable_tasks) - max_items) + tasks = exportable_tasks[:max_items] + else: + deferred_count = 0 + tasks = exportable_tasks + + total_pairs = len(tasks) + await _log_and_update( + task_id, + ( + f"Found {len(candidate_tasks)} candidate pairs; " + f"{skipped_existing_pair_count} already exported; " + f"{total_pairs} new pairs selected for this run." + ), + ) + if total_pairs == 0: + await task_service.update_task( + task_id, + status="COMPLETED", + message=( + "No new source bundle pairs to export. " + f"Existing pairs: {len(existing_pairs)}; " + f"already exported in request: {skipped_existing_pair_count}." + ), + progress=100, + ) + return + + scene_records: List[Dict[str, Any]] = [] + scene_entries_by_source: Dict[str, Dict[str, Any]] = {} + scene_entries_by_relative: Dict[str, Dict[str, Any]] = {} + scene_entries_by_id: Dict[str, Dict[str, Any]] = {} + orbit_records: List[Dict[str, Any]] = [] + orbit_entries_by_source: Dict[str, Dict[str, Any]] = {} + orbit_entries_by_relative: Dict[str, Dict[str, Any]] = {} + orbit_entries_by_id: Dict[str, Dict[str, Any]] = {} + next_scene_number = _next_source_bundle_id(existing_scene_list, "scene_id", "scene") + next_orbit_number = _next_source_bundle_id(existing_orbit_list, "orbit_id", "orbit") + next_pair_number = _next_source_bundle_id(existing_pairs, "pair_id", "pair") + copy_units_by_target: Dict[str, Tuple[str, str, str, str]] = {} + pairs: List[Dict[str, Any]] = [] + missing_sources: List[Dict[str, Any]] = [] + + def register_scene_entry(raw_entry: Dict[str, Any]) -> Dict[str, Any]: + nonlocal next_scene_number + entry = dict(raw_entry) + source_path = entry.get("source_path") + source_key = "" + if source_path: + source_path = os.path.normpath(os.path.abspath(str(source_path))) + source_key = os.path.normcase(source_path) + entry["source_path"] = source_path + relative_path = _normalize_bundle_relative_path(entry.get("relative_path")) + if not relative_path and source_path: + relative_path = _bundle_relative_scene_path(source_path) + if relative_path: + entry["relative_path"] = relative_path + + existing = None + if source_key: + existing = scene_entries_by_source.get(source_key) + if existing is None and relative_path: + existing = scene_entries_by_relative.get(relative_path) + scene_id = str(entry.get("scene_id") or "").strip() + if existing is None and scene_id: + existing = scene_entries_by_id.get(scene_id) + if existing is not None: + if source_path and not existing.get("source_path"): + existing["source_path"] = source_path + scene_entries_by_source[source_key] = existing + if relative_path and not existing.get("relative_path"): + existing["relative_path"] = relative_path + scene_entries_by_relative[relative_path] = existing + return existing + + if not scene_id: + scene_id = f"scene_{next_scene_number:04d}" + entry["scene_id"] = scene_id + next_scene_number += 1 + else: + next_scene_number = max(next_scene_number, _source_bundle_id_number(scene_id, "scene") + 1) + + scene_records.append(entry) + scene_entries_by_id[scene_id] = entry + if source_key: + scene_entries_by_source[source_key] = entry + if relative_path: + scene_entries_by_relative[relative_path] = entry + return entry + + def register_orbit_entry(raw_entry: Dict[str, Any]) -> Dict[str, Any]: + nonlocal next_orbit_number + entry = dict(raw_entry) + source_path = entry.get("source_path") + source_key = "" + if source_path: + source_path = os.path.normpath(os.path.abspath(str(source_path))) + source_key = os.path.normcase(source_path) + entry["source_path"] = source_path + relative_path = _normalize_bundle_relative_path(entry.get("relative_path")) + if not relative_path and source_path: + relative_path = os.path.join("orbit", _safe_bundle_entry_name(source_path, "orbit")).replace(os.sep, "/") + if relative_path: + entry["relative_path"] = relative_path + + existing = None + if source_key: + existing = orbit_entries_by_source.get(source_key) + if existing is None and relative_path: + existing = orbit_entries_by_relative.get(relative_path) + orbit_id = str(entry.get("orbit_id") or "").strip() + if existing is None and orbit_id: + existing = orbit_entries_by_id.get(orbit_id) + if existing is not None: + if source_path and not existing.get("source_path"): + existing["source_path"] = source_path + orbit_entries_by_source[source_key] = existing + if relative_path and not existing.get("relative_path"): + existing["relative_path"] = relative_path + orbit_entries_by_relative[relative_path] = existing + return existing + + if not orbit_id: + orbit_id = f"orbit_{next_orbit_number:04d}" + entry["orbit_id"] = orbit_id + next_orbit_number += 1 + else: + next_orbit_number = max(next_orbit_number, _source_bundle_id_number(orbit_id, "orbit") + 1) + if not isinstance(entry.get("used_by"), list): + entry["used_by"] = [] + + orbit_records.append(entry) + orbit_entries_by_id[orbit_id] = entry + if source_key: + orbit_entries_by_source[source_key] = entry + if relative_path: + orbit_entries_by_relative[relative_path] = entry + return entry + + def add_copy_unit(kind: str, unit_id: str, source_path: Optional[str], relative_path: Optional[str]) -> None: + if not source_path or not relative_path: + return + target_path = os.path.join(dest_dir, _normalize_bundle_relative_path(relative_path)) + target_key = os.path.normcase(os.path.abspath(target_path)) + copy_units_by_target.setdefault(target_key, (kind, unit_id, source_path, target_path)) + + def add_orbit_usage(entry: Dict[str, Any], scene_id: str, role: str) -> None: + used_by = entry.setdefault("used_by", []) + usage = {"scene_id": scene_id, "role": role} + if not any( + str(item.get("scene_id")) == scene_id and str(item.get("role")) == role + for item in used_by + if isinstance(item, dict) + ): + used_by.append(usage) + + for scene in existing_scene_list: + register_scene_entry(scene) + + for orbit in existing_orbit_list: + register_orbit_entry(orbit) + + for pair in existing_pairs: + for role in ("master", "slave"): + scene_relative = _normalize_bundle_relative_path(pair.get(f"{role}_data")) + if scene_relative: + register_scene_entry( + { + "scene_id": pair.get(f"{role}_scene_id"), + "source_path": pair.get(f"{role}_source_path"), + "relative_path": scene_relative, + } + ) + orbit_relative = _normalize_bundle_relative_path(pair.get(f"{role}_orbit")) + if orbit_relative: + register_orbit_entry( + { + "orbit_id": pair.get(f"{role}_orbit_id"), + "source_path": pair.get(f"{role}_orbit_source_path"), + "relative_path": orbit_relative, + "used_by": [ + { + "scene_id": pair.get(f"{role}_scene_id"), + "role": role, + } + ] if pair.get(f"{role}_scene_id") else [], + } + ) + + def ensure_scene_entry(scene_path: str, role_item: Dict[str, Any], prefix: str) -> Dict[str, Any]: + source_path = os.path.normpath(os.path.abspath(str(scene_path))) + source_key = os.path.normcase(source_path) + relative_path = _bundle_relative_scene_path(source_path) + existing = scene_entries_by_source.get(source_key) or scene_entries_by_relative.get(relative_path) + if existing: + if not existing.get("source_path"): + existing["source_path"] = source_path + scene_entries_by_source[source_key] = existing + if not existing.get("relative_path"): + existing["relative_path"] = relative_path + scene_entries_by_relative[relative_path] = existing + add_copy_unit("scene", existing["scene_id"], source_path, existing["relative_path"]) + return existing + entry = register_scene_entry( + { + "source_path": source_path, + "relative_path": relative_path, + "satellite": role_item.get("satellite"), + "imaging_date": role_item.get("imaging_date"), + "imaging_mode": role_item.get("imaging_mode"), + "polarization": role_item.get("polarization"), + } + ) + add_copy_unit("scene", entry["scene_id"], source_path, entry["relative_path"]) + return entry + + def ensure_orbit_entry(orbit_path: Optional[str], scene_id: str, role: str) -> Optional[Dict[str, Any]]: + if not include_orbit_files or not orbit_path: + return None + source_path = os.path.normpath(os.path.abspath(str(orbit_path))) + source_key = os.path.normcase(source_path) + relative_path = os.path.join("orbit", _safe_bundle_entry_name(source_path, "orbit")).replace(os.sep, "/") + existing = orbit_entries_by_source.get(source_key) or orbit_entries_by_relative.get(relative_path) + if existing: + if not existing.get("source_path"): + existing["source_path"] = source_path + orbit_entries_by_source[source_key] = existing + if not existing.get("relative_path"): + existing["relative_path"] = relative_path + orbit_entries_by_relative[relative_path] = existing + add_orbit_usage(existing, scene_id, role) + add_copy_unit("orbit", existing["orbit_id"], source_path, existing["relative_path"]) + return existing + entry = register_orbit_entry( + { + "source_path": source_path, + "relative_path": relative_path, + } + ) + add_orbit_usage(entry, scene_id, role) + add_copy_unit("orbit", entry["orbit_id"], source_path, entry["relative_path"]) + return entry + + for item in tasks: + task_name = item["task_name"] + task_alias = item["task_alias"] + master_entry = ensure_scene_entry( + item["master_path"], + { + "satellite": item.get("master_satellite"), + "imaging_date": item.get("master_imaging_date"), + "imaging_mode": item.get("master_imaging_mode"), + "polarization": item.get("master_polarization"), + }, + "scene", + ) + slave_entry = ensure_scene_entry( + item["slave_path"], + { + "satellite": item.get("slave_satellite"), + "imaging_date": item.get("slave_imaging_date"), + "imaging_mode": item.get("slave_imaging_mode"), + "polarization": item.get("slave_polarization"), + }, + "scene", + ) + master_orbit = ensure_orbit_entry( + item.get("master_orbit_file_path"), + master_entry["scene_id"], + "master", + ) + slave_orbit = ensure_orbit_entry( + item.get("slave_orbit_file_path"), + slave_entry["scene_id"], + "slave", + ) + master_source_path = os.path.normpath(os.path.abspath(str(item["master_path"]))) + slave_source_path = os.path.normpath(os.path.abspath(str(item["slave_path"]))) + pairs.append( + { + "pair_id": f"pair_{next_pair_number:04d}", + "identity_key": item.get("identity_key"), + "task_name": task_name, + "task_alias": task_alias, + "pair_key": item.get("pair_key"), + "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"), + "master_scene_id": master_entry["scene_id"], + "slave_scene_id": slave_entry["scene_id"], + "master_source_path": master_source_path, + "slave_source_path": slave_source_path, + "master_data": master_entry["relative_path"], + "slave_data": slave_entry["relative_path"], + "master_orbit_id": master_orbit.get("orbit_id") if master_orbit else None, + "slave_orbit_id": slave_orbit.get("orbit_id") if slave_orbit else None, + "master_orbit_source_path": master_orbit.get("source_path") if master_orbit else None, + "slave_orbit_source_path": slave_orbit.get("source_path") if slave_orbit else None, + "master_orbit": master_orbit.get("relative_path") if master_orbit else None, + "slave_orbit": slave_orbit.get("relative_path") if slave_orbit else None, + "master_imaging_date": item.get("master_imaging_date"), + "slave_imaging_date": item.get("slave_imaging_date"), + "time_baseline_days": item.get("time_baseline_days"), + "scene_center_distance_meters": item.get("scene_center_distance_meters"), + "spatial_baseline_meters": item.get("spatial_baseline_meters"), + "network_run_id": item.get("network_run_id"), + "network_edge_id": item.get("network_edge_id"), + "policy_version": item.get("policy_version"), + "selection_strategy": item.get("selection_strategy"), + } + ) + next_pair_number += 1 + + scene_list = sorted( + scene_records, + key=lambda entry: (_source_bundle_id_number(entry.get("scene_id"), "scene"), str(entry.get("scene_id") or "")), + ) + orbit_list = sorted( + orbit_records, + key=lambda entry: (_source_bundle_id_number(entry.get("orbit_id"), "orbit"), str(entry.get("orbit_id") or "")), + ) + copy_units = list(copy_units_by_target.values()) + + copied_count = 0 + skipped_count = 0 + failed_count = 0 + total_units = max(1, len(copy_units)) + for index, (kind, unit_id, source_path, target_path) in enumerate(copy_units, start=1): + prog = int(((index - 1) / total_units) * 90) + await task_service.update_task( + task_id, + progress=prog, + message=f"Bundling {kind} ({index}/{total_units}): {unit_id}", + ) + await _log_and_update(task_id, f"[{index}/{total_units}] Bundling {kind}: {unit_id}") + if not os.path.exists(source_path): + await _log_and_update(task_id, f" -> Missing source: {source_path}") + missing_sources.append({"kind": kind, "id": unit_id, "source_path": source_path}) + failed_count += 1 + continue + try: + action = await asyncio.to_thread(_copy_source_into_bundle, source_path, target_path, skip_existing) + if action == "skipped": + skipped_count += 1 + await _log_and_update(task_id, " -> Skipped (already exists in bundle)") + else: + copied_count += 1 + await _log_and_update(task_id, " -> Success") + except PermissionError: + await _log_and_update(task_id, " -> Failed: permission denied") + failed_count += 1 + except Exception as exc: + await _log_and_update(task_id, f" -> Failed: {exc}") + failed_count += 1 + + combined_pairs = existing_pairs + pairs + final_msg = ( + "D-InSAR source bundle export finished. " + f"Pairs {len(combined_pairs)} (+{len(pairs)}), Scenes {len(scene_list)}, " + f"Orbits {len(orbit_list)}, " + f"Copied {copied_count}, Skipped {skipped_count}, " + f"Already exported pairs {skipped_existing_pair_count}, " + f"Deferred {deferred_count}, Failed {failed_count}" + ) + if failed_count > 0: + await task_service.update_task(task_id, status="FAILED", message=final_msg, progress=100) + raise CopyTaskExecutionError(final_msg) + + exported_at = datetime.utcnow().isoformat(timespec="seconds") + "Z" + pairs_payload = { + "schema": "dinsar_source_bundle_pairs.v1", + "exported_at": exported_at, + "pairs": combined_pairs, + } + manifest_payload = { + "schema": "dinsar_source_bundle_manifest.v1", + "exported_at": exported_at, + "package_format": "source_bundle", + "destination": os.path.normpath(os.path.abspath(dest_dir)), + "pair_count": len(combined_pairs), + "new_pair_count": len(pairs), + "existing_pair_count": len(existing_pairs), + "candidate_pair_count": len(candidate_tasks), + "skipped_existing_pairs": skipped_existing_pair_count, + "scene_count": len(scene_list), + "orbit_count": len(orbit_list), + "include_orbit_files": bool(include_orbit_files), + "skip_existing": bool(skip_existing), + "max_items": max_items, + "deferred_pairs": deferred_count, + "copied_units": copied_count, + "skipped_units": skipped_count, + "failed_units": failed_count, + "missing_sources": missing_sources, + "directories": { + "data": "data", + "orbit": "orbit" if orbit_list else None, + }, + "scenes": scene_list, + "orbits": orbit_list, + } + await asyncio.to_thread(_write_json_file, pairs_path, pairs_payload) + await asyncio.to_thread(_write_json_file, manifest_path, manifest_payload) + + await _log_and_update(task_id, final_msg, progress=100) + await task_service.update_task(task_id, status="COMPLETED", message=final_msg, progress=100) + except CopyTaskExecutionError: + raise + except Exception as e: + fail_msg = f"D-InSAR source bundle fatal error: {e}" + try: + await task_service.update_task(task_id, status="FAILED", message=fail_msg) + except Exception: + pass + raise CopyTaskExecutionError(fail_msg) from e + + async def run_ps_copy_items(task_id: str, items: List[Dict[str, Any]], dest_dir: str) -> None: try: await task_service.start_task(task_id, message="Starting PS-InSAR copy task...") @@ -247,8 +967,18 @@ async def run_dinsar_copy_items( *, include_orbit_files: bool = False, export_zip: bool = False, + skip_existing: bool = True, + max_items: Optional[int] = None, ) -> None: try: + if max_items is not None: + try: + max_items = int(max_items) + except (TypeError, ValueError): + max_items = None + if max_items is not None and max_items <= 0: + max_items = None + await task_service.start_task(task_id, message="Starting D-InSAR copy task...") await _log_and_update(task_id, f"D-InSAR copy started. Dest: {dest_dir}") await _log_and_update( @@ -256,7 +986,9 @@ async def run_dinsar_copy_items( ( "D-InSAR copy options: " f"include_orbit_files={include_orbit_files}, " - f"export_zip={export_zip}" + f"export_zip={export_zip}, " + f"skip_existing={skip_existing}, " + f"max_items={max_items if max_items is not None else 'unlimited'}" ), ) @@ -295,7 +1027,10 @@ async def run_dinsar_copy_items( return success_count = 0 + skipped_count = 0 failed_count = 0 + deferred_count = 0 + attempted_count = 0 for i, item in enumerate(tasks, start=1): task_name = item["task_name"] task_alias = item["task_alias"] @@ -308,6 +1043,28 @@ async def run_dinsar_copy_items( staging_root: Optional[str] = None try: + final_task_dir = os.path.join(dest_dir, task_alias) + zip_path = os.path.join(dest_dir, f"{task_alias}.zip") if export_zip else None + + if skip_existing: + if export_zip and zip_path and _is_existing_dinsar_zip_complete(zip_path): + await _log_and_update(task_id, " -> Skipped (existing zip package)") + skipped_count += 1 + continue + if (not export_zip) and _is_existing_dinsar_folder_complete(final_task_dir): + await _log_and_update(task_id, " -> Skipped (existing master/slave folders)") + skipped_count += 1 + continue + + if max_items is not None and attempted_count >= max_items: + deferred_count = total - i + 1 + await _log_and_update( + task_id, + f"Reached max_items={max_items}; deferred {deferred_count} remaining candidates.", + ) + break + attempted_count += 1 + if export_zip: staging_root = await asyncio.to_thread( tempfile.mkdtemp, @@ -315,10 +1072,13 @@ async def run_dinsar_copy_items( dir=dest_dir, ) task_dir = os.path.join(staging_root, task_alias) - zip_path = os.path.join(dest_dir, f"{task_alias}.zip") else: - task_dir = os.path.join(dest_dir, task_alias) - zip_path = None + staging_root = await asyncio.to_thread( + tempfile.mkdtemp, + prefix="._dinsar_copy_", + dir=dest_dir, + ) + task_dir = os.path.join(staging_root, task_alias) master_dir = os.path.join(task_dir, "master") slave_dir = os.path.join(task_dir, "slave") @@ -345,40 +1105,26 @@ async def run_dinsar_copy_items( await asyncio.to_thread( write_pair_metadata, task_dir, - { - "pair_key": item.get("pair_key"), - "task_name": task_name, - "task_alias": task_alias, - "master_path": item.get("master_path"), - "slave_path": item.get("slave_path"), - "master_satellite": item.get("master_satellite"), - "slave_satellite": item.get("slave_satellite"), - "master_imaging_date": item.get("master_imaging_date"), - "slave_imaging_date": item.get("slave_imaging_date"), - "master_imaging_mode": item.get("master_imaging_mode"), - "slave_imaging_mode": item.get("slave_imaging_mode"), - "master_polarization": item.get("master_polarization"), - "slave_polarization": item.get("slave_polarization"), - "time_baseline_days": item.get("time_baseline_days"), - "spatial_baseline_meters": item.get("spatial_baseline_meters"), - "scene_center_distance_meters": item.get("scene_center_distance_meters"), - "package_format": "zip" if export_zip else "folder", - "include_orbit_files": bool(include_orbit_files), - "master_orbit_file_path": item.get("master_orbit_file_path"), - "slave_orbit_file_path": item.get("slave_orbit_file_path"), - "orbit_files": orbit_entries, - "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"), - "network_edge_id": item.get("network_edge_id"), - "policy_version": item.get("policy_version"), - "selection_strategy": item.get("selection_strategy"), - "copied_at": datetime.utcnow().isoformat(timespec="seconds") + "Z", - }, + _build_dinsar_pair_metadata( + item, + task_name, + task_alias, + "zip" if export_zip else "folder", + include_orbit_files, + orbit_entries, + ), ) if export_zip and zip_path: await asyncio.to_thread(_zip_task_directory, task_dir, zip_path) await _log_and_update(task_id, f" -> ZIP: {zip_path}") + elif not export_zip: + if os.path.exists(final_task_dir): + raise CopyTaskExecutionError( + "Destination folder already exists and cannot be atomically replaced: " + f"{final_task_dir}" + ) + await asyncio.to_thread(os.replace, task_dir, final_task_dir) + await _log_and_update(task_id, f" -> Folder: {final_task_dir}") await _log_and_update(task_id, " -> Success") success_count += 1 @@ -394,7 +1140,8 @@ async def run_dinsar_copy_items( final_msg = ( f"D-InSAR copy finished. Mode {'zip' if export_zip else 'folder'}. " - f"Success {success_count}, Failed {failed_count}" + f"Copied {success_count}, Skipped {skipped_count}, " + f"Deferred {deferred_count}, Failed {failed_count}" ) await _log_and_update(task_id, final_msg, progress=100) if failed_count > 0: diff --git a/backend/app/routers/tools.py b/backend/app/routers/tools.py index d33672b..e4e4701 100644 --- a/backend/app/routers/tools.py +++ b/backend/app/routers/tools.py @@ -27,6 +27,13 @@ COPY_BATCH_MAX_STATUS_COUNT = read_int_env( minimum=1, maximum=64, ) +COPY_BATCH_MAX_COPY_ITEMS = read_int_env( + "COPY_BATCH_MAX_COPY_ITEMS", + 5000, + minimum=1, + maximum=200000, +) +COPY_DINSAR_PACKAGE_MODES = {"task_folder", "task_zip", "source_bundle"} class CopyBatchRequest(BaseModel): @@ -35,6 +42,9 @@ class CopyBatchRequest(BaseModel): copy_statuses: Optional[List[str]] = None include_orbit_files: bool = False export_zip: bool = False + package_mode: str = "task_folder" + skip_existing: bool = True + max_items: Optional[int] = None @field_validator("batch_id", "dest_dir", mode="before") @classmethod @@ -57,6 +67,33 @@ class CopyBatchRequest(BaseModel): ) return value + @field_validator("max_items", mode="before") + @classmethod + def _normalize_max_items(cls, value): + if value in (None, ""): + return None + try: + parsed = int(value) + except (TypeError, ValueError) as exc: + raise ValueError("max_items must be an integer.") from exc + if parsed <= 0: + return None + if parsed > COPY_BATCH_MAX_COPY_ITEMS: + raise ValueError( + f"max_items exceeds max item count ({COPY_BATCH_MAX_COPY_ITEMS})." + ) + return parsed + + @field_validator("package_mode", mode="before") + @classmethod + def _normalize_package_mode(cls, value): + normalized = str(value or "task_folder").strip().lower() + if normalized not in COPY_DINSAR_PACKAGE_MODES: + raise ValueError( + f"package_mode must be one of: {sorted(COPY_DINSAR_PACKAGE_MODES)}." + ) + return normalized + def _normalize_copy_batch_statuses(copy_statuses: Optional[List[str]]) -> List[str]: if not copy_statuses: @@ -135,13 +172,19 @@ async def copy_dinsar_pairs_endpoint( _validate_export_path(request.dest_dir, "dest_dir") try: copy_statuses = _normalize_copy_batch_statuses(request.copy_statuses) + package_mode = request.package_mode + if bool(request.export_zip) and package_mode == "task_folder": + package_mode = "task_zip" params = { "dest_dir": request.dest_dir, "file_type": "DINSAR_PAIRS", "batch_id": request.batch_id, "copy_statuses": copy_statuses, "include_orbit_files": bool(request.include_orbit_files), - "export_zip": bool(request.export_zip), + "export_zip": package_mode == "task_zip", + "package_mode": package_mode, + "skip_existing": bool(request.skip_existing), + "max_items": request.max_items, } task_id = await task_service.create_task("COPY_DATA", f"D-InSAR 数据分发: {request.dest_dir}", params=params) @@ -151,7 +194,10 @@ async def copy_dinsar_pairs_endpoint( "batch_id": request.batch_id, "copy_statuses": copy_statuses, "include_orbit_files": bool(request.include_orbit_files), - "export_zip": bool(request.export_zip), + "export_zip": package_mode == "task_zip", + "package_mode": package_mode, + "skip_existing": bool(request.skip_existing), + "max_items": request.max_items, } await job_queue_service.create_job("COPY_DATA", payload=payload, task_id=task_id) await _add_operation_audit_log( @@ -165,7 +211,10 @@ async def copy_dinsar_pairs_endpoint( "dest_dir": request.dest_dir, "copy_statuses": copy_statuses, "include_orbit_files": bool(request.include_orbit_files), - "export_zip": bool(request.export_zip), + "export_zip": package_mode == "task_zip", + "package_mode": package_mode, + "skip_existing": bool(request.skip_existing), + "max_items": request.max_items, }, ) await db.commit() diff --git a/backend/app/services/job_handlers.py b/backend/app/services/job_handlers.py index 1f7e450..987691a 100644 --- a/backend/app/services/job_handlers.py +++ b/backend/app/services/job_handlers.py @@ -50,7 +50,7 @@ from .timeseries_service import ( timeseries_service, ) from .unpack_service import run_unpack_task -from ..copier import run_ps_copy_items, run_dinsar_copy_items +from ..copier import run_ps_copy_items, run_dinsar_copy_items, run_dinsar_source_bundle_items from ..ai_service import ( train_quality_model, predict_quality, @@ -116,6 +116,16 @@ def _normalize_copy_statuses(raw_statuses: Any) -> List[str]: return normalized or ["COMPLETED"] +def _normalize_positive_int(value: Any) -> Optional[int]: + if value in (None, ""): + return None + try: + parsed = int(value) + except (TypeError, ValueError): + return None + return parsed if parsed > 0 else None + + def _dedupe_existing_dirs(paths: Any) -> List[str]: ordered: List[str] = [] for raw_path in paths or []: @@ -289,6 +299,9 @@ async def _handle_copy_data(job: SystemJobORM) -> None: copy_statuses = _normalize_copy_statuses(payload.get("copy_statuses")) include_orbit_files = bool(payload.get("include_orbit_files")) export_zip = bool(payload.get("export_zip")) + package_mode = str(payload.get("package_mode") or ("task_zip" if export_zip else "task_folder")).strip().lower() + skip_existing = payload.get("skip_existing") is not False + max_items = _normalize_positive_int(payload.get("max_items")) if not batch_id: raise ValueError("COPY_DATA requires batch_id payload.") @@ -377,13 +390,25 @@ async def _handle_copy_data(job: SystemJobORM) -> None: raise ValueError( f"No D-InSAR items matched copy statuses: {', '.join(copy_statuses)}" ) - await run_dinsar_copy_items( - job.task_id, - items, - dest_dir, - include_orbit_files=include_orbit_files, - export_zip=export_zip, - ) + if package_mode in {"source_bundle", "bundle", "dedupe_source"}: + await run_dinsar_source_bundle_items( + job.task_id, + items, + dest_dir, + include_orbit_files=include_orbit_files, + skip_existing=skip_existing, + max_items=max_items, + ) + else: + await run_dinsar_copy_items( + job.task_id, + items, + dest_dir, + include_orbit_files=include_orbit_files, + export_zip=(package_mode == "task_zip" or export_zip), + skip_existing=skip_existing, + max_items=max_items, + ) return raise ValueError(f"Unknown COPY_DATA file_type: {file_type}") diff --git a/docs/DINSAR_PAIRING_DISTRIBUTION_LOGIC_20260508.md b/docs/DINSAR_PAIRING_DISTRIBUTION_LOGIC_20260508.md index 8982a22..673e8de 100644 --- a/docs/DINSAR_PAIRING_DISTRIBUTION_LOGIC_20260508.md +++ b/docs/DINSAR_PAIRING_DISTRIBUTION_LOGIC_20260508.md @@ -279,7 +279,10 @@ score = - `dest_dir` - `copy_statuses`，为空时默认 `["COMPLETED"]` - `include_orbit_files`，默认 `false`；为 `true` 时把 master/slave 精轨复制到 Task 内的 `orbit/` -- `export_zip`，默认 `false`；为 `true` 时每个 Task 输出为一个 `.zip` 包 +- `package_mode`，支持 `task_folder`、`task_zip`、`source_bundle` +- `export_zip`，兼容旧参数；为 `true` 且 `package_mode=task_folder` 时等价于 `task_zip` +- `skip_existing`，默认 `true` +- `max_items`，每次最多处理的新 Task 或新 pair 数量；为空或 0 表示不限制 后端动作： @@ -288,7 +291,9 @@ score = 3. 创建 `SystemJob`，job_type 也是 `COPY_DATA`。 4. worker 领取 job 后进入 `job_handlers._handle_copy_data()`。 5. `_handle_copy_data()` 根据 `batch_id` 查询 `dinsar_task_items`，只取 `copy_statuses` 命中的条目。 -6. 调用 [backend/app/copier.py](../backend/app/copier.py) 的 `run_dinsar_copy_items()`。 +6. 根据 `package_mode` 调用 [backend/app/copier.py](../backend/app/copier.py) 的 `run_dinsar_copy_items()` 或 `run_dinsar_source_bundle_items()`。 + +### 9.1 Task 文件夹 / ZIP 模式 `run_dinsar_copy_items()` 对每个 item 执行： @@ -298,8 +303,10 @@ score = - slave 目录：`/slave` - 如果启用 `include_orbit_files`，从 `radar_data.orbit_file_path` 找 master/slave 精轨并复制到 `/orbit/` - 直接复制配对时保存的原始产品目录；D-InSAR 分发不再优先使用 `envi_import/` -- 使用 `shutil.copytree(..., dirs_exist_ok=True)` 复制 master/slave +- 先复制到临时目录，完成后再替换为最终 Task 目录或 ZIP，避免留下半成品 - 写入 `/.dinsar_pair.json` +- `skip_existing=true` 时，文件夹模式检查 `/master` 和 `/slave` 非空即跳过；ZIP 模式只检查 `.zip` 存在且大小大于 0，不打开 ZIP 做深度校验 +- `max_items` 限制本次新复制数量；已跳过的既有 Task 不消耗本次额度 `.dinsar_pair.json` 是后续生产追踪的关键 sidecar，包含： @@ -319,7 +326,32 @@ score = - `selection_strategy` - `copied_at` -当前实现不会清空已有 Task 目录，而是合并复制；如果目标已有旧文件，需要人工确认目录状态。 +当前实现不再合并写入已有 Task 目录。目标 Task 已完整存在时跳过；目标同名目录存在但不完整时，为避免误覆盖，会报错并要求人工处理。 + +### 9.2 去重源数据包模式 + +`package_mode=source_bundle` 时，分发不生成每个 `Task_*`，而是在同一个目标目录内维护： + +```text +/ + data/ + orbit/ + pairs.json + manifest.json +``` + +规则： + +- `data/` 只复制唯一源影像目录或文件，命名为 `scene__`。 +- `orbit/` 只复制唯一精密轨道文件，命名为 `orbit__`。 +- `pairs.json` 记录每个 pair 的 master/slave 数据相对路径、轨道相对路径、pair 元数据和 `identity_key`。 +- `manifest.json` 记录 package 统计信息、场景清单、轨道清单和本次追加统计。 +- 同一目标目录再次分发时，系统先读取已有 `pairs.json/manifest.json`，根据 `identity_key`、`scene_pair_uid/pair_uid`、`pair_key`、`network_run_id + network_edge_id`、源路径或 bundle 相对路径识别已导出的 pair。 +- `max_items` 在跳过已导出 pair 后生效。因此 500 个 pair 第一次限制 100，第二次同一目录限制 100，会追加下一批未导出的 100 个 pair。 +- `data/` / `orbit/` 仍按文件存在性跳过重复复制；pair 级续跑以 `pairs.json` 为准。 +- `pairs.json` 和 `manifest.json` 写入时先写临时文件，再原子替换。 + +这个模式面向外部分发和后续离线还原，不直接作为本系统生产输入。反向还原工具任务书见 [DINSAR_SOURCE_BUNDLE_REVERSE_TOOL_TASK_20260511.md](DINSAR_SOURCE_BUNDLE_REVERSE_TOOL_TASK_20260511.md)。 ## 10. 生产提交与运行分发 diff --git a/docs/DINSAR_SOURCE_BUNDLE_REVERSE_TOOL_TASK_20260511.md b/docs/DINSAR_SOURCE_BUNDLE_REVERSE_TOOL_TASK_20260511.md new file mode 100644 index 0000000..96cf9b0 --- /dev/null +++ b/docs/DINSAR_SOURCE_BUNDLE_REVERSE_TOOL_TASK_20260511.md @@ -0,0 +1,168 @@ +# D-InSAR 去重源数据包反向还原工具任务书 + +日期：2026-05-11 + +## 背景 + +本系统新增“去重源数据包”分发模式。该模式不直接生成每个干涉对的 `Task_*` 目录，而是只分发唯一源影像、唯一精密轨道文件和配对关系文件，减少外部分发时的重复复制量。 + +反向还原工具由任务接收方本地运行，将去重源数据包还原为传统 D-InSAR `Task_*` 目录结构。 + +## 输入目录结构 + +```text +BundleRoot/ + data/ + scene__/ + ... + orbit/ + orbit__.txt + ... + pairs.json + manifest.json +``` + +`orbit/` 可能不存在，或 `pairs.json` 内某些配对的轨道字段为空。 + +## 输出目录结构 + +```text +OutputRoot/ + Task_YYYYMMDD_YYYYMMDD/ + master/ + + slave/ + + orbit/ + + .dinsar_pair.json +``` + +输出目录名称优先使用 `pairs.json` 内的 `task_alias`，若为空则使用 `task_name`，再为空则使用 `pair_id`。 + +## pairs.json 关键字段 + +```json +{ + "schema": "dinsar_source_bundle_pairs.v1", + "exported_at": "2026-05-11T00:00:00Z", + "pairs": [ + { + "pair_id": "pair_0001", + "identity_key": "uid:", + "task_name": "Task_20250101_20250113", + "task_alias": "Task_20250101_20250113", + "master_source_path": "D:/Source/master", + "slave_source_path": "D:/Source/slave", + "master_scene_id": "scene_0001", + "slave_scene_id": "scene_0002", + "master_data": "data/scene_xxx_master", + "slave_data": "data/scene_yyy_slave", + "master_orbit_id": "orbit_0001", + "slave_orbit_id": "orbit_0002", + "master_orbit_source_path": "D:/Orbit/master.EOF", + "slave_orbit_source_path": "D:/Orbit/slave.EOF", + "master_orbit": "orbit/orbit_xxx.txt", + "slave_orbit": "orbit/orbit_yyy.txt", + "master_imaging_date": "20250101", + "slave_imaging_date": "20250113", + "time_baseline_days": 12 + } + ] +} +``` + +## 本系统分发续跑规则 + +去重源数据包支持向同一个 `BundleRoot` 多次分发： + +- 每次启动时先读取目标目录内已有的 `pairs.json` 和 `manifest.json`。 +- 已导出的 pair 通过 `identity_key`、`scene_pair_uid/pair_uid`、`pair_key`、`network_run_id + network_edge_id`、`master/slave_source_path` 或 `master_data + slave_data` 识别。 +- 开启“每次最多追加新配对”时，系统会先跳过已导出的 pair，再从剩余 pair 中取下一批追加；例如 500 个 pair 第一次限制 100，第二次同一目录仍限制 100 时，会追加第 101-200 个未导出的 pair。 +- `data/` 和 `orbit/` 按源路径哈希命名，已有文件或目录在 `skip_existing` 开启时不会重复复制。 +- `pairs.json` 和 `manifest.json` 采用临时文件写入后原子替换，避免中途失败留下半写 JSON。 + +注意：如果用户手动删除了 `pairs.json`，系统无法再根据记录判断哪些 pair 已经分发，只能根据重新生成的 `data/` 路径做源数据级去重，pair 级续跑能力会丢失。 + +## 还原规则 + +1. 读取 `pairs.json`。 +2. 对每个 pair 创建目标 `Task` 目录。 +3. 将 `master_data` 指向的数据复制到 `Task/master/`。 +4. 将 `slave_data` 指向的数据复制到 `Task/slave/`。 +5. 如 `master_orbit` / `slave_orbit` 存在，将轨道文件复制到 `Task/orbit/`。 +6. 生成 `.dinsar_pair.json`，至少保留： + - `pair_id` + - `identity_key` + - `task_name` + - `task_alias` + - `master_scene_id` + - `slave_scene_id` + - `master_data` + - `slave_data` + - `master_orbit_id` + - `slave_orbit_id` + - `master_orbit` + - `slave_orbit` + - `master_imaging_date` + - `slave_imaging_date` + - `time_baseline_days` + - `restored_at` +7. 每个 Task 应采用临时目录还原，全部成功后再重命名为最终目录，避免半成品。 + +## 覆盖策略 + +工具应提供参数： + +- `--skip-existing`：默认开启。若目标 `Task/master` 和 `Task/slave` 均存在且非空，则跳过。 +- `--overwrite`：删除并重建已存在的目标 Task。 +- `--limit N`：最多还原 N 个 pair，便于分批执行。 +- `--dry-run`：只打印计划，不复制。 + +`--skip-existing` 与 `--overwrite` 同时出现时应报错。 + +## 校验要求 + +启动前： + +- 检查 `pairs.json` 是否存在且可解析。 +- 检查 `data/` 是否存在。 +- 检查每个 pair 的 `master_data` / `slave_data` 是否存在。 +- 轨道缺失不应阻断还原，但要记录 warning。 + +还原后： + +- `Task/master/` 非空。 +- `Task/slave/` 非空。 +- `.dinsar_pair.json` 存在。 + +## 日志与报告 + +工具结束后输出 `restore_report.json`： + +```json +{ + "started_at": "...", + "finished_at": "...", + "input_root": "...", + "output_root": "...", + "total_pairs": 20, + "restored": 18, + "skipped": 2, + "failed": 0, + "warnings": [] +} +``` + +同时建议输出人类可读日志 `restore.log`。 + +## 建议实现 + +建议使用 Python 3.10+： + +- `argparse` 处理命令行参数。 +- `pathlib.Path` 处理路径。 +- `shutil.copytree(..., dirs_exist_ok=True)` / `shutil.copy2()` 处理复制。 +- Windows 下注意长路径和权限异常。 + +该工具不需要连接本系统数据库，也不需要调用本系统 API。 diff --git a/docs/INDEX.md b/docs/INDEX.md index bfa8111..e93e2f1 100644 --- a/docs/INDEX.md +++ b/docs/INDEX.md @@ -68,6 +68,8 @@ Sentinel-1 D-InSAR 在不使用 ENVI + SARscape 核心时，基于 Gamma/PyINT 与 ISCE2 的可行性、接入边界和推荐实施顺序。 - [DINSAR_PAIRING_DISTRIBUTION_LOGIC_20260508.md](DINSAR_PAIRING_DISTRIBUTION_LOGIC_20260508.md) 2026-05-08 源码走读记录，梳理 D-InSAR 配对缓存、策略筛选、批次保存、数据分发和生产 worker 执行链路。 +- [DINSAR_SOURCE_BUNDLE_REVERSE_TOOL_TASK_20260511.md](DINSAR_SOURCE_BUNDLE_REVERSE_TOOL_TASK_20260511.md) + D-InSAR 去重源数据包的目录协议、续分发规则和外部反向还原工具任务书。 - [PAIRING_ENHANCEMENT_DESIGN.md](PAIRING_ENHANCEMENT_DESIGN.md) - [FRONTEND_NAVIGATION_ARCHITECTURE.md](FRONTEND_NAVIGATION_ARCHITECTURE.md) diff --git a/frontend/src/DataCopierPanel.jsx b/frontend/src/DataCopierPanel.jsx index ef7c3d2..3980743 100644 --- a/frontend/src/DataCopierPanel.jsx +++ b/frontend/src/DataCopierPanel.jsx @@ -16,8 +16,10 @@ const DataCopierPanel = ({ apiEndpoint, readOnly = false, onJobQueued }) => { const [activeTab, setActiveTab] = useState('dinsar'); const [destDir, setDestDir] = useState(''); const [copyStatuses, setCopyStatuses] = useState(['COMPLETED']); - const [includeDinsarOrbitFiles, setIncludeDinsarOrbitFiles] = useState(false); - const [dinsarExportZip, setDinsarExportZip] = useState(false); + const [includeDinsarOrbitFiles, setIncludeDinsarOrbitFiles] = useState(true); + const [dinsarPackageMode, setDinsarPackageMode] = useState('task_folder'); + const [skipExistingDinsarTasks, setSkipExistingDinsarTasks] = useState(true); + const [dinsarMaxItems, setDinsarMaxItems] = useState('200'); const [batches, setBatches] = useState([]); const [selectedBatchId, setSelectedBatchId] = useState(''); const [isUploading, setIsUploading] = useState(false); @@ -84,7 +86,10 @@ const DataCopierPanel = ({ apiEndpoint, readOnly = false, onJobQueued }) => { const fetchLogs = async () => { if (!taskId) return; try { - const response = await axios.get(`${apiEndpoint}/tools/copy-status/${taskId}`, { withCredentials: true }); + const response = await axios.get(`${apiEndpoint}/tools/copy-status/${taskId}`, { + withCredentials: true, + params: { limit: 1000 }, + }); setLogs(response.data.logs); const nextStatus = normalizeStatus(response.data.status); if (nextStatus && nextStatus !== 'UNKNOWN') { @@ -126,13 +131,19 @@ const DataCopierPanel = ({ apiEndpoint, readOnly = false, onJobQueued }) => { }; if (activeTab === 'dinsar') { payload.include_orbit_files = includeDinsarOrbitFiles; - payload.export_zip = dinsarExportZip; + payload.package_mode = dinsarPackageMode; + payload.export_zip = dinsarPackageMode === 'task_zip'; + payload.skip_existing = skipExistingDinsarTasks; + const parsedMaxItems = Number.parseInt(dinsarMaxItems, 10); + if (Number.isFinite(parsedMaxItems) && parsedMaxItems > 0) { + payload.max_items = parsedMaxItems; + } } const response = await axios.post(endpoint, payload, { withCredentials: true }); const taskId = response.data.task_id; setTaskId(taskId); - // 触发全局锁定 + // 通知全局任务状态；COPY_DATA 在全局控制里按非阻塞处理。 if (onJobQueued) { onJobQueued(taskId); } @@ -196,6 +207,41 @@ const DataCopierPanel = ({ apiEndpoint, readOnly = false, onJobQueued }) => { }} > +
+ + + +
+
+ + setDinsarMaxItems(event.target.value)} + disabled={status === 'RUNNING' || readOnly} + style={{ width: '110px', padding: '5px 7px' }} + /> + 0 或留空表示不限制 +
- 未勾选 ZIP 时直接导出 Task 文件夹；勾选后每个 Task 输出一个 .zip。 + 去重源数据包只复制唯一影像和精轨，并写出 pairs.json；再次分发到同一目录时会接着追加未导出的配对。
)} diff --git a/frontend/src/components/app/AppSidePanel.jsx b/frontend/src/components/app/AppSidePanel.jsx index c3822f1..6a393fb 100644 --- a/frontend/src/components/app/AppSidePanel.jsx +++ b/frontend/src/components/app/AppSidePanel.jsx @@ -262,7 +262,11 @@ export default function AppSidePanel({ taskPanel.onTaskStart(taskId, '数据分发任务已入队，正在处理...')} + onJobQueued={(taskId) => taskPanel.onTaskStart( + taskId, + '数据分发任务已入队，正在处理...', + { taskType: 'COPY_DATA', nonBlocking: true }, + )} /> diff --git a/frontend/src/hooks/useDinsarOperations.js b/frontend/src/hooks/useDinsarOperations.js index 4596f44..cfe2400 100644 --- a/frontend/src/hooks/useDinsarOperations.js +++ b/frontend/src/hooks/useDinsarOperations.js @@ -11,7 +11,7 @@ import { normalizePagePayload } from '../utils/appHelpers'; import { normalizeTaskStatus } from '../utils/appUiHelpers'; import { DEFAULT_LIST_PAGE_SIZE } from '../config/appConstants'; -const NON_BLOCKING_TASK_TYPES = new Set(['UNPACK_ARCHIVES']); +const NON_BLOCKING_TASK_TYPES = new Set(['UNPACK_ARCHIVES', 'COPY_DATA']); export default function useDinsarOperations({ onCleanupDinsarLayers, diff --git a/frontend/src/hooks/useGlobalTaskControl.js b/frontend/src/hooks/useGlobalTaskControl.js index 38b5134..afb7765 100644 --- a/frontend/src/hooks/useGlobalTaskControl.js +++ b/frontend/src/hooks/useGlobalTaskControl.js @@ -2,7 +2,7 @@ import { useCallback, useEffect, useRef, useState } from 'react'; import apiClient from '../api/client'; import { normalizeTaskStatus } from '../utils/appUiHelpers'; -const NON_BLOCKING_TASK_TYPES = new Set(['UNPACK_ARCHIVES']); +const NON_BLOCKING_TASK_TYPES = new Set(['UNPACK_ARCHIVES', 'COPY_DATA']); const isTaskNonBlocking = (taskId, taskType, nonBlockingTaskIds = []) => ( NON_BLOCKING_TASK_TYPES.has(String(taskType || '').toUpperCase()) diff --git a/nginx/nginx.conf b/nginx/nginx.conf index cf37188..a3fe236 100644 --- a/nginx/nginx.conf +++ b/nginx/nginx.conf @@ -18,6 +18,8 @@ http { sendfile on; keepalive_timeout 65; + # Large D-InSAR batch saves can exceed nginx's default 1m request body limit. + client_max_body_size 32m; server { listen 80;