deploy-test/modules/mangadex.module

{
    "name": "MangaDex Reader",
    "version": "1.1.5",
    "author": "Animex",
    "description": "MangaDex Reader - Forced Hosted Chapters Mode (Bypasses External Links).",
    "type": "MANGA_READER",
    "requirements": ["httpx"]
}
---
import asyncio
import httpx
import inspect
import urllib.parse
from typing import Optional, List, Dict, Any

# =========================
# SMART TUNNEL HELPER
# =========================

async def _smart_fetch(method: str, url: str, **kwargs) -> Any:
    func = getattr(httpx, method.lower())
    if inspect.iscoroutinefunction(func):
        return await func(url, **kwargs)
    async with httpx.AsyncClient(follow_redirects=True) as client:
        return await getattr(client, method.lower())(url, **kwargs)

async def _fetch_json(url: str, params: Dict[str, Any] = None, headers: Dict[str, str] = None, timeout: int = 15) -> Dict[str, Any]:
    try:
        resp = await _smart_fetch("GET", url, params=params, headers=headers, timeout=timeout)
        resp.raise_for_status()
        data = resp.json()
        return data
    except Exception as e:
        print(f" [MangaDex Debug] Request failed: {url}")
        raise

# =========================
# INTERNAL LOGIC
# =========================

async def get_title_from_mal(mal_id: int) -> Optional[str]:
    url = f"https://api.jikan.moe/v4/manga/{mal_id}"
    try:
        data = await _fetch_json(url)
        return data.get("data", {}).get("title_english") or data.get("data", {}).get("title")
    except Exception: return None

async def find_mangadex_id(mal_id: int, title: str) -> Optional[str]:
    search_url = (
        f"https://api.mangadex.org/manga"
        f"?title={urllib.parse.quote(title)}&limit=5"
        f"&contentRating[]=safe&contentRating[]=suggestive&contentRating[]=erotica&contentRating[]=pornographic"
    )
    try:
        data = await _fetch_json(search_url)
        for manga in data.get("data", []):
            if manga.get("attributes", {}).get("links", {}).get("mal") == str(mal_id):
                return manga["id"]
        return data["data"][0]["id"] if data.get("data") else None
    except Exception: return None

# =========================
# PUBLIC MODULE API
# =========================

async def get_chapters(mal_id: int) -> Optional[List[Dict[str, Any]]]:
    title = await get_title_from_mal(mal_id)
    if not title: return None
    md_id = await find_mangadex_id(mal_id, title)
    if not md_id: return None

    print(f" [MangaDex Debug] Fetching HOSTED ONLY feed for MD_ID: {md_id}")
    
    # CRITICAL CHANGE: includeExternalChapters=0 forces the API to return 
    # chapters actually hosted on MangaDex servers, ignoring official external redirects.
    feed_url = (
        f"https://api.mangadex.org/manga/{md_id}/feed"
        f"?translatedLanguage[]=en"
        f"&limit=500"
        f"&contentRating[]=safe&contentRating[]=suggestive&contentRating[]=erotica&contentRating[]=pornographic"
        f"&order[chapter]=asc"
        f"&includes[]=scanlation_group"
    )
    
    try:
        data = await _fetch_json(feed_url)
        raw_chapters = data.get("data", [])
        print(f" [MangaDex Debug] Found {len(raw_chapters)} HOSTED chapters.")
        
        formatted = []
        seen_numbers = set()

        for ch in raw_chapters:
            attr = ch.get("attributes", {})
            num = attr.get("chapter")
            
            if num is None or num in seen_numbers:
                continue
            
            # Find scanlation group name
            group_name = "Unknown Group"
            for rel in ch.get("relationships", []):
                if rel["type"] == "scanlation_group":
                    group_name = rel.get("attributes", {}).get("name", "Unknown Group")
                    break

            seen_numbers.add(num)
            formatted.append({
                "title": f"Ch. {num} - {attr.get('title') or group_name}",
                "url": ch["id"], 
                "chapter_number": str(num),
                "is_external": False
            })
            
        def safe_float(v):
            try: return float(v)
            except: return 0.0

        formatted.sort(key=lambda x: safe_float(x['chapter_number']), reverse=True)
        return formatted
    except Exception as e:
        print(f" [MangaDex Debug] Feed Error: {e}")
        return None

async def get_chapter_images(mal_id: int, chapter_num: str) -> Optional[List[str]]:
    print(f"🎬 MangaDex: Retrieving Images for MAL:{mal_id} Chapter:{chapter_num}")
    
    all_chapters = await get_chapters(mal_id)
    if not all_chapters:
        print("❌ MangaDex: No hosted chapters found in feed.")
        return None

    target = str(chapter_num)
    # Match via float to handle "1" vs "1.0"
    chapter_data = None
    try:
        target_f = float(target)
        chapter_data = next((ch for ch in all_chapters if float(ch["chapter_number"]) == target_f), None)
    except:
        chapter_data = next((ch for ch in all_chapters if ch["chapter_number"] == target), None)

    if not chapter_data:
        print(f"❌ MangaDex: Chapter {target} is not available in hosted mode.")
        return None

    chapter_uuid = chapter_data["url"]
    print(f"🔗 MangaDex: Target UUID: {chapter_uuid}")

    try:
        at_home_url = f"https://api.mangadex.org/at-home/server/{chapter_uuid}"
        data = await _fetch_json(at_home_url)
        
        base_url = data.get("baseUrl")
        chapter_hash = data.get("chapter", {}).get("hash")
        filenames = data.get("chapter", {}).get("data", [])

        if not filenames:
            print(f"⚠️ MangaDex: No images found in At-Home response.")
            return []

        print(f"✅ MangaDex: Found {len(filenames)} images.")
        return [f"{base_url}/data/{chapter_hash}/{f}" for f in filenames]
    except Exception as e:
        print(f"❌ MangaDex: At-Home API failed: {e}")
        return None