YggTorrent MCP Server

from datetime import datetime from math import floor, log from math import pow as pw from typing import Any from .categories import get_category_id, get_category_name from .models import Torrent def format_size(size_bytes: int | None) -> str: """Converts a size in bytes to a human-readable string.""" if size_bytes is None: return "N/A" if size_bytes == 0: return "0B" size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB") i = int(floor(log(size_bytes, 1024))) p = pw(1024, i) s = round(size_bytes / p, 2) return f"{s} {size_name[i]}" def format_date(date_str: str | None) -> str: """Converts an ISO date string to a human-readable string 'YYYY-MM-DD HH:MM:SS' or 'N/A' if input is None.""" if date_str is None: return "N/A" return datetime.fromisoformat(date_str).strftime("%Y-%m-%d %H:%M:%S") def format_torrent(torrent: dict[str, Any], torrent_id: int | None = None) -> Torrent: """Converts a torrent data dictionary from the API into a Torrent model instance.""" cat_id: int | str = torrent.get("category_id") or "N/A" return Torrent( id=torrent_id or torrent.get("id") or 0, filename=torrent.get("title") or "N/A", category=( get_category_name(cat_id) or str(cat_id) if isinstance(cat_id, int) else str(cat_id) ), size=format_size(torrent.get("size")), seeders=torrent.get("seeders") or 0, leechers=torrent.get("leechers") or 0, downloads=torrent.get("downloads") or 0, date=format_date(torrent.get("uploaded_at")), magnet_link=None, ) def check_categories(categories: list[int] | list[str] | None = None) -> list[int]: """Checks if the categories are valid.""" processed_category_ids: list[int] = [] if categories: if all(isinstance(cat, int) for cat in categories): processed_category_ids = list(set(categories)) # type: ignore elif all(isinstance(cat, str) for cat in categories): temp_ids: list[int] = [] for keyword_val in categories: cat_id = get_category_id(str(keyword_val).lower()) if cat_id is not None: temp_ids.append(cat_id) if temp_ids: processed_category_ids = list(set(temp_ids)) if not processed_category_ids: print( f"Warning: None of the provided category keywords matched: {categories}." " Proceeding without category filter." ) return processed_category_ids