Spaces:
Running
Running
| from __future__ import annotations | |
| import os, time, json, mimetypes, tempfile, logging | |
| from typing import Dict, Any, Optional | |
| from schema.pydantic_schema_video import AdAnalysis | |
| from prompt.analyser_prompt import ANALYSER_PROMPT | |
| from google import genai | |
| from dotenv import load_dotenv | |
| from database.operations import insert_video_analysis | |
| from helpers_function.helpers import get_video_thumbnail_base64 | |
| load_dotenv() | |
| logger = logging.getLogger(__name__) | |
| GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") | |
| def _configure_gemini() -> genai.Client: | |
| if not GEMINI_API_KEY: | |
| raise RuntimeError("GEMINI_API_KEY is not set") | |
| return genai.Client(api_key=GEMINI_API_KEY) | |
| def analyze_video_only(video_path: str) -> Dict[str, Any]: | |
| try: | |
| client = _configure_gemini() | |
| except Exception as e: | |
| logger.exception("Gemini configuration error") | |
| return {"__error__": str(e)} | |
| try: | |
| f = client.files.upload(file=video_path) | |
| while getattr(f.state, "name", "") == "PROCESSING": | |
| time.sleep(2); f = client.files.get(name=f.name) | |
| if getattr(f.state, "name", "") == "FAILED": | |
| return {"__error__": "Video indexing failed."} | |
| resp = client.models.generate_content( | |
| model=os.getenv("VIDEO_ANALYZER_MODEL", "gemini-2.0-flash"), | |
| contents=[ANALYSER_PROMPT, f], | |
| config={"response_mime_type": "application/json"}, | |
| ) | |
| raw = getattr(resp, "text", "") or "" | |
| if not raw.strip(): | |
| return {"__error__": "Empty response from model."} | |
| try: | |
| return AdAnalysis.model_validate_json(raw).model_dump() | |
| except Exception: | |
| try: | |
| return json.loads(raw) | |
| except Exception: | |
| return {"__error__": "Model response not valid JSON."} | |
| except Exception as e: | |
| logger.exception("Video analysis failed") | |
| return {"__error__": str(e)} | |
| def run_and_store_video_analysis( | |
| *, | |
| category: str, | |
| created_by: str, | |
| uploaded_file=None, | |
| uploaded_file_path: Optional[str] = None, | |
| analyzer_model: Optional[str] = None, | |
| ) -> Dict[str, Any]: | |
| if not uploaded_file and not uploaded_file_path: | |
| return {"_id": None, "results": {"__error__": "No video provided."}, "video_meta": {}, "thumbnail": ""} | |
| tmp_created: Optional[str] = None | |
| try: | |
| if uploaded_file_path: | |
| video_path = uploaded_file_path | |
| orig_name = os.path.basename(uploaded_file_path) | |
| mime = mimetypes.guess_type(orig_name)[0] or "" | |
| size_bytes = os.path.getsize(uploaded_file_path) if os.path.exists(uploaded_file_path) else None | |
| else: | |
| suffix = os.path.splitext(getattr(uploaded_file, "name", "upload.mp4"))[1] or ".mp4" | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp: | |
| tmp.write(uploaded_file.read()) | |
| video_path = tmp.name | |
| tmp_created = video_path | |
| orig_name = getattr(uploaded_file, "name", os.path.basename(video_path)) | |
| mime = getattr(uploaded_file, "type", None) or mimetypes.guess_type(orig_name)[0] or "" | |
| size_bytes = os.path.getsize(video_path) if os.path.exists(video_path) else None | |
| results = analyze_video_only(video_path) | |
| try: | |
| thumbnail_b64 = get_video_thumbnail_base64(video_path) or "" | |
| except Exception: | |
| thumbnail_b64 = "" | |
| video_meta = {"name": orig_name, "mimetype": mime, "size_bytes": size_bytes} | |
| model_label = analyzer_model or os.getenv("VIDEO_ANALYZER_MODEL", "gemini-2.0-flash") | |
| inserted_id = insert_video_analysis( | |
| video_name=orig_name, | |
| response=results, | |
| category=(category or "general"), | |
| created_by=created_by, | |
| analyzer_model=model_label, | |
| video_meta=video_meta, | |
| thumbnail=thumbnail_b64, | |
| ) | |
| return { | |
| "_id": inserted_id, | |
| "results": results, | |
| "video_meta": video_meta, | |
| "thumbnail": thumbnail_b64, | |
| } | |
| finally: | |
| if tmp_created and os.path.exists(tmp_created): | |
| try: | |
| os.remove(tmp_created) | |
| except Exception: | |
| pass | |