reference.py

# reference.py — محرك القارئ المرجعي (منقول من Node lib/reference.ts).
#
# الفكرة: مدة كل آية عند قارئ مرجعي تُجلب من CDN موقع alquran.cloud بطلب HEAD
# (الملفات 128kbps ثابتة، فالمدة = الحجم ÷ 16000 بايت/ثانية). النِّسَب بين الآيات
# هي المهمة فقط — كل حدّ يُثبَّت لاحقًا على أقرب وقفة حقيقية في تسجيل المستخدم.
#
# قائمة القرّاء: تُجلب ديناميكيًا من alquran.cloud (كل تلاوات «آية-آية» المتاحة)
# مع كاش أسبوعي على القرص، وقائمة مُنسّقة من 12 قارئًا مشهورًا كاحتياطي دائم.
import json
import os
import shutil
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import requests

BYTES_PER_SECOND = 16000  # 128 kbps CBR
PROBE_TIMEOUT_SEC = 10
PROBE_CONCURRENCY = 8
PROBE_BUDGET_SEC = 12
MIN_KNOWN_RATIO = 0.7
RECITERS_TTL_SEC = 7 * 24 * 3600

REF_CACHE_DIR = Path(os.environ.get("REF_CACHE_DIR", "data/ref-cache"))
_SEED_DIR = Path(__file__).parent / "ref-cache-seed"

DEFAULT_EDITION = "ar.alafasy"

# قائمة مضمونة (تلاوات آية-آية صالحة على alquran.cloud) — احتياطي دائم.
CURATED_RECITERS = [
    {"id": "ar.alafasy", "name": "مشاري راشد العفاسي"},
    {"id": "ar.husary", "name": "محمود خليل الحصري"},
    {"id": "ar.husarymujawwad", "name": "الحصري (المجوّد)"},
    {"id": "ar.abdulsamad", "name": "عبد الباسط عبد الصمد"},
    {"id": "ar.abdurrahmaansudais", "name": "عبد الرحمن السديس"},
    {"id": "ar.shaatree", "name": "أبو بكر الشاطري"},
    {"id": "ar.ahmedajamy", "name": "أحمد بن علي العجمي"},
    {"id": "ar.mahermuaiqly", "name": "ماهر المعيقلي"},
    {"id": "ar.saoodshuraym", "name": "سعود الشريم"},
    {"id": "ar.hudhaify", "name": "علي الحذيفي"},
    {"id": "ar.hanirifai", "name": "هاني الرفاعي"},
    {"id": "ar.muhammadayyoub", "name": "محمد أيوب"},
]

_lock = threading.Lock()
_reciters_cache: list[dict] | None = None
_reciters_loaded_at = 0.0


def init_cache_dir() -> None:
    """يجهّز مجلد الكاش ويزرعه من النسخة المشحونة مع الصورة (مرة واحدة)."""
    REF_CACHE_DIR.mkdir(parents=True, exist_ok=True)
    if _SEED_DIR.is_dir():
        for f in _SEED_DIR.glob("*.json"):
            dst = REF_CACHE_DIR / f.name
            if not dst.exists():
                shutil.copyfile(f, dst)


def _reciters_disk_path() -> Path:
    return REF_CACHE_DIR / "_reciters.json"


def _fetch_reciters_from_api() -> list[dict] | None:
    """كل تلاوات «آية-آية» الصوتية المتاحة على alquran.cloud (كل القرّاء الممكنين)."""
    try:
        r = requests.get(
            "https://api.alquran.cloud/v1/edition/format/audio", timeout=10
        )
        if r.status_code != 200:
            return None
        editions = r.json().get("data") or []
        out = []
        for e in editions:
            if e.get("format") == "audio" and e.get("type") == "versebyverse":
                out.append({
                    "id": e["identifier"],
                    # الاسم العربي إن وُجد، وإلا الاسم الإنجليزي
                    "name": e.get("name") or e.get("englishName") or e["identifier"],
                    "language": e.get("language", ""),
                })
        return out or None
    except Exception:
        return None


def get_reciters(force_refresh: bool = False) -> list[dict]:
    """قائمة القرّاء: من الذاكرة → القرص (أسبوع) → alquran.cloud → المضمونة."""
    global _reciters_cache, _reciters_loaded_at
    with _lock:
        now = time.time()
        if (not force_refresh and _reciters_cache
                and now - _reciters_loaded_at < RECITERS_TTL_SEC):
            return _reciters_cache

        disk = _reciters_disk_path()
        if not force_refresh and disk.exists():
            try:
                payload = json.loads(disk.read_text())
                if now - payload.get("at", 0) < RECITERS_TTL_SEC:
                    _reciters_cache = payload["reciters"]
                    _reciters_loaded_at = now
                    return _reciters_cache
            except Exception:
                pass

        fetched = _fetch_reciters_from_api()
        if fetched:
            # اضمن وجود القائمة المضمونة داخل النتيجة (لو الـ API شال حاجة)
            ids = {r["id"] for r in fetched}
            for c in CURATED_RECITERS:
                if c["id"] not in ids:
                    fetched.append(dict(c))
            _reciters_cache = fetched
            try:
                disk.write_text(json.dumps(
                    {"at": now, "reciters": fetched}, ensure_ascii=False))
            except Exception:
                pass
        else:
            _reciters_cache = [dict(c) for c in CURATED_RECITERS]
        _reciters_loaded_at = now
        return _reciters_cache


