| """Notebook export helpers. |
| |
| Spec references: |
| - `specs/04_interfaces.md`: implements `export_notebook_zip()`. |
| - `specs/07_security.md`: export remains scoped to one user's notebook root. |
| - `specs/10_test_plan.md`: export logic is explicit and unit-testable. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from datetime import datetime, timezone |
| from pathlib import Path |
| import zipfile |
|
|
| from notebooklm_clone.notebooks import get_notebook |
| from notebooklm_clone.storage import notebook_root, safe_join, user_root |
|
|
|
|
| class ExportError(Exception): |
| """Base exception for notebook export failures.""" |
|
|
|
|
| class ExportIOError(ExportError): |
| """Raised when notebook export files cannot be created.""" |
|
|
|
|
| def _utc_timestamp() -> str: |
| """Return a timestamp suitable for export filenames.""" |
|
|
| return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") |
|
|
|
|
| def _zip_name(notebook_id: str) -> str: |
| """Build a deterministic export filename for a notebook.""" |
|
|
| return f"{notebook_id}_{_utc_timestamp()}.zip" |
|
|
|
|
| def _should_exclude(path: Path) -> bool: |
| """Return whether a file should be excluded as a transient artifact. |
| |
| Spec references: |
| - User requirement: exclude large transient files if necessary. |
| """ |
|
|
| return path.name.endswith(".lock") or path.name.endswith(".sqlite-wal") or path.name.endswith( |
| ".sqlite-shm" |
| ) |
|
|
|
|
| def export_notebook_zip(username: str, notebook_id: str) -> Path: |
| """Zip one notebook directory and return the archive path. |
| |
| Spec references: |
| - `specs/04_interfaces.md`: implements `export_notebook_zip()`. |
| - `specs/07_security.md`: keeps export paths within the user's storage root. |
| |
| Raises: |
| ExportIOError: If the notebook archive cannot be created. |
| """ |
|
|
| |
| get_notebook(username, notebook_id) |
|
|
| source_root: Path = notebook_root(username, notebook_id) |
| destination_root: Path = user_root(username) |
| zip_path: Path = safe_join(destination_root, _zip_name(notebook_id)) |
|
|
| try: |
| with zipfile.ZipFile(zip_path, mode="w", compression=zipfile.ZIP_DEFLATED) as archive: |
| for file_path in sorted(source_root.rglob("*")): |
| if not file_path.is_file(): |
| continue |
| if _should_exclude(file_path): |
| continue |
| archive_name: Path = file_path.relative_to(source_root) |
| archive.write(file_path, arcname=str(archive_name)) |
| except OSError as exc: |
| raise ExportIOError(f"Failed to create notebook export archive: {zip_path}") from exc |
| except ValueError as exc: |
| raise ExportIOError(f"Failed to package notebook export archive: {zip_path}") from exc |
| except zipfile.BadZipFile as exc: |
| raise ExportIOError(f"Failed to finalize notebook export archive: {zip_path}") from exc |
|
|
| return zip_path |
|
|