#!/usr/bin/env python3 """OPTION C — pull single tracks out of MTG's remote tars without downloading them. **Why this exists, for a reader arriving cold.** The house-wide manifest's 486 tracks are spread across all 100 of MTG-Jamendo's ~5.4 GB tar archives, so the obvious pull moves **544.93 GB to extract 5.9 GB** — measured at 108-250 hours. A tar has no central directory, but its members are a linked list of 512-byte headers and the mirror answers HTTP Range requests. So: walk the headers, and range-fetch only the members the manifest names. **Why this is not a hack.** MTG publishes a sha256 for **every track**, not just every tar (``data/download/raw_30s_audio_sha256_tracks.txt``). Every member this script writes is verified against that digest before it is placed. A file that does not match is deleted, not kept. The provenance chain is therefore *exactly* the chain the full tar pull would have produced — the same bytes, the same publisher, the same digests — for 1 % of the transfer. **Measured on the archive box, 2026-09-05:** * a fresh TLS handshake per request gives 1.32 req/s; **one persistent connection gives 6.92 req/s (145 ms/request)** — the walk is latency-bound, so keep-alive is worth 5.2x and this script holds one connection open; * the walk needs **44,670 header reads**, not 55,801: the member order is known from the published sha256 list, so each tar stops at its last wanted member (80.1 % of a full walk); * at the house's polite ~3 req/s that is **~4.1 h of walking** plus ~1.2 h of member fetches — call it **5-6 hours** against 108-250. **Correctness does not depend on the order file.** Member names come from the real tar headers as they arrive. The order file only supplies the early-stop index; if it were ever wrong the script would simply keep walking to the end of the tar rather than miss a member (see ``_walk_one_tar``). **Resumable.** State lives in ``walk-state.json``: per tar, the byte offset and member index reached, and whether it finished. A killed run resumes from the offset it had, not from zero. A track already on disk whose sha256 matches the published digest is never refetched. Runs on the archive box under systemd-run, never a terminal scope:: systemd-run --user --unit jamendo-ranged-walk \\ --working-directory="$HOME" --setenv=HOME="$HOME" \\ /usr/bin/python3 $HOME/house-wide-intake/ranged_walk.py """ from __future__ import annotations import argparse import collections import hashlib import http.client import json import ssl import sys import time from pathlib import Path USER_AGENT = ("strata2signal-research/1.0 " "(private research; contact via strata2signal.com)") HOST = "cdn.freesound.org" PATH_TEMPLATE = "/mtg-jamendo/raw_30s/audio/{tar}" BLOCK = 512 #: Requests per second ceiling. CONTEXT.md's polite rate for metadata; these are #: 512-byte reads, so 3/s is ~1.5 KB/s against a university CDN. DEFAULT_RATE = 3.0 FETCH_CHUNK = 1 << 20 MAX_ATTEMPTS = 5 # -------------------------------------------------------------------------- class Mirror: """One persistent HTTPS connection to the mirror, rate-limited and self-healing. Reconnects on any transport error; the caller retries.""" def __init__(self, rate: float = DEFAULT_RATE, timeout: int = 120): self.min_gap = 1.0 / rate if rate > 0 else 0.0 self.timeout = timeout self._conn: http.client.HTTPSConnection | None = None self._last = 0.0 self.requests = 0 self.bytes = 0 def _connect(self): if self._conn is None: self._conn = http.client.HTTPSConnection( HOST, timeout=self.timeout, context=ssl.create_default_context()) return self._conn def close(self): if self._conn is not None: try: self._conn.close() except OSError: pass self._conn = None def get_range(self, path: str, start: int, end: int) -> bytes: """Bytes [start, end] inclusive. Raises after MAX_ATTEMPTS.""" last_error = None for attempt in range(MAX_ATTEMPTS): gap = self.min_gap - (time.monotonic() - self._last) if gap > 0: time.sleep(gap) self._last = time.monotonic() try: conn = self._connect() conn.request("GET", path, headers={ "User-Agent": USER_AGENT, "Range": f"bytes={start}-{end}", "Accept-Encoding": "identity", "Connection": "keep-alive", }) response = conn.getresponse() payload = response.read() if response.status not in (200, 206): raise OSError(f"HTTP {response.status} for bytes={start}-{end}") self.requests += 1 self.bytes += len(payload) return payload except Exception as exc: # noqa: BLE001 last_error = exc self.close() time.sleep(min(2 ** attempt, 30)) raise OSError(f"{path} bytes={start}-{end}: {last_error}") # -------------------------------------------------------------------------- def parse_header(block: bytes): """(name, size) from one 512-byte ustar header; None at the archive end.""" if len(block) < BLOCK: return None name = block[:100].rstrip(b"\0").decode("utf-8", "replace") if not name: return None size_field = block[124:136].rstrip(b"\0 ").decode("ascii", "replace") return name, int(size_field or "0", 8) def member_data_span(offset: int, size: int): """(first, last) byte offsets of a member's data, given its header offset.""" return offset + BLOCK, offset + BLOCK + size - 1 def next_offset(offset: int, size: int) -> int: return offset + BLOCK + ((size + BLOCK - 1) // BLOCK) * BLOCK def sha256_bytes(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() def sha256_file(path: Path, chunk: int = FETCH_CHUNK) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(chunk), b""): digest.update(block) return digest.hexdigest() # -------------------------------------------------------------------------- class Log: """Append-only run log with a running count line, flushed every write.""" def __init__(self, path: Path): self.path = path path.parent.mkdir(parents=True, exist_ok=True) self.handle = path.open("a", encoding="utf-8") def __call__(self, message: str): line = f"{time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())} {message}" self.handle.write(line + "\n") self.handle.flush() print(line, file=sys.stderr, flush=True) def count(self, done: int, total: int, started: float, mirror: Mirror): elapsed = time.time() - started rate = done / elapsed * 3600 if elapsed else 0 remaining = (total - done) / rate if rate else 0 self(f"COUNT {done}/{total} tracks {100 * done / total:5.1f}% " f"{mirror.requests:,} requests {mirror.bytes / 1e6:.1f} MB " f"{elapsed / 3600:.2f} h elapsed ETA {remaining:.2f} h") # -------------------------------------------------------------------------- def load_member_order(order_path: Path) -> dict[str, list[str]]: """{'14': ['14/12414.mp3', ...]} -- the order MTG packed each tar in.""" order: dict[str, list[str]] = collections.defaultdict(list) with order_path.open(encoding="utf-8") as handle: for line in handle: if line.strip(): _, name = line.split() order[name.split("/", 1)[0]].append(name) return order def already_good(dest: Path, published: str | None) -> bool: """True when dest exists and matches MTG's published digest.""" if not dest.exists() or dest.stat().st_size == 0: return False if published is None: return False return sha256_file(dest) == published def _walk_one_tar(tar: str, wanted: dict, mirror: Mirror, audio_dir: Path, track_sha: dict, state: dict, log: Log, stop_index: int | None): """Walk `tar`, fetching and verifying every member `wanted` names. `wanted` maps dataset path -> manifest record. `stop_index` is the member index after which nothing else is wanted -- an optimisation only: if the walk reaches it with members still outstanding, it keeps going. """ path = PATH_TEMPLATE.format(tar=tar) entry = state.setdefault(tar, {"offset": 0, "index": 0, "done": False, "placed": [], "failed": []}) offset, index = entry["offset"], entry["index"] outstanding = {p for p in wanted if wanted[p]["track_id"] not in entry["placed"]} while outstanding: header = parse_header(mirror.get_range(path, offset, offset + BLOCK - 1)) if header is None: log(f"END {tar}: archive ended at member {index} with " f"{len(outstanding)} member(s) unfound: {sorted(outstanding)}") break name, size = header if name in outstanding: record = wanted[name] published = track_sha.get(name) first, last = member_data_span(offset, size) payload = mirror.get_range(path, first, last) got = sha256_bytes(payload) if published and got != published: log(f"BAD {tar}:{name} sha256 {got[:12]}… != published " f"{published[:12]}… -- discarded, will retry on the next run") entry["failed"] = sorted(set(entry["failed"]) | {record["track_id"]}) else: dest = audio_dir / f"{record['track_id']}.mp3" part = dest.with_suffix(".part") part.write_bytes(payload) part.replace(dest) entry["placed"] = sorted(set(entry["placed"]) | {record["track_id"]}) entry["failed"] = [t for t in entry["failed"] if t != record["track_id"]] log(f"OK {tar}:{name} -> {dest.name} {len(payload):,} B " f"sha256 verified against MTG's published digest") outstanding.discard(name) offset = next_offset(offset, size) index += 1 entry["offset"], entry["index"] = offset, index if stop_index is not None and index > stop_index and outstanding: # The order file said we were done; we are not. Keep walking and say so. log(f"NOTE {tar}: past the expected last-wanted member ({stop_index}) " f"with {len(outstanding)} outstanding -- continuing to the end") stop_index = None entry["done"] = not outstanding return entry def main(argv=None): ap = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) home = Path.home() ap.add_argument("--manifest", type=Path, default=home / "house-wide-intake" / "manifest.jsonl") ap.add_argument("--order", type=Path, default=home / "house-wide-intake" / "raw_30s_audio_sha256_tracks.txt") ap.add_argument("--audio", type=Path, default=home / "music" / "corpus" / "house-wide" / "audio") ap.add_argument("--log", type=Path, default=home / "music" / "corpus" / "_mirror" / "ranged-walk.log") ap.add_argument("--state", type=Path, default=home / "music" / "corpus" / "_mirror" / "walk-state.json") ap.add_argument("--rate", type=float, default=DEFAULT_RATE, help=f"requests per second ceiling (default {DEFAULT_RATE})") ap.add_argument("--count-every", type=int, default=10, help="emit a running COUNT line every N tracks") args = ap.parse_args(argv) args.audio.mkdir(parents=True, exist_ok=True) log = Log(args.log) with args.manifest.open(encoding="utf-8") as handle: records = [json.loads(line) for line in handle if line.strip()] track_sha = {} order = load_member_order(args.order) with args.order.open(encoding="utf-8") as handle: for line in handle: if line.strip(): digest, name = line.split() track_sha[name] = digest state = {} if args.state.exists(): state = json.loads(args.state.read_text(encoding="utf-8")) by_tar: dict[str, dict] = collections.defaultdict(dict) for record in records: by_tar[record["mirror_tar"]][record["dataset_path"]] = record # Highest-yield tars first: an interrupted run leaves the most tracks on disk. tars = sorted(by_tar, key=lambda t: (-len(by_tar[t]), t)) # Anything already on disk and matching its published digest is done. done_ids = set() for record in records: dest = args.audio / f"{record['track_id']}.mp3" if already_good(dest, record.get("published_sha256")): done_ids.add(record["track_id"]) started = time.time() mirror = Mirror(rate=args.rate) log(f"START {len(records)} tracks across {len(tars)} tars " f"{len(done_ids)} already verified on disk rate<={args.rate} req/s " f"dest={args.audio}") try: for position, tar in enumerate(tars, start=1): wanted = {p: r for p, r in by_tar[tar].items() if r["track_id"] not in done_ids} if not wanted: continue nn = tar.split("-")[-1].split(".")[0] members = order.get(nn, []) indices = [i for i, n in enumerate(members) if n in by_tar[tar]] stop_index = (max(indices) + 1) if indices else None log(f"TAR [{position}/{len(tars)}] {tar} {len(wanted)} wanted " f"expected last-wanted member {stop_index} of {len(members)}") entry = _walk_one_tar(tar, wanted, mirror, args.audio, track_sha, state, log, stop_index) done_ids |= set(entry["placed"]) args.state.write_text(json.dumps(state, indent=1, sort_keys=True) + "\n", encoding="utf-8") if position % max(1, args.count_every // 5) == 0 or entry["done"]: log.count(len(done_ids), len(records), started, mirror) finally: mirror.close() args.state.write_text(json.dumps(state, indent=1, sort_keys=True) + "\n", encoding="utf-8") failed = sorted({t for e in state.values() for t in e.get("failed", [])}) log.count(len(done_ids), len(records), started, mirror) log(f"DONE placed={len(done_ids)}/{len(records)} failed={len(failed)} " f"{mirror.requests:,} requests {mirror.bytes / 1e6:.1f} MB transferred") if failed: log(f"FAILED {failed}") return 0 if len(done_ids) == len(records) else 1 def _selftest(): header = bytearray(BLOCK) header[:14] = b"14/12414.mp3\0\0" header[124:136] = b"00050000000\0" # octal 50000000 = 10,485,760 assert parse_header(bytes(header)) == ("14/12414.mp3", 0o50000000) assert parse_header(bytes(BLOCK)) is None assert member_data_span(512, 100) == (1024, 1123) assert next_offset(0, 1) == 1024 assert next_offset(0, 512) == 1024 assert next_offset(0, 513) == 1536 print("ranged_walk selftest OK") if __name__ == "__main__": if "--selftest" in sys.argv: _selftest() else: raise SystemExit(main())