59 lines
2.0 KiB
Python
59 lines
2.0 KiB
Python
import json
|
|
from pathlib import Path
|
|
from threading import RLock
|
|
from typing import Dict, Optional
|
|
|
|
from .models import SentenceBoundaryDocument
|
|
|
|
|
|
def normalize_video_hash(video_hash: str) -> str:
|
|
normalized = video_hash.strip().lower()
|
|
if len(normalized) != 64 or any(char not in "0123456789abcdef" for char in normalized):
|
|
raise ValueError("video_hash must be a SHA-256 hex digest")
|
|
return normalized
|
|
|
|
|
|
class BoundaryStore:
|
|
"""Read-only in-memory index loaded from a JSON file."""
|
|
|
|
def __init__(self, index_path: Path):
|
|
self.index_path = Path(index_path)
|
|
self._lock = RLock()
|
|
self._documents: Dict[str, SentenceBoundaryDocument] = {}
|
|
self.reload()
|
|
|
|
def reload(self) -> None:
|
|
raw = (
|
|
json.loads(self.index_path.read_text(encoding="utf-8"))
|
|
if self.index_path.is_file()
|
|
else {"videos": {}}
|
|
)
|
|
entries = raw.get("videos", raw)
|
|
if not isinstance(entries, dict):
|
|
raise ValueError("index JSON must contain a 'videos' object")
|
|
|
|
documents: Dict[str, SentenceBoundaryDocument] = {}
|
|
for video_hash, entry in entries.items():
|
|
normalized_hash = normalize_video_hash(video_hash)
|
|
if not isinstance(entry, dict):
|
|
raise ValueError(f"entry for {normalized_hash} must be an object")
|
|
document = SentenceBoundaryDocument(
|
|
video_hash=normalized_hash,
|
|
duration_ms=entry.get("duration_ms"),
|
|
algorithm_version=entry.get("algorithm_version", "unknown"),
|
|
sentences=entry.get("sentences", []),
|
|
)
|
|
documents[normalized_hash] = document
|
|
|
|
with self._lock:
|
|
self._documents = documents
|
|
|
|
def get(self, video_hash: str) -> Optional[SentenceBoundaryDocument]:
|
|
normalized_hash = normalize_video_hash(video_hash)
|
|
with self._lock:
|
|
return self._documents.get(normalized_hash)
|
|
|
|
def count(self) -> int:
|
|
with self._lock:
|
|
return len(self._documents)
|