#!/usr/bin/env python3 """Conservative, reproducible PDB restoration. Requires FFmpeg 7.1+. Creates separate derived files. Never changes input MP3s. Intermediate WAVs are 32-bit float: filtering can briefly produce peaks above full scale before the measured, peak-safe gain is applied. All measurements use decoded audio, not estimated durations in old MP3 headers. """ from __future__ import annotations import argparse import concurrent.futures import hashlib import json import math from pathlib import Path import re import subprocess import time PROFILE = { 'version': 'PDB_AB_ST_restoration_v2_2026-11-10', 'cleanup_filter': 'highpass=f=60:p=2,afftdn=nr=4:nf=-50:tn=1:ad=0.85:gs=10', 'target_lufs_mono': -19.0, 'preencode_true_peak_ceiling_dbtp': -2.5, 'maximum_positive_gain_db': 6.0, 'output_bitrate': '128k', 'output_sample_rate': 44100, 'output_channels': 1, 'intermediate_codec': 'pcm_f32le', 'dynamic_compression': False, 'silence_trimming': False, 'ai_speech_reconstruction': False, 'perceptual_listening_review': 'User A/B preview pending; technical QC is not a listening review.' } MEASURES = 'DC_offset+Peak_level+RMS_level+Number_of_samples+Number_of_NaNs+Number_of_Infs' def digest(p): h = hashlib.sha256() with open(p, 'rb') as f: for chunk in iter(lambda: f.read(4 * 1024 * 1024), b''): h.update(chunk) return h.hexdigest() def save(p, obj): p.parent.mkdir(parents=True, exist_ok=True) tmp = p.with_suffix(p.suffix + '.tmp') tmp.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding='utf-8') tmp.replace(p) def run(cmd, logpath): r = subprocess.run(cmd, capture_output=True, text=True) logpath.parent.mkdir(parents=True, exist_ok=True) logpath.write_text(r.stderr, encoding='utf-8') if r.returncode: raise RuntimeError(f'FFmpeg failed: {logpath.name}: {r.stderr[-1500:]}') return r.stderr def base(): return ['ffmpeg', '-nostdin', '-hide_banner', '-nostats', '-threads', '1', '-filter_threads', '1', '-filter_complex_threads', '1'] def meter(label): return f'ebur128@{label}=peak=true:framelog=quiet,astats@{label}=measure_perchannel=none:measure_overall={MEASURES}' def parse_metrics(log, label): # Named filters keep separate source, cleaned and exported measurements. e = re.search(r'\[ebur128@' + re.escape(label) + r'[^\]]*\] Summary:\s*(.*?)(?=\n\[|\Z)', log, re.S) if not e: raise ValueError(f'Missing ebur128 summary {label}') s = e.group(1) result = { 'integrated_lufs_mono': float(re.search(r'\bI:\s*([-\d.]+) LUFS', s).group(1)), 'lra_lu': float(re.search(r'\bLRA:\s*([-\d.]+) LU', s).group(1)), 'true_peak_dbtp': float(re.search(r'Peak:\s*([-\d.]+) dBFS', s).group(1)), } names = {'DC offset': 'dc_offset', 'Peak level dB': 'sample_peak_dbfs', 'RMS level dB': 'rms_dbfs', 'Number of samples': 'samples', 'Number of NaNs': 'nans', 'Number of Infs': 'infs'} for label_text, key in names.items(): pat = r'\[astats@' + re.escape(label) + r'[^\]]*\] ' + re.escape(label_text) + r':\s*([^\s]+)' m = re.search(pat, log) if not m: raise ValueError(f'Missing {key} for {label}') result[key] = float(m.group(1)) result['samples'] = int(result['samples']) result['duration_seconds'] = result['samples'] / PROFILE['output_sample_rate'] return result def probe(p): return json.loads(subprocess.check_output(['ffprobe', '-v', 'error', '-show_format', '-show_streams', '-of', 'json', str(p)], text=True)) def source_warnings(log): return [line for line in log.splitlines() if re.search( r'junk|inaccurate|invalid|error|corrupt|overread|underflow|Header missing', line, re.I)] def clean_one(item, a): ident = item['id'] src = a.sources / item['source_filename'] wav = a.work / 'float' / f'{ident}.wav' receipt = a.work / 'clean' / f'{ident}.json' if receipt.exists() and wav.exists(): old = json.loads(receipt.read_text()) if (old['source_sha256'] != digest(src) or old['profile'] != PROFILE or old.get('float_sha256') != digest(wav)): raise RuntimeError(f'Resume mismatch: {ident}') return old wav.parent.mkdir(parents=True, exist_ok=True) if wav.exists(): raise RuntimeError(f'Unreceipted intermediate exists; use a fresh work directory: {wav}') partial = wav.with_name(wav.stem + '.partial.wav') if partial.exists(): partial.unlink() info = probe(src) streams = [s for s in info['streams'] if s['codec_type'] == 'audio'] if len(streams) != 1 or int(streams[0]['sample_rate']) != 44100 or streams[0]['channels'] != 1: raise RuntimeError(f'Source format needs a separately evaluated profile: {ident}') graph = ('[0:a:0]asplit=2[original][work];[original]' + meter('source') + '[source_meter];[work]' + PROFILE['cleanup_filter'] + ',' + meter('clean') + '[cleaned]') cmd = base() + ['-i', str(src), '-filter_complex', graph, '-map', '[source_meter]', '-f', 'null', '-', '-map', '[cleaned]', '-c:a', 'pcm_f32le', '-ar', '44100', '-ac', '1', '-map_metadata', '-1', '-n', str(partial)] t = time.monotonic() log = run(cmd, a.work / 'logs' / f'{ident}-clean.log') orig, cleaned = parse_metrics(log, 'source'), parse_metrics(log, 'clean') if orig['samples'] != cleaned['samples'] or any(x['nans'] or x['infs'] for x in [orig, cleaned]): raise RuntimeError(f'Invalid cleanup length or samples: {ident}') partial.replace(wav) loudness_gain = PROFILE['target_lufs_mono'] - cleaned['integrated_lufs_mono'] peak_gain = PROFILE['preencode_true_peak_ceiling_dbtp'] - cleaned['true_peak_dbtp'] gain = min(loudness_gain, peak_gain, PROFILE['maximum_positive_gain_db']) result = {**item, 'profile': PROFILE, 'source_sha256': digest(src), 'source_bytes': src.stat().st_size, 'source_probe': info, 'source_metrics': orig, 'clean_metrics': cleaned, 'gain_db': round(gain, 3), 'gain_limited_by_peak': peak_gain < loudness_gain, 'decoder_warnings': source_warnings(log), 'float_wav': str(wav), 'float_sha256': digest(wav), 'cleanup_wall_seconds': round(time.monotonic() - t, 2)} save(receipt, result) print(f'CLEAN {ident}: {orig["duration_seconds"]:.2f}s; I={orig["integrated_lufs_mono"]:.1f} -> clean {cleaned["integrated_lufs_mono"]:.1f}; gain={gain:.2f}dB', flush=True) return result def encode_one(item, a): ident = item['id'] r = json.loads((a.work / 'clean' / f'{ident}.json').read_text()) src = a.sources / item['source_filename'] out = a.output / item['output_filename'] receipt = a.work / 'encoded' / f'{ident}.json' if (r['profile'] != PROFILE or digest(src) != r['source_sha256'] or digest(r['float_wav']) != r.get('float_sha256')): raise RuntimeError(f'Cleanup/source changed before encoding: {ident}') if receipt.exists() and out.exists(): old = json.loads(receipt.read_text()) if (old['output_sha256'] != digest(out) or old['profile'] != PROFILE or old['source_sha256'] != r['source_sha256'] or old.get('float_sha256') != r['float_sha256'] or old['gain_db'] != r['gain_db']): raise RuntimeError(f'Output changed: {ident}') return old a.output.mkdir(parents=True, exist_ok=True) if out.exists(): raise RuntimeError(f'Unreceipted output exists; use a fresh output directory: {out}') partial = out.with_name(out.stem + '.partial.mp3') if partial.exists(): partial.unlink() title = r['source_probe']['format'].get('tags', {}).get('title', item['label']) cmd = base() + ['-i', r['float_wav'], '-i', str(src), '-map', '0:a:0', '-map', '1:v?', '-map_metadata', '1', '-c:v', 'copy', '-af', f'volume={r["gain_db"]:.3f}dB', '-c:a', 'libmp3lame', '-b:a', PROFILE['output_bitrate'], '-ar', '44100', '-ac', '1', '-id3v2_version', '3', '-write_xing', '1', '-metadata', f'title={title} [Restored v2]', '-metadata', 'encoded_by=FFmpeg 7.1.5 / PDB conservative restoration v2', '-metadata', 'comment=Restored listening copy, 2026-11-10. Original preserved. Gentle rumble/noise reduction; measured fixed gain; no speech synthesis or edits.', '-metadata', 'replaygain_track_gain=', '-metadata', 'replaygain_album_gain=', '-metadata', 'iTunNORM=', '-metadata', 'iTunSMPB=', '-metadata', 'TLEN=', '-n', str(partial)] run(cmd, a.work / 'logs' / f'{ident}-encode.log') partial.replace(out) result = {**r, 'output_path': str(out), 'output_sha256': digest(out), 'output_bytes': out.stat().st_size, 'output_probe': probe(out)} save(receipt, result) print(f'ENCODE {ident}: {out.stat().st_size / 1e6:.2f} MB', flush=True) return result def verify_one(item, a): ident = item['id'] r = json.loads((a.work / 'encoded' / f'{ident}.json').read_text()) receipt = a.work / 'verified' / f'{ident}.json' if (r['profile'] != PROFILE or digest(a.sources / item['source_filename']) != r['source_sha256'] or digest(r['output_path']) != r['output_sha256']): raise RuntimeError(f'Source/output changed before verification: {ident}') if receipt.exists(): old = json.loads(receipt.read_text()) if (old['profile'] != PROFILE or old['source_sha256'] != r['source_sha256'] or old['output_sha256'] != r['output_sha256'] or old.get('float_sha256') != r.get('float_sha256') or old['gain_db'] != r['gain_db']): raise RuntimeError(f'Cached verification mismatch: {ident}') return old cmd = base() + ['-i', r['output_path'], '-map', '0:a:0', '-vn', '-af', meter('output'), '-f', 'null', '-'] log = run(cmd, a.work / 'logs' / f'{ident}-verify.log') m = parse_metrics(log, 'output') audio = [s for s in r['output_probe']['streams'] if s['codec_type'] == 'audio'][0] checks = { 'exact_decoded_sample_count': m['samples'] == r['source_metrics']['samples'], 'finite_samples': m['nans'] == 0 and m['infs'] == 0, 'true_peak_at_most_minus_1_5_dbtp': m['true_peak_dbtp'] <= -1.5, 'mono_44100': audio['channels'] == 1 and int(audio['sample_rate']) == 44100, 'mp3_128k': audio['codec_name'] == 'mp3' and int(audio['bit_rate']) == 128000, 'original_hash_unchanged': digest(a.sources / item['source_filename']) == r['source_sha256'], 'no_output_decoder_warnings': not source_warnings(log), } result = {**r, 'output_metrics': m, 'checks': checks, 'technical_qc_passed': all(checks.values()), 'output_decoder_warnings': source_warnings(log)} save(receipt, result) print(f'VERIFY {ident}: I={m["integrated_lufs_mono"]:.1f}LUFS TP={m["true_peak_dbtp"]:.1f}dBTP samples={m["samples"]} PASS={all(checks.values())}', flush=True) return result def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument('--sources', type=Path, required=True) p.add_argument('--manifest', type=Path, required=True) p.add_argument('--output', type=Path, required=True) p.add_argument('--work', type=Path, required=True) p.add_argument('--stage', choices=['clean', 'encode', 'verify', 'all'], default='all') p.add_argument('--jobs', type=int, default=4) p.add_argument('--only', help='Optional comma-separated semantic IDs') a = p.parse_args() items = json.loads(a.manifest.read_text())['files'] if a.only: selected = set(a.only.split(',')); items = [r for r in items if r['id'] in selected] for stage in (['clean', 'encode', 'verify'] if a.stage == 'all' else [a.stage]): func = {'clean': clean_one, 'encode': encode_one, 'verify': verify_one}[stage] with concurrent.futures.ThreadPoolExecutor(max_workers=a.jobs) as pool: results = list(pool.map(lambda x: func(x, a), items)) save(a.work / f'{stage}-summary.json', {'profile': PROFILE, 'files': results}) if a.stage in ('verify', 'all') and not all(r['technical_qc_passed'] for r in results): raise SystemExit('One or more outputs failed technical QC. Inspect receipts before delivery.') if __name__ == '__main__': main()