1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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])
|
