396 lines
22 KiB
Python
396 lines
22 KiB
Python
"""Find-Word v0.2.5 현장 녹음 파생본과 기계 판독 provenance를 만든다.
|
||
|
||
원본은 명시적으로 받은 /tmp 캐시에만 두고 네트워크를 호출하지 않는다.
|
||
전용 FFmpeg ebur128 실측에 따른 volume 스칼라만 적용하고 마지막 PCM을 재측정한다.
|
||
출처·구간·50ms fade는 보존하고 측정 실패나 목표 이탈은 원자 공개 전에 거부한다. 원본/출력 해시와 실제 변환 설정으로 다른 사람이 결과를 추적한다.
|
||
"""
|
||
from array import array
|
||
from datetime import datetime, timezone
|
||
import ctypes
|
||
import hashlib
|
||
import json
|
||
import math
|
||
import os
|
||
import re
|
||
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
|
||
|
||
# ebur128의 0.1 dB 보고 정밀도와 PCM16 양자화를 위해 내부 ceiling에 0.2 dB 여유를 둔다.
|
||
TARGET_LUFS = -32.5
|
||
TRUE_PEAK_DBTP = -6.0
|
||
INTERNAL_TRUE_PEAK_DBTP = -6.2
|
||
LOUDNESS_FILTER = "ebur128=peak=true"
|
||
NORMALIZATION = {
|
||
"algorithm": "FFmpeg ebur128 measurement; scalar volume in double precision; PCM16; fresh ebur128 verification",
|
||
"targetIntegratedLUFS": TARGET_LUFS, "truePeakCeilingDBTP": TRUE_PEAK_DBTP,
|
||
"internalTruePeakCeilingDBTP": INTERNAL_TRUE_PEAK_DBTP,
|
||
"toleranceLU": 0.2, "maximumSpreadLU": 0.4, "measurementFilter": LOUDNESS_FILTER,
|
||
"maximumIterations": 3,
|
||
}
|
||
STAT_KEYS = ("integratedLUFS", "truePeakDBTP", "loudnessRangeLU", "thresholdLUFS")
|
||
PCM_INPUT_ARGS = ["ffmpeg", "-nostdin", "-hide_banner", "-f", "f32le", "-ar", str(SAMPLE_RATE),
|
||
"-ac", "1", "-i", "pipe:0"]
|
||
|
||
|
||
def validate_stats(stats):
|
||
"""누락/NaN/inf는 통과값으로 대체하지 않는다."""
|
||
if set(stats) != set(STAT_KEYS) or not all(
|
||
type(stats[key]) in (int, float) and math.isfinite(stats[key]) for key in STAT_KEYS):
|
||
raise ValueError("유한한 ebur128 통계가 아닙니다.")
|
||
|
||
|
||
def parse_loudness(stderr):
|
||
"""프레임 진단값 대신 전용 ebur128의 마지막 전체 구간 Summary만 읽는다."""
|
||
if isinstance(stderr, bytes):
|
||
stderr = stderr.decode("utf-8")
|
||
try:
|
||
summary = stderr.rsplit("Summary:", 1)[1]
|
||
patterns = {"integratedLUFS": r"I:\s+([-\d.]+) LUFS",
|
||
"truePeakDBTP": r"Peak:\s+([-\d.]+) dBFS",
|
||
"loudnessRangeLU": r"LRA:\s+([-\d.]+) LU",
|
||
"thresholdLUFS": r"Threshold:\s+([-\d.]+) LUFS"}
|
||
result = {key: float(re.search(pattern, summary)[1]) for key, pattern in patterns.items()}
|
||
validate_stats(result)
|
||
return result
|
||
except (IndexError, TypeError, ValueError) as error:
|
||
raise ValueError("FFmpeg ebur128 측정 실패") from error
|
||
|
||
|
||
def measure_loudness(path=None, *, pcm=None):
|
||
"""매 호출 새 프로세스로 정확한 faded f32 PCM 또는 저장된 WAV 전체를 읽는다."""
|
||
if (path is None) == (pcm is None):
|
||
raise ValueError("하나의 측정 입력만 필요합니다.")
|
||
command = PCM_INPUT_ARGS if pcm is not None else ["ffmpeg", "-nostdin", "-hide_banner", "-i", str(path)]
|
||
result = subprocess.run([*command, "-af", LOUDNESS_FILTER, "-f", "null", "-"],
|
||
input=pcm, check=True, capture_output=True, timeout=90)
|
||
return parse_loudness(result.stderr)
|
||
|
||
|
||
def validate_loudness(stats):
|
||
"""최종 PCM의 ±0.2 LU 및 -6 dBTP 수용 조건은 자기 보고로 면제하지 않는다."""
|
||
validate_stats(stats)
|
||
if (abs(stats["integratedLUFS"] - TARGET_LUFS) > NORMALIZATION["toleranceLU"] + 1e-9
|
||
or stats["truePeakDBTP"] > TRUE_PEAK_DBTP):
|
||
raise ValueError("최종 loudness/true peak 계약 위반")
|
||
|
||
|
||
def volume_filter(gain):
|
||
return f"volume={gain:.12f}dB:precision=double"
|
||
|
||
|
||
def normalize_pcm(faded_pcm, destination):
|
||
"""각 시도는 같은 원본 PCM에서 시작한다. 누적 재양자화/리미터/압축은 사용하지 않는다.
|
||
|
||
입력 peak + gain을 변환 전에 검사하고 양자화 후 새 계기로 재측정한다.
|
||
목표 이탈 때만 최대 3회까지 스칼라를 보정하며 비수렴은 staging 검증 전에 실패한다.
|
||
"""
|
||
source = measure_loudness(pcm=faded_pcm)
|
||
validate_stats(source)
|
||
gain = round(TARGET_LUFS - source["integratedLUFS"], 12)
|
||
iterations = []
|
||
for _ in range(NORMALIZATION["maximumIterations"]):
|
||
if source["truePeakDBTP"] + gain > INTERNAL_TRUE_PEAK_DBTP + 1e-9:
|
||
raise ValueError("예측 true peak가 내부 ceiling을 초과하여 변환 중단")
|
||
exact_filter = volume_filter(gain)
|
||
encoded = subprocess.run([*PCM_INPUT_ARGS, "-af", exact_filter, "-c:a", "pcm_s16le",
|
||
"-f", "s16le", "pipe:1"], input=faded_pcm,
|
||
check=True, capture_output=True, timeout=90).stdout
|
||
if len(encoded) != len(faded_pcm) // 2:
|
||
raise ValueError("정규화 PCM 길이 오류")
|
||
with wave.open(str(destination), "wb") as stream:
|
||
stream.setparams((1, 2, SAMPLE_RATE, 0, "NONE", "not compressed"))
|
||
stream.writeframes(encoded)
|
||
final = measure_loudness(destination)
|
||
validate_stats(final)
|
||
iterations.append({"gainDB": gain, "filter": exact_filter,
|
||
"predictedTruePeakDBTP": round(source["truePeakDBTP"] + gain, 12),
|
||
"output": final})
|
||
if final["truePeakDBTP"] > INTERNAL_TRUE_PEAK_DBTP:
|
||
raise ValueError("양자화 후 true peak가 내부 ceiling 초과")
|
||
if abs(final["integratedLUFS"] - TARGET_LUFS) <= NORMALIZATION["toleranceLU"] + 1e-9:
|
||
validate_loudness(final)
|
||
return {"measurementFilter": LOUDNESS_FILTER, "fadedPcmSha256": hashlib.sha256(faded_pcm).hexdigest(),
|
||
"input": source, "iterations": iterations, "final": final}
|
||
gain = round(gain + TARGET_LUFS - final["integratedLUFS"], 12)
|
||
raise ValueError("PCM16 loudness 보정 비수렴; 공개 중단")
|
||
|
||
|
||
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):
|
||
"""source의 승인 start/duration 구간을 fade한 뒤 ebur128 실측에 따른 스칼라 gain만 적용한다.
|
||
|
||
source는 해시 확인한 원본, destination은 비공개 staging 파일이다. RMS 목표나
|
||
최대 증폭 제한은 쓰지 않으며 true peak/청감 계약을 만족하는 선형 변환만 공개한다.
|
||
"""
|
||
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))
|
||
if rms < 0.001:
|
||
raise ValueError("무음 또는 지나치게 작은 현장 녹음입니다.")
|
||
# 측정과 모든 변환 시도에 정확히 같은 faded float PCM을 전달한다.
|
||
faded_pcm = array("f", faded)
|
||
if sys.byteorder != "little": faded_pcm.byteswap()
|
||
loudness = normalize_pcm(faded_pcm.tobytes(), destination)
|
||
with wave.open(str(destination), "rb") as stream:
|
||
pcm = array("h", stream.readframes(stream.getnframes()))
|
||
if sys.byteorder != "little": pcm.byteswap()
|
||
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 pcm[0] != 0 or pcm[-1] != 0:
|
||
raise ValueError("정규화 후 fade 경계 손상")
|
||
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",
|
||
"loudness": loudness,
|
||
"operations": "excerpt, downmix, resample, linear end fades, FFmpeg double-precision scalar volume, PCM16 WAV encoding, fresh ebur128 integrated/true-peak measurement"}}
|
||
|
||
|
||
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.5 — 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.5" 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개가 필요합니다.")
|
||
if (manifest["normalization"] != NORMALIZATION
|
||
or not manifest["toolVersions"]["ffmpeg"].startswith("ffmpeg version ")):
|
||
raise ValueError("정규화 계약 또는 도구 버전 불일치")
|
||
measured_values = [] # 독립 재측정된 8개의 LUFS로 bundle 전체 spread를 검사한다.
|
||
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 > 10 ** (TRUE_PEAK_DBTP / 20) 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("현장음 실측값/변환 불일치")
|
||
actual = measure_loudness(path)
|
||
validate_loudness(actual)
|
||
loudness = entry["conversion"]["loudness"]
|
||
validate_provenance(loudness, actual)
|
||
measured_values.append(actual["integratedLUFS"])
|
||
if max(measured_values) - min(measured_values) > 0.4 + 1e-9:
|
||
raise ValueError("환경 간 loudness spread 초과")
|
||
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 validate_provenance(loudness, actual):
|
||
"""고정 계기·gain 공식·모든 시도·최종 실측을 서로 대조한다."""
|
||
source = loudness["input"]
|
||
validate_stats(source)
|
||
iterations = loudness["iterations"]
|
||
if (loudness["measurementFilter"] != LOUDNESS_FILTER
|
||
or not re.fullmatch(r"[a-f0-9]{64}", loudness["fadedPcmSha256"])
|
||
or not 1 <= len(iterations) <= NORMALIZATION["maximumIterations"]):
|
||
raise ValueError("선형 정규화 provenance 누락")
|
||
gain = round(TARGET_LUFS - source["integratedLUFS"], 12)
|
||
for index, iteration in enumerate(iterations):
|
||
validate_stats(iteration["output"])
|
||
if (iteration["gainDB"] != gain or iteration["filter"] != volume_filter(gain)
|
||
or iteration["predictedTruePeakDBTP"] != round(source["truePeakDBTP"] + gain, 12)
|
||
or source["truePeakDBTP"] + gain > INTERNAL_TRUE_PEAK_DBTP + 1e-9
|
||
or iteration["output"]["truePeakDBTP"] > INTERNAL_TRUE_PEAK_DBTP):
|
||
raise ValueError("선형 gain/peak provenance 불일치")
|
||
error = TARGET_LUFS - iteration["output"]["integratedLUFS"]
|
||
if index < len(iterations) - 1 and abs(error) <= NORMALIZATION["toleranceLU"] + 1e-9:
|
||
raise ValueError("불필요한 보정 반복")
|
||
gain = round(gain + error, 12)
|
||
if loudness["final"] != actual or iterations[-1]["output"] != actual:
|
||
raise ValueError("최종 ebur128 실측/provenance 불일치")
|
||
|
||
|
||
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.5", "normalization": NORMALIZATION, "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():
|
||
"""불변 v0.2.3 출처 계획을 읽되 v0.2.5 변환 계약으로만 새 bundle을 발행한다."""
|
||
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()
|