anicli/core/parser.py - master - dig/anicli_ru

anicli/core/parser.py
212 строк7 KB
An0nX
refactor: modernize type hints, improve aniskip merging and history fallback
09 янв 2026, 00:43
09 янв 2026, 00:43925b7f8
import contextlib
import json
from dataclasses import dataclass
from urllib.parse import quote

import jmespath
from selectolax.parser import HTMLParser

from anicli.config import SourceConfig
from anicli.core.network import NetworkClient


@dataclass
class SearchResult:
    title: str
    url: str
    source_name: str


@dataclass
class Episode:
    num: str  # Строка, т.к. бывают "OVA", "Фильм" или "12.5"
    url: str


class SourceEngine:
    """
    Experimental engine for parsing sources based on declarative configuration.
    Currently used for prototyping new parsers.
    """

    def __init__(self, network: NetworkClient) -> None:
        """
        Initialize the SourceEngine.

        Args:
            network: NetworkClient instance.
        """
        self.network = network

    async def search(self, query: str, cfg: SourceConfig) -> list[SearchResult]:
        """
        Performs a search using the provided configuration.

        Args:
            query: Search query string.
            cfg: Source configuration object.

        Returns:
            List[SearchResult]: List of found items.
        """
        url = cfg.search_url.format(query=quote(query))
        try:
            raw_data = await self.network.get(url, headers=cfg.headers)
        except Exception:
            return []

        if cfg.type == "json":
            return self._search_json(raw_data, cfg)
        return self._search_html(raw_data, cfg)

    async def get_episodes(self, anime_url: str, cfg: SourceConfig) -> list[Episode]:
        """
        Получает список эпизодов с детальной страницы.

        Args:
            anime_url: URL аниме.
            cfg: Конфигурация источника.

        Returns:
            List[Episode]: Список эпизодов.
        """
        # Если указан отдельный URL для деталей (например, API)
        # Нам нужно извлечь ID или Slug из anime_url
        target_url = anime_url
        if cfg.detail_url:
            # Простая эвристика: берем последнюю часть URL как ID
            # Реализация зависит от конкретного API, здесь упрощенно для примера
            slug = anime_url.rstrip("/").split("/")[-1]
            target_url = cfg.detail_url.format(id=slug, slug=slug)

        try:
            raw_data = await self.network.get(target_url, headers=cfg.headers)
        except Exception:
            return []

        if cfg.type == "json":
            return self._episodes_json(raw_data, cfg)
        return self._episodes_html(raw_data, cfg)

    # --- HTML Implementation ---
    def _search_html(self, html: str, cfg: SourceConfig) -> list[SearchResult]:
        tree = HTMLParser(html)
        results = []
        rules = cfg.rules

        for item in tree.css(rules.search_path):
            title_node = item.css_first(rules.search_title)
            link_node = item.css_first(rules.search_link)

            if title_node and link_node:
                link = link_node.attributes.get("href", "")
                if link and not link.startswith("http"):
                    link = cfg.base_url + link

                results.append(
                    SearchResult(
                        title=title_node.text(strip=True),
                        url=link,
                        source_name=cfg.name,
                    )
                )
        return results

    def _episodes_html(self, html: str, cfg: SourceConfig) -> list[Episode]:
        tree = HTMLParser(html)
        episodes = []
        rules = cfg.rules

        # Иногда эпизоды в HTML - это JSON внутри <script> (AnimeGo case)
        # Проверяем, если rules.episode_context_path указывает на скрипт
        if "script" in rules.episode_context_path:
            # Специфично для AnimeGo: ищем JSON внутри скрипта
            scripts = tree.css(rules.episode_context_path)
            for script in scripts:
                if not script.text():
                    continue
                # Ищем json массив [..., {id:..., episode:..., }]
                # Это упрощенный regex, в реальности может потребоваться сложнее
                # Для AnimeGo ищем players = {...}
                pass

        # Стандартный CSS парсинг
        items = tree.css(rules.episode_context_path)
        for item in items:
            num_node = item.css_first(rules.episode_num)
            # Ссылка может быть атрибутом data-src или href
            link_node = item.css_first(rules.episode_link)

            if num_node:
                # Link node может быть null,
                # если ссылка в data-аттрибуте самого item
                num = num_node.text(strip=True)

                # Попытка найти ссылку
                url = ""
                if link_node:
                    url = link_node.attributes.get(
                        "href"
                    ) or link_node.attributes.get("data-src")
                else:
                    url = item.attributes.get("href") or item.attributes.get("data-src")

                if url:
                    if not str(url).startswith("http") and not str(url).startswith(
                        "//"
                    ):
                        url = cfg.base_url + str(url)
                    if str(url).startswith("//"):
                        url = "https:" + str(url)

                    episodes.append(Episode(num=num, url=str(url)))

        return episodes

    # --- JSON Implementation ---
    def _search_json(self, json_str: str, cfg: SourceConfig) -> list[SearchResult]:
        try:
            data = json.loads(json_str)
        except json.JSONDecodeError:
            return []

        results = []
        items = jmespath.search(cfg.rules.search_path, data) or []

        for item in items:
            title = jmespath.search(cfg.rules.search_title, item)
            link_val = jmespath.search(cfg.rules.search_link, item)

            if title and link_val:
                url = (
                    f"{cfg.base_url}{cfg.rules.link_prefix}"
                    f"{link_val}{cfg.rules.link_postfix}"
                )
                results.append(
                    SearchResult(title=str(title), url=url, source_name=cfg.name)
                )
        return results

    def _episodes_json(self, json_str: str, cfg: SourceConfig) -> list[Episode]:
        try:
            data = json.loads(json_str)
        except json.JSONDecodeError:
            return []

        episodes = []
        items = jmespath.search(cfg.rules.episode_context_path, data) or []

        for item in items:
            num = jmespath.search(cfg.rules.episode_num, item)
            url = jmespath.search(cfg.rules.episode_link, item)

            if num and url:
                if str(url).startswith("//"):
                    url = "https:" + str(url)
                episodes.append(Episode(num=str(num), url=str(url)))

        # Сортируем по номеру, если это числа
        with contextlib.suppress(ValueError):
            episodes.sort(key=lambda x: float(x.num))

        return episodes