""" scrape_content.py Segunda pasada por el curso: extrae titulos reales, notas, URLs de audio (MP3) y documentos (PDF/ZIP) por leccion. Guarda resultado en course_content.json """ import asyncio import json import re import sys from pathlib import Path from playwright.async_api import async_playwright from bs4 import BeautifulSoup sys.stdout.reconfigure(encoding="utf-8") SCRIPTS_DIR = Path(__file__).parent COOKIES_FILE = SCRIPTS_DIR / ".course_cookies.json" STRUCTURE_FILE = SCRIPTS_DIR / "course_structure.json" CONTENT_FILE = SCRIPTS_DIR / "course_content.json" MEMBERS_URL = "https://kaleanders.clickfunnels.com/members-raio" BAD_TITLE_EXACT = {"working...", "working", "loading...", "loading", "importante"} BAD_TITLE_CONTAINS = {"hemos actualizado", "actualizado el curso", "tip:", "consejo:", "nota:"} DATE_PATTERN = re.compile( r'\b(20\d\d|enero|febrero|marzo|abril|mayo|junio|julio|' r'agosto|septiembre|octubre|noviembre|diciembre)\b', re.I ) def log(msg, level="INFO"): icons = {"INFO": " ", "OK": "[OK]", "SKIP": "[--]", "ERROR": "[!!]", "HEAD": "\n==="} print(f"{icons.get(level,' ')} {msg}", flush=True) def is_bad_title(text: str) -> bool: low = text.lower() if low in BAD_TITLE_EXACT: return True if any(b in low for b in BAD_TITLE_CONTAINS): return True if DATE_PATTERN.search(low): return True return False def extract_lesson_data(html: str, lesson_num: int) -> dict: soup = BeautifulSoup(html, "html.parser") # Remove noise for tag in soup(["script", "style", "noscript"]): tag.decompose() # --- Title --- title = None for tag in soup.find_all(["h1", "h2", "h3"]): text = tag.get_text(strip=True) if text and not is_bad_title(text) and 4 < len(text) < 200: title = text break # --- Audio MP3 links --- audio_urls = [] seen_audio: set[str] = set() for link in soup.find_all("a", href=True): href = str(link["href"]) if ".mp3" in href.lower() and href not in seen_audio: seen_audio.add(href) audio_urls.append({ "url": href, "label": link.get_text(strip=True) or "Audio MP3", }) for audio_tag in soup.find_all("audio"): for attr in ["src"]: src = str(audio_tag.get(attr, "")) if src and src not in seen_audio: seen_audio.add(src) audio_urls.append({"url": src, "label": "Audio"}) for source in audio_tag.find_all("source"): src = str(source.get("src", "")) if src and src not in seen_audio: seen_audio.add(src) audio_urls.append({"url": src, "label": "Audio"}) # --- Document links --- doc_urls = [] seen_docs: set[str] = set() doc_exts = (".pdf", ".zip", ".docx", ".pptx", ".xlsx") for link in soup.find_all("a", href=True): href = str(link["href"]) if any(href.lower().endswith(ext) for ext in doc_exts) and href not in seen_docs: seen_docs.add(href) doc_urls.append({ "url": href, "label": link.get_text(strip=True) or "Documento", }) # --- Notes (paragraphs, excluding very short or noisy ones) --- notes_lines = [] for p in soup.find_all("p"): text = p.get_text(strip=True) if text and len(text) > 20 and "working" not in text.lower(): notes_lines.append(text) notes = "\n\n".join(notes_lines[:25]) if notes_lines else "" return { "title": title or f"Leccion {lesson_num}", "audio_urls": audio_urls, "doc_urls": doc_urls, "notes": notes, } async def click_next(page) -> bool: for sel in ["a[href='#next-lesson']", "#next-lesson", "a:has-text('Siguiente')", "a:has-text('Next')"]: try: btn = await page.query_selector(sel) if btn and await btn.is_visible(): await btn.click() return True except Exception: continue return False async def wait_for_real_title(page, max_seconds: int = 12) -> str: """Espera hasta que aparezca un titulo que no sea 'Working...'""" for _ in range(max_seconds): html = await page.content() soup = BeautifulSoup(html[:30000], "html.parser") for tag in soup.find_all(["h1", "h2", "h3"]): text = tag.get_text(strip=True) if text and not is_bad_title(text) and len(text) > 4: return html await page.wait_for_timeout(1000) return await page.content() async def main(): log("RAIO Content Scraper", "HEAD") lessons = json.loads(STRUCTURE_FILE.read_text(encoding="utf-8")) cookies = json.loads(COOKIES_FILE.read_text(encoding="utf-8")) async with async_playwright() as pw: browser = await pw.chromium.launch(headless=True) ctx = await browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124 Safari/537.36", viewport={"width": 1280, "height": 800}, ) await ctx.add_cookies(cookies) page = await ctx.new_page() # Capture audio / doc network responses window_net: list[str] = [] async def on_response(response): url = response.url low = url.lower() if ".mp3" in low or ".pdf" in low or ".zip" in low: window_net.append(url) page.on("response", on_response) log("Cargando pagina del curso...") await page.goto(MEMBERS_URL, wait_until="domcontentloaded", timeout=30000) await page.wait_for_timeout(8000) content_data = [] for i, lesson in enumerate(lessons): lesson_num = lesson["num"] html = await wait_for_real_title(page, max_seconds=12) extracted = extract_lesson_data(html, lesson_num) # Enrich with network-captured resources seen_net_audio = {a["url"] for a in extracted["audio_urls"]} seen_net_docs = {d["url"] for d in extracted["doc_urls"]} for url in window_net: low = url.lower() if ".mp3" in low and url not in seen_net_audio: seen_net_audio.add(url) extracted["audio_urls"].append({"url": url, "label": "Audio MP3"}) elif ".pdf" in low and url not in seen_net_docs: seen_net_docs.add(url) extracted["doc_urls"].append({"url": url, "label": "Documento PDF"}) log(f"[{lesson_num:2d}] {extracted['title'][:55]}") if extracted["audio_urls"]: log(f" {len(extracted['audio_urls'])} audio(s): {extracted['audio_urls'][0]['url'][:70]}", "OK") if extracted["doc_urls"]: log(f" {len(extracted['doc_urls'])} doc(s): {extracted['doc_urls'][0]['url'][:70]}", "OK") if not extracted["audio_urls"] and not extracted["doc_urls"]: log(f" Solo video", "SKIP") content_data.append({ "num": lesson_num, "title": extracted["title"], "vidalytics_id": lesson.get("vidalytics_id"), "stream_url": lesson.get("stream_url"), "audio_urls": extracted["audio_urls"], "doc_urls": extracted["doc_urls"], "notes": extracted["notes"], }) if lesson_num % 10 == 0: CONTENT_FILE.write_text( json.dumps(content_data, ensure_ascii=False, indent=2), encoding="utf-8" ) log(f"Progreso guardado ({lesson_num} lecciones)") window_net.clear() if i < len(lessons) - 1: clicked = await click_next(page) if not clicked: log("Sin boton siguiente. Fin.", "OK") break await page.wait_for_timeout(6000) page.remove_listener("response", on_response) await browser.close() CONTENT_FILE.write_text( json.dumps(content_data, ensure_ascii=False, indent=2), encoding="utf-8" ) has_audio = sum(1 for l in content_data if l["audio_urls"]) has_docs = sum(1 for l in content_data if l["doc_urls"]) log(f"Total: {len(content_data)} | Con audio: {has_audio} | Con docs: {has_docs}", "OK") if __name__ == "__main__": asyncio.run(main())