import json import re import os import requests import time # ── Configuration ───────────────────────────────────────────────────────────── GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "") HF_API_KEY = os.environ.get("HF_API_KEY", "") # Cache file to remember which HF model worked last time HF_MODEL_CACHE = os.path.join(os.path.dirname(__file__), ".hf_model_cache") # Preferred HF models in order (updated periodically, auto-discovery as fallback) HF_PREFERRED_MODELS = [ "meta-llama/Llama-3.2-3B-Instruct", "mistralai/Mistral-7B-Instruct-v0.3", "microsoft/Phi-3-mini-4k-instruct", "google/gemma-2-2b-it", "HuggingFaceH4/zephyr-7b-beta", "TinyLlama/TinyLlama-1.1B-Chat-v1.0", ] PROMPT_TEMPLATE = """You are given a video transcript with timestamps. Find 3 engaging segments for short-form video. RULES: - Each clip MUST be 30 to 90 seconds long (end - start >= 30) - Use EXACT timestamps from the transcript - Return ONLY a raw JSON array, nothing else Example output: [ {{"start": 45.0, "end": 105.0, "title": "Catchy title", "score": 8, "reason": "Strong hook"}}, {{"start": 200.0, "end": 260.0, "title": "Another title", "score": 7, "reason": "Key insight"}} ] Transcript: {transcript} JSON array:""" # ── Helpers ─────────────────────────────────────────────────────────────────── def build_transcript_text(transcript_result: dict) -> str: segments = transcript_result.get("segments", []) lines = [] for seg in segments: lines.append(f"[{seg['start']:.0f}s-{seg['end']:.0f}s] {seg['text'].strip()}") return "\n".join(lines) def extract_json(text: str) -> list: """Extract and validate a JSON array from model response.""" match = re.search(r'\[.*?\]', text, re.DOTALL) if not match: return [] try: segments = json.loads(match.group()) valid = [] for seg in segments: if "start" in seg and "end" in seg: seg["start"] = float(seg["start"]) seg["end"] = float(seg["end"]) if seg["end"] > seg["start"]: valid.append(seg) return valid except (json.JSONDecodeError, ValueError): return [] def fallback_segments(transcript_result: dict) -> list: """Split video into equal 60-second clips as last resort.""" segments = transcript_result.get("segments", []) if not segments: return [] total = segments[-1]["end"] clips, start, i = [], 0.0, 1 while start + 60 <= total and i <= 5: clips.append({"start": start, "end": start + 60, "title": f"Highlight {i}", "score": 5, "reason": "Auto-generated"}) start += 60 i += 1 return clips # ── Provider: Groq ──────────────────────────────────────────────────────────── def try_groq(prompt: str) -> list: if not GROQ_API_KEY: return [] try: resp = requests.post( "https://api.groq.com/openai/v1/chat/completions", headers={"Authorization": f"Bearer {GROQ_API_KEY}", "Content-Type": "application/json"}, json={"model": "llama-3.1-8b-instant", "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, "max_tokens": 1024}, timeout=30, ) if resp.status_code == 200: content = resp.json()["choices"][0]["message"]["content"] return extract_json(content) except Exception: pass return [] # ── Provider: HuggingFace (single model) ───────────────────────────────────── def try_hf_model(model_id: str, prompt: str) -> list: headers = {"Content-Type": "application/json"} if HF_API_KEY: headers["Authorization"] = f"Bearer {HF_API_KEY}" try: resp = requests.post( f"https://api-inference.huggingface.co/models/{model_id}", headers=headers, json={"inputs": prompt, "parameters": {"max_new_tokens": 512, "temperature": 0.3, "return_full_text": False}}, timeout=45, ) if resp.status_code == 200: data = resp.json() if isinstance(data, list) and data: text = data[0].get("generated_text", "") result = extract_json(text) if result: return result except Exception: pass return [] # ── HuggingFace auto-discovery ──────────────────────────────────────────────── def discover_hf_models() -> list: """ Query HuggingFace API to find currently available text-generation models with active inference endpoints. Returns list of model IDs. """ try: resp = requests.get( "https://huggingface.co/api/models", params={ "pipeline_tag": "text-generation", "inference": "warm", "sort": "likes", "direction": "-1", "limit": 20, "filter": "conversational", }, timeout=15, ) if resp.status_code == 200: models = resp.json() return [m["modelId"] for m in models if "modelId" in m] except Exception: pass return [] def load_cached_hf_model() -> str: if os.path.exists(HF_MODEL_CACHE): with open(HF_MODEL_CACHE) as f: return f.read().strip() return "" def save_cached_hf_model(model_id: str): with open(HF_MODEL_CACHE, "w") as f: f.write(model_id) def try_huggingface(prompt: str) -> list: """ Try HuggingFace models in this order: 1. Last known working model (cached) 2. Preferred hardcoded list 3. Auto-discovered models from HF API """ candidates = [] # 1. Try cached model first cached = load_cached_hf_model() if cached: candidates.append(cached) # 2. Add preferred list (skip duplicates) for m in HF_PREFERRED_MODELS: if m not in candidates: candidates.append(m) # Try all candidates for model_id in candidates: result = try_hf_model(model_id, prompt) if result: save_cached_hf_model(model_id) return result # 3. Auto-discover new working models from HuggingFace API print("[analyzer] Preferred models failed, auto-discovering HuggingFace models...") discovered = discover_hf_models() for model_id in discovered: if model_id in candidates: continue result = try_hf_model(model_id, prompt) if result: save_cached_hf_model(model_id) print(f"[analyzer] Found working HF model: {model_id}") return result return [] # ── Main entry point ────────────────────────────────────────────────────────── def analyze_transcript(transcript_result: dict) -> list: """ Analyze transcript using: 1. Groq (fast, free tier) 2. HuggingFace (auto-selects best working model) 3. Rule-based fallback (always works) """ transcript_text = build_transcript_text(transcript_result) if not transcript_text.strip(): return fallback_segments(transcript_result) # Limit to 100 lines to avoid token limits lines = transcript_text.split("\n") if len(lines) > 100: transcript_text = "\n".join(lines[:100]) prompt = PROMPT_TEMPLATE.format(transcript=transcript_text) # 1. Try Groq result = try_groq(prompt) if result: print("[analyzer] Used Groq") return result # 2. Try HuggingFace (auto-discover) result = try_huggingface(prompt) if result: print("[analyzer] Used HuggingFace") return result # 3. Rule-based fallback print("[analyzer] Using rule-based fallback") return fallback_segments(transcript_result)