add sentence service

sentence_api/generate_boundaries.py

@@ -0,0 +1,116 @@
+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()