from __future__ import annotations from dataclasses import dataclass from pathlib import PurePosixPath from typing import Iterable from sqlalchemy import or_ from sqlalchemy.orm import Session from govoplan_files.backend.db.models import FileAsset, FileFolder from govoplan_files.backend.storage.access import ensure_owner_access from govoplan_files.backend.storage.common import FileConflictResolution, FileStorageError, RenamePlanItem, utcnow from govoplan_files.backend.storage.files import ( _active_asset_exists, _asset_owner_id, _asset_query_for_owner, _candidate_renamed_path, _copy_asset_to_path, _next_available_logical_path, _normalize_conflict_strategy, _resolution_by_path, _soft_delete_conflicting_asset, get_asset_for_user, rename_asset, ) from govoplan_files.backend.storage.folders import _active_folder_exists, _ensure_target_folder_hierarchy, _folder_query_for_owner from govoplan_files.backend.storage.paths import filename_from_path, join_folder_filename, normalize_folder, normalize_logical_path @dataclass(frozen=True, slots=True) class _NormalizedTransferRequest: operation: str source_owner_type: str target_owner_type: str source_folder_paths: list[str] target_folder: str conflict_strategy: str conflict_resolution_map: dict[str, FileConflictResolution] @dataclass(frozen=True, slots=True) class _NormalizedRenameRequest: mode: str selected_file_ids: list[str] selected_folder_roots: list[str] owner_type: str | None owner_id: str | None def transfer_selection( session: Session, *, tenant_id: str, user_id: str, operation: str, file_ids: list[str], folder_paths: list[str], source_owner_type: str, source_owner_id: str, target_owner_type: str, target_owner_id: str, target_folder: str, conflict_strategy: str = "reject", conflict_resolutions: Iterable[FileConflictResolution] | None = None, is_admin: bool = False, ) -> tuple[int, int]: """Move or copy files/folders between user/group file spaces. Folder transfers preserve the selected folder's basename below the target folder. File transfers place files directly in the target folder. Existing active target paths are handled by the requested conflict strategy. Copies create new file assets/versions that reference the existing immutable blob. """ normalized = _normalize_transfer_request( operation=operation, source_owner_type=source_owner_type, target_owner_type=target_owner_type, folder_paths=folder_paths, target_folder=target_folder, conflict_strategy=conflict_strategy, conflict_resolutions=conflict_resolutions, ) operation = normalized.operation source_owner_type = normalized.source_owner_type target_owner_type = normalized.target_owner_type source_folder_paths = normalized.source_folder_paths target_folder = normalized.target_folder conflict_strategy = normalized.conflict_strategy conflict_resolution_map = normalized.conflict_resolution_map _ensure_transfer_access_and_shape( session, tenant_id=tenant_id, user_id=user_id, source_owner_type=source_owner_type, source_owner_id=source_owner_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, source_folder_paths=source_folder_paths, target_folder=target_folder, operation=operation, is_admin=is_admin, ) assets_by_id = _selected_assets_for_owner( session, tenant_id=tenant_id, owner_type=source_owner_type, owner_id=source_owner_id, file_ids=file_ids, ) folder_target_paths, all_targets = _plan_transfer_targets( session, tenant_id=tenant_id, assets_by_id=assets_by_id, operation=operation, source_owner_type=source_owner_type, source_owner_id=source_owner_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, source_folder_paths=source_folder_paths, target_folder=target_folder, conflict_strategy=conflict_strategy, conflict_resolution_map=conflict_resolution_map, ) if not all_targets and not source_folder_paths: return 0, 0 all_targets = _resolve_transfer_targets( session, tenant_id=tenant_id, assets_by_id=assets_by_id, all_targets=all_targets, operation=operation, target_owner_type=target_owner_type, target_owner_id=target_owner_id, conflict_strategy=conflict_strategy, conflict_resolution_map=conflict_resolution_map, ) copied_or_moved_folders = _ensure_transfer_target_hierarchy( session, tenant_id=tenant_id, user_id=user_id, operation=operation, source_owner_type=source_owner_type, source_owner_id=source_owner_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, folder_target_paths=folder_target_paths, all_targets=all_targets, ) copied_or_moved_files = _apply_transfer_assets( session, tenant_id=tenant_id, user_id=user_id, operation=operation, assets_by_id=assets_by_id, all_targets=all_targets, target_owner_type=target_owner_type, target_owner_id=target_owner_id, ) _apply_transfer_folder_records( session, tenant_id=tenant_id, user_id=user_id, operation=operation, source_owner_type=source_owner_type, source_owner_id=source_owner_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, folder_target_paths=folder_target_paths, ) return copied_or_moved_files, copied_or_moved_folders def _plan_transfer_targets( session: Session, *, tenant_id: str, assets_by_id: dict[str, FileAsset], operation: str, source_owner_type: str, source_owner_id: str, target_owner_type: str, target_owner_id: str, source_folder_paths: list[str], target_folder: str, conflict_strategy: str, conflict_resolution_map: dict[str, FileConflictResolution], ) -> tuple[dict[str, str], dict[str, str]]: folder_target_paths, folder_asset_targets = _plan_folder_transfer_targets( session, tenant_id=tenant_id, assets_by_id=assets_by_id, operation=operation, source_owner_type=source_owner_type, source_owner_id=source_owner_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, source_folder_paths=source_folder_paths, target_folder=target_folder, conflict_strategy=conflict_strategy, conflict_resolution_map=conflict_resolution_map, ) direct_file_targets = _direct_file_transfer_targets( assets_by_id, folder_asset_targets=folder_asset_targets, target_folder=target_folder, ) return folder_target_paths, {**folder_asset_targets, **direct_file_targets} def _plan_folder_transfer_targets( session: Session, *, tenant_id: str, assets_by_id: dict[str, FileAsset], operation: str, source_owner_type: str, source_owner_id: str, target_owner_type: str, target_owner_id: str, source_folder_paths: list[str], target_folder: str, conflict_strategy: str, conflict_resolution_map: dict[str, FileConflictResolution], ) -> tuple[dict[str, str], dict[str, str]]: folder_asset_targets: dict[str, str] = {} folder_target_paths: dict[str, str] = {} for folder_path in source_folder_paths: target_prefix = _resolved_folder_target_prefix( session, tenant_id=tenant_id, operation=operation, source_owner_type=source_owner_type, source_owner_id=source_owner_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, folder_path=folder_path, target_folder=target_folder, conflict_strategy=conflict_strategy, conflict_resolution_map=conflict_resolution_map, ) if target_prefix is None: continue folder_target_paths[folder_path] = target_prefix for asset in _assets_under_folder(session, tenant_id=tenant_id, owner_type=source_owner_type, owner_id=source_owner_id, folder_path=folder_path): relative = asset.display_path[len(folder_path) + 1 :] folder_asset_targets[asset.id] = normalize_logical_path(f"{target_prefix}/{relative}") assets_by_id[asset.id] = asset return folder_target_paths, folder_asset_targets def _resolved_folder_target_prefix( session: Session, *, tenant_id: str, operation: str, source_owner_type: str, source_owner_id: str, target_owner_type: str, target_owner_id: str, folder_path: str, target_folder: str, conflict_strategy: str, conflict_resolution_map: dict[str, FileConflictResolution], ) -> str | None: folder_basename = PurePosixPath(folder_path).name target_prefix = normalize_folder(f"{target_folder}/{folder_basename}" if target_folder else folder_basename) same_owner_same_path = source_owner_type == target_owner_type and source_owner_id == target_owner_id and target_prefix == folder_path if operation == "move" and same_owner_same_path: return target_prefix if not _active_folder_exists(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, path=target_prefix): return target_prefix resolution = conflict_resolution_map.get(target_prefix) action = resolution.action if resolution else conflict_strategy if action == "skip": return None if action == "reject": raise FileStorageError(f"Target folder already exists: {target_prefix}") if action == "rename": return _next_available_folder_path(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, desired_path=target_prefix) return target_prefix def _next_available_folder_path( session: Session, *, tenant_id: str, owner_type: str, owner_id: str, desired_path: str, ) -> str: counter = 1 candidate = desired_path while _active_folder_exists(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, path=candidate): candidate = _candidate_renamed_path(desired_path, counter) counter += 1 return normalize_folder(candidate) def _assets_under_folder( session: Session, *, tenant_id: str, owner_type: str, owner_id: str, folder_path: str, ) -> list[FileAsset]: return _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id).filter( FileAsset.deleted_at.is_(None), FileAsset.display_path.like(f"{folder_path}/%"), ).all() def _direct_file_transfer_targets( assets_by_id: dict[str, FileAsset], *, folder_asset_targets: dict[str, str], target_folder: str, ) -> dict[str, str]: return { file_id: normalize_logical_path(join_folder_filename(target_folder, filename_from_path(asset.display_path))) for file_id, asset in assets_by_id.items() if file_id not in folder_asset_targets } def _resolve_transfer_targets( session: Session, *, tenant_id: str, assets_by_id: dict[str, FileAsset], all_targets: dict[str, str], operation: str, target_owner_type: str, target_owner_id: str, conflict_strategy: str, conflict_resolution_map: dict[str, FileConflictResolution], ) -> dict[str, str]: resolved_targets: dict[str, str] = {} reserved_targets: set[str] = set() for asset_id, target_path in all_targets.items(): normalized_target = _resolve_transfer_asset_target( session, tenant_id=tenant_id, source_asset=assets_by_id[asset_id], target_path=target_path, operation=operation, target_owner_type=target_owner_type, target_owner_id=target_owner_id, conflict_strategy=conflict_strategy, conflict_resolution_map=conflict_resolution_map, reserved_targets=reserved_targets, ) if normalized_target is None: continue reserved_targets.add(normalized_target) resolved_targets[asset_id] = normalized_target return resolved_targets def _resolve_transfer_asset_target( session: Session, *, tenant_id: str, source_asset: FileAsset, target_path: str, operation: str, target_owner_type: str, target_owner_id: str, conflict_strategy: str, conflict_resolution_map: dict[str, FileConflictResolution], reserved_targets: set[str], ) -> str | None: same_owner = source_asset.owner_type == target_owner_type and _asset_owner_id(source_asset) == target_owner_id exclude = source_asset.id if operation == "move" and same_owner else None normalized_target = normalize_logical_path(target_path) resolution = conflict_resolution_map.get(normalized_target) action = resolution.action if resolution else conflict_strategy if resolution and resolution.new_path and action == "rename": normalized_target = normalize_logical_path(resolution.new_path) conflict = _active_asset_exists( session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, path=normalized_target, exclude_asset_id=exclude, ) or normalized_target in reserved_targets if not conflict and action != "rename": return normalized_target if conflict and action == "reject": raise FileStorageError(f"Target file already exists: {normalized_target}") if conflict and action == "skip": return None if conflict and action == "overwrite": _soft_delete_conflicting_asset( session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, path=normalized_target, exclude_asset_id=exclude, ) return normalized_target if action == "rename": return _next_available_logical_path( session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, desired_path=normalized_target, reserved_paths=reserved_targets, exclude_asset_id=exclude, ) return normalized_target def _ensure_transfer_target_hierarchy( session: Session, *, tenant_id: str, user_id: str, operation: str, source_owner_type: str, source_owner_id: str, target_owner_type: str, target_owner_id: str, folder_target_paths: dict[str, str], all_targets: dict[str, str], ) -> int: same_owner_transfer = source_owner_type == target_owner_type and source_owner_id == target_owner_id for target_path in folder_target_paths.values(): path_to_create = target_path if operation == "move" and same_owner_transfer: parent = str(PurePosixPath(target_path).parent) path_to_create = "" if parent == "." else parent if path_to_create: _ensure_target_folder_hierarchy(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, user_id=user_id, path=path_to_create) for target_path in all_targets.values(): parent = str(PurePosixPath(target_path).parent) if parent and parent != ".": _ensure_target_folder_hierarchy(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, user_id=user_id, path=parent) return len(folder_target_paths) def _apply_transfer_assets( session: Session, *, tenant_id: str, user_id: str, operation: str, assets_by_id: dict[str, FileAsset], all_targets: dict[str, str], target_owner_type: str, target_owner_id: str, ) -> int: if operation == "copy": for asset_id, target_path in all_targets.items(): _copy_asset_to_path(session, assets_by_id[asset_id], tenant_id=tenant_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, target_path=target_path, user_id=user_id) return len(all_targets) return _move_transfer_assets( session, tenant_id=tenant_id, user_id=user_id, assets_by_id=assets_by_id, all_targets=all_targets, target_owner_type=target_owner_type, target_owner_id=target_owner_id, ) def _move_transfer_assets( session: Session, *, tenant_id: str, user_id: str, assets_by_id: dict[str, FileAsset], all_targets: dict[str, str], target_owner_type: str, target_owner_id: str, ) -> int: now = utcnow() moved = 0 for asset_id, target_path in all_targets.items(): asset = assets_by_id[asset_id] same_owner = asset.owner_type == target_owner_type and _asset_owner_id(asset) == target_owner_id if same_owner: if normalize_logical_path(asset.display_path) == target_path: continue rename_asset(asset, new_path=target_path) session.add(asset) else: _copy_asset_to_path(session, asset, tenant_id=tenant_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, target_path=target_path, user_id=user_id) asset.deleted_at = now session.add(asset) moved += 1 return moved def _apply_transfer_folder_records( session: Session, *, tenant_id: str, user_id: str, operation: str, source_owner_type: str, source_owner_id: str, target_owner_type: str, target_owner_id: str, folder_target_paths: dict[str, str], ) -> None: now = utcnow() same_owner_transfer = operation == "move" and source_owner_type == target_owner_type and source_owner_id == target_owner_id for source_path, target_prefix in folder_target_paths.items(): for folder in _folder_rows_under_path(session, tenant_id=tenant_id, owner_type=source_owner_type, owner_id=source_owner_id, source_path=source_path): new_path = _folder_transfer_target_path(folder.path, source_path=source_path, target_prefix=target_prefix) if same_owner_transfer: folder.path = new_path session.add(folder) continue _ensure_target_folder_hierarchy(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, user_id=user_id, path=new_path) if operation == "move": folder.deleted_at = now session.add(folder) def _folder_rows_under_path( session: Session, *, tenant_id: str, owner_type: str, owner_id: str, source_path: str, ) -> list[FileFolder]: return _folder_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id).filter( FileFolder.deleted_at.is_(None), or_(FileFolder.path == source_path, FileFolder.path.like(f"{source_path}/%")), ).all() def _folder_transfer_target_path(path: str, *, source_path: str, target_prefix: str) -> str: relative = "" if path == source_path else path[len(source_path) + 1 :] return normalize_folder(f"{target_prefix}/{relative}" if relative else target_prefix) def _normalize_transfer_request( *, operation: str, source_owner_type: str, target_owner_type: str, folder_paths: list[str], target_folder: str, conflict_strategy: str, conflict_resolutions: Iterable[FileConflictResolution] | None, ) -> _NormalizedTransferRequest: operation = operation.lower().strip() if operation not in {"move", "copy"}: raise FileStorageError("Unsupported transfer operation") return _NormalizedTransferRequest( operation=operation, source_owner_type=source_owner_type.lower().strip(), target_owner_type=target_owner_type.lower().strip(), source_folder_paths=[normalized for path in folder_paths if (normalized := normalize_folder(path))], target_folder=normalize_folder(target_folder), conflict_strategy=_normalize_conflict_strategy(conflict_strategy), conflict_resolution_map=_resolution_by_path(conflict_resolutions), ) def _ensure_transfer_access_and_shape( session: Session, *, tenant_id: str, user_id: str, source_owner_type: str, source_owner_id: str, target_owner_type: str, target_owner_id: str, source_folder_paths: list[str], target_folder: str, operation: str, is_admin: bool, ) -> None: ensure_owner_access(session, tenant_id=tenant_id, owner_type=source_owner_type, owner_id=source_owner_id, user_id=user_id, is_admin=is_admin) ensure_owner_access(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, user_id=user_id, is_admin=is_admin) if operation != "move" or source_owner_type != target_owner_type or source_owner_id != target_owner_id: return for folder_path in source_folder_paths: if target_folder == folder_path or target_folder.startswith(f"{folder_path}/"): raise FileStorageError("Cannot move a folder into itself or one of its child folders") def _rename_basename(name: str, *, mode: str, new_name: str | None = None, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "") -> str: if mode == "direct": cleaned = normalize_logical_path(new_name or "", fallback_filename="item") if "/" in cleaned: raise FileStorageError("Rename expects a name, not a path") return cleaned stem_path = PurePosixPath(name) suffixes = "".join(stem_path.suffixes) stem = name[: -len(suffixes)] if suffixes else name if mode == "prefix": return prefix + name if mode == "suffix": return f"{stem}{suffix}{suffixes}" if mode == "replace": return name.replace(find or "", replacement) if find else name raise FileStorageError("Unsupported rename mode") def _rename_path(path: str, *, mode: str, new_name: str | None = None, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "") -> str: normalized = normalize_logical_path(path) logical = PurePosixPath(normalized) folder = "" if str(logical.parent) == "." else str(logical.parent) next_name = _rename_basename(logical.name, mode=mode, new_name=new_name, find=find, replacement=replacement, prefix=prefix, suffix=suffix) return normalize_logical_path(f"{folder}/{next_name}" if folder else next_name) def _rename_relative_recursive(relative_path: str, *, mode: str, new_name: str | None = None, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "") -> str: parts = normalize_logical_path(relative_path).split("/") return normalize_logical_path( "/".join( _rename_basename(part, mode=mode, new_name=new_name if len(parts) == 1 else None, find=find, replacement=replacement, prefix=prefix, suffix=suffix) for part in parts if part ) ) def build_rename_preview(asset: FileAsset, *, mode: str, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "") -> str: return _rename_path(asset.display_path, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix) def _collapse_folder_roots(folder_paths: list[str]) -> list[str]: roots = sorted({normalize_folder(path) for path in folder_paths if normalize_folder(path)}, key=lambda item: (item.count("/"), item)) collapsed: list[str] = [] for path in roots: if any(path == root or path.startswith(f"{root}/") for root in collapsed): continue collapsed.append(path) return collapsed def _selected_assets_for_owner( session: Session, *, tenant_id: str, owner_type: str, owner_id: str, file_ids: list[str], ) -> dict[str, FileAsset]: selected_file_ids = list(dict.fromkeys(file_ids)) if not selected_file_ids: return {} assets = { asset.id: asset for asset in _asset_query_for_owner( session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, ) .filter(FileAsset.id.in_(selected_file_ids), FileAsset.deleted_at.is_(None)) .all() } if len(assets) != len(selected_file_ids): raise FileStorageError("File not found") return assets def _path_under_root(path: str, root: str) -> bool: normalized = normalize_logical_path(path) return normalized == root or normalized.startswith(f"{root}/") def _folder_new_path_for_root(path: str, root: str, new_root: str, *, recursive: bool, mode: str, find: str | None, replacement: str, prefix: str, suffix: str) -> str: if path == root: return normalize_folder(new_root) relative = path[len(root) + 1 :] if recursive: relative = _rename_relative_recursive(relative, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix) return normalize_folder(f"{new_root}/{relative}") def _file_new_path_for_root(path: str, root: str, new_root: str, *, recursive: bool, mode: str, find: str | None, replacement: str, prefix: str, suffix: str) -> str: relative = path[len(root) + 1 :] if recursive: relative = _rename_relative_recursive(relative, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix) return normalize_logical_path(f"{new_root}/{relative}") def _rename_selection( session: Session, *, tenant_id: str, user_id: str, file_ids: list[str], folder_paths: list[str], owner_type: str | None, owner_id: str | None, mode: str, new_name: str | None = None, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "", recursive: bool = False, dry_run: bool = True, is_admin: bool = False, allow_legacy_without_owner_context: bool = False, ) -> list[RenamePlanItem]: normalized = _normalize_rename_request( file_ids=file_ids, folder_paths=folder_paths, owner_type=owner_type, owner_id=owner_id, mode=mode, ) mode = normalized.mode selected_file_ids = normalized.selected_file_ids selected_folder_roots = normalized.selected_folder_roots owner_type_norm = normalized.owner_type owner_id_norm = normalized.owner_id assets = _collect_rename_assets( session, tenant_id=tenant_id, user_id=user_id, selected_file_ids=selected_file_ids, owner_type=owner_type_norm, owner_id=owner_id_norm, is_admin=is_admin, allow_legacy_without_owner_context=allow_legacy_without_owner_context, ) folder_rows_by_path, affected_folder_paths = _collect_rename_folder_scope( session, tenant_id=tenant_id, user_id=user_id, assets=assets, selected_folder_roots=selected_folder_roots, owner_type=owner_type_norm, owner_id=owner_id_norm, is_admin=is_admin, ) folder_plan, root_target_paths = _build_folder_rename_plan( selected_folder_roots, affected_folder_paths=affected_folder_paths, mode=mode, new_name=new_name, find=find, replacement=replacement, prefix=prefix, suffix=suffix, recursive=recursive, ) file_plan = _build_file_rename_plan( assets, selected_folder_roots=selected_folder_roots, root_target_paths=root_target_paths, mode=mode, new_name=new_name, find=find, replacement=replacement, prefix=prefix, suffix=suffix, recursive=recursive, ) _validate_rename_plans( session, tenant_id=tenant_id, assets=assets, file_plan=file_plan, folder_plan=folder_plan, selected_folder_roots=selected_folder_roots, owner_type=owner_type_norm, owner_id=owner_id_norm, ) plan = _rename_plan_items(assets=assets, file_plan=file_plan, folder_plan=folder_plan) if dry_run: return plan _apply_rename_plans(session, assets=assets, folder_rows_by_path=folder_rows_by_path, file_plan=file_plan, folder_plan=folder_plan) return plan def _collect_rename_assets( session: Session, *, tenant_id: str, user_id: str, selected_file_ids: list[str], owner_type: str | None, owner_id: str | None, is_admin: bool, allow_legacy_without_owner_context: bool, ) -> dict[str, FileAsset]: if not selected_file_ids: return {} if owner_type and owner_id: ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, user_id=user_id, is_admin=is_admin) return _selected_assets_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, file_ids=selected_file_ids) if not allow_legacy_without_owner_context: raise FileStorageError("Bulk rename requires an owner file space") assets: dict[str, FileAsset] = {} for file_id in selected_file_ids: asset = get_asset_for_user(session, tenant_id=tenant_id, user_id=user_id, asset_id=file_id, require_write=True, is_admin=is_admin) assets[asset.id] = asset return assets def _collect_rename_folder_scope( session: Session, *, tenant_id: str, user_id: str, assets: dict[str, FileAsset], selected_folder_roots: list[str], owner_type: str | None, owner_id: str | None, is_admin: bool, ) -> tuple[dict[str, FileFolder], set[str]]: if not selected_folder_roots or not owner_type or not owner_id: return {}, set() ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, user_id=user_id, is_admin=is_admin) folder_rows_by_path: dict[str, FileFolder] = {} affected_folder_paths: set[str] = set() for root in selected_folder_roots: rows = _folder_rows_under_path(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, source_path=root) if not rows and not _folder_has_assets(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, root=root): raise FileStorageError(f"Folder not found: {root}") for row in rows: folder_rows_by_path[row.path] = row affected_folder_paths.add(row.path) for asset in _assets_under_folder(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, folder_path=root): assets[asset.id] = asset return folder_rows_by_path, affected_folder_paths def _folder_has_assets(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, root: str) -> bool: return bool( _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id) .filter(FileAsset.deleted_at.is_(None), FileAsset.display_path.like(f"{root}/%")) .first() ) def _build_folder_rename_plan( selected_folder_roots: list[str], *, affected_folder_paths: set[str], mode: str, new_name: str | None, find: str | None, replacement: str, prefix: str, suffix: str, recursive: bool, ) -> tuple[dict[str, str], dict[str, str]]: folder_plan: dict[str, str] = {} root_target_paths: dict[str, str] = {} for root in selected_folder_roots: new_root = _rename_root_target(root, mode=mode, new_name=new_name, find=find, replacement=replacement, prefix=prefix, suffix=suffix) root_target_paths[root] = new_root if root in affected_folder_paths: folder_plan[root] = new_root for path in sorted(affected_folder_paths, key=lambda item: item.count("/")): if path != root and _path_under_root(path, root): folder_plan[path] = _folder_new_path_for_root(path, root, new_root, recursive=recursive, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix) return folder_plan, root_target_paths def _rename_root_target( root: str, *, mode: str, new_name: str | None, find: str | None, replacement: str, prefix: str, suffix: str, ) -> str: if mode != "direct": return normalize_folder(_rename_path(root, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix)) root_parent = "" if str(PurePosixPath(root).parent) == "." else str(PurePosixPath(root).parent) root_new_name = _rename_basename(PurePosixPath(root).name, mode=mode, new_name=new_name) return normalize_folder(f"{root_parent}/{root_new_name}" if root_parent else root_new_name) def _build_file_rename_plan( assets: dict[str, FileAsset], *, selected_folder_roots: list[str], root_target_paths: dict[str, str], mode: str, new_name: str | None, find: str | None, replacement: str, prefix: str, suffix: str, recursive: bool, ) -> dict[str, str]: file_plan: dict[str, str] = {} folder_roots_by_length = sorted(selected_folder_roots, key=len, reverse=True) for asset in assets.values(): matched_root = next((root for root in folder_roots_by_length if _path_under_root(asset.display_path, root)), None) if matched_root: root_new = root_target_paths.get(matched_root) if root_new: file_plan[asset.id] = _file_new_path_for_root(asset.display_path, matched_root, root_new, recursive=recursive, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix) else: file_plan[asset.id] = _rename_path(asset.display_path, mode=mode, new_name=new_name, find=find, replacement=replacement, prefix=prefix, suffix=suffix) return file_plan def _validate_rename_plans( session: Session, *, tenant_id: str, assets: dict[str, FileAsset], file_plan: dict[str, str], folder_plan: dict[str, str], selected_folder_roots: list[str], owner_type: str | None, owner_id: str | None, ) -> None: _validate_file_rename_targets(session, tenant_id=tenant_id, assets=assets, file_plan=file_plan) if selected_folder_roots and owner_type and owner_id: _validate_folder_rename_targets(session, tenant_id=tenant_id, folder_plan=folder_plan, owner_type=owner_type, owner_id=owner_id) def _validate_file_rename_targets( session: Session, *, tenant_id: str, assets: dict[str, FileAsset], file_plan: dict[str, str], ) -> None: target_files: set[str] = set() for asset_id, target in file_plan.items(): normalized = normalize_logical_path(target) if normalized in target_files: raise FileStorageError(f"Rename would create duplicate file path: {normalized}") target_files.add(normalized) asset = assets[asset_id] owner_id = _asset_owner_id(asset) if _active_asset_exists(session, tenant_id=tenant_id, owner_type=asset.owner_type, owner_id=owner_id, path=normalized, exclude_asset_id=asset.id): raise FileStorageError(f"Target file already exists: {normalized}") if _active_folder_exists(session, tenant_id=tenant_id, owner_type=asset.owner_type, owner_id=owner_id, path=normalized): raise FileStorageError(f"Target path is already a folder: {normalized}") def _validate_folder_rename_targets( session: Session, *, tenant_id: str, folder_plan: dict[str, str], owner_type: str, owner_id: str, ) -> None: target_folders: set[str] = set() for target in folder_plan.values(): normalized = normalize_folder(target) if normalized in target_folders: raise FileStorageError(f"Rename would create duplicate folder path: {normalized}") target_folders.add(normalized) if _active_folder_exists(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, path=normalized, exclude_paths=set(folder_plan.keys())): raise FileStorageError(f"Target folder already exists: {normalized}") if _active_asset_exists(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, path=normalized): raise FileStorageError(f"Target path is already a file: {normalized}") def _rename_plan_items( *, assets: dict[str, FileAsset], file_plan: dict[str, str], folder_plan: dict[str, str], ) -> list[RenamePlanItem]: plan = [RenamePlanItem(kind="folder", id=old_path, old_path=old_path, new_path=new_path) for old_path, new_path in sorted(folder_plan.items())] for asset_id, new_path in file_plan.items(): asset = assets[asset_id] plan.append(RenamePlanItem(kind="file", id=asset.id, old_path=asset.display_path, new_path=new_path)) return plan def _apply_rename_plans( session: Session, *, assets: dict[str, FileAsset], folder_rows_by_path: dict[str, FileFolder], file_plan: dict[str, str], folder_plan: dict[str, str], ) -> None: for old_path, new_path in folder_plan.items(): folder = folder_rows_by_path.get(old_path) if folder: folder.path = normalize_folder(new_path) session.add(folder) for asset_id, new_path in file_plan.items(): rename_asset(assets[asset_id], new_path=new_path) session.add(assets[asset_id]) def _normalize_rename_request( *, file_ids: list[str], folder_paths: list[str], owner_type: str | None, owner_id: str | None, mode: str, ) -> _NormalizedRenameRequest: clean_mode = mode.lower().strip() if clean_mode not in {"direct", "prefix", "suffix", "replace"}: raise FileStorageError("Unsupported rename mode") selected_file_ids = list(dict.fromkeys(file_ids)) selected_folder_roots = _collapse_folder_roots(folder_paths) if not selected_file_ids and not selected_folder_roots: raise FileStorageError("No files or folders selected") if selected_folder_roots and (not owner_type or not owner_id): raise FileStorageError("Folder rename requires an owner file space") if clean_mode == "direct" and (len(selected_file_ids) + len(selected_folder_roots)) != 1: raise FileStorageError("Direct rename requires exactly one selected item") return _NormalizedRenameRequest( mode=clean_mode, selected_file_ids=selected_file_ids, selected_folder_roots=selected_folder_roots, owner_type=owner_type.lower().strip() if owner_type else None, owner_id=owner_id, ) def rename_selection( session: Session, *, tenant_id: str, user_id: str, file_ids: list[str], folder_paths: list[str], owner_type: str, owner_id: str, mode: str, new_name: str | None = None, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "", recursive: bool = False, dry_run: bool = True, is_admin: bool = False, ) -> list[RenamePlanItem]: return _rename_selection( session, tenant_id=tenant_id, user_id=user_id, file_ids=file_ids, folder_paths=folder_paths, owner_type=owner_type, owner_id=owner_id, mode=mode, new_name=new_name, find=find, replacement=replacement, prefix=prefix, suffix=suffix, recursive=recursive, dry_run=dry_run, is_admin=is_admin, ) def legacy_rename_selection_without_owner_context( session: Session, *, tenant_id: str, user_id: str, file_ids: list[str], mode: str, new_name: str | None = None, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "", dry_run: bool = True, is_admin: bool = False, ) -> list[RenamePlanItem]: """Compatibility path for historical file-only callers that lack owner scope.""" return _rename_selection( session, tenant_id=tenant_id, user_id=user_id, file_ids=file_ids, folder_paths=[], owner_type=None, owner_id=None, mode=mode, new_name=new_name, find=find, replacement=replacement, prefix=prefix, suffix=suffix, recursive=False, dry_run=dry_run, is_admin=is_admin, allow_legacy_without_owner_context=True, )