258 lines
14 KiB
Python
258 lines
14 KiB
Python
"""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()
|