Spaces:
Sleeping
Sleeping
| import re | |
| from dataclasses import dataclass | |
| from viral_script_engine.platforms.platform_spec import PlatformRegistry | |
| _PACING_THRESHOLDS = {"very_fast": 8, "fast": 12, "moderate": 18} | |
| _CTA_TARGETS = { | |
| "last_5_percent": 0.95, | |
| "last_8_percent": 0.92, | |
| "last_10_percent": 0.90, | |
| "last_15_percent": 0.85, | |
| } | |
| class PacingRewardResult: | |
| score: float | |
| pacing_score: float | |
| ratio_score: float | |
| cta_score: float | |
| platform: str | |
| def _split_sections(script: str): | |
| """Split script into hook, body, cta by sentence position.""" | |
| sentences = re.split(r'(?<=[.!?])\s+', script.strip()) | |
| if not sentences: | |
| return "", "", "" | |
| n = len(sentences) | |
| if n == 1: | |
| return sentences[0], "", "" | |
| if n == 2: | |
| return sentences[0], sentences[1], "" | |
| # hook = first ~20%, body = next ~70%, cta = last ~10% | |
| hook_end = max(1, int(n * 0.2)) | |
| cta_start = max(hook_end + 1, int(n * 0.9)) | |
| hook = " ".join(sentences[:hook_end]) | |
| body = " ".join(sentences[hook_end:cta_start]) | |
| cta = " ".join(sentences[cta_start:]) | |
| return hook, body, cta | |
| def _avg_words_per_sentence(text: str) -> float: | |
| if not text.strip(): | |
| return 0.0 | |
| sentences = [s for s in re.split(r'(?<=[.!?])\s+', text.strip()) if s] | |
| if not sentences: | |
| return 0.0 | |
| total_words = sum(len(s.split()) for s in sentences) | |
| return total_words / len(sentences) | |
| class PlatformPacingReward: | |
| """ | |
| Rule-based pacing reward — zero LLM calls. | |
| Checks sentence length distribution, section ratios, and CTA position | |
| against the target platform's spec. | |
| """ | |
| def __init__(self): | |
| self.platform_registry = PlatformRegistry() | |
| def score(self, script: str, platform: str) -> PacingRewardResult: | |
| spec = self.platform_registry.get(platform) | |
| hook, body, cta = _split_sections(script) | |
| # Check 1: avg words per sentence in hook vs pacing norm | |
| hook_avg_words = _avg_words_per_sentence(hook) if hook else 0.0 | |
| threshold = _PACING_THRESHOLDS.get(spec.pacing_norm, 12) | |
| if hook_avg_words == 0.0: | |
| pacing_score = 0.5 | |
| elif hook_avg_words <= threshold: | |
| pacing_score = 1.0 | |
| else: | |
| pacing_score = max(0.0, 1.0 - (hook_avg_words - threshold) / threshold) | |
| # Check 2: section length ratio vs optimal | |
| hook_words = len(hook.split()) if hook else 0 | |
| body_words = len(body.split()) if body else 0 | |
| cta_words = len(cta.split()) if cta else 0 | |
| total_words = max(hook_words + body_words + cta_words, 1) | |
| section_totals = sum(spec.optimal_sentences_per_section.values()) | |
| optimal_hook_ratio = spec.optimal_sentences_per_section["hook"] / section_totals | |
| actual_hook_ratio = hook_words / total_words | |
| deviation = abs(actual_hook_ratio - optimal_hook_ratio) | |
| ratio_score = max(0.0, 1.0 - min(1.0, deviation / max(optimal_hook_ratio, 0.01))) | |
| # Check 3: CTA position | |
| cta_start_pos = (hook_words + body_words) / total_words | |
| cta_target = _CTA_TARGETS.get(spec.cta_position, 0.90) | |
| cta_score = 1.0 if cta_start_pos >= cta_target else 0.5 | |
| final_score = pacing_score * 0.4 + ratio_score * 0.4 + cta_score * 0.2 | |
| return PacingRewardResult( | |
| score=round(final_score, 4), | |
| pacing_score=round(pacing_score, 4), | |
| ratio_score=round(ratio_score, 4), | |
| cta_score=round(cta_score, 4), | |
| platform=platform, | |
| ) | |