import os import time import json from typing import Dict, Any, List, Literal import tempfile from urllib.request import urlretrieve from pydantic import BaseModel, constr from google import genai from dotenv import load_dotenv load_dotenv() GEMINI_API_KEY = os.getenv("GEMENI_KEY", "") def configure_gemini() -> genai.Client: if not GEMINI_API_KEY: raise RuntimeError("GEMENI_KEY is not set in environment variables.") return genai.Client(api_key=GEMINI_API_KEY) Timestamp = constr(pattern=r'^\d{2}:\d{2}$') RangeTimestamp = constr(pattern=r'^\d{2}:\d{2}-\d{2}:\d{2}$') Score010 = constr(pattern=r'^(?:10|[0-9])\/10$') class Hook(BaseModel): hook_text: str principle: str advantages: List[str] class StoryboardItem(BaseModel): timeline: Timestamp scene: str visuals: str dialogue: str camera: str sound_effects: str class ScriptLine(BaseModel): timeline: Timestamp dialogue: str class VideoMetric(BaseModel): timestamp: RangeTimestamp element: str current_approach: str effectiveness_score: Score010 notes: str class VideoAnalysis(BaseModel): effectiveness_factors: str psychological_triggers: str target_audience: str video_metrics: List[VideoMetric] class TimestampImprovement(BaseModel): timestamp: RangeTimestamp current_element: str improvement_type: str recommended_change: str expected_impact: str priority: Literal["High", "Medium", "Low"] class AdAnalysis(BaseModel): brief: str caption_details: str hook: Hook framework_analysis: str storyboard: List[StoryboardItem] script: List[ScriptLine] video_analysis: VideoAnalysis timestamp_improvements: List[TimestampImprovement] analyser_prompt = """You are an expert video advertisement analyst. Analyze the provided video and give response conforms EXACTLY to the schema below with no extra text or markdown. Populate: 1. **brief** → A concise summary covering visual style, speaker, target audience, and marketing objective. 2. **caption_details** → Description of captions (color/style/position) or exactly the string `"None"` if not visible. 3. **hook** → - `"hook_text"`: Exact opening line or, if no speech, the precise description of the opening visual. - `"principle"`: Psychological/marketing principle that makes this hook effective. - `"advantages"`: ARRAY of 3–6 concise benefit statements tied to the ad’s value proposition. 4. **framework_analysis** → A detailed block identifying copywriting/psychology/storytelling frameworks (e.g., PAS, AIDA). Highlight use of social proof, urgency, fear, authority, scroll-stopping hooks, loop openers, value positioning, and risk reversals. 5. **storyboard** → ARRAY of 4–10 objects. Each must include: - `"timeline"` in `"MM:SS"` (zero-padded) - `"scene"` (brief) - `"visuals"` (detailed) - `"dialogue"` (exact words; use `""` if none) - `"camera"` (shot/angle) - `"sound_effects"` (or `"None"`) 6. **script** → ARRAY of dialogue objects, each with `"timeline"` (`"MM:SS"`) and `"dialogue"` (exact spoken line). 7. **video_analysis** → OBJECT with: - `"effectiveness_factors"`: Key factors that influence effectiveness - `"psychological_triggers"`: Triggers used (e.g., scarcity, authority) - `"target_audience"`: Audience profile inferred - `"video_metrics"`: ARRAY of objects with: - `"timestamp"`: `"MM:SS-MM:SS"` - `"element"`: The aspect being evaluated (e.g., Hook Strategy) - `"current_approach"`: Description of current execution - `"effectiveness_score"`: String score `"X/10"` (integer X) - `"notes"`: Analytical notes 8. **timestamp_improvements** → ARRAY of recommendation objects with: - `"timestamp"`: `"MM:SS-MM:SS"` - `"current_element"`: Current content of the segment - `"improvement_type"`: Category (e.g., Hook Enhancement) - `"recommended_change"`: Specific recommendation - `"expected_impact"`: Projected effect on metrics or perception - `"priority"`: `"High"`, `"Medium"`, or `"Low"` ⚠️ The output must be strictly matching field names and types, no additional keys, and all timestamps must be zero-padded (`"MM:SS"` for single points, `"MM:SS-MM:SS"` for ranges). """ def analyse_video_only(video_path: str) -> Dict[str, Any]: client = configure_gemini() try: video_file = client.files.upload(file=video_path) while getattr(video_file.state, "name", "") == "PROCESSING": time.sleep(2) video_file = client.files.get(name=video_file.name) if getattr(video_file.state, "name", "") == "FAILED": return {} print("Uploaded the video") resp = client.models.generate_content( model="gemini-2.0-flash", contents=[analyser_prompt, video_file], config={"response_mime_type": "application/json", "response_schema": AdAnalysis,} ) raw = getattr(resp, "text", "") or "" try: model_obj = AdAnalysis.model_validate_json(raw) return model_obj.model_dump() except Exception: try: return json.loads(raw) except Exception: return {} except Exception: return {} def download_video_to_temp(video_url): temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") temp_file_path = temp_file.name temp_file.close() urlretrieve(video_url, temp_file_path) print("Downloaded the video") return temp_file_path def get_analysis(video_url): file_path = download_video_to_temp(video_url) analyses = analyse_video_only(file_path) return analyses