diff --git a/src/govoplan_files/backend/storage/campaign_attachments.py b/src/govoplan_files/backend/storage/campaign_attachments.py index 6b57ec8..305cf6a 100644 --- a/src/govoplan_files/backend/storage/campaign_attachments.py +++ b/src/govoplan_files/backend/storage/campaign_attachments.py @@ -12,8 +12,8 @@ from typing import Any, Iterator from sqlalchemy.orm import Session -from govoplan_files.backend.db.models import FileAsset, FileShare -from govoplan_files.backend.storage.backends import StorageBackendError, get_storage_backend +from govoplan_files.backend.db.models import FileAsset, FileBlob, FileShare, FileVersion +from govoplan_files.backend.storage.backends import StorageBackend, StorageBackendError, get_storage_backend from govoplan_files.backend.storage.common import FileStorageError from govoplan_files.backend.storage.files import current_versions_and_blobs, get_asset_for_user, list_assets_for_user, share_files from govoplan_files.backend.storage.access import ensure_owner_access @@ -232,6 +232,177 @@ def _selected_base_path( return None +def _prepared_attachment_config(raw_json: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any], list[Any]]: + prepared_json = copy.deepcopy(raw_json if isinstance(raw_json, dict) else {}) + attachments = prepared_json.get("attachments") + if not isinstance(attachments, dict): + attachments = {} + prepared_json["attachments"] = attachments + base_paths = attachments.get("base_paths") + if not isinstance(base_paths, list): + base_paths = [] + return prepared_json, attachments, base_paths + + +def _campaign_snapshot_assets( + session: Session, + *, + tenant_id: str, + campaign_id: str, + prepared_json: dict[str, Any], + include_unlinked_candidates: bool, + user_id: str, + is_admin: bool, +) -> tuple[list[FileAsset], list[FileAsset], set[str]]: + shared_assets = list_assets_for_user( + session, + tenant_id=tenant_id, + user_id="", + campaign_id=campaign_id, + is_admin=True, + ) + if not include_unlinked_candidates: + return shared_assets, shared_assets, {asset.id for asset in shared_assets} + candidate_assets, linked_asset_ids = _candidate_assets_for_managed_sources( + session, + tenant_id=tenant_id, + campaign_id=campaign_id, + raw_json=prepared_json, + user_id=user_id, + is_admin=is_admin, + shared_assets=shared_assets, + ) + return shared_assets, candidate_assets, linked_asset_ids + + +def _assets_grouped_by_owner(assets: list[FileAsset]) -> dict[tuple[str, str], list[FileAsset]]: + assets_by_owner: dict[tuple[str, str], list[FileAsset]] = defaultdict(list) + for asset in assets: + owner_id = _asset_owner_id(asset) + if owner_id: + assets_by_owner[(asset.owner_type, owner_id)].append(asset) + return assets_by_owner + + +def _materialize_managed_asset( + asset: FileAsset, + *, + owner_id: str, + logical_root: str, + local_root: Path, + version_blobs: dict[str, tuple[FileVersion, FileBlob]], + backend: StorageBackend | None, + include_bytes: bool, + linked_asset_ids: set[str], +) -> tuple[str, ManagedAttachmentFile] | None: + relative_path = _relative_asset_path(asset, logical_root) + if not relative_path: + return None + target = _safe_local_target(local_root, relative_path) + target.parent.mkdir(parents=True, exist_ok=True) + version, blob = version_blobs[asset.id] + if include_bytes: + try: + data = backend.get_bytes(blob.storage_key) if backend else b"" + except StorageBackendError as exc: + raise FileStorageError(str(exc)) from exc + target.write_bytes(data) + else: + target.touch() + local_key = str(target.resolve()) + return local_key, ManagedAttachmentFile( + local_path=local_key, + asset_id=asset.id, + version_id=version.id, + blob_id=blob.id, + display_path=asset.display_path, + relative_path=normalize_logical_path(relative_path), + filename=asset.filename, + owner_type=asset.owner_type, + owner_id=owner_id, + checksum_sha256=blob.checksum_sha256, + size_bytes=blob.size_bytes, + content_type=blob.content_type, + linked_to_campaign=asset.id in linked_asset_ids, + source_provenance=source_provenance_from_metadata(asset.metadata_), + source_revision=source_revision_from_metadata(asset.metadata_), + ) + + +def _prepare_managed_base_paths( + base_paths: list[Any], + *, + materialized_root: Path, + assets_by_owner: dict[tuple[str, str], list[FileAsset]], + version_blobs: dict[str, tuple[FileVersion, FileBlob]], + backend: StorageBackend | None, + include_bytes: bool, + linked_asset_ids: set[str], +) -> tuple[ + dict[str, ManagedAttachmentFile], + dict[str, tuple[str, str]], + dict[str, list[tuple[str, str]]], + tuple[str, str] | None, +]: + manifest: dict[str, ManagedAttachmentFile] = {} + prepared_by_id: dict[str, tuple[str, str]] = {} + prepared_by_old_path: dict[str, list[tuple[str, str]]] = {} + first_prepared: tuple[str, str] | None = None + for index, item in enumerate(base_paths): + if not isinstance(item, dict): + continue + parsed_source = parse_managed_source(item.get("source")) + if parsed_source is None: + continue + owner_type, owner_id = parsed_source + old_path = str(item.get("path") or ".").strip() or "." + logical_root = "" if old_path in {"", ".", "/"} else normalize_folder(old_path) + base_path_id = str(item.get("id") or f"base-path-{index + 1}") + local_root = materialized_root / f"{index + 1:03d}-{safe_storage_component(base_path_id)}" + local_root.mkdir(parents=True, exist_ok=True) + local_root_string = str(local_root.resolve()) + prepared = (base_path_id, local_root_string) + prepared_by_id[base_path_id] = prepared + prepared_by_old_path.setdefault(old_path, []).append(prepared) + if first_prepared is None: + first_prepared = prepared + item["path"] = local_root_string + for asset in assets_by_owner.get((owner_type, owner_id), []): + materialized = _materialize_managed_asset( + asset, + owner_id=owner_id, + logical_root=logical_root, + local_root=local_root, + version_blobs=version_blobs, + backend=backend, + include_bytes=include_bytes, + linked_asset_ids=linked_asset_ids, + ) + if materialized is not None: + local_key, managed_file = materialized + manifest[local_key] = managed_file + return manifest, prepared_by_id, prepared_by_old_path, first_prepared + + +def _rewrite_managed_attachment_rules( + attachments: dict[str, Any], + prepared_json: dict[str, Any], + *, + prepared_by_id: dict[str, tuple[str, str]], + prepared_by_old_path: dict[str, list[tuple[str, str]]], + first_prepared: tuple[str, str] | None, +) -> None: + for rule in _iter_rule_dicts(attachments, prepared_json): + selected = _selected_base_path(rule, prepared_by_id, prepared_by_old_path, first_prepared) + if selected is None: + continue + base_path_id, local_root_string = selected + rule["base_path_id"] = base_path_id + rule["base_dir"] = local_root_string + if first_prepared is not None: + attachments["base_path"] = first_prepared[1] + + def prepare_campaign_snapshot( session: Session, *, @@ -258,114 +429,35 @@ def prepare_campaign_snapshot( materialized_root = destination / "managed-attachments" materialized_root.mkdir(parents=True, exist_ok=True) - prepared_json = copy.deepcopy(raw_json if isinstance(raw_json, dict) else {}) - attachments = prepared_json.get("attachments") - if not isinstance(attachments, dict): - attachments = {} - prepared_json["attachments"] = attachments - base_paths = attachments.get("base_paths") - if not isinstance(base_paths, list): - base_paths = [] - - shared_assets = list_assets_for_user( + prepared_json, attachments, base_paths = _prepared_attachment_config(raw_json) + shared_assets, candidate_assets, linked_asset_ids = _campaign_snapshot_assets( session, tenant_id=tenant_id, - user_id="", campaign_id=campaign_id, - is_admin=True, + prepared_json=prepared_json, + include_unlinked_candidates=include_unlinked_candidates, + user_id=user_id, + is_admin=is_admin, ) - candidate_assets = shared_assets - linked_asset_ids = {asset.id for asset in shared_assets} - if include_unlinked_candidates: - candidate_assets, linked_asset_ids = _candidate_assets_for_managed_sources( - session, - tenant_id=tenant_id, - campaign_id=campaign_id, - raw_json=prepared_json, - user_id=user_id, - is_admin=is_admin, - shared_assets=shared_assets, - ) - - assets_by_owner: dict[tuple[str, str], list[FileAsset]] = defaultdict(list) - for asset in candidate_assets: - owner_id = _asset_owner_id(asset) - if owner_id: - assets_by_owner[(asset.owner_type, owner_id)].append(asset) + assets_by_owner = _assets_grouped_by_owner(candidate_assets) version_blobs = current_versions_and_blobs(session, candidate_assets) backend = get_storage_backend() if include_bytes else None - - manifest: dict[str, ManagedAttachmentFile] = {} - prepared_by_id: dict[str, tuple[str, str]] = {} - prepared_by_old_path: dict[str, list[tuple[str, str]]] = {} - first_prepared: tuple[str, str] | None = None - - for index, item in enumerate(base_paths): - if not isinstance(item, dict): - continue - parsed_source = parse_managed_source(item.get("source")) - if parsed_source is None: - continue - owner_type, owner_id = parsed_source - old_path = str(item.get("path") or ".").strip() or "." - logical_root = "" if old_path in {"", ".", "/"} else normalize_folder(old_path) - base_path_id = str(item.get("id") or f"base-path-{index + 1}") - local_root = materialized_root / f"{index + 1:03d}-{safe_storage_component(base_path_id)}" - local_root.mkdir(parents=True, exist_ok=True) - local_root_string = str(local_root.resolve()) - - prepared = (base_path_id, local_root_string) - prepared_by_id[base_path_id] = prepared - prepared_by_old_path.setdefault(old_path, []).append(prepared) - if first_prepared is None: - first_prepared = prepared - - item["path"] = local_root_string - - for asset in assets_by_owner.get((owner_type, owner_id), []): - relative_path = _relative_asset_path(asset, logical_root) - if not relative_path: - continue - target = _safe_local_target(local_root, relative_path) - target.parent.mkdir(parents=True, exist_ok=True) - version, blob = version_blobs[asset.id] - if include_bytes: - try: - data = backend.get_bytes(blob.storage_key) if backend else b"" - except StorageBackendError as exc: - raise FileStorageError(str(exc)) from exc - target.write_bytes(data) - else: - target.touch() - local_key = str(target.resolve()) - manifest[local_key] = ManagedAttachmentFile( - local_path=local_key, - asset_id=asset.id, - version_id=version.id, - blob_id=blob.id, - display_path=asset.display_path, - relative_path=normalize_logical_path(relative_path), - filename=asset.filename, - owner_type=asset.owner_type, - owner_id=owner_id, - checksum_sha256=blob.checksum_sha256, - size_bytes=blob.size_bytes, - content_type=blob.content_type, - linked_to_campaign=asset.id in linked_asset_ids, - source_provenance=source_provenance_from_metadata(asset.metadata_), - source_revision=source_revision_from_metadata(asset.metadata_), - ) - - for rule in _iter_rule_dicts(attachments, prepared_json): - selected = _selected_base_path(rule, prepared_by_id, prepared_by_old_path, first_prepared) - if selected is None: - continue - base_path_id, local_root_string = selected - rule["base_path_id"] = base_path_id - rule["base_dir"] = local_root_string - - if first_prepared is not None: - attachments["base_path"] = first_prepared[1] + manifest, prepared_by_id, prepared_by_old_path, first_prepared = _prepare_managed_base_paths( + base_paths, + materialized_root=materialized_root, + assets_by_owner=assets_by_owner, + version_blobs=version_blobs, + backend=backend, + include_bytes=include_bytes, + linked_asset_ids=linked_asset_ids, + ) + _rewrite_managed_attachment_rules( + attachments, + prepared_json, + prepared_by_id=prepared_by_id, + prepared_by_old_path=prepared_by_old_path, + first_prepared=first_prepared, + ) snapshot_path = destination / "campaign.json" snapshot_path.write_text(json.dumps(prepared_json, ensure_ascii=False, indent=2), encoding="utf-8")