app.py

#!/usr/bin/env python3
"""
Anna's Archives API - Hugging Face Space Edition
Optimized for HF Free Tier (CPU-only, minimal resources)
"""

import os
import re
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
from typing import Optional, Any, Literal
import html

from flask import Flask, jsonify, request, Response
from curl_cffi import requests
from bs4 import BeautifulSoup
import csv
import io

# ============================================================================
# CONFIGURATION
# ============================================================================

class Config:
    PORT = int(os.getenv("PORT", 7860))
    HOST = "0.0.0.0"
    MIRRORS_URL = "https://shadowlibraries.github.io/DirectDownloads/AnnasArchive/"
    DEFAULT_BASE_URL = "https://annas-archive.gs"
    # Welib est un miroir stable avec l'endpoint /popular — on le cible directement.
    WELIB_BASE_URL = "https://fr.welib.org"
    BROWSER_IMPERSONATE = "chrome110"
    CACHE_TTL_MINUTES = 10
    REQUEST_TIMEOUT = 20

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)

# ============================================================================
# SIMPLE CACHE
# ============================================================================

class SimpleCache:
    def __init__(self, ttl_minutes: int):
        self._cache = {}
        self._ttl = timedelta(minutes=ttl_minutes)
    
    def get(self, key: str) -> Optional[Any]:
        if key in self._cache:
            value, timestamp = self._cache[key]
            if datetime.now() - timestamp < self._ttl:
                return value
            del self._cache[key]
        return None
    
    def set(self, key: str, value: Any):
        if len(self._cache) > 100:
            oldest = min(self._cache.items(), key=lambda x: x[1][1])[0]
            del self._cache[oldest]
        self._cache[key] = (value, datetime.now())
    
    def clear(self):
        self._cache.clear()
    
    def size(self):
        return len(self._cache)

cache = SimpleCache(Config.CACHE_TTL_MINUTES)

# ============================================================================
# DATA MODELS
# ============================================================================

@dataclass
class Book:
    md5: Optional[str]
    title: str
    author: str
    publisher: str
    year: Optional[int]
    format: str
    language: str
    size_mb: float
    url: str
    cover_url: Optional[str] = None
    description: Optional[str] = None

    def to_dict(self):
        return asdict(self)

# ============================================================================
# UTILITIES
# ============================================================================

def clean_url(url: str) -> str:
    if not url:
        return ""
    from urllib.parse import urlparse, urlunparse
    parsed = urlparse(url)
    return urlunparse((parsed.scheme, parsed.netloc, parsed.path.rstrip('/'), '', '', ''))

def clean_text(text: str) -> str:
    if not text:
        return ""
    text = html.unescape(text)
    text = re.sub(r'[👤🏢📘🚀✅❌⭐]', '', text)
    text = re.sub(r'\s+', ' ', text)
    return text.strip()

def parse_size(size_str: str) -> float:
    if not size_str:
        return 0.0
    match = re.search(r'([\d.]+)\s*([KMGT]?B)', size_str, re.I)
    if not match:
        return 0.0
    num = float(match.group(1))
    unit = match.group(2).upper()
    multipliers = {'B': 1/1024/1024, 'KB': 1/1024, 'MB': 1, 'GB': 1024, 'TB': 1024*1024}
    return round(num * multipliers.get(unit, 1), 2)

# ============================================================================
# MIRROR MANAGER
# ============================================================================

