| """Construcción ligera de patrones de guion a partir de transcripciones públicas. |
| |
| Este módulo no descarga datos en runtime. Lee transcripciones ya cargadas en: |
| data/raw/youtube_transcripts_900/Transcripts/*.json |
| |
| También deja preparado el punto de extensión para Hugging Face: |
| from datasets import load_dataset |
| dataset = load_dataset("ZelonPrograms/Youtube") |
| |
| En la demo pública no se descarga ZelonPrograms/Youtube para evitar fallos por red, |
| tiempo o cambios externos. La referencia queda documentada y el pipeline funciona |
| con el dataset Kaggle de transcripciones incluido en el repositorio. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import re |
| from collections import Counter |
| from pathlib import Path |
| from typing import Any, Dict, Iterable, List |
|
|
| import pandas as pd |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| RAW_TRANSCRIPTS_DIR = ROOT / "data" / "raw" / "youtube_transcripts_900" / "Transcripts" |
| PROCESSED_DIR = ROOT / "data" / "processed" |
| FEATURES_CSV = PROCESSED_DIR / "youtube_transcripts_script_features.csv" |
| PATTERNS_JSON = PROCESSED_DIR / "script_patterns_reference.json" |
|
|
| CTA_TERMS = { |
| "subscribe", "like", "comment", "share", "click", "download", "visit", "buy", "try", "watch", |
| "learn more", "sign up", "follow", "save", "join", "comenta", "suscríbete", "comparte", "guarda", |
| "haz clic", "visita", "compra", "descarga", "regístrate", "sígueme", "conoce más", |
| } |
| BENEFIT_TERMS = { |
| "learn", "improve", "discover", "avoid", "reduce", "increase", "optimize", "how to", "why", "tips", |
| "strategy", "mistake", "secret", "guide", "aprende", "mejora", "descubre", "evita", "reduce", |
| "aumenta", "optimiza", "cómo", "por qué", "consejos", "estrategia", "errores", "guía", |
| } |
| URGENCY_TERMS = { |
| "now", "today", "limited", "urgent", "last chance", "hoy", "ahora", "urgente", "última oportunidad", "limitado" |
| } |
| TRUST_TERMS = { |
| "official", "verified", "study", "evidence", "expert", "safe", "guarantee", "case study", "oficial", |
| "verificado", "estudio", "evidencia", "experto", "seguro", "caso real", |
| } |
| EXAGGERATED_TERMS = { |
| "guaranteed", "100%", "without effort", "miracle", "instant", "get rich", "lose weight fast", |
| "garantizado", "100 %", "sin esfuerzo", "milagroso", "resultado inmediato", "gana dinero rápido", |
| "pierde peso rápido", "cura definitiva", |
| } |
| GENERIC_OPENINGS = { |
| "hello", "hi guys", "hey guys", "welcome", "today we are going", "in this video", |
| "hola", "hola amigos", "bienvenidos", "en este video", "hoy vamos", |
| } |
|
|
|
|
| def _normalize_text(text: str) -> str: |
| text = re.sub(r"\s+", " ", str(text or "")).strip() |
| return text |
|
|
|
|
| def _count_terms(text: str, terms: Iterable[str]) -> int: |
| low = text.lower() |
| return sum(1 for t in terms if t.lower() in low) |
|
|
|
|
| def _first_words(text: str, n: int = 30) -> str: |
| return " ".join(text.split()[:n]) |
|
|
|
|
| def _read_transcript_json(path: Path) -> str: |
| try: |
| data = json.loads(path.read_text(encoding="utf-8")) |
| except Exception: |
| return "" |
| if isinstance(data, list): |
| return _normalize_text(" ".join(str(x.get("text", "")) if isinstance(x, dict) else str(x) for x in data)) |
| if isinstance(data, dict): |
| if "transcript" in data: |
| return _normalize_text(data.get("transcript", "")) |
| if "text" in data: |
| return _normalize_text(data.get("text", "")) |
| if "segments" in data and isinstance(data["segments"], list): |
| return _normalize_text(" ".join(str(x.get("text", "")) for x in data["segments"] if isinstance(x, dict))) |
| return "" |
|
|
|
|
| def _script_quality_features(transcript: str) -> Dict[str, Any]: |
| transcript = _normalize_text(transcript) |
| words = transcript.split() |
| hook = _first_words(transcript, 30) |
| low_hook = hook.lower() |
| low_text = transcript.lower() |
|
|
| word_count = len(words) |
| sentence_parts = [s.strip() for s in re.split(r"[.!?]+", transcript) if s.strip()] |
| avg_sentence_len = (sum(len(s.split()) for s in sentence_parts) / max(len(sentence_parts), 1)) if sentence_parts else 0 |
|
|
| hook_has_question = "?" in hook or any(low_hook.startswith(x) for x in ["how", "why", "what", "when", "where", "cómo", "por qué", "qué"]) |
| hook_has_benefit = _count_terms(hook, BENEFIT_TERMS) > 0 |
| generic_opening = any(p in low_hook[:160] for p in GENERIC_OPENINGS) |
| cta_count = _count_terms(transcript, CTA_TERMS) |
| benefit_count = _count_terms(transcript, BENEFIT_TERMS) |
| urgency_count = _count_terms(transcript, URGENCY_TERMS) |
| trust_count = _count_terms(transcript, TRUST_TERMS) |
| exaggerated_count = _count_terms(transcript, EXAGGERATED_TERMS) |
|
|
| hook_score = 50 + 20 * int(hook_has_question) + 25 * int(hook_has_benefit) - 20 * int(generic_opening) |
| hook_score = max(0, min(100, hook_score)) |
| cta_score = max(0, min(100, 35 + min(cta_count, 3) * 20)) |
| value_score = max(0, min(100, 45 + min(benefit_count, 4) * 12 + min(trust_count, 2) * 8)) |
| clarity_score = 78 |
| if word_count < 25: |
| clarity_score -= 15 |
| if word_count > 2500: |
| clarity_score -= 12 |
| if avg_sentence_len > 28: |
| clarity_score -= 15 |
| if transcript.count("!") > 6: |
| clarity_score -= 6 |
| clarity_score = max(0, min(100, clarity_score)) |
| policy_claim_risk = "alto" if exaggerated_count >= 2 else ("medio" if exaggerated_count == 1 else "bajo") |
| script_quality_score = int(round(0.30 * hook_score + 0.25 * clarity_score + 0.25 * value_score + 0.20 * cta_score)) |
|
|
| return { |
| "transcript_word_count": word_count, |
| "hook_first_30_words": hook, |
| "hook_has_question": int(hook_has_question), |
| "hook_has_benefit": int(hook_has_benefit), |
| "generic_opening": int(generic_opening), |
| "cta_count": cta_count, |
| "benefit_count": benefit_count, |
| "urgency_count": urgency_count, |
| "trust_count": trust_count, |
| "exaggerated_claim_count": exaggerated_count, |
| "avg_sentence_length": round(avg_sentence_len, 2), |
| "hook_score": hook_score, |
| "cta_score": cta_score, |
| "value_proposition_score": value_score, |
| "clarity_score": clarity_score, |
| "policy_claim_risk": policy_claim_risk, |
| "script_quality_score": script_quality_score, |
| } |
|
|
|
|
| def build_from_local_transcripts(raw_dir: Path = RAW_TRANSCRIPTS_DIR) -> pd.DataFrame: |
| rows: List[Dict[str, Any]] = [] |
| for path in sorted(raw_dir.glob("*.json")): |
| transcript = _read_transcript_json(path) |
| if not transcript: |
| continue |
| feats = _script_quality_features(transcript) |
| rows.append({ |
| "video_id": path.stem, |
| "source_dataset": "kaggle_youtube_trending_videos_transcripts_900", |
| "title": "", |
| "views": None, |
| "transcript": transcript, |
| **feats, |
| }) |
| return pd.DataFrame(rows) |
|
|
|
|
| def build_patterns_reference(df: pd.DataFrame) -> Dict[str, Any]: |
| if df.empty: |
| return default_patterns_reference() |
| top = df.sort_values("script_quality_score", ascending=False).head(min(150, len(df))) |
| hook_terms = Counter() |
| cta_terms = Counter() |
| benefit_terms = Counter() |
| for _, row in top.iterrows(): |
| hook = str(row.get("hook_first_30_words", "")).lower() |
| transcript = str(row.get("transcript", "")).lower() |
| for term in BENEFIT_TERMS: |
| if term in hook: |
| hook_terms[term] += 1 |
| if term in transcript: |
| benefit_terms[term] += 1 |
| for term in CTA_TERMS: |
| if term in transcript: |
| cta_terms[term] += 1 |
| return { |
| "source": "youtube_trending_videos_transcripts_900", |
| "n_transcripts": int(len(df)), |
| "quality_score_mean": float(round(df["script_quality_score"].mean(), 2)), |
| "quality_score_p75": float(round(df["script_quality_score"].quantile(0.75), 2)), |
| "top_hook_patterns": [x for x, _ in hook_terms.most_common(20)] or sorted(list(BENEFIT_TERMS))[:20], |
| "top_cta_patterns": [x for x, _ in cta_terms.most_common(20)] or sorted(list(CTA_TERMS))[:20], |
| "top_benefit_patterns": [x for x, _ in benefit_terms.most_common(20)] or sorted(list(BENEFIT_TERMS))[:20], |
| "risk_terms": sorted(list(EXAGGERATED_TERMS)), |
| "generic_openings": sorted(list(GENERIC_OPENINGS)), |
| "notes": "Patrones extraídos de transcripciones públicas de videos populares. No garantizan viralidad; se usan como referencia semántica para mejorar guion.", |
| } |
|
|
|
|
| def default_patterns_reference() -> Dict[str, Any]: |
| return { |
| "source": "default_rules", |
| "n_transcripts": 0, |
| "quality_score_mean": 0, |
| "quality_score_p75": 70, |
| "top_hook_patterns": sorted(list(BENEFIT_TERMS))[:20], |
| "top_cta_patterns": sorted(list(CTA_TERMS))[:20], |
| "top_benefit_patterns": sorted(list(BENEFIT_TERMS))[:20], |
| "risk_terms": sorted(list(EXAGGERATED_TERMS)), |
| "generic_openings": sorted(list(GENERIC_OPENINGS)), |
| "notes": "Patrones por defecto; ejecutar script_dataset_builder.py para usar transcripciones reales.", |
| } |
|
|
|
|
| def build_and_save() -> Dict[str, Any]: |
| PROCESSED_DIR.mkdir(parents=True, exist_ok=True) |
| df = build_from_local_transcripts() |
| df.to_csv(FEATURES_CSV, index=False) |
| patterns = build_patterns_reference(df) |
| PATTERNS_JSON.write_text(json.dumps(patterns, ensure_ascii=False, indent=2), encoding="utf-8") |
| return {"rows": int(len(df)), "features_csv": str(FEATURES_CSV), "patterns_json": str(PATTERNS_JSON)} |
|
|
|
|
| if __name__ == "__main__": |
| print(json.dumps(build_and_save(), ensure_ascii=False, indent=2)) |
|
|
