import json import re from pathlib import Path from typing import List, Optional from pydantic import BaseModel from viral_script_engine.platforms.platform_spec import PlatformRegistry _FILLER_PHRASES = [ "hey guys", "welcome back", "today i want to", "so today", "in this video", "what's up everyone", "hey everyone", "guys today", "hello everyone", "so basically", "you know", "kind of", "sort of", "basically", "um ", "uh ", ] _COMMON_WORDS = { 'i', 'the', 'a', 'an', 'my', 'your', 'its', 'it', 'is', 'are', 'was', 'were', 'be', 'been', "i've", "i'm", "it's", "here's", 'today', 'and', 'but', 'so', 'that', 'this', 'these', 'those', 'to', 'of', 'in', 'for', 'on', 'with', 'at', 'by', 'from', 'or', 'not', 'you', 'we', 'they', 'he', 'she', 'if', 'do', 'get', 'just', 'up', 'out', 'about', 'what', 'all', 'some', } _PROMISE_PATTERNS = [ r'\d', r'\bhow to\b', r'\bwhy\b', r'\bwhat happens when\b', r'\bi made\b', r'\bwill\b', r'\bguaranteed\b', r'\bstep\b', r'\btips?\b', r'\bsecrets?\b', r'\bprove[sd]?\b', r'\bhere\'?s\b', ] _KNOWN_PLATFORMS = ["Reels", "Shorts", "Feed", "TikTok"] class ScriptFeatures(BaseModel): # Hook features (predicts early drop-off 0–5s) hook_word_count: int hook_has_number: bool hook_has_question: bool hook_has_promise: bool hook_filler_score: float # 0=no filler, 1=all filler # Pacing features (predicts mid-video retention 5–30s) avg_words_per_sentence: float sentence_count: int short_sentence_ratio: float # sentences < 8 words / total sentences section_balance_score: float # how evenly hook:body:cta matches platform spec # Content features (predicts late retention 30s+) specificity_score: float # ratio of specific nouns/numbers to total words cultural_ref_count: int cta_position_ratio: float # word offset of CTA start / total words # Platform fit features platform: str word_count: int length_vs_optimal: float # word_count / optimal_script_length for platform def to_vector(self) -> List[float]: platform_one_hot = [1.0 if self.platform == p else 0.0 for p in _KNOWN_PLATFORMS] return [ float(self.hook_word_count), 1.0 if self.hook_has_number else 0.0, 1.0 if self.hook_has_question else 0.0, 1.0 if self.hook_has_promise else 0.0, float(self.hook_filler_score), float(self.avg_words_per_sentence), float(self.sentence_count), float(self.short_sentence_ratio), float(self.section_balance_score), float(self.specificity_score), float(self.cultural_ref_count), float(self.cta_position_ratio), float(self.word_count), float(self.length_vs_optimal), ] + platform_one_hot class FeatureExtractor: def __init__(self, cultural_kb_path: Optional[str] = None): self.platform_registry = PlatformRegistry() self._cultural_kb_path = cultural_kb_path self._cultural_kb: Optional[dict] = None def _load_kb(self) -> None: if self._cultural_kb is not None: return kb_path = self._cultural_kb_path or str( Path(__file__).parent.parent / "data" / "cultural_kb.json" ) with open(kb_path, "r", encoding="utf-8") as f: self._cultural_kb = json.load(f) def extract(self, script: str, platform: str, region: str) -> ScriptFeatures: self._load_kb() spec = self.platform_registry.get(platform) sentences = [s for s in re.split(r'(?<=[.!?])\s+', script.strip()) if s.strip()] if not sentences: sentences = [script] total_words = len(script.split()) # --- Hook: first ~20% of sentences (min 1, max 3) --- n = len(sentences) hook_end = max(1, min(3, int(n * 0.2))) if n >= 5 else max(1, min(2, n)) hook_text = " ".join(sentences[:hook_end]) hook_lower = hook_text.lower() hook_words = hook_text.split() hook_word_count = len(hook_words) hook_has_number = bool(re.search(r'\d', hook_text)) hook_has_question = '?' in hook_text hook_has_promise = any(re.search(p, hook_lower) for p in _PROMISE_PATTERNS) filler_hits = sum(1 for phrase in _FILLER_PHRASES if phrase in hook_lower) hook_filler_score = min(1.0, filler_hits / max(hook_word_count, 1) * 4) # --- Pacing --- sentence_count = n words_per_sent = [len(s.split()) for s in sentences] avg_words_per_sentence = sum(words_per_sent) / max(n, 1) short_sentence_ratio = sum(1 for w in words_per_sent if w < 8) / max(n, 1) # Section balance: compare actual word distribution to platform spec cta_start_idx = max(hook_end + 1, n - max(1, int(n * 0.1))) hook_w = sum(len(s.split()) for s in sentences[:hook_end]) body_w = sum(len(s.split()) for s in sentences[hook_end:cta_start_idx]) cta_w = sum(len(s.split()) for s in sentences[cta_start_idx:]) total_w = max(hook_w + body_w + cta_w, 1) opt = spec.optimal_sentences_per_section opt_total = max(sum(opt.values()), 1) opt_hook_r = opt.get("hook", 2) / opt_total opt_body_r = opt.get("body", 6) / opt_total act_hook_r = hook_w / total_w act_body_r = body_w / total_w balance_dev = (abs(act_hook_r - opt_hook_r) + abs(act_body_r - opt_body_r)) / 2 section_balance_score = max(0.0, 1.0 - balance_dev * 4) # --- Content features --- words = script.split() specific_count = sum( 1 for w in words if ( re.search(r'\d', w) or (len(w) > 1 and w[0].isupper() and w.lower().strip('.,!?;:\'"') not in _COMMON_WORDS) ) ) specificity_score = min(1.0, specific_count / max(total_words, 1)) cultural_ref_count = 0 if self._cultural_kb and region in self._cultural_kb: kb = self._cultural_kb[region] script_lower = script.lower() cultural_ref_count = ( sum(1 for r in kb.get("valid_refs", []) if r.lower() in script_lower) + sum(1 for i in kb.get("correct_idioms", []) if i.lower() in script_lower) ) cta_word_offset = hook_w + body_w cta_position_ratio = cta_word_offset / max(total_words, 1) # --- Platform fit --- length_vs_optimal = total_words / max(spec.optimal_script_length_words, 1) return ScriptFeatures( hook_word_count=hook_word_count, hook_has_number=hook_has_number, hook_has_question=hook_has_question, hook_has_promise=hook_has_promise, hook_filler_score=round(hook_filler_score, 4), avg_words_per_sentence=round(avg_words_per_sentence, 4), sentence_count=sentence_count, short_sentence_ratio=round(short_sentence_ratio, 4), section_balance_score=round(section_balance_score, 4), specificity_score=round(specificity_score, 4), cultural_ref_count=cultural_ref_count, cta_position_ratio=round(cta_position_ratio, 4), platform=platform, word_count=total_words, length_vs_optimal=round(length_vs_optimal, 4), )