Spaces:
Running
Running
File size: 4,315 Bytes
9bc1376 801dcf0 9bc1376 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 |
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
|