class MirrorManager:
    _mirrors_cache: Optional[list] = None
    _current_mirror: Optional[str] = None

    def get_mirrors(self) -> list[dict]:
        if MirrorManager._mirrors_cache is not None:
            return MirrorManager._mirrors_cache
        logger.info("Fetching mirrors...")
        try:
            resp = requests.get(
                Config.MIRRORS_URL,
                impersonate=Config.BROWSER_IMPERSONATE,
                timeout=Config.REQUEST_TIMEOUT
            )
            soup = BeautifulSoup(resp.text, "html.parser")
            article = soup.find("article", class_="book-article")
            if not article:
                return []
            heading = article.find("h3", id="links")
            if not heading:
                return []
            ul = heading.find_next_sibling("ul")
            if not ul:
                return []
            mirrors = []
            for li in ul.find_all("li"):
                a = li.find("a", href=True)
                if a:
                    mirrors.append({
                        "label": a.get_text(strip=True),
                        "url": clean_url(a["href"])
                    })
            logger.info(f"Found {len(mirrors)} mirrors")
            MirrorManager._mirrors_cache = mirrors
            return mirrors
        except Exception as e:
            logger.error(f"Failed to fetch mirrors: {e}")
            return []

    def get_active_mirror(self) -> str:
        if MirrorManager._current_mirror:
            return MirrorManager._current_mirror
        mirrors = self.get_mirrors()
        for mirror in mirrors:
            try:
                logger.info(f"Testing mirror: {mirror['url']}")
                resp = requests.get(
                    mirror['url'],
                    impersonate=Config.BROWSER_IMPERSONATE,
                    timeout=10
                )
                if resp.status_code == 200:
                    MirrorManager._current_mirror = mirror['url']
                    logger.info(f"✅ Active mirror: {MirrorManager._current_mirror}")
                    return MirrorManager._current_mirror
            except Exception as e:
                logger.warning(f"Mirror {mirror['url']} failed: {e}")
                continue
        logger.warning("No active mirror, using default")
        MirrorManager._current_mirror = Config.DEFAULT_BASE_URL
        return MirrorManager._current_mirror

    def reset(self):
        MirrorManager._mirrors_cache = None
        MirrorManager._current_mirror = None

mirror_manager = MirrorManager()

# ============================================================================
# SCRAPER — SEARCH (Anna's Archive)
# ============================================================================

def scrape_search(query: str, page: int = 1, **filters) -> dict:
    cache_key = f"search_{query}_{page}_{sorted(filters.items())}"
    cached = cache.get(cache_key)
    if cached:
        logger.info(f"Cache HIT: {query} (page {page})")
        return cached
    
    logger.info(f"Scraping: {query} (page {page})")
    base_url = mirror_manager.get_active_mirror()
    search_url = f"{base_url}/search"
    params = {"q": query}
    if page > 1:
        params["page"] = page
    for key in ['lang', 'content', 'ext', 'sort']:
        if key in filters and filters[key]:
            params[key] = filters[key]
    
    try:
        resp = requests.get(
            search_url,
            params=params,
            impersonate=Config.BROWSER_IMPERSONATE,
            timeout=Config.REQUEST_TIMEOUT,
            headers={"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"}
        )
        resp.raise_for_status()
        books = parse_books(resp.text, base_url)
        has_more = check_next_page(resp.text)
        result = {
            "books": [b.to_dict() for b in books],
            "total": len(books),
            "has_more": has_more,
            "timestamp": datetime.now().isoformat()
        }
        cache.set(cache_key, result)
        logger.info(f"Found {len(books)} books")
        return result
    except Exception as e:
        logger.error(f"Scraping error: {e}")
        return {"books": [], "total": 0, "has_more": False, "error": str(e)}


# ============================================================================
# SCRAPER — POPULAR (Welib /popular endpoint)
# ============================================================================

# Intervalles valides côté serveur welib
PopularInterval = Literal["24h", "week", "month", "random"]

