Add library push to website from the content editor

tools/content/library_push.py

@@ -0,0 +1,97 @@
+import sys
+import urllib.parse
+from dataclasses import dataclass, field
+from typing import Callable
+
+import library_api
+
+SECTIONS = {
+    "media": {"label": "shows and movies", "path": "/api/library/media/"},
+    "books": {"label": "books", "path": "/api/library/books/"},
+    "games": {"label": "games", "path": "/api/library/games/"},
+}
+
+
+@dataclass
+class PushPlan:
+    source: library_api.Site
+    target: library_api.Site
+    entries: dict[str, list[dict]]
+    removals: dict[str, list[str]]
+
+    def describe(self) -> str:
+        pushed = ", ".join(f"{len(self.entries[name])} {section['label']}" for name, section in SECTIONS.items())
+        removed = sum(len(identifiers) for identifiers in self.removals.values())
+        removal_text = f" and remove {removed} entries that only exist there" if removed else ""
+        return f"Push {pushed} from {self.source.url} to {self.target.url}{removal_text}?"
+
+
+@dataclass
+class PushResult:
+    pushed: int = 0
+    removed: int = 0
+    errors: list[str] = field(default_factory=list)
+
+
+def entry_path(section: str, identifier: str) -> str:
+    return f"{SECTIONS[section]['path']}{urllib.parse.quote(identifier, safe='')}"
+
+
+def plan_push() -> PushPlan:
+    source = library_api.local_site()
+    target = library_api.push_site()
+    if source.url == target.url:
+        raise library_api.LibraryError("LIBRARY_PUSH_URL is the same site as LIBRARY_API_URL")
+
+    local = library_api.fetch_export(source)
+    if not any(local[section] for section in SECTIONS):
+        raise library_api.LibraryError(f"The library at {source.url} is empty, so nothing was pushed")
+    remote = library_api.fetch_export(target)
+
+    removals = {}
+    for section in SECTIONS:
+        local_ids = {entry["id"] for entry in local[section]}
+        removals[section] = [entry["id"] for entry in remote[section] if entry["id"] not in local_ids]
+    return PushPlan(source, target, {section: local[section] for section in SECTIONS}, removals)
+
+
+def run_push(plan: PushPlan, on_progress: Callable[[str], None] = lambda message: None) -> PushResult:
+    result = PushResult()
+    for section, identifiers in plan.removals.items():
+        for identifier in identifiers:
+            try:
+                library_api.send("DELETE", entry_path(section, identifier), site=plan.target)
+                result.removed += 1
+            except library_api.LibraryError as error:
+                result.errors.append(f"remove {identifier}: {error}")
+
+    for section, entries in plan.entries.items():
+        on_progress(f"Pushing {len(entries)} {SECTIONS[section]['label']}...")
+        for entry in entries:
+            try:
+                library_api.send("PUT", entry_path(section, entry["id"]), entry["fields"], site=plan.target)
+                result.pushed += 1
+            except library_api.LibraryError as error:
+                result.errors.append(f"{entry['id']}: {error}")
+    return result
+
+
+def main() -> int:
+    try:
+        plan = plan_push()
+    except (OSError, ValueError) as error:
+        print(f"Push stopped: {error}", file=sys.stderr)
+        return 1
+    print(plan.describe())
+    if input("Type yes to continue: ").strip().lower() != "yes":
+        print("Nothing was pushed")
+        return 0
+    result = run_push(plan, print)
+    for message in result.errors:
+        print(f"Error: {message}", file=sys.stderr)
+    print(f"Pushed {result.pushed}, removed {result.removed}, {len(result.errors)} failed")
+    return 1 if result.errors else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())