271 lines
7.6 KiB
Python
271 lines
7.6 KiB
Python
"""
|
|
Import functionality for recordings and transcriptions.
|
|
|
|
Supports idempotent import of:
|
|
- JSON transcription files (from transcribe.py output)
|
|
- Audio files from various sources
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import database
|
|
|
|
|
|
def parse_timestamp_from_filename(filename: str) -> Optional[datetime]:
|
|
"""
|
|
Extract timestamp from filename like 'recording_2026-04-26_19-52-47.json'.
|
|
Returns None if pattern doesn't match.
|
|
"""
|
|
pattern = r'recording_(\d{4})-(\d{2})-(\d{2})_(\d{2})-(\d{2})-(\d{2})'
|
|
match = re.search(pattern, filename)
|
|
if match:
|
|
year, month, day, hour, minute, second = map(int, match.groups())
|
|
return datetime(year, month, day, hour, minute, second)
|
|
return None
|
|
|
|
|
|
def import_transcription_json(
|
|
json_path: Path,
|
|
source_name: str = "phone",
|
|
audio_base_path: Optional[Path] = None,
|
|
) -> dict:
|
|
"""
|
|
Import a single transcription JSON file.
|
|
|
|
Returns dict with:
|
|
- status: 'created', 'exists', or 'error'
|
|
- recording_id: ID if created or exists
|
|
- message: description
|
|
"""
|
|
try:
|
|
with open(json_path) as f:
|
|
data = json.load(f)
|
|
except (json.JSONDecodeError, IOError) as e:
|
|
return {"status": "error", "message": f"Failed to read JSON: {e}"}
|
|
|
|
source_file = data.get("source_file", str(json_path))
|
|
|
|
# Check if already imported (idempotent)
|
|
existing = database.get_recording_by_source_file(source_file)
|
|
if existing:
|
|
return {
|
|
"status": "exists",
|
|
"recording_id": existing["id"],
|
|
"message": f"Recording already imported: {existing['filename']}",
|
|
}
|
|
|
|
# Parse filename and timestamp
|
|
filename = Path(source_file).name
|
|
recorded_at = parse_timestamp_from_filename(filename)
|
|
|
|
# Determine audio path
|
|
audio_path = None
|
|
if audio_base_path:
|
|
# Try to find audio file relative to base path
|
|
audio_file = Path(source_file)
|
|
if audio_file.exists():
|
|
audio_path = str(audio_file)
|
|
else:
|
|
# Try in audio_base_path
|
|
potential = audio_base_path / audio_file.name
|
|
if potential.exists():
|
|
audio_path = str(potential)
|
|
|
|
# Create recording entry
|
|
recording_id = database.create_recording(
|
|
source_file=source_file,
|
|
filename=filename,
|
|
source_name=source_name,
|
|
audio_path=audio_path,
|
|
duration=data.get("duration"),
|
|
language=data.get("language"),
|
|
language_probability=data.get("language_probability"),
|
|
recorded_at=recorded_at,
|
|
transcribed_at=datetime.now(),
|
|
)
|
|
|
|
# Import segments
|
|
segments = data.get("segments", [])
|
|
segment_count = database.create_segments(recording_id, segments)
|
|
|
|
return {
|
|
"status": "created",
|
|
"recording_id": recording_id,
|
|
"message": f"Imported {filename} with {segment_count} segments",
|
|
"segments": segment_count,
|
|
}
|
|
|
|
|
|
def import_transcription_directory(
|
|
directory: Path,
|
|
source_name: str = "phone",
|
|
audio_base_path: Optional[Path] = None,
|
|
) -> dict:
|
|
"""
|
|
Import all JSON transcription files from a directory.
|
|
|
|
Returns summary with counts of created, existing, and errors.
|
|
"""
|
|
results = {
|
|
"created": 0,
|
|
"exists": 0,
|
|
"errors": 0,
|
|
"details": [],
|
|
}
|
|
|
|
json_files = sorted(directory.glob("*.json"))
|
|
|
|
for json_path in json_files:
|
|
result = import_transcription_json(json_path, source_name, audio_base_path)
|
|
results["details"].append({
|
|
"file": json_path.name,
|
|
**result,
|
|
})
|
|
|
|
if result["status"] == "created":
|
|
results["created"] += 1
|
|
elif result["status"] == "exists":
|
|
results["exists"] += 1
|
|
else:
|
|
results["errors"] += 1
|
|
|
|
return results
|
|
|
|
|
|
def import_audio_file(
|
|
audio_path: Path,
|
|
source_name: str,
|
|
move_to: Optional[Path] = None,
|
|
) -> dict:
|
|
"""
|
|
Import a single audio file (without transcription).
|
|
|
|
If move_to is provided, the file will be moved there.
|
|
Otherwise, just creates a database entry pointing to current location.
|
|
"""
|
|
if not audio_path.exists():
|
|
return {"status": "error", "message": f"File not found: {audio_path}"}
|
|
|
|
source_file = str(audio_path.resolve())
|
|
|
|
# Check if already imported
|
|
existing = database.get_recording_by_source_file(source_file)
|
|
if existing:
|
|
return {
|
|
"status": "exists",
|
|
"recording_id": existing["id"],
|
|
"message": f"Audio already imported: {existing['filename']}",
|
|
}
|
|
|
|
filename = audio_path.name
|
|
recorded_at = parse_timestamp_from_filename(filename)
|
|
|
|
# Get file size
|
|
file_size = audio_path.stat().st_size
|
|
|
|
# Move file if requested
|
|
final_path = str(audio_path.resolve())
|
|
if move_to:
|
|
move_to.mkdir(parents=True, exist_ok=True)
|
|
dest = move_to / filename
|
|
audio_path.rename(dest)
|
|
final_path = str(dest.resolve())
|
|
|
|
recording_id = database.create_recording(
|
|
source_file=source_file,
|
|
filename=filename,
|
|
source_name=source_name,
|
|
audio_path=final_path,
|
|
recorded_at=recorded_at,
|
|
)
|
|
|
|
# Update file size
|
|
database.update_recording(recording_id, file_size_bytes=file_size)
|
|
|
|
return {
|
|
"status": "created",
|
|
"recording_id": recording_id,
|
|
"message": f"Imported audio: {filename}",
|
|
}
|
|
|
|
|
|
def import_audio_directory(
|
|
directory: Path,
|
|
source_name: str,
|
|
extensions: tuple[str, ...] = (".ogg", ".mp3", ".wav", ".m4a", ".flac", ".webm"),
|
|
move_to: Optional[Path] = None,
|
|
) -> dict:
|
|
"""
|
|
Import all audio files from a directory.
|
|
"""
|
|
results = {
|
|
"created": 0,
|
|
"exists": 0,
|
|
"errors": 0,
|
|
"details": [],
|
|
}
|
|
|
|
# Recursively find all audio files
|
|
audio_files = sorted(
|
|
p for ext in extensions
|
|
for p in directory.rglob(f"*{ext}")
|
|
)
|
|
|
|
for audio_path in audio_files:
|
|
result = import_audio_file(audio_path, source_name, move_to)
|
|
results["details"].append({
|
|
"file": audio_path.name,
|
|
**result,
|
|
})
|
|
|
|
if result["status"] == "created":
|
|
results["created"] += 1
|
|
elif result["status"] == "exists":
|
|
results["exists"] += 1
|
|
else:
|
|
results["errors"] += 1
|
|
|
|
return results
|
|
|
|
|
|
def link_audio_to_transcriptions(audio_dir: Path) -> dict:
|
|
"""
|
|
Link audio files to existing transcription records that are missing audio_path.
|
|
|
|
Matches by filename pattern (e.g., recording_2026-04-26_19-52-47.ogg).
|
|
"""
|
|
updated = 0
|
|
not_found = 0
|
|
|
|
with database.get_db() as conn:
|
|
# Get all recordings without audio_path
|
|
cur = conn.execute(
|
|
"SELECT id, filename, source_file FROM recordings WHERE audio_path IS NULL"
|
|
)
|
|
recordings = cur.fetchall()
|
|
|
|
for rec in recordings:
|
|
# Try to find matching audio file
|
|
base_name = Path(rec["filename"]).stem
|
|
for ext in (".ogg", ".mp3", ".wav", ".m4a", ".flac", ".webm"):
|
|
potential = audio_dir / f"{base_name}{ext}"
|
|
if potential.exists():
|
|
conn.execute(
|
|
"UPDATE recordings SET audio_path = ? WHERE id = ?",
|
|
(str(potential.resolve()), rec["id"])
|
|
)
|
|
updated += 1
|
|
break
|
|
else:
|
|
not_found += 1
|
|
|
|
return {
|
|
"updated": updated,
|
|
"not_found": not_found,
|
|
"message": f"Linked {updated} recordings to audio files, {not_found} still missing",
|
|
}
|