def scrape_popular(interval: PopularInterval, offset: int = 0, limit: int = 10) -> dict:
    """
    Scrape GET /popular?interval={interval}&offset={offset}&limit={limit} sur fr.welib.org.
    Le HTML retourné est un fragment de liste de livres (pas une page complète).
    interval : "24h" | "week" | "month" | "random"
    """
    cache_key = f"popular_{interval}_{offset}_{limit}"
    # Pour "random" (surprenez-moi), le cache est volontairement court (1 min).
    ttl = 1 if interval == "random" else Config.CACHE_TTL_MINUTES
    cached = cache.get(cache_key)
    if cached and interval != "random":
        logger.info(f"Cache HIT: popular/{interval}")
        return cached

    logger.info(f"Fetching popular books: interval={interval}, offset={offset}, limit={limit}")
    url = f"{Config.WELIB_BASE_URL}/popular"
    params = {"interval": interval, "offset": offset, "limit": limit}

    try:
        resp = requests.get(
            url,
            params=params,
            impersonate=Config.BROWSER_IMPERSONATE,
            timeout=Config.REQUEST_TIMEOUT,
            headers={
                "Accept": "*/*",
                "Referer": f"{Config.WELIB_BASE_URL}/",
                "Accept-Language": "fr,fr-FR;q=0.9,en-US;q=0.8,en;q=0.7",
            }
        )
        resp.raise_for_status()
        books = parse_welib_books(resp.text)
        result = {
            "interval": interval,
            "offset": offset,
            "limit": limit,
            "books": [b.to_dict() for b in books],
            "total": len(books),
            "timestamp": datetime.now().isoformat()
        }
        cache.set(cache_key, result)
        logger.info(f"Got {len(books)} popular books ({interval})")
        return result
    except Exception as e:
        logger.error(f"Popular scraping error: {e}")
        return {"interval": interval, "books": [], "total": 0, "error": str(e)}


def parse_welib_books(html_text: str) -> list[Book]:
    """
    Parse le fragment HTML retourné par /popular sur fr.welib.org.
    Structure des cartes :
      .book-card
        img[data-author][data-title][src]   → cover, author, title
        a[href=/md5/...]                     → md5, url
        h2.font-semibold                     → title (fallback)
        a[href=/search?q=...]               → author
        p.text-gray-600                     → description
        div.mb-1 > span (×4)               → format · langue · année · taille
    """
    soup = BeautifulSoup(html_text, "html.parser")
    books = []
    seen_md5s: set[str] = set()

    for card in soup.find_all("div", class_="book-card"):
        try:
            # — MD5 & URL —
            md5 = None
            url = ""
            anchor = card.find("a", href=re.compile(r'/md5/'))
            if anchor:
                href = anchor.get("href", "")
                md5_match = re.search(r'/md5/([a-f0-9]{32})', href)
                if md5_match:
                    md5 = md5_match.group(1)
                url = f"{Config.WELIB_BASE_URL}{href}" if href.startswith('/') else href

            if md5:
                if md5 in seen_md5s:
                    continue
                seen_md5s.add(md5)

            # — Titre —
            title = ""
            img = card.find("img", attrs={"data-title": True})
            if img:
                title = clean_text(img["data-title"])
            if not title:
                h2 = card.find("h2", class_=lambda x: x and "font-semibold" in x)
                if h2:
                    title = clean_text(h2.get_text())
            if not title:
                continue  # carte invalide

            # — Auteur —
            author = "Unknown"
            if img and img.get("data-author"):
                author = clean_text(img["data-author"])
            else:
                author_link = card.find("a", href=re.compile(r'search\?q='))
                if author_link:
                    author = clean_text(author_link.get_text())

            # — Cover URL —
            cover_url = None
            if img:
                src = img.get("src", "")
                cover_url = src if src else None

            # — Description —
            description = None
            desc_p = card.find("p", class_=lambda x: x and "text-gray-600" in x)
            if desc_p:
                # Exclure le bouton "Lire plus…"
                for btn in desc_p.find_all("button"):
                    btn.decompose()
                description = clean_text(desc_p.get_text()) or None

            # — Métadonnées (format · langue · année · taille) —
            # Dans le HTML welib, ces 4 infos sont dans des <span> inside div.mb-1
            fmt = "UNKNOWN"
            language = "xx"
            year = None
            size_mb = 0.0

            meta_div = card.find("div", class_="mb-1")
            if meta_div:
                spans = [clean_text(s.get_text()) for s in meta_div.find_all("span")]
                # spans typiques : ["PDF", "· français", "· 2017", "· 13.6 MB"]
                # On nettoie les "· " en tête et on parse chaque span
                for span in spans:
                    span = re.sub(r'^[·\s]+', '', span).strip()
                    if not span:
                        continue
                    if re.match(r'^\d{4}$', span):
                        year = int(span)
                    elif re.search(r'[\d.]+\s*[KMGT]?B', span, re.I):
                        size_mb = parse_size(span)
                    elif re.match(r'^[A-Z0-9]{2,6}$', span):
                        fmt = span
                    else:
                        # langue : peut être "français", "english", "deutsch", etc.
                        language = span

            books.append(Book(
                md5=md5,
                title=title,
                author=author,
                publisher="Unknown",   # pas exposé dans ce fragment HTML
                year=year,
                format=fmt,
                language=language,
                size_mb=size_mb,
                url=url,
                cover_url=cover_url,
                description=description,
            ))
        except Exception as e:
            logger.warning(f"Error parsing welib book card: {e}")
            continue

    return books


