summaryrefslogtreecommitdiff
path: root/tools
diff options
context:
space:
mode:
Diffstat (limited to 'tools')
-rw-r--r--tools/chaprecheck.py106
-rw-r--r--tools/chapsurvey.py122
2 files changed, 228 insertions, 0 deletions
diff --git a/tools/chaprecheck.py b/tools/chaprecheck.py
new file mode 100644
index 0000000..c130acf
--- /dev/null
+++ b/tools/chaprecheck.py
@@ -0,0 +1,106 @@
+"""Second pass over the feeds whose ID3 tag did not fit the first window.
+
+Reads the declared tag size from the 10-byte header, then fetches exactly that much, so the
+answer does not depend on a guessed window. Also handles ID3v2.2, whose frame headers are
+6 bytes with 3-byte ids rather than 10 with 4 -- misparsing those is what made some of them
+look truncated in the first pass.
+"""
+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")
+
+WANTED = {
+ "Cruising the Waves Podcast", "Conversations with Tyler", "Darknet Diaries",
+ "Dan Carlin's Hardcore History", "1619", "Common Sense with Dan Carlin", "Reply All",
+ "Dan Carlin's Hardcore History: Addendum", "Flesh and Code",
+ "Talk Easy with Sam Fragoso", "This American Life", "Hooked On Wooden Boats", "Humans",
+ "Dream Chasers and Eccentrics", "Paranormal Investigations Hawaii",
+}
+
+
+def get(url, rangebytes=None, timeout=40):
+ req = urllib.request.Request(url, headers={"User-Agent": UA})
+ if rangebytes:
+ req.add_header("Range", f"bytes={rangebytes[0]}-{rangebytes[1]}")
+ with urllib.request.urlopen(req, timeout=timeout) as r:
+ return r.read()
+
+
+def enclosure_of(feed_xml):
+ root = ET.fromstring(feed_xml)
+ for item in root.iter():
+ if item.tag.split('}')[-1] not in ("item", "entry"):
+ continue
+ for child in item:
+ if child.tag.split('}')[-1] == "enclosure":
+ return child.get("url")
+ return None
+
+
+def synchsafe(b):
+ return (b[0] << 21) | (b[1] << 14) | (b[2] << 7) | b[3]
+
+
+def scan(blob, major):
+ """Walk the frame list. v2.2 uses 3-byte ids and 3-byte sizes; v2.3/2.4 use 4 and 4."""
+ pos, ids = 10, []
+ while True:
+ if major == 2:
+ if pos + 6 > len(blob):
+ break
+ fid = blob[pos:pos + 3]
+ if not fid.strip(b"\x00"):
+ break
+ size = int.from_bytes(blob[pos + 3:pos + 6], "big")
+ step = 6 + size
+ else:
+ if pos + 10 > len(blob):
+ break
+ fid = blob[pos:pos + 4]
+ if not fid.strip(b"\x00"):
+ break
+ raw = blob[pos + 4:pos + 8]
+ size = synchsafe(raw) if major == 4 else int.from_bytes(raw, "big")
+ step = 10 + size
+ if size <= 0:
+ break
+ ids.append(fid.decode("latin-1"))
+ pos += step
+ return ids
+
+
+def check(name, url):
+ feed = get(url)
+ enclosure = enclosure_of(feed)
+ if not enclosure:
+ return name, "no enclosure", []
+ head = get(enclosure, rangebytes=(0, 9))
+ if not head.startswith(b"ID3"):
+ return name, "no ID3 tag at all", []
+ major = head[3]
+ size = synchsafe(head[6:10])
+ blob = get(enclosure, rangebytes=(0, 10 + size - 1))
+ got = len(blob) >= 10 + size
+ ids = scan(blob, major)
+ verdict = "CHAP PRESENT" if "CHAP" in ids else "no CHAP"
+ detail = f"v2.{major}, tag {size / 1024:.0f}KB, {'complete' if got else 'SHORT'}, frames: {sorted(set(ids))}"
+ return name, verdict, detail
+
+
+def main(path):
+ tree = ET.parse(path)
+ feeds = [(o.get("text"), o.get("xmlUrl"))
+ for o in tree.iter("outline") if o.get("xmlUrl") and o.get("text") in WANTED]
+ for name, url in feeds:
+ try:
+ n, verdict, detail = check(name, url)
+ print(f"{verdict:<14} {n[:42]:<42} {detail}", flush=True)
+ except Exception as e:
+ print(f"{'error':<14} {name[:42]:<42} {type(e).__name__}", flush=True)
+
+
+if __name__ == "__main__":
+ main(sys.argv[1])
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])
generated by cgit v1.2.3 (git 2.39.1) at 2026-11-14 21:42:58 +0000