Add Surah/Ayah browse with Gharaati Tafsir Noor crawl.
build-and-deploy / deploy (push) Failing after 11m0s
build-and-deploy / deploy (push) Failing after 11m0s
Replace episode bookmarks with surah/ayah bookmarks and ayah favorites; import ayah text, notes, messages, and official per-ayah audio from gharaati.ir.
This commit is contained in:
Executable
+161
@@ -0,0 +1,161 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user