# ============================================================================
# SCRAPER — RECENT DOWNLOADS (Anna's Archive /dyn/recent_downloads/)
# ============================================================================

def scrape_recent_downloads() -> dict:
    """
    Endpoint /dyn/recent_downloads/ — retourne les 50 derniers téléchargements globaux.
    Requiert Accept: text/css (trick anti-DDoS Guard, même pattern que /dyn/search_counts).
    Réponse : JSON array de {path, title}, cachée 60s côté serveur.
    """
    cache_key = "recent_downloads"
    cached = cache.get(cache_key)
    if cached:
        logger.info("Cache HIT: recent_downloads")
        return cached

    logger.info("Fetching recent downloads...")
    base_url = mirror_manager.get_active_mirror()
    url = f"{base_url}/dyn/recent_downloads/"

    try:
        resp = requests.get(
            url,
            impersonate=Config.BROWSER_IMPERSONATE,
            timeout=Config.REQUEST_TIMEOUT,
            headers={"Accept": "text/css"}  # Requis — bypass DDoS Guard cache
        )
        resp.raise_for_status()
        items = resp.json()
        enriched = []
        for item in items:
            md5_match = re.search(r'/md5/([a-f0-9]{32})', item.get("path", ""))
            enriched.append({
                "md5": md5_match.group(1) if md5_match else None,
                "title": item.get("title", ""),
                "path": item.get("path", ""),
                "url": f"{base_url}{item['path']}" if item.get("path") else None
            })
        result = {
            "items": enriched,
            "total": len(enriched),
            "timestamp": datetime.now().isoformat()
        }
        cache.set(cache_key, result)
        logger.info(f"Got {len(enriched)} recent downloads")
        return result
    except Exception as e:
        logger.error(f"Recent downloads error: {e}")
        return {"items": [], "total": 0, "error": str(e)}


# ============================================================================
# SCRAPER — SEARCH PARSER (Anna's Archive)
# ============================================================================

