Support split Internet Archive items

.github/scripts/lint_ia.py

@@ -20,35 +20,90 @@
 from internetarchive import get_item
 
 
-def extract_ia_entries(readme_path: str) -> list[tuple[str, str, bool]]:
+def extract_ia_entries(readme_path: str) -> list[tuple[str, list[str], bool, str]]:
     """Extract Internet Archive entries from README.md.
 
-    Returns list of (log_origin, item_identifier, has_torrent) tuples.
+    Returns (log_origin, item_identifiers, has_torrent, archive_location)
+    tuples.
     """
     content = Path(readme_path).read_text()
     entries = []
 
-    # Match table rows with archive.org URLs, optionally with torrent links
-    # Format: | log_origin | https://archive.org/details/item_id ... | [.torrent](...) |
-    pattern = r'\|\s*([^\|]+?)\s*\|\s*https://archive\.org/details/(\S+?)\s[^\|]*\|([^\n]*)'
+    for line in content.splitlines():
+        if not line.startswith("|"):
+            continue
 
-    for match in re.finditer(pattern, content):
-        log_origin = match.group(1).strip()
-        item_id = match.group(2).strip()
-        has_torrent = ".torrent" in match.group(3)
-        entries.append((log_origin, item_id, has_torrent))
+        cells = line.split("|")
+        if len(cells) < 4:
+            continue
+
+        log_origin = cells[1].strip()
+        archive_location = cells[2].strip()
+        item_ids = re.findall(
+            r"https://archive\.org/details/([^\s|]+)", archive_location
+        )
+        if not item_ids:
+            continue
+
+        has_torrent = ".torrent" in cells[3]
+        entries.append((log_origin, item_ids, has_torrent, archive_location))
 
     return entries
 
 
