117 lines
4.0 KiB
Python
117 lines
4.0 KiB
Python
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from sentence_analysis import detect_sentence_boundaries
|
|
from sentence_api.store import normalize_video_hash
|
|
|
|
|
|
ALGORITHM_VERSION = "silence-rms-v1"
|
|
DEFAULT_INDEX_PATH = Path(__file__).resolve().parent / "data" / "sentence_boundaries.json"
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as media_file:
|
|
for chunk in iter(lambda: media_file.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def media_duration_ms(path: Path) -> int:
|
|
try:
|
|
import av
|
|
except ImportError as exc:
|
|
raise RuntimeError("PyAV is required to generate sentence boundaries.") from exc
|
|
|
|
container = av.open(str(path))
|
|
try:
|
|
if not container.duration or container.duration <= 0:
|
|
raise RuntimeError("Could not determine media duration.")
|
|
return int(round(container.duration / 1000))
|
|
finally:
|
|
container.close()
|
|
|
|
|
|
def make_entry(path: Path, video_hash: Optional[str] = None, min_silence: float = 0.30,
|
|
min_sentence: float = 0.35) -> Tuple[Dict[str, Any], str]:
|
|
duration_ms = media_duration_ms(path)
|
|
starts = detect_sentence_boundaries(
|
|
path,
|
|
min_silence=min_silence,
|
|
min_sentence=min_sentence,
|
|
)
|
|
if not starts:
|
|
raise RuntimeError("No sentence boundaries could be detected.")
|
|
|
|
starts_ms = sorted({max(0, int(round(start * 1000))) for start in starts})
|
|
starts_ms = [start for start in starts_ms if start < duration_ms]
|
|
sentences: List[Dict[str, Any]] = []
|
|
for index, start_ms in enumerate(starts_ms):
|
|
end_ms = starts_ms[index + 1] if index + 1 < len(starts_ms) else duration_ms
|
|
if end_ms <= start_ms:
|
|
continue
|
|
sentences.append({
|
|
"index": len(sentences),
|
|
"start_ms": start_ms,
|
|
"end_ms": end_ms,
|
|
"text": None,
|
|
})
|
|
if not sentences:
|
|
raise RuntimeError("Detected boundaries do not form valid sentence ranges.")
|
|
|
|
entry = {
|
|
"duration_ms": duration_ms,
|
|
"algorithm_version": ALGORITHM_VERSION,
|
|
"sentences": sentences,
|
|
}
|
|
resolved_hash = normalize_video_hash(video_hash) if video_hash else sha256_file(path)
|
|
return entry, resolved_hash
|
|
|
|
|
|
def update_index(index_path: Path, video_hash: str, entry: Dict[str, Any]) -> None:
|
|
if index_path.exists():
|
|
raw = json.loads(index_path.read_text(encoding="utf-8"))
|
|
else:
|
|
raw = {"videos": {}}
|
|
videos = raw.setdefault("videos", {})
|
|
if not isinstance(videos, dict):
|
|
raise ValueError("index JSON must contain a 'videos' object")
|
|
videos[normalize_video_hash(video_hash)] = entry
|
|
index_path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary_path = index_path.with_suffix(index_path.suffix + ".tmp")
|
|
temporary_path.write_text(
|
|
json.dumps(raw, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
temporary_path.replace(index_path)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Generate API sentence boundaries for a video.")
|
|
parser.add_argument("video", type=Path)
|
|
parser.add_argument("--index", type=Path, default=DEFAULT_INDEX_PATH)
|
|
parser.add_argument("--video-hash", help="Override the calculated SHA-256 hash.")
|
|
parser.add_argument("--min-silence", type=float, default=0.30)
|
|
parser.add_argument("--min-sentence", type=float, default=0.35)
|
|
args = parser.parse_args()
|
|
|
|
if not args.video.is_file():
|
|
parser.error(f"Video file does not exist: {args.video}")
|
|
entry, video_hash = make_entry(
|
|
args.video,
|
|
video_hash=args.video_hash,
|
|
min_silence=args.min_silence,
|
|
min_sentence=args.min_sentence,
|
|
)
|
|
update_index(args.index, video_hash, entry)
|
|
print(f"video_hash={video_hash}")
|
|
print(f"sentences={len(entry['sentences'])}")
|
|
print(f"index={args.index}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|