def parse_books(html_text: str, base_url: str) -> list[Book]:
    soup = BeautifulSoup(html_text, 'html.parser')
    books = []
    seen_md5s = set()

    blocks = soup.find_all(
        'div',
        class_=lambda x: x and 'flex' in x and 'pt-3' in x and 'pb-3' in x
    )
    
    for block in blocks:
        try:
            md5 = None
            md5_div = block.find('div', class_='hidden')
            if md5_div:
                match = re.search(r'md5:([a-f0-9]{32})', md5_div.text)
                if match:
                    md5 = match.group(1)

            title_link = block.find('a', class_=lambda x: x and 'js-vim-focus' in x and 'font-semibold' in x)
            if not title_link:
                continue
            title = clean_text(title_link.text)
            url = title_link.get('href', '')
            if url.startswith('/'):
                url = f"{base_url}{url}"

            if not md5:
                match = re.search(r'/md5/([a-f0-9]{32})', url)
                if match:
                    md5 = match.group(1)
            if md5:
                if md5 in seen_md5s:
                    continue
                seen_md5s.add(md5)

            cover_img = block.find('img')
            cover_url = None
            if cover_img:
                cover_url = cover_img.get('src', '')
                if cover_url and cover_url.startswith('/'):
                    cover_url = f"{base_url}{cover_url}"

            author = "Unknown"
            for link in block.find_all('a', href=re.compile(r'search\?q=')):
                if 'user-edit' in str(link):
                    author = clean_text(link.get_text())
                    break

            publisher = "Unknown"
            year = None
            for link in block.find_all('a', href=re.compile(r'search\?q=')):
                if 'company' in str(link):
                    pub_text = clean_text(link.get_text())
                    year_match_pub = re.search(r'(\d{4})$', pub_text)
                    if year_match_pub:
                        year = int(year_match_pub.group(1))
                        publisher = re.sub(r',\s*(?:\w+\s+\d+,\s*)?\d{4}$', '', pub_text).strip()
                    else:
                        publisher = pub_text
                    break

            info_div = block.find('div', class_=re.compile(r'text-gray-800'))
            info_text = info_div.get_text() if info_div else ""
            format_match = re.search(r'·\s*([A-Z0-9]+)\s*·', info_text)
            lang_match = re.search(r'\[([a-z]{2,4})\]', info_text)
            size_match = re.search(r'([\d.]+\s*[KMGT]?B)', info_text)
            year_match = re.search(r'·\s*(\d{4})\s*·', info_text)

            book = Book(
                md5=md5,
                title=title,
                author=author,
                publisher=publisher,
                year=year or (int(year_match.group(1)) if year_match else None),
                format=format_match.group(1) if format_match else "UNKNOWN",
                language=lang_match.group(1) if lang_match else "xx",
                size_mb=parse_size(size_match.group(1)) if size_match else 0.0,
                url=url,
                cover_url=cover_url
            )
            books.append(book)
        except Exception as e:
            logger.warning(f"Error parsing book: {e}")
            continue
    return books


def check_next_page(html_text: str) -> bool:
    soup = BeautifulSoup(html_text, 'html.parser')
    return soup.find('a', string=re.compile(r'Next|→|»')) is not None

# ============================================================================
# FLASK APP
# ============================================================================

app = Flask(__name__)

@app.route('/')
def index():
    return jsonify({
        "name": "Anna's Archives API",
        "version": "1.2.0",
        "description": "HF Space Edition - Free Tier Optimized",
        "browser": Config.BROWSER_IMPERSONATE,
        "endpoints": {
            "GET /": "Documentation",
            "GET /search": "Search books",
            "GET /recent": "Recent global downloads (live feed)",
            "GET /popular": "Popular books by interval (24h / week / month)",
            "GET /surprise": "Random book selection (surprenez-moi)",
            "GET /health": "Health check",
            "GET /mirrors": "List mirrors",
            "POST /cache/clear": "Clear cache"
        },
        "examples": {
            "search": "/search?q=python",
            "filters": "/search?q=machine+learning&ext=pdf&lang=en",
            "pagination": "/search?q=python&page=2",
            "csv": "/search?q=python&format=csv",
            "recent": "/recent",
            "popular_day": "/popular?interval=24h",
            "popular_week": "/popular?interval=week",
            "popular_month": "/popular?interval=month",
            "popular_paged": "/popular?interval=week&offset=10&limit=10",
            "surprise": "/surprise"
        }
    })


