1 files changed, 122 insertions, 0 deletions
diff --git a/tools/chapsurvey.py b/tools/chapsurvey.py
new file mode 100644
index 0000000..da4aa84
--- /dev/null
+++ b/tools/chapsurvey.py
@@ -0,0 +1,122 @@
+"""Survey Peter's real subscriptions for chapter markers.
+
+Answers one question: of the shows he actually listens to, how many publish Podcasting 2.0
+chapter JSON (already supported), how many only have ID3 CHAP frames (not supported), and how
+many have nothing at all.
+
+ID3v2 tags sit at the very start of an MP3, so a Range request for the first chunk is enough --
+no need to pull a whole episode.
+"""
+import concurrent.futures as cf
+import re
+import sys
+import urllib.request
+import xml.etree.ElementTree as ET
+
+UA = ("Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36 (KHTML, like Gecko) "
+      "Chrome/120.0.0.0 Mobile Safari/537.36 Listen/1.0")
+HEAD_BYTES = 400_000
+
+
+def get(url, timeout=25, rangebytes=None):
+    req = urllib.request.Request(url, headers={"User-Agent": UA})
+    if rangebytes:
+        req.add_header("Range", f"bytes=0-{rangebytes - 1}")
+    with urllib.request.urlopen(req, timeout=timeout) as r:
+        return r.read()
+
+
+def newest_item(feed_xml):
+    """The newest item's enclosure URL and whether the feed declares chapter JSON."""
+    root = ET.fromstring(feed_xml)
+    for item in root.iter():
+        if item.tag.split('}')[-1] not in ("item", "entry"):
+            continue
+        enclosure, chapters, mime = None, False, None
+        for child in item:
+            tag = child.tag.split('}')[-1]
+            if tag == "enclosure":
+                enclosure = child.get("url")
+                mime = child.get("type")
+            elif tag == "chapters":
+                chapters = True
+        return enclosure, chapters, mime
+    return None, False, None
+
+
+def synchsafe(b):
+    return (b[0] << 21) | (b[1] << 14) | (b[2] << 7) | b[3]
+
+
+def has_chap(blob):
+    """Parse the ID3v2 frame list looking for a CHAP frame. Returns (found, note)."""
+    if not blob.startswith(b"ID3"):
+        return False, "no ID3 tag"
+    major = blob[3]
+    tag_size = synchsafe(blob[6:10])
+    end = min(10 + tag_size, len(blob))
+    pos = 10
+    saw_any = False
+    while pos + 10 <= end:
+        fid = blob[pos:pos + 4]
+        if not re.fullmatch(rb"[A-Z0-9]{4}", fid):
+            break  # padding, or the end of the frame list
+        raw = blob[pos + 4:pos + 8]
+        size = synchsafe(raw) if major == 4 else int.from_bytes(raw, "big")
+        if size <= 0:
+            break
+        saw_any = True
+        if fid == b"CHAP":
+            return True, f"ID3v2.{major}"
+        pos += 10 + size
+    truncated = 10 + tag_size > len(blob)
+    if truncated:
+        return False, f"ID3v2.{major}, tag larger than fetched window"
+    return False, f"ID3v2.{major}" if saw_any else "ID3 tag unreadable"
+
+
+def survey(name, url):
+    try:
+        feed = get(url)
+    except Exception as e:
+        return name, "feed-error", type(e).__name__, ""
+    try:
+        enclosure, json_chapters, mime = newest_item(feed)
+    except Exception as e:
+        return name, "parse-error", type(e).__name__, ""
+    if not enclosure:
+        return name, "no-enclosure", "", ""
+    if json_chapters:
+        return name, "json-chapters", "", mime or ""
+    if mime and "mpeg" not in mime and "mp3" not in mime:
+        return name, "not-mp3", mime, mime or ""
+    try:
+        blob = get(enclosure, rangebytes=HEAD_BYTES)
+    except Exception as e:
+        return name, "media-error", type(e).__name__, mime or ""
+    found, note = has_chap(blob)
+    return name, ("id3-chap" if found else "none"), note, mime or ""
+
+
+def main(path):
+    tree = ET.parse(path)
+    feeds = [(o.get("text"), o.get("xmlUrl"))
+             for o in tree.iter("outline") if o.get("xmlUrl")]
+    print(f"{len(feeds)} feeds\n", flush=True)
+
+    results = []
+    with cf.ThreadPoolExecutor(max_workers=8) as pool:
+        for r in pool.map(lambda f: survey(*f), feeds):
+            results.append(r)
+            print(f"  {r[1]:<14} {r[0][:52]:<52} {r[2]}", flush=True)
+
+    print("\n=== summary ===")
+    counts = {}
+    for _, verdict, _, _ in results:
+        counts[verdict] = counts.get(verdict, 0) + 1
+    for k in sorted(counts, key=lambda k: -counts[k]):
+        print(f"  {counts[k]:>3}  {k}")
+
+
+if __name__ == "__main__":
+    main(sys.argv[1])