-def lint_item(log_origin: str, item_id: str, has_torrent: bool) -> list[str]:
-    """Lint a single Internet Archive item.
+def lint_item(
+    log_origin: str,
+    item_ids: list[str],
+    has_torrent: bool,
+    archive_location: str,
+) -> list[str]:
+    """Lint all Internet Archive items for a single log.
 
     Returns a list of error messages (empty if all checks pass).
     """
     errors = []
+    items = [(item_id, get_item(item_id)) for item_id in item_ids]
+
+    expected_archive_location = " ".join(
+        f"https://archive.org/details/{item_id}" for item_id in item_ids
+    )
+    actual_archive_location = archive_location.removesuffix(" †")
+    if actual_archive_location != expected_archive_location:
+        errors.append(
+            f"Archive location should be '{expected_archive_location}'"
+        )
 
-    item = get_item(item_id)
+    expected_extension_ids = [
+        f"{item_ids[0]}_ext{index}" for index in range(1, len(item_ids))
+    ]
+    if item_ids[1:] != expected_extension_ids:
+        errors.append(
+            "Extension items must be named consecutively: expected "
+            f"{expected_extension_ids}, found {item_ids[1:]}"
+        )
+
+    total_zip_count = sum(
+        1
+        for _, item in items
+        for f in item.files
+        if f.get("name", "").endswith(".zip")
+    )
+
+    for item_id, item in items:
+        item_errors = lint_item_part(
+            log_origin, item_id, item, total_zip_count,
+            has_torrent and item_id == item_ids[0],
+        )
+        errors.extend(f"{item_id}: {error}" for error in item_errors)
+
+    return errors
+
+
+def lint_item_part(
+    log_origin: str, item_id: str, item, total_zip_count: int, has_torrent: bool
+) -> list[str]:
+    """Lint one IA item that is part of a possibly split log archive."""
+    errors = []
     metadata = item.metadata
 
     if not metadata:
@@ -89,22 +144,18 @@ def lint_item(log_origin: str, item_id: str, has_torrent: bool) -> list[str]:
             f"Collection should be one of {allowed_collections} (has: {collection})"
         )
 
-    # Check 4: Number of zip files matches ceil(ctlogsize / 256^3)
+    # Check 4: Number of zip files across all extension items matches
+    # ceil(ctlogsize / 256^3).
     ctlogsize = metadata.get("ctlogsize")
     if ctlogsize:
         try:
             ctlogsize = int(ctlogsize)
             expected_zips = math.ceil(ctlogsize / (256**3))
 
-            # Count zip files in item
-            zip_count = sum(
-                1 for f in item.files if f.get("name", "").endswith(".zip")
-            )
-
-            if zip_count != expected_zips:
+            if total_zip_count != expected_zips:
                 errors.append(
-                    f"Expected {expected_zips} zip files based on ctlogsize "
-                    f"{ctlogsize}, found {zip_count}"
+                    f"Expected {expected_zips} zip files across all items based on "
+                    f"ctlogsize {ctlogsize}, found {total_zip_count}"
                 )
         except ValueError:
             errors.append(f"Invalid ctlogsize value: {ctlogsize}")
@@ -118,7 +169,17 @@ def lint_item(log_origin: str, item_id: str, has_torrent: bool) -> list[str]:
             f.get("name") for f in item.files if f.get("name", "").endswith(".zip")
         )
         if zip_files:
-            first_zip = zip_files[0]
+            if re.search(r"_ext[1-9]\d*$", item_id):
+                first_zip = zip_files[0]
+            elif "000.zip" in zip_files:
+                first_zip = "000.zip"
+            else:
+                errors.append("No 000.zip found in base item")
+                first_zip = None
+
+            if first_zip is None:
+                return errors
+
             base_url = f"https://archive.org/download/{item_id}/{first_zip}"
 
             # Fetch and verify log.v3.json
@@ -293,9 +354,11 @@ def main():
     print(f"Found {len(entries)} Internet Archive entries")
 
     all_passed = True
-    for log_origin, item_id, has_torrent in entries:
-        print(f"\nLinting {item_id} ({log_origin})...")
-        errors = lint_item(log_origin, item_id, has_torrent)
+    for log_origin, item_ids, has_torrent, archive_location in entries:
+        print(f"\nLinting {', '.join(item_ids)} ({log_origin})...")
+        errors = lint_item(
+            log_origin, item_ids, has_torrent, archive_location
+        )
 
         if errors:
             all_passed = False

README.md

@@ -181,6 +181,10 @@ files can be uploaded to a new Internet Archive item using the `ia` CLI. The
 `ia-metadata.py` script populates the metadata from the contents of the item.
 
 Use `ct_operator_nameYYYYhN` as the item identifier, e.g. `ct_sectigo_sabre2024h1`.
+For a split archive, use the base identifier for the first item and consecutive
+`_extN` suffixes for the rest, for example `ct_sectigo_elephant2026h1`,
+`ct_sectigo_elephant2026h1_ext1`, and `ct_sectigo_elephant2026h1_ext2`. Run
+`ia-metadata.py` on every item.
 
     uv tool install internetarchive
     ia configure

cmd/ia-metadata.py

@@ -16,7 +16,9 @@
 Run the script with: uv run --script cmd/photocamera-archiver/ia-metadata.py <IDENTIFIER>
 """
 
+import re
 import sys
+
 import requests
 from internetarchive import get_item, modify_metadata
 
@@ -59,9 +61,24 @@ def main():
     item = get_item(identifier)
     metadata = item.metadata
 
-    print("Fetching checkpoint and log info...", file=sys.stderr)
-    checkpoint_url = f"https://archive.org/download/{identifier}/000.zip/checkpoint"
-    log_info_url = f"https://archive.org/download/{identifier}/000.zip/log.v3.json"
+    zip_files = sorted(
+        f.get("name") for f in item.files if f.get("name", "").endswith(".zip")
+    )
+    if re.search(r"_ext[1-9]\d*$", identifier):
+        if not zip_files:
+            print(f"Error: no zip files found in item {identifier}", file=sys.stderr)
+            sys.exit(1)
+        metadata_zip = zip_files[0]
+    elif "000.zip" in zip_files:
+        metadata_zip = "000.zip"
+    else:
+        print(f"Error: no 000.zip found in base item {identifier}", file=sys.stderr)
+        sys.exit(1)
+
+    print(f"Fetching checkpoint and log info from {metadata_zip}...", file=sys.stderr)
+    base_url = f"https://archive.org/download/{identifier}/{metadata_zip}"
+    checkpoint_url = f"{base_url}/checkpoint"
+    log_info_url = f"{base_url}/log.v3.json"
 
     checkpoint_resp = requests.get(checkpoint_url)
     checkpoint_resp.raise_for_status()