@app.route('/search')
def search():
    query = request.args.get('q', '').strip()
    if not query:
        return jsonify({"error": "Parameter 'q' is required"}), 400
    try:
        page = max(1, int(request.args.get('page', 1)))
    except ValueError:
        return jsonify({"error": "Invalid page number"}), 400
    
    filters = {
        'lang': request.args.get('lang'),
        'ext': request.args.get('ext'),
        'content': request.args.get('content'),
        'sort': request.args.get('sort')
    }
    result = scrape_search(query, page, **filters)
    
    if request.args.get('format') == 'csv':
        output = io.StringIO()
        if result['books']:
            writer = csv.DictWriter(output, fieldnames=result['books'][0].keys())
            writer.writeheader()
            writer.writerows(result['books'])
        return Response(
            output.getvalue(),
            mimetype='text/csv',
            headers={'Content-Disposition': f'attachment; filename=search_{query}.csv'}
        )
    
    return jsonify({
        "query": query,
        "page": page,
        **result,
        "filters": filters
    })


@app.route('/popular')
def popular():
    """
    Livres populaires par période.
    Paramètres :
      interval : "24h" | "week" | "month"  (défaut : "week")
      offset   : int  (défaut : 0)
      limit    : int  (défaut : 10, max : 50)
    Source : fr.welib.org/popular
    """
    interval = request.args.get('interval', 'week').lower()
    valid_intervals = {"24h", "week", "month"}
    if interval not in valid_intervals:
        return jsonify({
            "error": f"Invalid interval '{interval}'. Must be one of: {', '.join(sorted(valid_intervals))}"
        }), 400

    try:
        offset = max(0, int(request.args.get('offset', 0)))
        limit = min(50, max(1, int(request.args.get('limit', 10))))
    except ValueError:
        return jsonify({"error": "Invalid offset or limit"}), 400

    result = scrape_popular(interval, offset, limit)
    return jsonify(result)


@app.route('/surprise')
def surprise():
    """
    Sélection aléatoire de livres — "Surprenez-moi".
    Paramètres :
      limit : int  (défaut : 10, max : 50)
    Source : fr.welib.org/popular?interval=random
    Le résultat N'EST PAS mis en cache (chaque appel retourne une sélection fraîche).
    """
    try:
        limit = min(50, max(1, int(request.args.get('limit', 10))))
    except ValueError:
        return jsonify({"error": "Invalid limit"}), 400

    result = scrape_popular("random", offset=0, limit=limit)
    # On rebaptise l'interval pour l'utilisateur
    result["interval"] = "random"
    result["description"] = "Random book selection — surprenez-moi!"
    return jsonify(result)


@app.route('/recent')
def recent_downloads():
    result = scrape_recent_downloads()
    return jsonify(result)


@app.route('/health')
def health():
    try:
        mirror = mirror_manager.get_active_mirror()
        status = "healthy"
    except Exception:
        mirror = "unavailable"
        status = "degraded"
    return jsonify({
        "status": status,
        "mirror": mirror,
        "welib": Config.WELIB_BASE_URL,
        "cache_size": cache.size(),
        "browser": Config.BROWSER_IMPERSONATE
    })


@app.route('/mirrors')
def mirrors():
    return jsonify({
        "mirrors": mirror_manager.get_mirrors(),
        "current": mirror_manager.get_active_mirror()
    })


@app.route('/cache/clear', methods=['POST'])
def clear_cache():
    cache.clear()
    mirror_manager.reset()
    return jsonify({"message": "Cache cleared", "size": 0})


if __name__ == "__main__":
    logger.info("=" * 70)
    logger.info("🚀 Anna's Archives API - HF Space Edition v1.2.0")
    logger.info("=" * 70)
    logger.info(f"Port: {Config.PORT}")
    logger.info(f"Browser: {Config.BROWSER_IMPERSONATE}")
    logger.info(f"Popular source: {Config.WELIB_BASE_URL}")
    logger.info("=" * 70)
    mirror_manager.get_active_mirror()
    app.run(host=Config.HOST, port=Config.PORT)