"""Find-Word v0.2.3 현장 녹음 파생본과 기계 판독 provenance를 만든다. 원본은 명시적으로 받은 /tmp 캐시에만 두고 네트워크를 호출하지 않는다. FFmpeg는 구간 선택·다운믹스·리샘플만 담당하며 fade와 선형 gain은 여기서 결정한다. 원본/출력 해시와 실제 변환 설정으로 다른 사람이 결과를 추적한다. """ from array import array from datetime import datetime, timezone import ctypes import hashlib import json import math import os from pathlib import Path import shutil import subprocess import sys import tempfile import wave ROOT = Path(__file__).resolve().parent.parent # 환경 목록만 고정한다. 구간 시작점은 승인 계획의 불변 provenance에서만 읽는다. ENVIRONMENT_IDS = frozenset(("street", "mall", "theater", "department", "market", "subway", "cafe", "rain")) EXCERPT_SECONDS = 12 SAMPLE_RATE = 24000 FADE_SECONDS = 0.05 def digest(path): """큰 원본도 메모리에 전부 적재하지 않고 SHA-256을 계산한다.""" with Path(path).open("rb") as stream: return hashlib.file_digest(stream, "sha256").hexdigest() def convert(source, destination, start, duration=EXCERPT_SECONDS): """실제 PCM에 짧은 양끝 fade와 공통 gain을 적용해 24kHz mono PCM16으로 쓴다. start/duration은 원본의 초 단위 구간이다. RMS 목표 .08, peak .85, 최대 증폭 4배 중 가장 작은 gain으로 과증폭을 막고 파형의 비율을 보존한다. """ if not 8 <= duration <= 20 or not math.isfinite(start) or start < 0: raise ValueError("현장음 구간은 8~20초, 시작점은 0 이상이어야 합니다.") output_options = ["-t", str(duration), "-ar", str(SAMPLE_RATE), "-ac", "1", "-f", "f32le", "pipe:1"] seek_method = "input seek" decoded = subprocess.run([ "ffmpeg", "-v", "error", "-ss", str(start), "-i", str(source), *output_options, ], check=True, capture_output=True, timeout=90).stdout if len(decoded) < duration * SAMPLE_RATE * 4: # 일부 Ogg seek/timestamp 조합은 몇 프레임이 짧아진다. 처음부터 디코드하여 # 승인 초를 24kHz sample 경계로 선택한다. PCM 추가·패딩·시간 변형은 하지 않는다. # 출력 PTS를 sample 수로 다시 매겨 원본 PTS 불연속이 -t 길이를 줄이지 않게 한다. seek_method = "full decode then sample trim" first_frame = round(start * SAMPLE_RATE) last_frame = first_frame + round(duration * SAMPLE_RATE) selection = f"aresample={SAMPLE_RATE},atrim=start_sample={first_frame}:end_sample={last_frame},asetpts=N/SR/TB" decoded = subprocess.run([ "ffmpeg", "-v", "error", "-i", str(source), "-af", selection, *output_options, ], check=True, capture_output=True, timeout=90).stdout samples = array("f", decoded) if sys.byteorder != "little": samples.byteswap() if len(samples) != duration * SAMPLE_RATE or not all(math.isfinite(x) for x in samples): raise ValueError("원본 구간이 잘렸거나 PCM이 올바르지 않습니다.") # 경계의 DC도 함께 0으로 내려 루프 접합점의 불연속 클릭을 막는다. fade_frames = round(FADE_SECONDS * SAMPLE_RATE) faded = [x * min(1, i / fade_frames, (len(samples) - 1 - i) / fade_frames) for i, x in enumerate(samples)] rms = math.sqrt(sum(x*x for x in faded) / len(faded)) peak = max(abs(x) for x in faded) if rms < 0.001: raise ValueError("무음 또는 지나치게 작은 현장 녹음입니다.") gain = min(0.08 / rms, 0.85 / peak, 4) pcm = array("h", (round(x * gain * 32767) for x in faded)) measured_rms = math.sqrt(sum(x*x for x in pcm) / len(pcm)) / 32768 measured_peak = max(abs(x) for x in pcm) / 32768 if sys.byteorder != "little": pcm.byteswap() with wave.open(str(destination), "wb") as stream: stream.setparams((1, 2, SAMPLE_RATE, 0, "NONE", "not compressed")) stream.writeframes(pcm.tobytes()) return {"sha256": digest(destination), "bytes": Path(destination).stat().st_size, "duration": len(samples) / SAMPLE_RATE, "rms": measured_rms, "peak": measured_peak, "conversion": {"startSeconds": start, "durationSeconds": duration, "seekMethod": seek_method, "fadeSeconds": FADE_SECONDS, "sampleRate": SAMPLE_RATE, "channels": 1, "encoding": "PCM16 little-endian", "linearGain": gain, "targetRms": 0.08, "peakLimit": 0.85, "maxGain": 4, "operations": "excerpt, downmix, resample, linear end fades, bounded linear normalization, WAV encoding"}} def source_path(directory, environment): """승인된 환경의 직접 미디어 확장자를 사용한다.""" extension = ".mp3" if environment in ("theater", "department") else ".flac" if environment == "rain" else ".ogg" return Path(directory) / (environment + extension) def validate_plan(plan): """선택 구간은 직접 미디어 길이 안의 유한한 수여야 하며 보정하거나 추정하지 않는다.""" assets = plan["assets"] if len(assets) != 8 or {a["environmentId"] for a in assets} != ENVIRONMENT_IDS: raise ValueError("정확한 승인 원본 8개가 필요합니다.") for asset in assets: start = asset.get("selectedSegmentStartSeconds") duration = asset.get("directMediaDurationSeconds") if (any(type(value) not in (int, float) or not math.isfinite(value) for value in (start, duration)) or start < 0 or duration <= 0 or start + EXCERPT_SECONDS > duration): raise ValueError(f"{asset['environmentId']}: 승인 구간은 직접 미디어 길이 안의 12초여야 합니다.") def verify_sources(plan, directory): """여덟 원본 모두의 불변 승인값을 확인한 뒤에만 첫 변환을 허용한다.""" validate_plan(plan) assets = plan["assets"] for asset in assets: source = source_path(directory, asset["environmentId"]) if (not source.is_file() or source.is_symlink() or source.stat().st_size != asset.get("sourceBytes") or digest(source) != asset.get("expectedSourceSha256")): raise ValueError(f"{asset['environmentId']}: 승인 원본 SHA-256/bytes 불일치") def render_notice(files): """각 항목 자체에 저작자·두 URL·원본/파생 라이선스·변경 고지를 완결한다.""" lines = ["# Find-Word v0.2.3 — Field recording notices", "", "These third-party recordings retain their own licenses. The private project license does not apply to them.", "All files are modified excerpts: trimmed, faded, downmixed, resampled, normalized and converted to WAV.", "The market derivative remains CC BY-SA 4.0. No additional restrictions are imposed on that audio.", ""] for entry in files: conversion = entry["conversion"] start = conversion["startSeconds"] end = start + conversion["durationSeconds"] lines += [f"## {entry['displayName']} — {entry['environmentId']}", "", entry["attribution"], "", f"Creator: {entry['creator']}", f"Source: {entry['pageUrl']}", f"Download: {entry['directUrl']}", f"License: {entry['license']} — {entry['licenseUrl']}", f"Output license: {entry['outputLicense']}", f"Changes: {conversion['operations']}; source seconds {start}–{end}.", ""] return "\n".join(lines) def validate_bundle(directory, plan): """공개 전 staged 파일을 다시 읽어 manifest 자기 보고와 실제 PCM을 대조한다.""" try: validate_plan(plan) expected = {f"{name}.wav" for name in ENVIRONMENT_IDS} | {"manifest.json", "AMBIENCE_NOTICES.md"} if {p.name for p in directory.iterdir()} != expected or any( p.is_symlink() or not p.is_file() for p in directory.iterdir()): raise ValueError("현장음 파일 누락·추가·symlink") manifest = json.loads((directory / "manifest.json").read_text()) entries = manifest["files"] if (manifest["version"] != "0.2.3" or len(entries) != 8 or {e["environmentId"] for e in entries} != ENVIRONMENT_IDS or len({e["sha256"] for e in entries}) != 8): raise ValueError("고유한 승인 현장음 8개가 필요합니다.") approved = {a["environmentId"]: a for a in plan["assets"]} for entry in entries: environment = entry["environmentId"] asset = approved[environment] if (any(entry.get(key) != value for key, value in asset.items()) or entry["file"] != f"{environment}.wav" or entry["sourceSha256"] != asset["expectedSourceSha256"] or entry["outputLicense"] != asset["license"] or not entry["conversion"]["operations"].strip()): raise ValueError("승인 출처 또는 파생본 메타데이터 불일치") path = directory / entry["file"] if path.stat().st_size != entry["bytes"] or digest(path) != entry["sha256"]: raise ValueError("현장음 해시/bytes 불일치") with wave.open(str(path), "rb") as stream: if (stream.getframerate(), stream.getnchannels(), stream.getsampwidth(), stream.getcomptype()) != (SAMPLE_RATE, 1, 2, "NONE"): raise ValueError("24kHz mono PCM16이 아닙니다.") frames = stream.getnframes() raw = stream.readframes(frames) samples = array("h", raw) if sys.byteorder != "little": samples.byteswap() if (frames != EXCERPT_SECONDS * SAMPLE_RATE or len(raw) != frames * 2 or path.stat().st_size != len(raw) + 44 or samples[0] != 0 or samples[-1] != 0): raise ValueError("12초 PCM 또는 fade 경계 불일치") rms = math.sqrt(sum(x*x for x in samples) / frames) / 32768 peak = max(abs(x) for x in samples) / 32768 if (rms <= 0.003 or peak > 0.85 or not math.isclose(rms, entry["rms"], abs_tol=1e-12) or not math.isclose(peak, entry["peak"], abs_tol=1e-12) or entry["duration"] != frames / SAMPLE_RATE or entry["conversion"]["startSeconds"] != asset["selectedSegmentStartSeconds"] or entry["conversion"]["durationSeconds"] != EXCERPT_SECONDS or entry["conversion"]["fadeSeconds"] != FADE_SECONDS): raise ValueError("현장음 실측값/변환 불일치") if (directory / "AMBIENCE_NOTICES.md").read_text() != render_notice(entries): raise ValueError("항목별 출처/변경 고지 불일치") except (KeyError, TypeError, json.JSONDecodeError, wave.Error, EOFError) as error: raise ValueError("현장음 bundle 검증 실패") from error def atomic_publish(staged, output): """Linux의 단일 renameat2 교환이 commit 지점이다. 실패 시 기존 이름은 유지된다. 성공 전에는 기존 디렉터리를 건드리지 않는다. 성공 후에는 완성된 새 디렉터리만 그 이름에 보이며 이전 트리는 staging에 남는다. 두 번 rename하는 fallback은 없다. """ if output.is_symlink() or (output.exists() and not output.is_dir()): raise ValueError("현장음 출력은 symlink가 아닌 디렉터리여야 합니다.") if not output.exists(): os.rename(staged, output) return libc = ctypes.CDLL(None, use_errno=True) exchange = getattr(libc, "renameat2", None) if exchange is None: raise OSError("renameat2 원자 교환을 지원하지 않습니다; 이전 디렉터리 보존") exchange.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] exchange.restype = ctypes.c_int if exchange(-100, os.fsencode(staged), -100, os.fsencode(output), 2) != 0: raise OSError(ctypes.get_errno(), "원자 교환 실패; 이전 디렉터리 보존") def generate(plan, source_directory, output, *, converter=None): """모든 원본과 파생본/manifest/notice를 검증한 뒤 디렉터리 전체를 공개한다.""" output = Path(output) if output.is_symlink() or (output.exists() and not output.is_dir()): raise ValueError("현장음 출력은 symlink가 아닌 디렉터리여야 합니다.") verify_sources(plan, source_directory) converter = converter or convert ffmpeg = subprocess.check_output(["ffmpeg", "-version"], text=True).splitlines()[0] output.parent.mkdir(parents=True, exist_ok=True) # 같은 부모 아래 만들어 EXDEV를 피한다. 기존 파일을 overlay하지 않으므로 stale 파일도 사라진다. working = Path(tempfile.mkdtemp(prefix=".ambience-stage-", dir=output.parent)) ready = working / "ready" ready.mkdir() files = [] try: for asset in plan["assets"]: environment = asset["environmentId"] name = environment + ".wav" info = converter(source_path(source_directory, environment), ready / name, asset["selectedSegmentStartSeconds"]) entry = dict(asset) entry.update({"file": name, "sourceSha256": asset["expectedSourceSha256"], "outputLicense": asset["license"], "verificationDateUtc": "2026-10-14", **info}) files.append(entry) manifest = {"version": "0.2.3", "generatedAtUtc": datetime.now(timezone.utc).isoformat(), "toolVersions": {"ffmpeg": ffmpeg}, "sampleRate": SAMPLE_RATE, "channels": 1, "encoding": "PCM16", "files": files} (ready / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n") (ready / "AMBIENCE_NOTICES.md").write_text(render_notice(files)) validate_bundle(ready, plan) atomic_publish(ready, output) return manifest finally: # commit 뒤 정리 실패를 발행 실패로 오인하지 않는다. SIGKILL은 숨은 stage를 남길 수 있으나 # 공개 이름은 syscall 전의 전체 이전 트리 또는 syscall 후의 전체 새 트리다. shutil.rmtree(working, ignore_errors=True) def main(): plan = json.loads((ROOT / ".agent/AMBIENCE-SOURCES-v0.2.3.json").read_text()) manifest = generate(plan, Path("/tmp/findword-ambience-sources"), ROOT / "public/ambience") print(json.dumps({"files": len(manifest["files"]), "bytes": sum(f["bytes"] for f in manifest["files"])}, indent=2)) if __name__ == "__main__": main()