1"""Turn an Internet Archive live-recording link into a playable Terrapin Trail tour.
2
3Pipeline:
4
51. ``parse_identifier`` pulls the item id out of a pasted archive.org URL.
62. ``fetch_metadata`` reads that item's metadata and ``band_collection`` picks the
7 band-specific collection (e.g. ``GratefulDead``), filtering out the generic
8 umbrella collections every LMA item shares.
93. ``load_shows`` bulk-fetches every recording in that collection (paginated
10 scrape API, cached per band under ``DATA_DIR``) so we can list touring years
11 and build tours without re-hitting the network.
124. ``year_list`` aggregates distinct show-nights per year.
135. ``build_tour`` turns one year's nights into a tour: dedupe to one canonical
14 stop per night, order by date, geocode the legs into miles, mark the
15 best-circulating night "legendary" and the last night the "finale".
16
17The output tour dict matches the shape the game's ``TOURS`` catalog uses, so it
18drops straight into the engine.
19"""
20
21import json
22import re
23import time
24import urllib.parse
25import urllib.request
26from datetime import date as date_cls
27from pathlib import Path
28
29from terrapin_trail import geocode
30
31_UA = "TerrapinTrailGenerator/1.0 (Minds workspace app)"
32
33# Collections every Live Music Archive item carries -- not the band.
34_GENERIC_COLLECTIONS = {
35 "etree",
36 "stream_only",
37 "stream",
38 "samples_only",
39 "audio",
40 "opensource_audio",
41 "community",
42 "communityaudio",
43}
44
45# A tour with more nights than this is sampled down to a playable length.
46MAX_STOPS = 12
47
48# On-disk caches grow by at most one file per run (one band scrape, one seed
49# metadata snapshot). Cap how many of each we keep and evict oldest-first on
50# write so the caches stay bounded no matter how many links are pasted over time.
51MAX_CACHED_BANDS = 40
52MAX_CACHED_ITEMS = 300
53
54# Consecutive shows more than this many days apart belong to different tours.
55_TOUR_GAP_DAYS = 11
56
57_MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
58
59# Non-concert items (studio work, compilations) share the collection but aren't
60# shows -- they pollute tours and, being highly rated by a few voters, hijack the
61# "legendary" pick. Filter them by venue text and by a missing/placeholder city.
62_JUNK_VENUE = re.compile(
63 r"studio|outtake|rehears|session|soundcheck|sound check|demo|compilation|"
64 r"interview|sampler|acoustic mix|remaster",
65 re.I,
66)
67_JUNK_CITY = {"", "various", "various locations", "unknown", "studio", "n/a", "-"}
68
69
70class ArchiveError(Exception):
71 """A user-facing problem resolving or building from an archive.org link."""
72
73
74def parse_identifier(url: str) -> str:
75 """Extract the archive.org item identifier from a pasted URL (or bare id)."""
76 raw = (url or "").strip()
77 if not raw:
78 raise ArchiveError("Paste an archive.org recording link to get started.")
79 # Bare identifier (no slashes, no scheme).
80 if "/" not in raw and "." not in raw.split("?")[0][:6]:
81 if re.fullmatch(r"[A-Za-z0-9._-]+", raw):
82 return raw
83 parsed = urllib.parse.urlparse(raw if "//" in raw else "https://" + raw)
84 if parsed.netloc and "archive.org" not in parsed.netloc:
85 raise ArchiveError("That doesn't look like an archive.org link.")
86 parts = [p for p in parsed.path.split("/") if p]
87 for anchor in ("details", "metadata", "download", "embed"):
88 if anchor in parts:
89 i = parts.index(anchor)
90 if i + 1 < len(parts):
91 return urllib.parse.unquote(parts[i + 1])
92 if parts:
93 return urllib.parse.unquote(parts[-1])
94 raise ArchiveError("Couldn't find a recording id in that link.")
95
96
97def _get_json(url: str, timeout: int = 45) -> dict:
98 req = urllib.request.Request(url, headers={"User-Agent": _UA})
99 with urllib.request.urlopen(req, timeout=timeout) as r:
100 return json.load(r)
101
102
103def _safe_filename(key: str) -> str:
104 return re.sub(r"[^A-Za-z0-9._-]", "_", key) + ".json"
105
106
107def _prune_cache(dir_path: Path, max_files: int) -> None:
108 """Keep only the newest ``max_files`` JSON snapshots in ``dir_path``, evicting the oldest."""
109 files = sorted(dir_path.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True)
110 for stale in files[max_files:]:
111 stale.unlink(missing_ok=True)
112
113
114def _items_dir(data_dir: Path) -> Path:
115 return data_dir / "items"
116
117
118def fetch_metadata(identifier: str, data_dir: Path) -> dict:
119 """Fetch an item's metadata (raise if it doesn't exist), caching the raw payload.
120
121 The raw metadata response is persisted under ``DATA_DIR/items`` keyed by the
122 item identifier. Item metadata is immutable, so a re-paste reads the cached
123 snapshot instead of re-hitting the network; keeping the raw record also lets
124 the band/year view be re-derived later without a refetch (preserve the
125 source record the view is built from). The cache is bounded by
126 ``MAX_CACHED_ITEMS`` (oldest evicted on write).
127 """
128 cache = _items_dir(data_dir) / _safe_filename(identifier)
129 if cache.exists():
130 try:
131 return json.loads(cache.read_text())
132 except ValueError:
133 pass
134 data = _get_json("https://archive.org/metadata/" + urllib.parse.quote(identifier))
135 if not data or not data.get("metadata"):
136 raise ArchiveError(
137 "That recording isn't on the Internet Archive (or the id is wrong)."
138 )
139 cache.parent.mkdir(parents=True, exist_ok=True)
140 cache.write_text(json.dumps(data))
141 _prune_cache(cache.parent, MAX_CACHED_ITEMS)
142 return data
143
144
145def _as_list(value: object) -> list[str]:
146 if value is None:
147 return []
148 if isinstance(value, list):
149 return [str(v) for v in value]
150 return [str(value)]
151
152
153def band_collection(meta: dict) -> str:
154 """Pick the band-specific collection id from an item's metadata."""
155 md = meta.get("metadata", {})
156 collections = _as_list(md.get("collection"))
157 candidates = [
158 c for c in collections if c.lower() not in _GENERIC_COLLECTIONS and not c.lower().startswith("fav-")
159 ]
160 if not candidates:
161 raise ArchiveError(
162 "That recording isn't filed under a band collection we can build a tour from."
163 )
164 # If several remain, the first listed is the item's primary band collection.
165 return candidates[0]
166
167
168def band_name(meta: dict, collection: str) -> str:
169 md = meta.get("metadata", {})
170 creator = md.get("creator")
171 if isinstance(creator, list):
172 creator = creator[0] if creator else None
173 return str(creator) if creator else collection
174
175
176def _bands_dir(data_dir: Path) -> Path:
177 return data_dir / "bands"
178
179
180def load_shows(collection: str, data_dir: Path, max_age_days: int = 30) -> list[dict]:
181 """All recordings in a collection, cached per band under DATA_DIR.
182
183 Each entry: ``{identifier, date, coverage, venue, avg_rating}``. Cached to
184 ``DATA_DIR/bands/<collection>.json`` so year lists and tour builds are
185 offline after the first fetch.
186 """
187 cache = _bands_dir(data_dir) / _safe_filename(collection)
188 if cache.exists():
189 age = time.time() - cache.stat().st_mtime
190 if age < max_age_days * 86400:
191 try:
192 return json.loads(cache.read_text())["shows"]
193 except (ValueError, KeyError):
194 pass
195 shows = _scrape_shows(collection)
196 cache.parent.mkdir(parents=True, exist_ok=True)
197 cache.write_text(json.dumps({"collection": collection, "fetched": time.time(), "shows": shows}))
198 _prune_cache(cache.parent, MAX_CACHED_BANDS)
199 return shows
200
201
202def _parse_scrape_items(data: dict) -> list[dict]:
203 """Normalize one scrape-API page into show records (drops dateless items).
204
205 The scrape API returns ISO datetimes (``1977-05-08T00:00:00Z``) and omits
206 ``avg_rating``/``venue``/``coverage`` on items that have none, so trim the
207 date to ``YYYY-MM-DD`` and default the rest.
208 """
209 out: list[dict] = []
210 for it in data.get("items", []):
211 if it.get("date"):
212 out.append(
213 {
214 "identifier": it.get("identifier"),
215 "date": it["date"][:10],
216 "coverage": it.get("coverage") or "",
217 "venue": it.get("venue") or "",
218 "avg_rating": float(it.get("avg_rating") or 0),
219 }
220 )
221 return out
222
223
224def _scrape_shows(collection: str) -> list[dict]:
225 fields = "identifier,date,coverage,venue,avg_rating"
226 base = "https://archive.org/services/search/v1/scrape"
227 q = "collection:" + collection
228 cursor: str | None = None
229 out: list[dict] = []
230 for _ in range(50): # hard page cap (50 * 10k) guards against runaway loops
231 params = {"q": q, "fields": fields, "count": 10000}
232 if cursor:
233 params["cursor"] = cursor
234 data = _get_json(base + "?" + urllib.parse.urlencode(params))
235 out.extend(_parse_scrape_items(data))
236 cursor = data.get("cursor")
237 if not cursor:
238 break
239 return out
240
241
242def _canonical_night(recordings: list[dict]) -> dict:
243 """Pick one canonical stop (venue/city/rating) for a night's recordings.
244
245 ``sources`` counts how many recordings exist for the night -- a strong proxy
246 for how legendary/widely-circulated the show is (Cornell '77 has dozens).
247 """
248 best = max(recordings, key=lambda r: r["avg_rating"])
249 # Prefer a non-empty venue/coverage even if it comes from another recording.
250 venue = best["venue"] or next((r["venue"] for r in recordings if r["venue"]), "")
251 city = best["coverage"] or next((r["coverage"] for r in recordings if r["coverage"]), "")
252 return {
253 "ymd": best["date"],
254 "venue": venue,
255 "city": city,
256 "rating": best["avg_rating"],
257 "sources": len(recordings),
258 "identifier": best["identifier"],
259 }
260
261
262def _is_concert(night: dict) -> bool:
263 if night["city"].strip().lower() in _JUNK_CITY:
264 return False
265 if night["venue"] and _JUNK_VENUE.search(night["venue"]):
266 return False
267 return True
268
269
270def concert_nights(shows: list[dict]) -> list[dict]:
271 """One canonical concert per date across all years, sorted, junk removed."""
272 by_night: dict[str, list[dict]] = {}
273 for s in shows:
274 by_night.setdefault(s["date"], []).append(s)
275 nights = [_canonical_night(recs) for _, recs in sorted(by_night.items())]
276 return [n for n in nights if _is_concert(n)]
277
278
279def year_list(shows: list[dict]) -> list[dict]:
280 """Concert-nights per year: ``[{year, count, span}]`` in chronological order."""
281 nights = concert_nights(shows)
282 by_year: dict[int, list[dict]] = {}
283 for n in nights:
284 by_year.setdefault(int(n["ymd"][:4]), []).append(n)
285 out = []
286 for y in sorted(by_year):
287 ms = sorted({int(n["ymd"][5:7]) for n in by_year[y]})
288 span = _MONTHS[ms[0] - 1] + (" – " + _MONTHS[ms[-1] - 1] if ms[-1] != ms[0] else "")
289 out.append({"year": y, "count": len(by_year[y]), "span": span})
290 return out
291
292
293def _split_into_tours(nights: list[dict]) -> list[list[dict]]:
294 """Split date-sorted nights into runs, breaking where a long gap appears."""
295 runs: list[list[dict]] = []
296 current: list[dict] = []
297 prev: date_cls | None = None
298 for n in nights:
299 d = date_cls.fromisoformat(n["ymd"])
300 if prev is not None and (d - prev).days > _TOUR_GAP_DAYS:
301 runs.append(current)
302 current = []
303 current.append(n)
304 prev = d
305 if current:
306 runs.append(current)
307 return runs
308
309
310def _sample_nights(nights: list[dict], must_keep: set[str]) -> list[dict]:
311 """Evenly sample a long run down to MAX_STOPS, always keeping ``must_keep``.
312
313 ``must_keep`` is a set of ymds (first, last, legendary, the seed show) that
314 must survive so the tour still contains its anchor shows.
315 """
316 if len(nights) <= MAX_STOPS:
317 return nights
318 keep_idx = {i for i, n in enumerate(nights) if n["ymd"] in must_keep}
319 keep_idx |= {0, len(nights) - 1}
320 remaining = MAX_STOPS - len(keep_idx)
321 if remaining > 0:
322 middle = [i for i in range(1, len(nights) - 1) if i not in keep_idx]
323 if middle:
324 step = len(middle) / remaining
325 for k in range(remaining):
326 keep_idx.add(middle[int(k * step)])
327 # If must_keep already exceeds MAX_STOPS, honor it (don't drop anchors).
328 return [nights[i] for i in sorted(keep_idx)]
329
330
331def _fmt_date(ymd: str) -> str:
332 try:
333 d = date_cls.fromisoformat(ymd)
334 return f"{_MONTHS[d.month - 1]} {d.day}"
335 except ValueError:
336 return ymd
337
338
339def build_tour(
340 shows: list[dict], year: int, band: str, collection: str, seed_ymd: str | None = None
341) -> dict:
342 """Build one year's tour (TOURS-catalog shape) from a band's shows.
343
344 A jam band's year holds several tours plus studio dates, so we take a single
345 *contiguous* run of concert-nights: the run containing the seed show if it
346 falls in this year, otherwise the year's longest run (its marquee tour).
347 """
348 nights_all = [n for n in concert_nights(shows) if n["ymd"][:4] == str(year)]
349 if not nights_all:
350 raise ArchiveError(f"No {band} concerts on the archive for {year}.")
351
352 runs = _split_into_tours(nights_all)
353 chosen = None
354 if seed_ymd and seed_ymd[:4] == str(year):
355 chosen = next((r for r in runs if any(n["ymd"] == seed_ymd for n in r)), None)
356 if chosen is None:
357 chosen = max(runs, key=lambda r: (len(r), r[-1]["ymd"]))
358 nights = chosen
359
360 # Legendary = the most-recorded night (widely-circulated = legendary),
361 # tie-broken by rating -- far more reliable than rating alone (many tie at 5).
362 legendary = max(nights, key=lambda n: (n["sources"], n["rating"]))
363 must_keep = {nights[0]["ymd"], nights[-1]["ymd"], legendary["ymd"]}
364 if seed_ymd:
365 must_keep.add(seed_ymd)
366 nights = _sample_nights(nights, must_keep)
367
368 stops = []
369 prev_place: geocode.Place | None = None
370 for i, n in enumerate(nights):
371 place = geocode.resolve(n["city"])
372 miles = 0 if i == 0 else (geocode.road_miles(prev_place, place) or 0)
373 prev_place = place if place is not None else prev_place
374 stop = {
375 "venue": n["venue"] or "Unknown venue",
376 "city": n["city"] or "Somewhere on the road",
377 "date": _fmt_date(n["ymd"]),
378 "ymd": n["ymd"],
379 "miles": miles,
380 "identifier": n["identifier"],
381 "rating": round(n["rating"], 2),
382 }
383 if n["ymd"] == legendary["ymd"]:
384 stop["legendary"] = True
385 stops.append(stop)
386 if stops:
387 stops[-1]["finale"] = True
388
389 total = sum(int(str(s["miles"])) for s in stops)
390 return {
391 "year": year,
392 "label": f"{year}: {band}",
393 "blurb": (
394 f"{band}, {year}. {len(stops)} nights built live from the Internet "
395 f"Archive — real venues, real dates, the miles between them yours to drive."
396 ),
397 "collection": collection,
398 "band": band,
399 "total_miles": total,
400 "stops": stops,
401 }
402
403
404# ---------------------------------------------------------------- browser data
405# The archive browser lists bands without a pasted link, splits a year into its
406# tours, and plays a night's recording. These read the same caches as above.
407
408MAX_BAND_LIST_AGE = 86400
409
410
411def list_bands(data_dir: Path) -> list[dict]:
412 """Every band collection on the Live Music Archive, most downloaded first.
413
414 Cached under ``DATA_DIR/bandlist.json`` for a day: ``[{id, name, items}]``.
415 """
416 cache = data_dir / "bandlist.json"
417 if cache.exists() and time.time() - cache.stat().st_mtime < MAX_BAND_LIST_AGE:
418 try:
419 return json.loads(cache.read_text())["bands"]
420 except (ValueError, KeyError):
421 pass
422 params = [
423 ("q", "collection:etree AND mediatype:collection"),
424 ("fl[]", "identifier"),
425 ("fl[]", "title"),
426 ("fl[]", "item_count"),
427 ("sort[]", "downloads desc"),
428 ("rows", "20000"),
429 ("output", "json"),
430 ]
431 data = _get_json("https://archive.org/advancedsearch.php?" + urllib.parse.urlencode(params))
432 bands = [
433 {"id": d["identifier"], "name": d.get("title") or d["identifier"], "items": int(d.get("item_count") or 0)}
434 for d in data.get("response", {}).get("docs", [])
435 if d.get("identifier")
436 ]
437 bands = [b for b in bands if b["items"] > 0]
438 cache.parent.mkdir(parents=True, exist_ok=True)
439 cache.write_text(json.dumps({"fetched": time.time(), "bands": bands}))
440 return bands
441
442
443def first_show_year(data_dir: Path) -> int:
444 """Year of the oldest dated recording on the Live Music Archive, cached for a day."""
445 cache = data_dir / "firstshow.json"
446 if cache.exists() and time.time() - cache.stat().st_mtime < MAX_BAND_LIST_AGE:
447 try:
448 return int(json.loads(cache.read_text())["year"])
449 except (ValueError, KeyError):
450 pass
451 params = [
452 ("q", "collection:etree AND mediatype:etree AND date:[1900-01-01 TO 2100-01-01]"),
453 ("fl[]", "date"),
454 ("fl[]", "identifier"),
455 ("sort[]", "date asc"),
456 ("rows", "1"),
457 ("output", "json"),
458 ]
459 docs = _get_json("https://archive.org/advancedsearch.php?" + urllib.parse.urlencode(params)).get("response", {}).get("docs", [])
460 if not docs or not docs[0].get("date"):
461 raise ArchiveError("No dated recordings found.")
462 year = int(docs[0]["date"][:4])
463 cache.parent.mkdir(parents=True, exist_ok=True)
464 cache.write_text(json.dumps({"year": year, "identifier": docs[0].get("identifier"), "fetched": time.time()}))
465 return year
466
467
468def sort_name(name: str) -> str:
469 """How a band sorts A to Z: no leading 'The', punctuation or spaces ('...And You Will
470 Know Us' under A, 'The Allman Brothers Band' under A, 'moe.' under M)."""
471 n = re.sub(r"^\W+", "", name.strip()).lower()
472 n = re.sub(r"^the\s+", "", n)
473 return re.sub(r"[^a-z0-9]+", " ", n).strip() or name.lower()
474
475
476def tours_for_year(shows: list[dict], year: int) -> list[dict]:
477 """A year's tours: ``[{start, end, nights, first_city, last_city}]`` in date order."""
478 nights = [n for n in concert_nights(shows) if n["ymd"][:4] == str(year)]
479 return [
480 {
481 "start": run[0]["ymd"],
482 "end": run[-1]["ymd"],
483 "nights": len(run),
484 "first_city": run[0]["city"],
485 "last_city": run[-1]["city"],
486 }
487 for run in _split_into_tours(nights)
488 ]
489
490
491def _first(value) -> str:
492 if isinstance(value, list):
493 return str(value[0]) if value else ""
494 return str(value or "")
495
496
497def _clock(length: str) -> str:
498 """A song length as m:ss. The archive stores most as "09:40" but some as seconds ("759.07")."""
499 if re.fullmatch(r"\d+(\.\d+)?", length or ""):
500 secs = int(float(length))
501 h, rest = divmod(secs, 3600)
502 return (f"{h}:{rest // 60:02d}:{rest % 60:02d}") if h else f"{rest // 60:02d}:{rest % 60:02d}"
503 return length or ""
504
505
506# Track titles that aren't songs: tuning, crowd noise, breaks, chatter.
507_NOT_A_SONG = re.compile(
508 r"^(tuning|crowd|banter|intro(duction)?|set ?(i+|one|two|1|2)? ?break|encore break|applause|stage (announcements?|banter)"
509 r"|dead air|soundcheck|talk(ing)?|tape flip|silence|noise|drums? ?/ ?space intro|untitled|unknown|take a step back)\b",
510 re.IGNORECASE,
511)
512
513
514def _song_title(raw: str) -> str:
515 """A track title as a setlist line: no track numbers, segue arrows or timings."""
516 t = re.sub(r"^\s*(d\d+)?t?\d{1,3}\s*[-.:)]\s*", "", raw or "")
517 t = re.sub(r"\s*(->|>|\*+|\(\d+:\d+\))\s*$", "", t)
518 return t.strip(" -")
519
520
521def _secs(clock: str) -> int:
522 parts = [int(p) for p in re.findall(r"\d+", clock or "")]
523 out = 0
524 for p in parts[-3:]:
525 out = out * 60 + p
526 return out
527
528
529def best_recording(collection: str, ymd: str) -> str | None:
530 """The best-rated recording of one night in a band's collection, by archive id."""
531 params = [
532 ("q", f'collection:"{collection}" AND date:{ymd}'),
533 ("fl[]", "identifier"),
534 ("sort[]", "avg_rating desc"),
535 ("sort[]", "downloads desc"),
536 ("rows", "1"),
537 ("output", "json"),
538 ]
539 docs = _get_json("https://archive.org/advancedsearch.php?" + urllib.parse.urlencode(params), timeout=20)
540 docs = docs.get("response", {}).get("docs", [])
541 return docs[0]["identifier"] if docs and docs[0].get("identifier") else None
542
543
544def setlist(identifier: str, data_dir: Path) -> list[dict]:
545 """The songs played on a recording, in order, each ``{title, secs}``."""
546 out = []
547 for t in show_tracks(identifier, data_dir)["tracks"]:
548 title = _song_title(t["title"])
549 if title and not _NOT_A_SONG.match(title):
550 out.append({"title": title[:60], "secs": _secs(t["length"])})
551 return out
552
553
554def show_tracks(identifier: str, data_dir: Path) -> dict:
555 """A recording's playable songs plus who taped it, for the in-game player."""
556 meta = fetch_metadata(identifier, data_dir)
557 md = meta.get("metadata", {})
558 files = meta.get("files", [])
559 mp3 = [f for f in files if f.get("format") in ("VBR MP3", "MP3", "64Kbps MP3", "128Kbps MP3")]
560 # Prefer the VBR derivatives (one per song); fall back to whatever MP3s exist.
561 vbr = [f for f in mp3 if f.get("format") == "VBR MP3"]
562 picked = vbr or mp3
563 tracks = []
564 for f in picked:
565 tracks.append(
566 {
567 "title": _first(f.get("title")) or f["name"].rsplit(".", 1)[0],
568 "length": _clock(_first(f.get("length"))),
569 "track": _first(f.get("track")),
570 "url": "https://archive.org/download/" + urllib.parse.quote(identifier) + "/" + urllib.parse.quote(f["name"]),
571 }
572 )
573
574 def track_key(t: dict):
575 m = re.match(r"\d+", t["track"] or "")
576 return (int(m.group()) if m else 9999, t["url"])
577
578 if all(t["track"] for t in tracks):
579 tracks.sort(key=track_key)
580 return {
581 "identifier": identifier,
582 "title": _first(md.get("title")),
583 "date": _first(md.get("date"))[:10],
584 "venue": _first(md.get("venue")),
585 "city": _first(md.get("coverage")),
586 "taper": _first(md.get("taper")),
587 "transferer": _first(md.get("transferer")),
588 "source": _first(md.get("source"))[:200],
589 "details": "https://archive.org/details/" + urllib.parse.quote(identifier),
590 "tracks": tracks,
591 }
592