""" main.py - FastAPI application with endpoints, database, and video generation Multilingual support with Korean and English translation SCALABLE VERSION: Async database, Redis caching, rate limiting, thread-pool video generation """ import os import base64 import hashlib import asyncio import re from typing import Dict, List, Optional, Any from datetime import datetime from fastapi import FastAPI, Query, HTTPException, Request, BackgroundTasks from fastapi.responses import Response, JSONResponse from fastapi.middleware.cors import CORSMiddleware from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded import uvicorn from dotenv import load_dotenv import replicate from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.orm import declarative_base from sqlalchemy import Column, String, LargeBinary, DateTime, Integer, select from redis import asyncio as aioredis # Import from workout_list module (unchanged) from workout_app.workout_list import ( WorkoutDatabase, UserProfile, WeeklyPlanUI, ExerciseUI, WorkoutDayUI, normalize_blood_type, parse_duration_to_minutes, parse_calories ) # Import Pydantic models for API from pydantic import BaseModel, Field # Import language middleware and translation service from middleware import LanguageMiddleware, get_language_from_request from translation_service import translation_service load_dotenv() # ========== SCALABILITY: Rate limiting setup ========== limiter = Limiter(key_func=get_remote_address, storage_uri=os.getenv("REDIS_URL", "redis://localhost:6379")) app = FastAPI(title="Personalized Blood Type Workout Planner with Videos", version="8.0.0") app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Add Language Detection Middleware app.add_middleware(LanguageMiddleware, supported_languages=['en', 'ko']) # ========== SCALABILITY: Async SQLite with aiosqlite ========== DATABASE_URL = os.getenv("DATABASE_URL", "sqlite+aiosqlite:///./exercise_videos.db") engine = create_async_engine(DATABASE_URL, echo=False, pool_size=20, max_overflow=40) AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) Base = declarative_base() # ========== Database model for exercise videos ========== class ExerciseVideo(Base): """Database model for storing exercise videos""" __tablename__ = "exercise_videos" id = Column(Integer, primary_key=True, autoincrement=True) exercise_name = Column(String, unique=True, index=True, nullable=False) exercise_hash = Column(String, unique=True, index=True, nullable=False) video_data = Column(LargeBinary, nullable=False) prompt_used = Column(String, nullable=False) file_size = Column(Integer, nullable=False) created_at = Column(DateTime, default=datetime.utcnow) last_accessed = Column(DateTime, default=datetime.utcnow) access_count = Column(Integer, default=0) class VideoGenerationRequest(BaseModel): exercise_name: str = Field(..., description="Name of the exercise") force_regenerate: bool = Field(False, description="Force regenerate even if cached") class VideoResponse(BaseModel): exercise_name: str video_base64: Optional[str] = None video_url: Optional[str] = None status: str from_cache: bool file_size: int created_at: Optional[datetime] = None message: Optional[str] = None # Create tables on startup async def init_db(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) # ========== SCALABILITY: Redis cache setup ========== _redis_client = None async def get_redis(): global _redis_client if _redis_client is None: redis_url = os.getenv("REDIS_URL", "redis://localhost:6379") _redis_client = await aioredis.from_url(redis_url, decode_responses=True) return _redis_client # ========== SCALABILITY: Concurrency semaphores ========== LLM_SEMAPHORE = asyncio.Semaphore(1000) # OpenAI calls VIDEO_SEMAPHORE = asyncio.Semaphore(100) # Replicate calls # Initialize Workout Database (synchronous part remains unchanged) workout_db = WorkoutDatabase() # ========== Database helper functions (async) ========== async def get_db(): async with AsyncSessionLocal() as session: yield session def get_exercise_hash(exercise_name: str) -> str: return hashlib.md5(exercise_name.lower().strip().encode()).hexdigest() async def get_cached_video(db: AsyncSession, exercise_name: str) -> Optional[ExerciseVideo]: exercise_hash = get_exercise_hash(exercise_name) result = await db.execute( select(ExerciseVideo).where(ExerciseVideo.exercise_hash == exercise_hash) ) video_record = result.scalar_one_or_none() if video_record: video_record.last_accessed = datetime.utcnow() video_record.access_count += 1 await db.commit() print(f"Cache HIT for '{exercise_name}' (accessed {video_record.access_count} times)") return video_record async def save_video_to_cache(db: AsyncSession, exercise_name: str, video_data: bytes, prompt: str) -> ExerciseVideo: exercise_hash = get_exercise_hash(exercise_name) result = await db.execute( select(ExerciseVideo).where(ExerciseVideo.exercise_hash == exercise_hash) ) existing = result.scalar_one_or_none() if existing: existing.video_data = video_data existing.prompt_used = prompt existing.file_size = len(video_data) existing.last_accessed = datetime.utcnow() await db.commit() print(f"Updated cache for '{exercise_name}'") return existing else: new_video = ExerciseVideo( exercise_name=exercise_name, exercise_hash=exercise_hash, video_data=video_data, prompt_used=prompt, file_size=len(video_data) ) db.add(new_video) await db.commit() await db.refresh(new_video) print(f"Saved new video to cache for '{exercise_name}'") return new_video # ========== Exercise name validation (async, using LLM) ========== async def validate_and_clean_exercise_name(exercise_name: str) -> str: """Use LLM to validate and clean exercise name without changing its meaning""" if workout_db.validation_llm is None: workout_db.initialize_llms() validation_prompt = f"""You are an exercise name validator. Your job is to clean and standardize exercise names WITHOUT changing their meaning. Exercise Name: "{exercise_name}" Rules: 1. Fix obvious typos and spelling errors 2. Standardize capitalization (Title Case for exercise names) 3. Remove extra spaces, special characters that don't belong 4. Keep the core exercise name EXACTLY the same 5. Don't add or remove words that change the exercise type 6. If the name is already valid, return it as-is Examples: - "push up" -> "Push Up" - "burpee's" -> "Burpees" - "mountain-climbers" -> "Mountain Climbers" - "JUMPING JACKS" -> "Jumping Jacks" - "squat" -> "Squat" Return ONLY the cleaned exercise name, nothing else.""" try: async with LLM_SEMAPHORE: response = await workout_db.validation_llm.ainvoke(validation_prompt) cleaned_name = response.content.strip() if cleaned_name != exercise_name: print(f"Cleaned: '{exercise_name}' -> '{cleaned_name}'") return cleaned_name except Exception as e: print(f"Validation error for '{exercise_name}': {e}") return exercise_name.strip() # ========== Video generation functions (async, thread pool for Replicate) ========== def create_exercise_prompt(exercise_name: str) -> str: clean_name = exercise_name.strip() prompt = f"A person performing {clean_name} exercise, proper form, fitness demonstration, athletic movement, gym environment, professional fitness video, clear movement" return prompt async def generate_video_with_replicate_async(exercise_name: str) -> bytes: """Generate video using Replicate's SeedDance model (async wrapper)""" replicate_api_key = os.getenv("REPLICATE_API_TOKEN") if not replicate_api_key: raise HTTPException(status_code=500, detail="REPLICATE_API_TOKEN not configured") cleaned_name = await validate_and_clean_exercise_name(exercise_name) prompt = create_exercise_prompt(cleaned_name) print(f"Generating video for '{cleaned_name}'...") print(f" Prompt: {prompt}") loop = asyncio.get_running_loop() async with VIDEO_SEMAPHORE: try: output = await loop.run_in_executor( None, lambda: replicate.run( "bytedance/seedance-1-pro-fast", input={ "fps": 24, "prompt": prompt, "duration": 3, "resolution": "480p", "aspect_ratio": "16:9", "camera_fixed": True } ) ) video_data = output.read() print(f"Generated video for '{cleaned_name}' ({len(video_data)} bytes)") return video_data except Exception as e: error_msg = f"Replicate API error: {str(e)}" print(error_msg) raise HTTPException(status_code=500, detail=error_msg) async def get_or_generate_video_async(exercise_name: str, force_regenerate: bool = False) -> 'VideoResponse': """Get video from cache or generate new one (async)""" cleaned_name = await validate_and_clean_exercise_name(exercise_name) async for db in get_db(): if not force_regenerate: cached = await get_cached_video(db, cleaned_name) if cached: video_base64 = base64.b64encode(cached.video_data).decode('utf-8') return VideoResponse( exercise_name=cleaned_name, video_base64=video_base64, video_url=f"/workout/api/exercise_video/{cached.exercise_hash}", status="cached", from_cache=True, file_size=cached.file_size, created_at=cached.created_at, message="Retrieved from cache" ) video_data = await generate_video_with_replicate_async(cleaned_name) prompt = create_exercise_prompt(cleaned_name) saved = await save_video_to_cache(db, cleaned_name, video_data, prompt) video_base64 = base64.b64encode(video_data).decode('utf-8') return VideoResponse( exercise_name=cleaned_name, video_base64=video_base64, video_url=f"/workout/api/exercise_video/{saved.exercise_hash}", status="generated", from_cache=False, file_size=len(video_data), created_at=saved.created_at, message="Generated and cached" ) # Fallback (should never reach) raise RuntimeError("Database session not available") # ========== Translation functions (unchanged) ========== def translate_exercise(exercise: Dict, language: str) -> Dict: if language == 'en' or not translation_service.is_translation_enabled(): return exercise try: if 'name' in exercise and exercise['name']: exercise['name'] = translation_service.translate_text(exercise['name'], language) if 'duration' in exercise and exercise['duration']: duration = exercise['duration'] match = re.match(r'(\d+(?:\.\d+)?)\s*(.+)', duration) if match: number = match.group(1) unit = match.group(2) translated_unit = translation_service.translate_text(unit, language) exercise['duration'] = f"{number} {translated_unit}" else: exercise['duration'] = translation_service.translate_text(duration, language) if 'sets' in exercise and exercise['sets']: sets = exercise['sets'] match = re.match(r'(\d+(?:\.\d+)?)\s*(.+)', sets) if match: number = match.group(1) unit = match.group(2) translated_unit = translation_service.translate_text(unit, language) exercise['sets'] = f"{number} {translated_unit}" else: exercise['sets'] = translation_service.translate_text(sets, language) if 'calories' in exercise and exercise['calories']: calories = exercise['calories'] match = re.match(r'(\d+(?:\.\d+)?)\s*(.+)', calories) if match: number = match.group(1) unit = match.group(2) translated_unit = translation_service.translate_text(unit, language) exercise['calories'] = f"{number} {translated_unit}" else: exercise['calories'] = translation_service.translate_text(calories, language) if 'video_status' in exercise and exercise['video_status']: exercise['video_status'] = translation_service.translate_text(exercise['video_status'], language) except Exception as e: print(f"Exercise translation error: {e}") return exercise def translate_workout_plan(plan_data: Dict, language: str) -> Dict: if language == 'en' or not translation_service.is_translation_enabled(): return plan_data try: profile_fields = ['body_shape', 'activity_level', 'workout_level', 'main_goal', 'bmi_category'] for field in profile_fields: if field in plan_data and plan_data[field]: plan_data[field] = translation_service.translate_text(plan_data[field], language) if 'focus_areas' in plan_data and isinstance(plan_data['focus_areas'], list): translated_areas = [] for area in plan_data['focus_areas']: if area: translated_areas.append(translation_service.translate_text(area, language)) plan_data['focus_areas'] = translated_areas if 'plan_summary' in plan_data and plan_data['plan_summary']: summary = plan_data['plan_summary'] if 'average_duration' in summary: avg_duration = summary['average_duration'] if 'Min' in avg_duration: number = avg_duration.replace('Min', '').strip() translated_unit = translation_service.translate_text('Min', language) summary['average_duration'] = f"{number} {translated_unit}" for key in summary: if isinstance(summary[key], str) and key != 'average_duration': if not summary[key].isdigit(): summary[key] = translation_service.translate_text(summary[key], language) if 'weekly_workouts' in plan_data and plan_data['weekly_workouts']: for workout_day in plan_data['weekly_workouts']: if 'title' in workout_day: workout_day['title'] = translation_service.translate_text(workout_day['title'], language) if 'intensity' in workout_day: workout_day['intensity'] = translation_service.translate_text(workout_day['intensity'], language) if 'category' in workout_day: workout_day['category'] = translation_service.translate_text(workout_day['category'], language) if 'duration' in workout_day: duration = workout_day['duration'] if 'Min' in duration: number = duration.replace('Min', '').strip() translated_unit = translation_service.translate_text('Min', language) workout_day['duration'] = f"{number} {translated_unit}" else: workout_day['duration'] = translation_service.translate_text(duration, language) if 'total_calories' in workout_day: calories = workout_day['total_calories'] if 'Calories' in calories: number = calories.replace('Calories', '').strip() translated_unit = translation_service.translate_text('Calories', language) workout_day['total_calories'] = f"{number} {translated_unit}" else: workout_day['total_calories'] = translation_service.translate_text(calories, language) if 'warmup_exercises' in workout_day: for exercise in workout_day['warmup_exercises']: translate_exercise(exercise, language) if 'main_exercises' in workout_day: for exercise in workout_day['main_exercises']: translate_exercise(exercise, language) if 'cooldown_exercises' in workout_day: for exercise in workout_day['cooldown_exercises']: translate_exercise(exercise, language) plan_data['language'] = language plan_data['translation_service'] = translation_service.is_translation_enabled() except Exception as e: print(f"Workout plan translation error: {e}") return plan_data # ========== Convert LLM Output to UI Format (async version) ========== async def convert_to_ui_format(plan_dict: Dict, user_profile: UserProfile, language: str) -> WeeklyPlanUI: """Convert LLM output to UI format with video generation and auto-corrected totals (async)""" weekly_workouts = [] days_order = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] videos_generated = 0 videos_cached = 0 for day in days_order: if day not in plan_dict: continue day_data = plan_dict[day] display_day = day if language == 'ko': day_translations = { "Monday": "월요일", "Tuesday": "화요일", "Wednesday": "수요일", "Thursday": "목요일", "Friday": "금요일", "Saturday": "토요일", "Sunday": "일요일" } display_day = day_translations.get(day, day) warmup = [] warmup_total_duration = 0 warmup_total_calories = 0 for ex in day_data.get("warmup", []): cleaned_name = await validate_and_clean_exercise_name(ex.get("name", "").strip()) exercise_ui = ExerciseUI( name=cleaned_name, duration=ex.get("duration", "0 Min"), sets=ex.get("sets", "1 Set"), calories=ex.get("calories", "0 Calories") ) warmup_total_duration += parse_duration_to_minutes(exercise_ui.duration) warmup_total_calories += parse_calories(exercise_ui.calories) if user_profile.generate_videos: video_response = await get_or_generate_video_async(cleaned_name) exercise_ui.video_url = video_response.video_url exercise_ui.video_status = video_response.status if video_response.from_cache: videos_cached += 1 elif video_response.status == "generated": videos_generated += 1 warmup.append(exercise_ui) main = [] main_total_duration = 0 main_total_calories = 0 for ex in day_data.get("main", []): cleaned_name = await validate_and_clean_exercise_name(ex.get("name", "").strip()) exercise_ui = ExerciseUI( name=cleaned_name, duration=ex.get("duration", "0 Min"), sets=ex.get("sets", "1 Set"), calories=ex.get("calories", "0 Calories") ) main_total_duration += parse_duration_to_minutes(exercise_ui.duration) main_total_calories += parse_calories(exercise_ui.calories) if user_profile.generate_videos: video_response = await get_or_generate_video_async(cleaned_name) exercise_ui.video_url = video_response.video_url exercise_ui.video_status = video_response.status if video_response.from_cache: videos_cached += 1 elif video_response.status == "generated": videos_generated += 1 main.append(exercise_ui) cooldown = [] cooldown_total_duration = 0 cooldown_total_calories = 0 for ex in day_data.get("cooldown", []): cleaned_name = await validate_and_clean_exercise_name(ex.get("name", "").strip()) exercise_ui = ExerciseUI( name=cleaned_name, duration=ex.get("duration", "0 Min"), sets=ex.get("sets", "1 Set"), calories=ex.get("calories", "0 Calories") ) cooldown_total_duration += parse_duration_to_minutes(exercise_ui.duration) cooldown_total_calories += parse_calories(exercise_ui.calories) if user_profile.generate_videos: video_response = await get_or_generate_video_async(cleaned_name) exercise_ui.video_url = video_response.video_url exercise_ui.video_status = video_response.status if video_response.from_cache: videos_cached += 1 elif video_response.status == "generated": videos_generated += 1 cooldown.append(exercise_ui) total_duration = warmup_total_duration + main_total_duration + cooldown_total_duration total_calories = warmup_total_calories + main_total_calories + cooldown_total_calories title = day_data.get("title", f"{day} Workout") if language == 'ko': title = translation_service.translate_text(title, language) workout_day = WorkoutDayUI( day=display_day, title=title, duration=f"{total_duration} Min", intensity=day_data.get("intensity", "Medium"), total_calories=f"{total_calories} Calories", category=day_data.get("category", "GENERAL"), warmup_exercises=warmup, main_exercises=main, cooldown_exercises=cooldown ) weekly_workouts.append(workout_day) print(f"\n{display_day}: {total_duration} min, {total_calories} cal") total_weekly_calories = sum(parse_calories(w.total_calories) for w in weekly_workouts) plan_summary = { "total_workouts": len(weekly_workouts), "total_calories_per_week": total_weekly_calories, "average_duration": f"{sum(parse_duration_to_minutes(w.duration) for w in weekly_workouts) // len(weekly_workouts)} Min" if weekly_workouts else "0 Min", "intensity_distribution": { "high": sum(1 for w in weekly_workouts if "High" in w.intensity), "medium": sum(1 for w in weekly_workouts if "Medium" in w.intensity), "low": sum(1 for w in weekly_workouts if "Low" in w.intensity) } } return WeeklyPlanUI( user_id=user_profile.user_id, blood_type=user_profile.blood_type, age=user_profile.age, bmi=user_profile.calculate_bmi(), bmi_category=user_profile.get_bmi_category(), body_shape=user_profile.body_shape, activity_level=user_profile.activity_level, main_goal=user_profile.main_goal, workout_level=user_profile.workout_level, focus_areas=user_profile.focus_areas, plan_summary=plan_summary, weekly_workouts=weekly_workouts, videos_generated=videos_generated, videos_cached=videos_cached ) # ========== Personalized plan generation (async, with caching) ========== async def generate_personalized_plan(user_profile: UserProfile, language: str = 'en') -> WeeklyPlanUI: """Generate personalized workout plan (async, with Redis cache)""" # Initialize LLMs if needed if workout_db.llm is None: workout_db.initialize_llms() # Create context and prompt context = workout_db.create_personalized_context(user_profile) prompt = workout_db.create_personalized_prompt() # Use async chain chain = prompt | workout_db.llm | workout_db.WorkoutPlanOutputParser() print(f"\n{'='*60}") print(f"Generating plan for {user_profile.user_id}") print(f"Blood Type: {user_profile.blood_type}") print(f"Language: {language}") if user_profile.age: print(f"Age: {user_profile.age}") if user_profile.body_shape: print(f"Body Shape: {user_profile.body_shape}") if user_profile.activity_level: print(f"Activity Level: {user_profile.activity_level}") if user_profile.main_goal: print(f"Main Goal: {user_profile.main_goal}") if user_profile.focus_areas: print(f"Focus Areas: {', '.join(user_profile.focus_areas)}") print(f"Context length: {len(context)} characters") print(f"{'='*60}\n") try: async with LLM_SEMAPHORE: plan_dict = await chain.ainvoke({"personalized_context": context}) return await convert_to_ui_format(plan_dict, user_profile, language) except Exception as e: print(f"Error: {e}") raise HTTPException(status_code=500, detail=str(e)) # ========== Startup event (async) ========== async def startup_event(): print("="*60) print("Starting Personalized Workout Planner v1.0 (Scalable)") print("="*60) print("\nInitializing database...") await init_db() print("Database ready") print("\nLoading workout databases...") workout_db.load_and_parse_all_pdfs() print("\nInitializing AI...") workout_db.initialize_llms() print("AI initialized!") print("\nChecking translation service...") if translation_service.is_translation_enabled(): print("Translation service enabled") print(f" Supported languages: {translation_service.supported_languages}") else: print("Translation service disabled") app.add_event_handler("startup", startup_event) # ========== API Endpoints (with rate limiting) ========== @app.get("/") @limiter.limit("30/minute") async def root(request: Request): """API Documentation""" language = get_language_from_request(request) async for db in get_db(): result = await db.execute(select(ExerciseVideo)) video_count = len(result.all()) break base_response = { "service": "Personalized Blood Type Workout Planner with Videos", "version": "8.0.0 (Scalable)", "language": language, "translation_enabled": translation_service.is_translation_enabled(), "features": { "workout_plans": "AI-generated personalized workout plans", "user_profiling": "Comprehensive user profiling (body shape, activity level, goals, focus areas)", "video_generation": "Exercise videos using Replicate SeedDance", "video_caching": f"{video_count} videos cached", "name_validation": "LLM-powered exercise name cleaning", "smart_planning": "Goal-based and focus-area-driven workout generation", "multilingual": "Supports English and Korean" }, "endpoints": { "weekly_workout_plan_ui": "/workout/weekly_workout_plan_ui", "generate_exercise_video": "/workout/api/generate_video", "get_exercise_video": "/workout/api/exercise_video/{exercise_hash}", "video_cache_stats": "/workout/api/video_cache/stats", "health": "/workout/health" }, "language_support": { "detection_methods": ["query_param", "header", "cookie"], "supported_languages": ["en", "ko"], "default_language": "en" } } if language != 'en' and translation_service.is_translation_enabled(): base_response["service"] = translation_service.translate_text(base_response["service"], language) for feature_key, feature_desc in base_response["features"].items(): if feature_key != "video_caching": base_response["features"][feature_key] = translation_service.translate_text(feature_desc, language) endpoint_descriptions = { "weekly_workout_plan_ui": "주간 운동 계획 생성", "generate_exercise_video": "운동 비디오 생성", "get_exercise_video": "운동 비디오 조회", "validate_exercise_name": "운동 이름 검증", "video_cache_stats": "비디오 캐시 통계", "health": "서버 상태 확인" } if language == 'ko': for endpoint, korean_desc in endpoint_descriptions.items(): if endpoint in base_response["endpoints"]: base_response["endpoints"][endpoint] = korean_desc base_response["language_support"]["detection_methods"] = [ translation_service.translate_text(method, language) for method in base_response["language_support"]["detection_methods"] ] return JSONResponse(content=base_response) @app.get("/health") @limiter.limit("60/minute") async def health(request: Request): language = get_language_from_request(request) async for db in get_db(): result = await db.execute(select(ExerciseVideo)) video_count = len(result.all()) break health_data = { "status": "healthy", "language": language, "data_loaded": bool(workout_db.parsed_workout_data), "llm_ready": workout_db.llm is not None, "validation_llm_ready": workout_db.validation_llm is not None, "video_cache_size": video_count, "replicate_configured": bool(os.getenv("REPLICATE_API_TOKEN")), "translation_enabled": translation_service.is_translation_enabled(), "supported_languages": ['en', 'ko'] } if language != 'en' and translation_service.is_translation_enabled(): health_data["status"] = translation_service.translate_text("healthy", language) return JSONResponse(content=health_data) @app.get("/weekly_workout_plan_ui") @limiter.limit("10/minute") async def get_weekly_workout_plan_ui( request: Request, user_id: str = Query(..., description="User ID"), blood_type: str = Query(..., description="Blood Type (O, A, B, AB)"), age: Optional[int] = Query(None, ge=10, le=100, description="Age"), weight: Optional[float] = Query(None, gt=0, le=300, description="Weight (kg)"), height: Optional[float] = Query(None, gt=0, le=300, description="Height (cm)"), body_shape: Optional[str] = Query(None, description="Body Shape: Medium, Flabby, Skinny, Muscular"), activity_level: Optional[str] = Query(None, description="Activity Level: Sedentary, Lightly Active, Moderately Active, Very Active"), workout_level: Optional[str] = Query(None, description="Workout Level: Easy To Start, Break A Light Sweat, Challenging"), main_goal: Optional[str] = Query(None, description="Main Goal: Lose Weight, Gain Muscle, Stay Fit"), desired_weight: Optional[float] = Query(None, gt=0, le=300, description="Desired Weight (kg)"), focus_areas: Optional[str] = Query(None, description="Focus Areas (comma-separated): Arms, Upper Body, Abs, Butt, Legs"), generate_videos: bool = Query(True, description="Generate videos for exercises") ): language = get_language_from_request(request) if language not in ['en', 'ko']: language = 'en' blood_type = normalize_blood_type(blood_type) valid_blood_types = ["O", "A", "B", "AB"] if blood_type not in valid_blood_types: error_msg = "Invalid blood type. Must be O, A, B, or AB." if language == 'ko': error_msg = "잘못된 혈액형입니다. O, A, B, AB 중 하나여야 합니다." raise HTTPException(status_code=400, detail=error_msg) focus_areas_list = [] if focus_areas: focus_areas_list = [area.strip() for area in focus_areas.split(",") if area.strip()] user_profile = UserProfile( user_id=user_id, blood_type=blood_type, age=age, weight=weight, height=height, body_shape=body_shape, activity_level=activity_level, workout_level=workout_level, main_goal=main_goal, desired_weight=desired_weight, focus_areas=focus_areas_list, generate_videos=generate_videos ) plan = await generate_personalized_plan(user_profile, language) plan_dict = plan.dict() if language != 'en' and translation_service.is_translation_enabled(): plan_dict = translate_workout_plan(plan_dict, language) return JSONResponse(content=plan_dict) # ========== Video API Endpoints (async) ========== @app.post("/api/generate_video") @limiter.limit("20/minute") async def generate_exercise_video(request: Request, video_request: VideoGenerationRequest): language = get_language_from_request(request) response = await get_or_generate_video_async( video_request.exercise_name, force_regenerate=video_request.force_regenerate ) if language != 'en' and translation_service.is_translation_enabled() and response.message: response.message = translation_service.translate_text(response.message, language) return response @app.get("/api/exercise_video/{exercise_hash}") @limiter.limit("60/minute") async def get_exercise_video_by_hash(request: Request, exercise_hash: str): async for db in get_db(): result = await db.execute( select(ExerciseVideo).where(ExerciseVideo.exercise_hash == exercise_hash) ) video_record = result.scalar_one_or_none() if not video_record: raise HTTPException(status_code=404, detail="Video not found") video_record.last_accessed = datetime.utcnow() video_record.access_count += 1 await db.commit() video_data = video_record.video_data break return Response( content=video_data, media_type="video/mp4", headers={"Content-Disposition": f"inline; filename={exercise_hash}.mp4"} ) @app.get("/api/video_cache/stats") @limiter.limit("30/minute") async def get_video_cache_stats(request: Request): language = get_language_from_request(request) async for db in get_db(): result = await db.execute(select(ExerciseVideo)) all_videos = result.scalars().all() total_videos = len(all_videos) total_bytes = sum(v.file_size for v in all_videos) most_accessed = sorted(all_videos, key=lambda v: v.access_count, reverse=True)[:10] recent_videos = sorted(all_videos, key=lambda v: v.created_at, reverse=True)[:10] break stats_data = { "total_videos": total_videos, "total_size_mb": round(total_bytes / (1024 * 1024), 2), "average_size_kb": round((total_bytes / total_videos) / 1024, 2) if total_videos > 0 else 0, "most_accessed": [ { "exercise": video.exercise_name, "access_count": video.access_count, "last_accessed": video.last_accessed } for video in most_accessed ], "recent_additions": [ { "exercise": video.exercise_name, "created_at": video.created_at, "file_size_kb": round(video.file_size / 1024, 2) } for video in recent_videos ] } if language != 'en' and translation_service.is_translation_enabled(): for item in stats_data["most_accessed"]: item["exercise"] = translation_service.translate_text(item["exercise"], language) for item in stats_data["recent_additions"]: item["exercise"] = translation_service.translate_text(item["exercise"], language) return JSONResponse(content=stats_data) @app.delete("/api/video_cache/clear") @limiter.limit("5/minute") async def clear_video_cache( request: Request, confirm: bool = Query(False, description="Confirm deletion") ): language = get_language_from_request(request) if not confirm: message = "Add ?confirm=true to actually clear the cache" warning = "This will delete all cached videos" if language == 'ko' and translation_service.is_translation_enabled(): message = "캐시를 실제로 지우려면 ?confirm=true를 추가하세요" warning = "이 작업은 모든 캐시된 비디오를 삭제합니다" return JSONResponse(content={"message": message, "warning": warning}) async for db in get_db(): result = await db.execute(select(ExerciseVideo)) all_videos = result.scalars().all() deleted = len(all_videos) for video in all_videos: await db.delete(video) await db.commit() break message = "Cache cleared successfully" if language == 'ko' and translation_service.is_translation_enabled(): message = "캐시가 성공적으로 삭제되었습니다" return JSONResponse(content={"message": message, "videos_deleted": deleted})