TafsirQuran/scripts/crawl-gharaati.py

#!/usr/bin/env python3
"""Crawl Tafsir Noor from gharaati.ir into AUDIO_DIR/gharaati + content JSON."""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
from html import unescape
from pathlib import Path

BASE = "https://gharaati.ir"
UA = "TafsirQuranCrawler/1.0 (+local; educational mirror)"


def fetch(url: str, retries: int = 5) -> bytes:
    last: Exception | None = None
    for i in range(retries):
        try:
            req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "*/*"})
            with urllib.request.urlopen(req, timeout=60) as resp:
                return resp.read()
        except Exception as ex:  # noqa: BLE001
            last = ex
            time.sleep(1.5 * (i + 1))
    raise RuntimeError(f"fetch failed {url}: {last}")


def strip_tags(html: str) -> str:
    text = re.sub(r"<br\s*/?>", "\n", html, flags=re.I)
    text = re.sub(r"</p\s*>", "\n", text, flags=re.I)
    text = re.sub(r"<[^>]+>", "", text)
    return unescape(re.sub(r"\s+\n", "\n", text)).strip()


def section_after_title(html: str, title: str) -> str | None:
    # Match heading text then first content cell with class tafsir or tafsir_ayeh
    pattern = (
        rf'<span class="post-title h-text">\s*{re.escape(title)}\s*</span>.*?'
        rf'<td[^>]*class=[\'"]?(?:tafsir_ayeh|tafsir)[\'"]?[^>]*>(.*?)</td>'
    )
    m = re.search(pattern, html, flags=re.I | re.S)
    if not m:
        # alternate spellings
        return None
    return m.group(1).strip()


def parse_show3(html: str) -> dict:
    # Normalize common title variants
    ayah_html = section_after_title(html, "آيه") or section_after_title(html, "آیه")
    tarjome = section_after_title(html, "ترجمه")
    nokteh = section_after_title(html, "نکته ها") or section_after_title(html, "نكته ها")
    payam = (
        section_after_title(html, "پيام ها")
        or section_after_title(html, "پیام ها")
        or section_after_title(html, "پيامها")
    )
    video = None
    vm = re.search(r'src=["\'](https?://[^"\']*aparat\.com[^"\']*)["\']', html, flags=re.I)
    if vm:
        video = vm.group(1)

    return {
        "textAr": strip_tags(ayah_html) if ayah_html else "",
        "textFa": strip_tags(tarjome) if tarjome else "",
        "notesHtml": nokteh or "",
        "messagesHtml": payam or "",
        "videoEmbedUrl": video,
    }


def ayah_list(surah: int) -> list[int]:
    raw = fetch(f"{BASE}/mydata.php?op=getayehlist&id={surah}").decode("utf-8", "replace")
    nums = [int(x) for x in re.findall(r"value=['\"]?(\d+)", raw)]
    return sorted(set(nums)) if nums else []


def download_mp3(surah: int, ayah: int, dest: Path) -> bool:
    if dest.exists() and dest.stat().st_size > 1000:
        return False
    dest.parent.mkdir(parents=True, exist_ok=True)
    url = f"{BASE}/archive/tafsir/{surah}/{ayah}.mp3"
    try:
        data = fetch(url, retries=3)
        if len(data) < 500:
            return False
        tmp = dest.with_suffix(".part")
        tmp.write_bytes(data)
        tmp.replace(dest)
        return True
    except Exception as ex:  # noqa: BLE001
        print(f"  mp3 miss {surah}/{ayah}: {ex}", file=sys.stderr)
        return False


def crawl_one(surah: int, ayah: int, audio_dir: Path, delay: float) -> dict:
    html = fetch(f"{BASE}/show3.php?page=tafsir3&numsooreh={surah}&numayeh={ayah}").decode(
        "utf-8", "replace"
    )
    parsed = parse_show3(html)
    rel = f"gharaati/{surah}/{ayah}.mp3"
    mp3_path = audio_dir / rel
    download_mp3(surah, ayah, mp3_path)
    if mp3_path.exists():
        parsed["audioPath"] = rel
    else:
        parsed["audioPath"] = None
    parsed["surah"] = surah
    parsed["ayah"] = ayah
    content_path = audio_dir / "gharaati" / "content" / str(surah) / f"{ayah}.json"
    content_path.parent.mkdir(parents=True, exist_ok=True)
    content_path.write_text(json.dumps(parsed, ensure_ascii=False, indent=2), encoding="utf-8")
    time.sleep(delay)
    return parsed


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("audio_dir", nargs="?", default=os.environ.get("AUDIO_DIR", "/data/audio"))
    ap.add_argument("--surah-from", type=int, default=1)
    ap.add_argument("--surah-to", type=int, default=114)
    ap.add_argument("--delay", type=float, default=0.35)
    ap.add_argument("--skip-existing-json", action="store_true")
    args = ap.parse_args()
    audio_dir = Path(args.audio_dir)
    audio_dir.mkdir(parents=True, exist_ok=True)

    total = 0
    for s in range(args.surah_from, args.surah_to + 1):
        ayahs = ayah_list(s)
        if not ayahs:
            print(f"surah {s}: no ayah list", file=sys.stderr)
            continue
        print(f"surah {s}: {len(ayahs)} ayahs")
        for a in ayahs:
            jpath = audio_dir / "gharaati" / "content" / str(s) / f"{a}.json"
            if args.skip_existing_json and jpath.exists():
                # still try mp3 if missing
                mp3 = audio_dir / "gharaati" / str(s) / f"{a}.mp3"
                if not mp3.exists():
                    download_mp3(s, a, mp3)
                continue
            try:
                crawl_one(s, a, audio_dir, args.delay)
                total += 1
                if total % 25 == 0:
                    print(f"  crawled {total} ayahs (at {s}:{a})")
            except Exception as ex:  # noqa: BLE001
                print(f"FAIL {s}:{a} {ex}", file=sys.stderr)
                time.sleep(2)
    print(f"done. crawled={total}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())