"""The research-package import — one half of the file contract with the Claude research session (locked decision D6). The zip layout the contract promises: manifest.json screenshots/-desktop.png screenshots/-mobile.png Import is idempotent by option slug: a re-upload updates the matching options in place (fields and screenshots, including on a published option) and NEVER touches collected reviews — republishing an updated screenshot keeps the ratings that already exist. New slugs become DRAFT options appended after the existing ones; nothing is auto-published (D14: clients see options only after an admin deliberately publishes). """ import json import re import zipfile from dataclasses import dataclass, field from io import BytesIO from sqlalchemy.orm import Session from app.constants import IMPORT_SCHEMA_VERSION, MAX_UPLOAD_BYTES, OptionStatus from app.models.option import Option from app.models.project import Project from app.services import options as options_service from app.services import storage _SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,78}$") class ImportError_(Exception): """A package the import refuses. `code` becomes the response error_code.""" def __init__(self, code: str, summary: str, details: str = "") -> None: super().__init__(summary) self.code = code self.summary = summary self.details = details @dataclass class ImportResult: created: list[str] = field(default_factory=list) updated: list[str] = field(default_factory=list) def _safe_entry_names(archive: zipfile.ZipFile) -> set[str]: """The zip-slip guard: refuse absolute paths and any `..` traversal before a single byte is extracted.""" names = set() for info in archive.infolist(): name = info.filename if name.endswith("/"): # directory entries carry no content continue if name.startswith(("/", "\\")) or ".." in name.split("/") or ".." in name.split("\\"): raise ImportError_( "UNSAFE_ZIP_ENTRY", f"The package contains an unsafe path: '{name}'.", "Entry names must be relative, with no '..' components.", ) if info.file_size > MAX_UPLOAD_BYTES: raise ImportError_( "UPLOAD_TOO_LARGE", f"'{name}' is larger than the {MAX_UPLOAD_BYTES // (1024 * 1024)} MB per-file limit.", ) names.add(name) return names def _read_manifest(archive: zipfile.ZipFile, names: set[str]) -> dict: if "manifest.json" not in names: raise ImportError_( "MANIFEST_MISSING", "The package has no manifest.json at its root.", "The research package must follow the import contract embedded in the briefing export.", ) try: manifest = json.loads(archive.read("manifest.json")) except (json.JSONDecodeError, UnicodeDecodeError) as exc: raise ImportError_("MANIFEST_INVALID", "manifest.json is not valid JSON.") from exc if not isinstance(manifest, dict): raise ImportError_("MANIFEST_INVALID", "manifest.json must be a JSON object.") if manifest.get("schema_version") != IMPORT_SCHEMA_VERSION: raise ImportError_( "UNSUPPORTED_SCHEMA_VERSION", f"This build understands schema_version {IMPORT_SCHEMA_VERSION}, " f"got {manifest.get('schema_version')!r}.", ) return manifest def _validate_options(manifest: dict, project: Project, names: set[str]) -> list[dict]: if manifest.get("project_slug") != project.slug: raise ImportError_( "PROJECT_SLUG_MISMATCH", f"The package is for project '{manifest.get('project_slug')}', not '{project.slug}'.", "Re-export the briefing for this project, or import into the right one.", ) raw_options = manifest.get("options") if not isinstance(raw_options, list) or not raw_options: raise ImportError_("NO_OPTIONS", "The manifest lists no options.") seen: set[str] = set() for entry in raw_options: if not isinstance(entry, dict): raise ImportError_("OPTION_INVALID", "Every options[] entry must be an object.") slug = entry.get("slug", "") if not isinstance(slug, str) or not _SLUG_RE.match(slug): raise ImportError_( "OPTION_SLUG_INVALID", f"Option slug {slug!r} is invalid.", "Slugs are kebab-case: lowercase letters, digits, hyphens.", ) if slug in seen: raise ImportError_("OPTION_SLUG_DUPLICATE", f"Slug '{slug}' appears twice.") seen.add(slug) screenshots = entry.get("screenshots") or {} for kind in ("desktop", "mobile"): path = screenshots.get(kind, "") if not path: raise ImportError_( "SCREENSHOT_MISSING", f"Option '{slug}' declares no {kind} screenshot.", "Every option needs both a desktop and a mobile capture.", ) if path not in names: raise ImportError_( "SCREENSHOT_MISSING", f"'{path}' (option '{slug}', {kind}) is not in the package.", ) return raw_options def import_package(db: Session, project: Project, payload: bytes) -> ImportResult: """Validate then apply a research package. Validation is completed for the whole package before anything is written, so a bad manifest cannot leave a half-imported project behind.""" try: archive = zipfile.ZipFile(BytesIO(payload)) except zipfile.BadZipFile as exc: raise ImportError_("NOT_A_ZIP", "The upload is not a readable zip file.") from exc with archive: names = _safe_entry_names(archive) manifest = _read_manifest(archive, names) entries = _validate_options(manifest, project, names) existing = { option.slug: option for option in options_service.list_for_admin(db, project.id) } result = ImportResult() sort_order = options_service.next_sort_order(db, project.id) for entry in entries: slug = entry["slug"] screenshots = entry.get("screenshots") or {} desktop = storage.store_screenshot( project.id, slug, "desktop", archive.read(screenshots["desktop"]) ) mobile = storage.store_screenshot( project.id, slug, "mobile", archive.read(screenshots["mobile"]) ) fields = { "display_label": str(entry.get("display_label") or "").strip(), "descriptor": str(entry.get("descriptor") or "").strip(), "source_name": str(entry.get("source_name") or "").strip(), "source_url": str(entry.get("source_url") or "").strip(), "why_selected": str(entry.get("why_selected") or "").strip(), "design_notes": str(entry.get("design_notes") or "").strip(), } option = existing.get(slug) if option is not None: # Update in place. Status and reviews are deliberately untouched: # a re-import refreshes content, it never un-publishes and never # discards what reviewers already said. for key, value in fields.items(): if value: setattr(option, key, value) option.screenshot_desktop = desktop option.screenshot_mobile = mobile result.updated.append(slug) else: db.add( Option( project_id=project.id, slug=slug, display_label=fields["display_label"] or options_service.next_display_label(db, project.id), descriptor=fields["descriptor"], source_name=fields["source_name"], source_url=fields["source_url"], why_selected=fields["why_selected"], design_notes=fields["design_notes"], status=OptionStatus.DRAFT, sort_order=sort_order, screenshot_desktop=desktop, screenshot_mobile=mobile, ) ) # Flush so next_display_label sees this row when the next entry # also arrives without a display_label of its own. db.flush() sort_order += 1 result.created.append(slug) db.commit() return result