def is_valid_reciter(edition: str) -> bool:
    return any(r["id"] == edition for r in get_reciters())


# ----------------------- مدد الآيات المرجعية -----------------------

def _cdn_url(edition: str, ayah_number: int) -> str:
    return f"https://cdn.islamic.network/quran/audio/128/{edition}/{ayah_number}.mp3"


def _probe_duration(url: str) -> float:
    """مدة تقديرية من Content-Length، أو NaN عند أي فشل (بدون رمي أخطاء)."""
    try:
        r = requests.head(url, timeout=PROBE_TIMEOUT_SEC, allow_redirects=True)
        if r.status_code != 200:
            return float("nan")
        length = int(r.headers.get("content-length") or 0)
        d = length / BYTES_PER_SECOND
        return d if d > 0 else float("nan")
    except Exception:
        return float("nan")


def _cache_path(edition: str) -> Path:
    safe = "".join(c for c in edition if c.isalnum() or c in "._-")
    return REF_CACHE_DIR / f"{safe}.json"


def _load_cache(edition: str) -> dict[int, float]:
    try:
        raw = json.loads(_cache_path(edition).read_text())
        return {int(k): float(v) for k, v in raw.items()}
    except Exception:
        return {}


def _save_cache(edition: str, cache: dict[int, float]) -> None:
    try:
        _cache_path(edition).write_text(
            json.dumps({str(k): v for k, v in cache.items()}))
    except Exception:
        pass  # فشل الكتابة يكلّف إعادة فحص لاحقًا فقط


def _is_known(v) -> bool:
    return isinstance(v, float) and v == v and v > 0  # ليس NaN


def _fill_missing_from_cdn(edition: str, numbers: list[int],
                           cache: dict[int, float]) -> None:
    """يفحص الأرقام الناقصة على الـ CDN. عيّنة صغيرة أولًا: لو فشلت كلها،
    الـ CDN غير قابل للوصول هنا فنتوقف فورًا بدل مهلة لكل آية."""
    missing = [n for n in numbers if not _is_known(cache.get(n))]
    if not missing:
        return

    sample = missing[:min(4, len(missing))]
    with ThreadPoolExecutor(max_workers=len(sample)) as ex:
        sampled = list(ex.map(
            lambda n: (n, _probe_duration(_cdn_url(edition, n))), sample))
    reachable = False
    for n, d in sampled:
        if _is_known(d):
            cache[n] = d
            reachable = True
    if not reachable:
        return

    rest = [n for n in missing if not _is_known(cache.get(n))]
    if rest:
        deadline = time.time() + PROBE_BUDGET_SEC

        def probe(n: int):
            if time.time() > deadline:
                return n, float("nan")
            return n, _probe_duration(_cdn_url(edition, n))

        with ThreadPoolExecutor(max_workers=PROBE_CONCURRENCY) as ex:
            for n, d in ex.map(probe, rest):
                if _is_known(d):
                    cache[n] = d
    _save_cache(edition, cache)


def _known_ratio(numbers: list[int], cache: dict[int, float]) -> float:
    if not numbers:
        return 1.0
    return sum(1 for n in numbers if _is_known(cache.get(n))) / len(numbers)


def _fill_with_mean(numbers: list[int], cache: dict[int, float]) -> list[float]:
    known = [cache[n] for n in numbers if _is_known(cache.get(n))]
    mean = (sum(known) / len(known)) if known else 1.0
    return [cache[n] if _is_known(cache.get(n)) else mean for n in numbers]


def get_reference_durations(edition: str, start_num: int, end_num: int) -> list[float]:
    """مدد الآيات المرجعية للأرقام العالمية [start..end] (1-based).

    تسلسل الحلّ — لا يفشل أبدًا لأن النِّسَب فقط هي المهمة:
      1. كاش القارئ المطلوب + فحص حي لما ينقص.
      2. كاش القارئ الافتراضي (النِّسَب متقاربة بين القرّاء).
      3. نِسَب متساوية (يتحوّل التقسيم لتثبيت حدود متساوية على وقفات المستخدم).
    """
    if end_num < start_num:
        raise ValueError("نطاق غير صحيح")
    numbers = list(range(start_num, end_num + 1))

    cache = _load_cache(edition)
    _fill_missing_from_cdn(edition, numbers, cache)
    if _known_ratio(numbers, cache) >= MIN_KNOWN_RATIO:
        return _fill_with_mean(numbers, cache)

    if edition != DEFAULT_EDITION:
        fallback = _load_cache(DEFAULT_EDITION)
        _fill_missing_from_cdn(DEFAULT_EDITION, numbers, fallback)
        if _known_ratio(numbers, fallback) >= MIN_KNOWN_RATIO:
            return _fill_with_mean(numbers, fallback)

    return [1.0] * len(numbers)