Update the fetch-lcp command to have two subcommands. One to fetch th…

…e manifest and one to fetch the underlying LCP file and lcpl license file.

README.md

@@ -11,8 +11,11 @@
 - `audiobook-manifest-summary` (`summarize_rwpm_audio_manifest.py`)
     - Produce a summary description from a [Readium Web Publication Manifest (RWPM)](https://github.com/readium/webpub-manifest)
 manifest conforming to the [Audiobook Profile](https://github.com/readium/webpub-manifest/blob/master/profiles/audiobook.md).
-- `fetch-lcp-audiobook-manifest`
-    - Given an LCP audiobook fulfillment URL, retrieve it and store/print its manifest.
+- `fetch-lcp`
+    - `audiobook-manifest`
+        - Given an LCP audiobook fulfillment URL, retrieve it and store/print its manifest.
+    - `files`
+        - Given an LCP audiobook fulfillment URL, retrieve and store the lcp and lcpl files.
 - `patron-bookshelf`
     - Print a patron's bookshelf as either a summary or in full as JSON.
 - `validate-audiobook-manifests`

pyproject.toml

@@ -47,7 +47,7 @@ types-pytz = "^2024.1.0.20240203"
 
 [tool.poetry.scripts]
 audiobook-manifest-summary = "palace_tools.cli.summarize_rwpm_audio_manifest:main"
-fetch-lcp-audiobook-manifest = "palace_tools.cli.download_lcp_manifest:main"
+fetch-lcp = "palace_tools.cli.fetch_lcp:main"
 palace-terminal = "palace_tools.cli.palace_terminal:main"
 patron-bookshelf = "palace_tools.cli.patron_bookshelf:main"
 validate-audiobook-manifests = "palace_tools.cli.validate_manifests:main"

src/palace_tools/cli/download_lcp_manifest.py → src/palace_tools/cli/fetch_lcp.py

@@ -7,6 +7,7 @@
 import zipfile
 from io import BytesIO
 from pathlib import Path
+from typing import BinaryIO, TextIO
 
 import typer
 
@@ -23,11 +24,55 @@
 
 
 def main() -> None:
-    run_typer_app_as_main(app, prog_name="download-lcp-audio-manifest")
+    run_typer_app_as_main(app, prog_name="fetch-lcp")
 
 
 @app.command()
-def command(
+def files(
+    fulfillment_url: str = typer.Argument(..., help="LCP audiobook fulfillment URL"),
+    output_dir: Path = typer.Argument(
+        ...,
+        dir_okay=True,
+        file_okay=False,
+        help="Manifest output file.",
+    ),
+    username: str = typer.Option(..., "--username", "-u", help="Username or barcode."),
+    password: str = typer.Option(None, "--password", "-p", help="Password or PIN."),
+) -> None:
+    asyncio.run(
+        process_files_command(
+            fulfillment_url=fulfillment_url,
+            output_dir=output_dir,
+            username=username,
+            password=password,
+        )
+    )
+
+
+async def process_files_command(
+    fulfillment_url: str,
+    output_dir: Path,
+    username: str,
+    password: str | None,
+) -> None:
+    if not output_dir.exists():
+        print(f"Creating directory {output_dir}.")
+        output_dir.mkdir(parents=True)
+
+    license_file = output_dir / "license.lcpl"
+    lcp_file = output_dir / "audiobook.lcp"
+
+    with (
+        open(license_file, "w") as license_file_io,
+        open(lcp_file, "wb") as lcp_file_io,
+    ):
+        await process_command(
+            fulfillment_url, license_file_io, lcp_file_io, username, password
+        )
+
+
+@app.command()
+def audiobook_manifest(
     fulfillment_url: str = typer.Argument(..., help="LCP audiobook fulfillment URL"),
     manifest_file: Path = typer.Option(
         ...,
@@ -48,10 +93,8 @@ def command(
         help="Pretty print the result.",
     ),
 ) -> None:
-    # print(stdout if manifest_file == Path("-") else manifest_file)
-    # return
     asyncio.run(
-        process_command(
+        process_audiobook_manifest_command(
             fulfillment_url=fulfillment_url,
             manifest_file=STDOUT if manifest_file == Path("-") else manifest_file,
             username=username,
@@ -61,13 +104,34 @@ def command(
     )
 
 
-async def process_command(
+async def process_audiobook_manifest_command(
     fulfillment_url: str,
     manifest_file: Path | str | int,
     username: str,
     password: str | None,
     manifest_member_name: str = "manifest.json",
     pretty_print: bool = False,
+) -> None:
+    file = BytesIO()
+
+    await process_command(fulfillment_url, None, file, username, password)
+
+    zf = zipfile.ZipFile(file)
+    manifest = zf.read(name=manifest_member_name)
+    print(f"Sending output to {manifest_file}.")
+    with open(manifest_file, "w") as f:
+        if pretty_print:
+            json.dump(json.loads(manifest, strict=False), f, indent=2)
+        else:
+            f.write(manifest.decode("utf-8"))
+
+
+async def process_command(
+    fulfillment_url: str,
+    license_file: TextIO | None,
+    lcp_file: BinaryIO,
+    username: str,
+    password: str | None,
 ) -> None:
     client_headers = {"User-Agent": "Palace"}
     token: BaseAuthorizationToken = BasicAuthToken.from_username_and_password(
@@ -77,6 +141,10 @@ async def process_command(
     async with HTTPXAsyncClient(headers=client_headers) as client:
         response = await client.get(fulfillment_url, headers=token.as_http_headers)
         response.raise_for_status()
+
+        if license_file:
+            license_file.write(response.text)
+
         lcp_license = LCPLicenseDocument.model_validate(response.json())
         lcp_audiobook_links = match_links(
             lcp_license.links,
@@ -86,21 +154,14 @@ async def process_command(
         if not lcp_audiobook_links:
             return
         lcp_audiobook_url = lcp_audiobook_links[0].href
+
         lcp_audiobook_response = await streaming_fetch_with_progress(
             str(lcp_audiobook_url),
-            file := BytesIO(),
+            lcp_file,
             task_label="Downloading audiobook zip...",
             http_client=client,
         )
         lcp_audiobook_response.raise_for_status()
-        zf = zipfile.ZipFile(file)
-        manifest = zf.read(name=manifest_member_name)
-        print(f"Sending output to {manifest_file}.")
-        with open(manifest_file, "w") as f:
-            if pretty_print:
-                json.dump(json.loads(manifest, strict=False), f, indent=2)
-            else:
-                f.write(manifest.decode("utf-8"))
 
 
 if __name__ == "__main__":

src/palace_tools/cli/patron_bookshelf.py

@@ -26,12 +26,8 @@ def main() -> None:
 )
 def patron_bookshelf(
     *,
-    username: str = typer.Argument(
-        None, help="Username to present to the OPDS server."
-    ),
-    password: str = typer.Argument(
-        None, help="Password to present to the OPDS server."
-    ),
+    username: str = typer.Option(..., "--username", "-u", help="Username or barcode."),
+    password: str = typer.Option(None, "--password", "-p", help="Password or PIN."),
     auth_doc_url: str = typer.Option(
         None,
         "--auth_doc",

src/palace_tools/constants.py

@@ -1,6 +1,6 @@
 # Some defaults
 DEFAULT_ASYNC_TIMEOUT = 30.0
-DEFAULT_AUTH_DOC_PATH_SUFFIX = "/authentication-document"
+DEFAULT_AUTH_DOC_PATH_SUFFIX = "authentication_document"
 DEFAULT_REGISTRY_URL = "https://registry.thepalaceproject.org"
 DEFAULT_USER_AGENT = "Palace"
 

src/palace_tools/models/internal/bookshelf.py

@@ -18,13 +18,15 @@ def print_bookshelf_summary(bookshelf: OPDS2Feed) -> None:
 
     print("\n", "  Loans:" if loans else "  No loans.", sep="")
     for p in loans:
-        print(f"\n    {p.metadata.title}")
+        print(f"\n    {p.metadata.title}  ({p.metadata.author.name})")
         for link in p.acquisition_links:
             print(f"      Fulfillment url: {link.href}")
             for indirect in (
                 lnk for lnk in link.indirect_acquisition_links if lnk.get("type")
             ):
                 print(f"        Indirect type: {indirect['type']}")
+            if hashed_pw := link.properties.get("lcp_hashed_passphrase"):
+                print(f"      LCP hashed passphrase: {hashed_pw}")
     print("\n", "  Holds:" if holds else "  No holds.", sep="")
     for p in holds:
         print(f"    {p.metadata.title}")