mirror of
https://github.com/jamiepine/voicebox.git
synced 2026-09-29 15:15:27 -07:00
Initialize voicebox project with backend, frontend, and Tauri setup. Added configuration files, dependencies, and basic structure for components, hooks, and utilities. Included README and setup documentation for guidance.
This commit is contained in:
@@ -0,0 +1,219 @@
|
||||
"""
|
||||
Generation history management module.
|
||||
"""
|
||||
|
||||
from typing import List, Optional, Tuple
|
||||
from datetime import datetime
|
||||
import uuid
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import or_
|
||||
|
||||
from .models import GenerationRequest, GenerationResponse, HistoryQuery
|
||||
from .database import Generation as DBGeneration
|
||||
|
||||
|
||||
# Generations storage directory
|
||||
GENERATIONS_DIR = Path("data/generations")
|
||||
GENERATIONS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
async def create_generation(
|
||||
profile_id: str,
|
||||
text: str,
|
||||
language: str,
|
||||
audio_path: str,
|
||||
duration: float,
|
||||
seed: Optional[int],
|
||||
db: Session,
|
||||
) -> GenerationResponse:
|
||||
"""
|
||||
Create a new generation history entry.
|
||||
|
||||
Args:
|
||||
profile_id: Profile ID used for generation
|
||||
text: Generated text
|
||||
language: Language code
|
||||
audio_path: Path where audio was saved
|
||||
duration: Audio duration in seconds
|
||||
seed: Random seed used (if any)
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
Created generation entry
|
||||
"""
|
||||
db_generation = DBGeneration(
|
||||
id=str(uuid.uuid4()),
|
||||
profile_id=profile_id,
|
||||
text=text,
|
||||
language=language,
|
||||
audio_path=audio_path,
|
||||
duration=duration,
|
||||
seed=seed,
|
||||
created_at=datetime.utcnow(),
|
||||
)
|
||||
|
||||
db.add(db_generation)
|
||||
db.commit()
|
||||
db.refresh(db_generation)
|
||||
|
||||
return GenerationResponse.model_validate(db_generation)
|
||||
|
||||
|
||||
async def get_generation(
|
||||
generation_id: str,
|
||||
db: Session,
|
||||
) -> Optional[GenerationResponse]:
|
||||
"""
|
||||
Get a generation by ID.
|
||||
|
||||
Args:
|
||||
generation_id: Generation ID
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
Generation or None if not found
|
||||
"""
|
||||
generation = db.query(DBGeneration).filter_by(id=generation_id).first()
|
||||
if not generation:
|
||||
return None
|
||||
|
||||
return GenerationResponse.model_validate(generation)
|
||||
|
||||
|
||||
async def list_generations(
|
||||
query: HistoryQuery,
|
||||
db: Session,
|
||||
) -> Tuple[List[GenerationResponse], int]:
|
||||
"""
|
||||
List generations with optional filters.
|
||||
|
||||
Args:
|
||||
query: Query parameters (filters, pagination)
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
Tuple of (generations, total_count)
|
||||
"""
|
||||
# Build base query
|
||||
q = db.query(DBGeneration)
|
||||
|
||||
# Apply profile filter
|
||||
if query.profile_id:
|
||||
q = q.filter(DBGeneration.profile_id == query.profile_id)
|
||||
|
||||
# Apply search filter (searches in text content)
|
||||
if query.search:
|
||||
search_pattern = f"%{query.search}%"
|
||||
q = q.filter(DBGeneration.text.like(search_pattern))
|
||||
|
||||
# Get total count before pagination
|
||||
total_count = q.count()
|
||||
|
||||
# Apply ordering (newest first)
|
||||
q = q.order_by(DBGeneration.created_at.desc())
|
||||
|
||||
# Apply pagination
|
||||
q = q.offset(query.offset).limit(query.limit)
|
||||
|
||||
# Execute query
|
||||
generations = q.all()
|
||||
|
||||
return (
|
||||
[GenerationResponse.model_validate(g) for g in generations],
|
||||
total_count,
|
||||
)
|
||||
|
||||
|
||||
async def delete_generation(
|
||||
generation_id: str,
|
||||
db: Session,
|
||||
) -> bool:
|
||||
"""
|
||||
Delete a generation.
|
||||
|
||||
Args:
|
||||
generation_id: Generation ID
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
True if deleted, False if not found
|
||||
"""
|
||||
generation = db.query(DBGeneration).filter_by(id=generation_id).first()
|
||||
if not generation:
|
||||
return False
|
||||
|
||||
# Delete audio file
|
||||
audio_path = Path(generation.audio_path)
|
||||
if audio_path.exists():
|
||||
audio_path.unlink()
|
||||
|
||||
# Delete from database
|
||||
db.delete(generation)
|
||||
db.commit()
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def delete_generations_by_profile(
|
||||
profile_id: str,
|
||||
db: Session,
|
||||
) -> int:
|
||||
"""
|
||||
Delete all generations for a profile.
|
||||
|
||||
Args:
|
||||
profile_id: Profile ID
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
Number of generations deleted
|
||||
"""
|
||||
generations = db.query(DBGeneration).filter_by(profile_id=profile_id).all()
|
||||
|
||||
count = 0
|
||||
for generation in generations:
|
||||
# Delete audio file
|
||||
audio_path = Path(generation.audio_path)
|
||||
if audio_path.exists():
|
||||
audio_path.unlink()
|
||||
|
||||
# Delete from database
|
||||
db.delete(generation)
|
||||
count += 1
|
||||
|
||||
db.commit()
|
||||
|
||||
return count
|
||||
|
||||
|
||||
async def get_generation_stats(db: Session) -> dict:
|
||||
"""
|
||||
Get generation statistics.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
Statistics dictionary
|
||||
"""
|
||||
from sqlalchemy import func
|
||||
|
||||
total = db.query(func.count(DBGeneration.id)).scalar()
|
||||
|
||||
total_duration = db.query(func.sum(DBGeneration.duration)).scalar() or 0
|
||||
|
||||
# Get generations by profile
|
||||
by_profile = db.query(
|
||||
DBGeneration.profile_id,
|
||||
func.count(DBGeneration.id).label('count')
|
||||
).group_by(DBGeneration.profile_id).all()
|
||||
|
||||
return {
|
||||
"total_generations": total,
|
||||
"total_duration_seconds": total_duration,
|
||||
"generations_by_profile": {
|
||||
profile_id: count for profile_id, count in by_profile
|
||||
},
|
||||
}
|
||||
Reference in New Issue
Block a user