"""Validate and package the three fixed FT1 audio prompts.""" from __future__ import annotations import argparse import hashlib import json import shutil import subprocess import tempfile import zipfile from datetime import datetime, timezone from pathlib import Path EXPECTED_FILES = ( "factory_boot_zh.mp3", "factory_pass_zh.mp3", "factory_shutdown_zh.mp3", ) IMAGE_SIZE = 262144 def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest().upper() def _skip_id3(data: bytes) -> int: if not data.startswith(b"ID3") or len(data) < 10: return 0 size = 0 for value in data[6:10]: if value & 0x80: raise ValueError("ID3 使用了无效的 synchsafe size") size = (size << 7) | value return 10 + size def inspect_mp3(path: Path) -> dict: data = path.read_bytes() offset = _skip_id3(data) while offset + 4 <= len(data) and not (data[offset] == 0xFF and data[offset + 1] & 0xE0 == 0xE0): offset += 1 if offset + 4 > len(data): raise ValueError(f"{path.name}: 找不到 MPEG 音频帧") header = int.from_bytes(data[offset : offset + 4], "big") version_bits = (header >> 19) & 0x3 layer_bits = (header >> 17) & 0x3 bitrate_index = (header >> 12) & 0xF sample_index = (header >> 10) & 0x3 channel_mode = (header >> 6) & 0x3 if version_bits == 1 or layer_bits != 1 or bitrate_index in (0, 15) or sample_index == 3: raise ValueError(f"{path.name}: MP3 帧头不受支持") version = {3: "MPEG1", 2: "MPEG2", 0: "MPEG2.5"}[version_bits] sample_rates = { 3: (44100, 48000, 32000), 2: (22050, 24000, 16000), 0: (11025, 12000, 8000), } mpeg1_bitrates = (0, 32, 40, 48, 56, 64, 80, 96, 112, 128, 160, 192, 224, 256, 320) mpeg2_bitrates = (0, 8, 16, 24, 32, 40, 48, 56, 64, 80, 96, 112, 128, 144, 160) bitrate = (mpeg1_bitrates if version_bits == 3 else mpeg2_bitrates)[bitrate_index] sample_rate = sample_rates[version_bits][sample_index] channels = 1 if channel_mode == 3 else 2 if (sample_rate, channels, bitrate) != (16000, 1, 48): raise ValueError( f"{path.name}: 需要 16000Hz/单声道/48kbps，实际为 {sample_rate}Hz/{channels}ch/{bitrate}kbps" ) duration_ms = round(len(data) * 8 / (bitrate * 1000) * 1000) return { "name": path.name, "bytes": len(data), "sha256": sha256(path), "mpeg": version, "sample_rate_hz": sample_rate, "channels": channels, "bitrate_kbps": bitrate, "estimated_duration_ms": duration_ms, } def build_package(root: Path, output: Path) -> tuple[Path, Path, dict]: source = root / "audio_prompts" / "factory_test" / "customer_fs" tool = root / "ql-cross-tool" / "win32" / "mklfs-v2.exe" if not source.is_dir(): raise FileNotFoundError(f"音频目录不存在: {source}") if not tool.is_file(): raise FileNotFoundError(f"mklfs-v2.exe 不存在: {tool}") actual = sorted(path.name for path in source.iterdir() if path.is_file()) if actual != sorted(EXPECTED_FILES): raise ValueError(f"音频目录必须且只能包含 {EXPECTED_FILES}，实际为 {actual}") output.mkdir(parents=True, exist_ok=True) image = output / "customer_fs.bin" if image.exists(): image.unlink() audio = [inspect_mp3(source / name) for name in EXPECTED_FILES] result = subprocess.run( [str(tool), "-m", "-c", str(source), "-s", str(IMAGE_SIZE), "-i", str(image)], cwd=str(root), capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, ) if result.returncode != 0: raise RuntimeError(f"mklfs-v2 失败 ({result.returncode}): {result.stdout}\n{result.stderr}") if not image.is_file() or image.stat().st_size != IMAGE_SIZE: raise RuntimeError(f"customer_fs.bin 大小错误: {image.stat().st_size if image.exists() else '不存在'}") verified_files = [] with tempfile.TemporaryDirectory(prefix="verify-", dir=output) as verify_dir: verify_path = Path(verify_dir) verify = subprocess.run( [str(tool), "-o", "-c", str(verify_path), "-s", str(IMAGE_SIZE), "-i", str(image)], cwd=str(root), capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, ) if verify.returncode != 0: raise RuntimeError(f"customer_fs 解包验证失败 ({verify.returncode}): {verify.stdout}\n{verify.stderr}") extracted = sorted(path.name for path in verify_path.iterdir() if path.is_file()) if extracted != sorted(EXPECTED_FILES): raise RuntimeError(f"customer_fs 文件清单错误: {extracted}") for name in EXPECTED_FILES: source_hash = sha256(source / name) extracted_hash = sha256(verify_path / name) if source_hash != extracted_hash: raise RuntimeError(f"customer_fs 解包内容不一致: {name}") verified_files.append({"name": name, "sha256": extracted_hash}) try: revision = subprocess.run( ["git", "rev-parse", "HEAD"], cwd=str(root), capture_output=True, text=True, check=True ).stdout.strip() except (OSError, subprocess.SubprocessError): revision = "" manifest = { "format": "banban-factory-audio-v1", "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "git_revision": revision, "filesystem": { "name": "customer_fs.bin", "bytes": image.stat().st_size, "sha256": sha256(image), "extract_verified": True, "verified_files": verified_files, }, "mount_paths": [f"U:/{name}" for name in EXPECTED_FILES], "audio": audio, } manifest_path = output / "manifest.json" manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8-sig") readme_path = output / "README.txt" readme_path.write_text( "班班 FT1 产测音频包\n\n" "customer_fs.bin 固定为 262144 字节，用于 EG800AKCN_11LG 的 customer_fs 分区。\n" "三个 MP3 必须位于 U:/ 根目录；不要将此镜像混入正式业务固件。\n" "烧录后以 FT1 BOOT_AUDIO 和最终 session_result 作为设备侧验证。\n", encoding="utf-8-sig", ) package = output / "EG800AKCN_11LG-FT1-factory-audio.zip" if package.exists(): package.unlink() with zipfile.ZipFile(package, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive: archive.write(image, "customer_fs.bin") archive.write(manifest_path, "manifest.json") archive.write(readme_path, "README.txt") for name in EXPECTED_FILES: archive.write(source / name, f"customer_fs/{name}") manifest["package"] = {"name": package.name, "bytes": package.stat().st_size, "sha256": sha256(package)} manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8-sig") return image, package, manifest def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parent.parent) parser.add_argument("--output", type=Path, default=Path(__file__).resolve().parent / "dist" / "factory_audio") args = parser.parse_args() image, package, manifest = build_package(args.root.resolve(), args.output.resolve()) print(f"customer_fs={image}") print(f"package={package}") print(f"customer_fs_sha256={manifest['filesystem']['sha256']}") print(f"package_sha256={manifest['package']['sha256']}") if __name__ == "__main__": main()