upstream-archive/VoiceStudio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(persona): .ovsvoice build/parse core + embed_watermark(force=) (#29 slice A) (#460)

Browse Source 
Extends the merged persona-bundle nucleus (constants, normalize_spdx,
build_manifest, build_consent_json) with the model-coupled core that the
export/import router (next slice) will sit on:

- `build_persona_bundle(profile, *, license_spdx, tags, include_reference,
  embed_fn, …)` → assembles the .ovsvoice ZIP in memory: a watermarked
  preview.wav (24 kHz mono 16-bit, downmixed + resampled + trimmed ≤8 s),
  manifest.json, a legacy-shaped metadata.json (so an older OmniVoice can still
  import the ref audio), optional consent.json, and the raw ref/locked/consent
  members unless include_reference=False (privacy / preview-only, A12). Raises
  NoPreviewSource (router → 503) when no source clip is readable (A2-A5).
- `parse_persona_bundle(bytes)` → validates the ZIP, prefers manifest.json and
  falls back to legacy metadata.json, resolves audio members by prefix
  (last-wins, B9; member names never build paths — zip-slip safe), normalizes
  the SPDX id, flags preview-only / future-schema_version. Raises
  BundleError(400|413) for B1-B11. No DB, no file writes.
- `ParsedPersona` dataclass with `extract_member(prefix, dest_path)` — the
  router derives dest_path from the server-generated id, never the member name.
- `embed_watermark(..., *, force=False)`: keyword-only flag that bypasses the
  user's invisible-watermark preference for the mandatory persona preview, but
  still no-ops without AudioSeal. All existing positional call sites are
  unchanged (default force=False) — default cross-platform behaviour identical.

All heavy imports (torch/torchaudio/watermark/audio_io) are lazy so the module
stays model-free at collection (avoids the local torch/Triton segfault).

tests/test_persona_bundle.py: +31 cases — parse validation (manifest/legacy
selection, preview-only, future-schema, missing/malformed/no-audio → 400,
oversize → 413, bad-SPDX normalize, last-wins dup, advisory consent), build
round-trip (identity fields, metadata sibling, no-source → NoPreviewSource,
include_reference=False, stereo/off-rate downmix+resample), and the force=
unit (D1/D3). 25 pure cases pass locally; the 6 torchaudio-coupled cases run on
CI (local torch+pytest segfault is pre-existing). CJK guard green.
This commit is contained in:
Palash Debnath
2026-06-14 17:55:17 +05:30
committed by GitHub
parent ca8a2e8eb8
commit 4500dcb6b4
3 changed files with 530 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/persona_bundle.py
+312 -2
View File
@@ -9,7 +9,13 @@ pack/unpack (which lazily import torchaudio/watermark) layer on top of these.
"""
from __future__ import annotations
import io
import json
import os
import re
import time
import zipfile
from dataclasses import dataclass
from typing import Optional
# ── Format constants ─────────────────────────────────────────────────────────
@@ -18,6 +24,15 @@ OVSVOICE_SCHEMA_VERSION = 1
MAX_BUNDLE_BYTES = 100 * 1024 * 1024 # 100 MB (mirrors marketplace cap)
_MIN_CONSENT_AUDIO_BYTES = 1000 # the consent-recording floor
DEFAULT_LICENSE = "LicenseRef-OmniVoice-Personal"
PREVIEW_MAX_SECONDS = 8.0 # preview length cap (A6)
PREVIEW_SAMPLE_RATE = 24_000 # preview rate; mono, 16-bit PCM (A8)
# Audio member prefixes the importer recognises. The ZIP member NAME is never
# used to build an output path (zip-slip safe, B10) — only its extension, and
# only after the linear allowlist below.
_AUDIO_MEMBER_PREFIXES = ("ref_audio", "locked_audio", "consent_audio", "preview")
# Reused verbatim from profiles.py:306 — single linear quantifier, no ReDoS.
_MEMBER_EXT_RE = re.compile(r"^\.[A-Za-z0-9]{1,8}$")
# Membership allowlist for SPDX validation — a fixed-string set + the
# ``LicenseRef-`` prefix. NO regex over the (user-supplied) SPDX string, so this
@@ -37,6 +52,19 @@ class BundleError(Exception):
self.detail = detail
class NoPreviewSource(Exception):
"""No readable source clip exists to build a preview from (A2/A3/A4/A5/A12).
The router maps this to HTTP 503."""
def _safe_member_ext(member_name: str) -> str:
"""The extension for a ZIP member, allowlisted to ``^\\.[A-Za-z0-9]{1,8}$``
(else ``.wav``). Used ONLY to choose the output extension — never the path
(B11). Linear regex, no ReDoS."""
ext = os.path.splitext(member_name)[1]
return ext if _MEMBER_EXT_RE.match(ext) else ".wav"
def normalize_spdx(spdx: Optional[str]) -> str:
"""Return a safe SPDX id: the value if it's allowlisted or a ``LicenseRef-``
custom id, else :data:`DEFAULT_LICENSE`. Never raises, never 400s — a junk
@@ -119,8 +147,290 @@ def build_consent_json(profile: dict, *, has_recording: bool) -> Optional[dict]:
}
def _legacy_metadata(profile: dict, omnivoice_version: str) -> dict:
"""A ``metadata.json`` payload shaped like marketplace ``_bundle_metadata`` so
an OLDER OmniVoice (which only reads metadata.json) can still import the ref
audio from a ``.ovsvoice`` bundle."""
return {
"bundle_version": 1,
"profile_name": profile.get("name") or "",
"ref_text": profile.get("ref_text") or "",
"instruct": profile.get("instruct") or "",
"language": profile.get("language") or "Auto",
"personality": profile.get("personality") or "",
"seed": profile.get("seed"),
"kind": profile.get("kind") or "clone",
"vd_states": profile.get("vd_states"),
"is_locked": bool(profile.get("is_locked")),
"omnivoice_version": omnivoice_version or "",
}
def _resolve_voice_file(filename: Optional[str]) -> Optional[str]:
"""Resolve a DB-stored audio filename strictly inside VOICES_DIR, returning
an absolute path only if the file actually exists. None on missing/escape —
mirrors profiles._voices_path (basename + realpath confinement, E1)."""
if not filename or os.path.basename(filename) != filename:
return None
from core.config import VOICES_DIR
root = os.path.realpath(VOICES_DIR)
path = os.path.realpath(os.path.join(root, filename))
if not path.startswith(root + os.sep):
return None
return path if os.path.isfile(path) else None
def _generate_preview(profile: dict, embed_fn) -> tuple[bytes, bool, float]:
"""Load the profile's source clip, downmix→mono, resample→24 kHz, trim ≤8 s,
watermark (forced), and return ``(wav_bytes, watermarked, duration_s)``.
Source precedence is locked-over-ref (profiles.py:230). Raises
:class:`NoPreviewSource` when neither clip is readable (A2-A5). All heavy
imports (torch/torchaudio/watermark) are lazy so the module stays model-free
at collection time (avoids the known local torch/Triton segfault)."""
import torch # noqa: F401 (torchaudio needs it loaded)
import torchaudio
from services.audio_io import _safe_torchaudio_save
from services.watermark import _check_available
candidates = [profile.get("locked_audio_path"), profile.get("ref_audio_path")]
wav = None
for name in candidates:
path = _resolve_voice_file(name)
if not path:
continue
try:
waveform, sr = torchaudio.load(path)
except Exception: # noqa: BLE001 — try the next candidate (A4)
continue
if waveform.numel() == 0: # empty/zero-length (A5)
continue
if waveform.shape[0] > 1: # downmix to mono (A7)
waveform = waveform.mean(dim=0, keepdim=True)
if sr != PREVIEW_SAMPLE_RATE: # resample (A8)
waveform = torchaudio.functional.resample(waveform, sr, PREVIEW_SAMPLE_RATE)
cap = int(PREVIEW_SAMPLE_RATE * PREVIEW_MAX_SECONDS)
waveform = waveform[:, :cap] # trim, shorter used whole (A6)
wav = waveform
break
if wav is None or wav.numel() == 0:
raise NoPreviewSource("no readable reference or locked audio for a preview")
# Forced watermark — bypasses the user pref but still no-ops without AudioSeal.
fn = embed_fn or _default_embed
wav = fn(wav, PREVIEW_SAMPLE_RATE)
watermarked = bool(_check_available()) # best-effort honesty (A11)
duration_s = round(wav.shape[-1] / PREVIEW_SAMPLE_RATE, 3)
buf = io.BytesIO()
_safe_torchaudio_save(buf, wav, PREVIEW_SAMPLE_RATE, format="wav", bits_per_sample=16)
return buf.getvalue(), watermarked, duration_s
def _default_embed(wav, sample_rate):
"""Default preview watermarker: services.watermark.embed_watermark(force=True)."""
from services.watermark import embed_watermark
return embed_watermark(wav, sample_rate, force=True)
def build_persona_bundle(
profile: dict,
*,
license_spdx: str = DEFAULT_LICENSE,
tags: Optional[list[str]] = None,
custom_license_text: Optional[str] = None,
include_reference: bool = True,
engine_id: str = "",
omnivoice_version: str = "",
embed_fn=None,
) -> bytes:
"""Assemble a ``.ovsvoice`` ZIP in memory and return its bytes.
Always writes a watermarked ``preview.wav`` + ``manifest.json`` +
(legacy-shaped) ``metadata.json``. Writes ``consent.json`` when there's
something to attest, the raw ``ref_audio``/``locked_audio`` members unless
``include_reference=False`` (privacy / preview-only, A12), and
``consent_audio`` when a recording exists. Raises :class:`NoPreviewSource`
(router → 503) when no source clip is readable."""
preview_bytes, watermarked, duration_s = _generate_preview(profile, embed_fn)
members: dict = {"ref_audio": None, "locked_audio": None, "consent_audio": None}
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
if include_reference:
ref_path = _resolve_voice_file(profile.get("ref_audio_path"))
if ref_path:
name = f"ref_audio{os.path.splitext(ref_path)[1] or '.wav'}"
zf.write(ref_path, name)
members["ref_audio"] = name
locked_path = _resolve_voice_file(profile.get("locked_audio_path"))
if locked_path:
name = f"locked_audio{os.path.splitext(locked_path)[1] or '.wav'}"
zf.write(locked_path, name)
members["locked_audio"] = name
# Consent recording travels only when it exists and clears the floor.
consent_path = _resolve_voice_file(profile.get("consent_audio_path"))
has_recording = False
if consent_path and os.path.getsize(consent_path) >= _MIN_CONSENT_AUDIO_BYTES:
name = f"consent_audio{os.path.splitext(consent_path)[1] or '.wav'}"
zf.write(consent_path, name)
members["consent_audio"] = name
has_recording = True
zf.writestr("preview.wav", preview_bytes)
preview_block = {
"file": "preview.wav", "watermarked": watermarked,
"duration_s": duration_s, "sample_rate": PREVIEW_SAMPLE_RATE,
}
manifest = build_manifest(
profile, license_spdx=license_spdx, tags=tags or [],
engine_id=engine_id, custom_license_text=custom_license_text,
preview=preview_block, members=members,
omnivoice_version=omnivoice_version,
)
zf.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2))
zf.writestr("metadata.json",
json.dumps(_legacy_metadata(profile, omnivoice_version),
ensure_ascii=False, indent=2))
consent = build_consent_json(profile, has_recording=has_recording)
if consent is not None:
zf.writestr("consent.json", json.dumps(consent, ensure_ascii=False, indent=2))
return buf.getvalue()
@dataclass
class ParsedPersona:
manifest: dict # parsed manifest.json OR synthesized from metadata.json
consent: Optional[dict] # parsed consent.json, or None
is_legacy: bool # only metadata.json was found (B6/B23)
schema_version_ahead: bool # manifest.schema_version > OVSVOICE_SCHEMA_VERSION (B7)
license_spdx: str # normalized (B21)
preview_only: bool # only preview.wav, no ref/locked member (A12/B8)
members: dict # {prefix: member_name} for audio members present
watermarked_preview: bool # manifest.preview.watermarked (False for legacy)
_zip: zipfile.ZipFile # open handle; router extracts via extract_member()
def member_ext(self, prefix: str) -> str:
name = self.members.get(prefix)
return _safe_member_ext(name) if name else ".wav"
def extract_member(self, prefix: str, dest_path: str) -> bool:
"""Stream the audio member named by ``prefix`` to ``dest_path`` (a path
the CALLER derived from a server-generated id — never from the member
name). Returns False when the member is absent. Last-wins on dup (B9)."""
name = self.members.get(prefix)
if not name:
return False
import shutil
with self._zip.open(name) as src, open(dest_path, "wb") as dst:
shutil.copyfileobj(src, dst)
return True
def parse_persona_bundle(content: bytes) -> ParsedPersona:
"""Validate the ZIP and read manifest/consent WITHOUT touching the DB or
writing files. Raises :class:`BundleError` (400|413) for B1-B11. The caller
must use the returned ``ParsedPersona`` while the process holds ``content``
(the open ZIP reads from the in-memory bytes)."""
if len(content) > MAX_BUNDLE_BYTES:
raise BundleError(413, f"Bundle too large. Max is {MAX_BUNDLE_BYTES} bytes.")
try:
zf = zipfile.ZipFile(io.BytesIO(content))
except zipfile.BadZipFile:
raise BundleError(400, "not a valid ZIP bundle")
names = [n for n in zf.namelist() if not n.endswith("/")]
# Manifest selection: prefer manifest.json, fall back to legacy metadata.json.
manifest: dict = {}
is_legacy = False
if "manifest.json" in names:
try:
manifest = json.loads(zf.read("manifest.json"))
except (ValueError, UnicodeDecodeError):
raise BundleError(400, "manifest is not valid JSON")
if not isinstance(manifest, dict):
raise BundleError(400, "manifest is not valid JSON")
# A bundle whose format is neither ovsvoice nor absent → still read
# leniently (B6); we only branch on schema_version below.
elif "metadata.json" in names:
is_legacy = True
try:
legacy = json.loads(zf.read("metadata.json"))
except (ValueError, UnicodeDecodeError):
raise BundleError(400, "manifest is not valid JSON")
if not isinstance(legacy, dict):
raise BundleError(400, "manifest is not valid JSON")
manifest = {
"format": "omnivoice-legacy",
"schema_version": OVSVOICE_SCHEMA_VERSION,
"persona": {
"name": legacy.get("profile_name") or legacy.get("name") or "Imported Voice",
"kind": legacy.get("kind") or "clone",
"language": legacy.get("language") or "Auto",
"personality": legacy.get("personality") or "",
"instruct": legacy.get("instruct") or "",
"ref_text": legacy.get("ref_text") or "",
"seed": legacy.get("seed"),
"is_locked": bool(legacy.get("is_locked")),
"vd_states": legacy.get("vd_states"),
},
"license": {"spdx": DEFAULT_LICENSE, "custom_text": None},
"tags": [],
"preview": None,
"members": {},
}
else:
raise BundleError(400, "bundle is missing a manifest")
# Audio members by prefix (last-wins on duplicates, B9). The member NAME is
# retained only to read bytes + pick an extension — never to build a path.
members: dict = {}
for name in names:
for prefix in _AUDIO_MEMBER_PREFIXES:
if os.path.basename(name).startswith(prefix):
members[prefix] = name
has_audio = any(p in members for p in ("ref_audio", "locked_audio", "preview"))
if not has_audio:
raise BundleError(400, "bundle has no audio member")
consent = None
if "consent.json" in names:
try:
parsed = json.loads(zf.read("consent.json"))
if isinstance(parsed, dict):
consent = parsed
except (ValueError, UnicodeDecodeError):
consent = None # advisory only — a bad consent.json never 400s
schema_version = manifest.get("schema_version", OVSVOICE_SCHEMA_VERSION)
try:
ahead = int(schema_version) > OVSVOICE_SCHEMA_VERSION
except (TypeError, ValueError):
ahead = False
preview_block = manifest.get("preview") or {}
watermarked_preview = bool(preview_block.get("watermarked")) if isinstance(preview_block, dict) else False
license_spdx = normalize_spdx((manifest.get("license") or {}).get("spdx"))
preview_only = ("preview" in members
and "ref_audio" not in members and "locked_audio" not in members)
return ParsedPersona(
manifest=manifest, consent=consent, is_legacy=is_legacy,
schema_version_ahead=ahead, license_spdx=license_spdx,
preview_only=preview_only, members=members,
watermarked_preview=watermarked_preview, _zip=zf,
)
__all__ = [
"OVSVOICE_FORMAT", "OVSVOICE_SCHEMA_VERSION", "MAX_BUNDLE_BYTES",
"DEFAULT_LICENSE", "BundleError", "normalize_spdx", "build_manifest",
"build_consent_json",
"PREVIEW_MAX_SECONDS", "PREVIEW_SAMPLE_RATE", "DEFAULT_LICENSE",
"BundleError", "NoPreviewSource", "ParsedPersona",
"normalize_spdx", "build_manifest", "build_consent_json",
"build_persona_bundle", "parse_persona_bundle",
]
backend/services/watermark.py
+9 -1
View File
@@ -97,6 +97,8 @@ def embed_watermark(
waveform: torch.Tensor,
sample_rate: int,
message: Optional[list[int]] = None,
*,
force: bool = False,
) -> torch.Tensor:
"""
Embed an imperceptible watermark into the audio waveform.
@@ -105,11 +107,17 @@ def embed_watermark(
waveform: Audio tensor of shape (channels, samples) or (1, channels, samples)
sample_rate: Sample rate of the audio
message: Optional 16-bit message (list of 0/1). Defaults to OMNI_MESSAGE.
force: Keyword-only. When True, bypass the user's invisible-watermark
preference (``is_enabled()``) and embed regardless — used by the
persona-preview path, which mandates a watermark at package time.
It does NOT bypass availability: when AudioSeal isn't installed the
call still no-ops and returns the input unchanged. Existing
positional call sites default to ``force=False`` (unchanged).
Returns:
Watermarked waveform (same shape as input).
"""
if not is_enabled() or not _check_available():
if (not force and not is_enabled()) or not _check_available():
return waveform
try:
tests/test_persona_bundle.py
+209
View File
@@ -6,13 +6,24 @@ separate slice (and run on CI for the torch-coupled paths).
"""
from __future__ import annotations
import io
import json
import zipfile
import pytest
from services.persona_bundle import (
DEFAULT_LICENSE,
MAX_BUNDLE_BYTES,
OVSVOICE_FORMAT,
OVSVOICE_SCHEMA_VERSION,
BundleError,
NoPreviewSource,
build_consent_json,
build_manifest,
build_persona_bundle,
normalize_spdx,
parse_persona_bundle,
)
_PROFILE = {
@@ -104,3 +115,201 @@ def test_consent_recorded_at_coerced_when_missing_or_bad():
c = build_consent_json({"kind": "clone", "consent_text": "ok", "consent_recorded_at": "nope"},
has_recording=True)
assert isinstance(c["recorded_at"], float) # coerced to now, not a crash
# ── parse_persona_bundle: pure ZIP validation (no torch) ─────────────────────
def _zip(members: dict) -> bytes:
"""members: {arcname: bytes|str}."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
for name, body in members.items():
zf.writestr(name, body if isinstance(body, (bytes, bytearray)) else str(body))
return buf.getvalue()
def _manifest_bytes(**over) -> str:
base = build_manifest(_PROFILE, license_spdx="CC-BY-4.0", tags=["narration"],
preview={"file": "preview.wav", "watermarked": True,
"duration_s": 6.2, "sample_rate": 24000},
members={"ref_audio": "ref_audio.wav", "locked_audio": None,
"consent_audio": None})
base.update(over)
return json.dumps(base)
def test_parse_prefers_manifest_and_normalizes():
content = _zip({"manifest.json": _manifest_bytes(),
"ref_audio.wav": b"\x00" * 100, "preview.wav": b"\x00" * 100})
parsed = parse_persona_bundle(content)
assert parsed.is_legacy is False
assert parsed.manifest["format"] == OVSVOICE_FORMAT
assert parsed.license_spdx == "CC-BY-4.0"
assert parsed.watermarked_preview is True
assert parsed.preview_only is False
assert parsed.members.get("ref_audio") == "ref_audio.wav"
def test_parse_legacy_metadata_only():
legacy = {"profile_name": "Old Voice", "kind": "clone", "language": "English"}
content = _zip({"metadata.json": json.dumps(legacy), "ref_audio.wav": b"\x00" * 100})
parsed = parse_persona_bundle(content)
assert parsed.is_legacy is True
assert parsed.manifest["persona"]["name"] == "Old Voice"
assert parsed.license_spdx == DEFAULT_LICENSE
assert parsed.watermarked_preview is False
def test_parse_preview_only_bundle():
content = _zip({"manifest.json": _manifest_bytes(members={"ref_audio": None}),
"preview.wav": b"\x00" * 100})
parsed = parse_persona_bundle(content)
assert parsed.preview_only is True
def test_parse_future_schema_version_flagged():
content = _zip({"manifest.json": _manifest_bytes(schema_version=99),
"ref_audio.wav": b"\x00" * 100})
parsed = parse_persona_bundle(content)
assert parsed.schema_version_ahead is True
def test_parse_missing_manifest_400():
with pytest.raises(BundleError) as e:
parse_persona_bundle(_zip({"ref_audio.wav": b"\x00" * 100}))
assert e.value.status == 400
def test_parse_no_audio_member_400():
with pytest.raises(BundleError) as e:
parse_persona_bundle(_zip({"manifest.json": _manifest_bytes()}))
assert e.value.status == 400
def test_parse_malformed_manifest_json_400():
with pytest.raises(BundleError) as e:
parse_persona_bundle(_zip({"manifest.json": "{not json",
"ref_audio.wav": b"\x00" * 100}))
assert e.value.status == 400
def test_parse_not_a_zip_400():
with pytest.raises(BundleError) as e:
parse_persona_bundle(b"definitely not a zip")
assert e.value.status == 400
def test_parse_oversize_413():
# Header check fires before ZIP parsing — a non-zip blob over the cap is 413.
with pytest.raises(BundleError) as e:
parse_persona_bundle(b"\x00" * (MAX_BUNDLE_BYTES + 1))
assert e.value.status == 413
def test_parse_bad_consent_json_is_advisory_not_fatal():
content = _zip({"manifest.json": _manifest_bytes(), "ref_audio.wav": b"\x00" * 100,
"consent.json": "{broken"})
parsed = parse_persona_bundle(content)
assert parsed.consent is None # ignored, not a 400
def test_parse_bad_spdx_in_manifest_normalized():
content = _zip({"manifest.json": _manifest_bytes(license={"spdx": "haha; rm -rf", "custom_text": None}),
"ref_audio.wav": b"\x00" * 100})
assert parse_persona_bundle(content).license_spdx == DEFAULT_LICENSE
def test_parse_last_wins_on_duplicate_members():
content = _zip({"manifest.json": _manifest_bytes(),
"ref_audio.wav": b"\x00" * 100, "ref_audio_2.wav": b"\x11" * 100})
parsed = parse_persona_bundle(content)
# Whichever sorts last in the namelist wins; either is a valid prefix match.
assert parsed.members["ref_audio"].startswith("ref_audio")
# ── build_persona_bundle round-trip (torchaudio; runs on CI, often local) ────
def _write_wav(path, *, seconds=1.0, sr=16000, channels=1):
import numpy as np
import soundfile as sf
n = int(seconds * sr)
data = (0.1 * np.sin(2 * np.pi * 220 * np.arange(n) / sr)).astype("float32")
if channels > 1:
data = np.stack([data] * channels, axis=1)
sf.write(str(path), data, sr)
@pytest.fixture
def voices_dir(tmp_path, monkeypatch):
import core.config as cfg
d = tmp_path / "voices"
d.mkdir()
monkeypatch.setattr(cfg, "VOICES_DIR", str(d))
return d
def _identity_embed(wav, sr):
return wav # avoid loading AudioSeal in unit tests
def test_build_roundtrip_identity_fields(voices_dir):
_write_wav(voices_dir / "abc.wav")
profile = {**_PROFILE, "kind": "clone", "ref_audio_path": "abc.wav"}
content = build_persona_bundle(profile, license_spdx="CC-BY-4.0", tags=["x"],
embed_fn=_identity_embed)
parsed = parse_persona_bundle(content)
p = parsed.manifest["persona"]
assert p["name"] == "Aria Narration" and p["seed"] == 42
assert parsed.manifest["preview"]["sample_rate"] == 24000
assert isinstance(parsed.manifest["preview"]["duration_s"], float)
# legacy-reader compat: a metadata.json sibling is always written.
with zipfile.ZipFile(io.BytesIO(content)) as zf:
assert "metadata.json" in zf.namelist()
assert "preview.wav" in zf.namelist()
assert any(n.startswith("ref_audio") for n in zf.namelist())
def test_build_no_source_raises_no_preview_source(voices_dir):
profile = {**_PROFILE, "kind": "clone", "ref_audio_path": None, "locked_audio_path": None}
with pytest.raises(NoPreviewSource):
build_persona_bundle(profile, embed_fn=_identity_embed)
def test_build_missing_file_raises_no_preview_source(voices_dir):
profile = {**_PROFILE, "ref_audio_path": "gone.wav"}
with pytest.raises(NoPreviewSource):
build_persona_bundle(profile, embed_fn=_identity_embed)
def test_build_include_reference_false_is_preview_only(voices_dir):
_write_wav(voices_dir / "abc.wav")
profile = {**_PROFILE, "kind": "clone", "ref_audio_path": "abc.wav"}
content = build_persona_bundle(profile, include_reference=False, embed_fn=_identity_embed)
with zipfile.ZipFile(io.BytesIO(content)) as zf:
names = zf.namelist()
assert "preview.wav" in names
assert not any(n.startswith("ref_audio") for n in names)
assert parse_persona_bundle(content).preview_only is True
def test_build_stereo_offrate_source_downmixed_resampled(voices_dir):
_write_wav(voices_dir / "st.wav", sr=48000, channels=2, seconds=12.0)
profile = {**_PROFILE, "kind": "clone", "ref_audio_path": "st.wav"}
content = build_persona_bundle(profile, embed_fn=_identity_embed)
parsed = parse_persona_bundle(content)
assert parsed.manifest["preview"]["sample_rate"] == 24000
assert parsed.manifest["preview"]["duration_s"] <= 8.0 # trimmed to cap
# ── embed_watermark(force=) unit (D1-D3) ─────────────────────────────────────
def test_embed_watermark_force_keyword(monkeypatch):
import torch
from services import watermark
monkeypatch.setattr(watermark, "_check_available", lambda: False) # AudioSeal absent
wav = torch.zeros(1, 100)
# force=True still no-ops without AudioSeal (D3) — returns input unchanged.
out = watermark.embed_watermark(wav, 24000, force=True)
assert out is wav
# default force=False also unchanged for existing positional callers (D1).
assert watermark.embed_watermark(wav, 24000) is wav