# Purpose: One Space that offers six tools/tabs (all exposed as MCP tools):
#   1) Fetch — convert webpages to clean Markdown format
#   2) DuckDuckGo Search — compact JSONL search output (short keys to minimize tokens)
#   3) Python Code Executor — run Python code and capture stdout/errors
#   4) Kokoro TTS — synthesize speech from text using Kokoro-82M with 54 voice options
#   5) Image Generation - HF serverless inference providers (Default: FLUX.1-Krea-dev)
#   6) Video Generation - HF serverless inference providers (Default: Wan2.2-T2V-A14B)

from __future__ import annotations

import re
import json
import sys
import os
import random
from io import StringIO
from typing import List, Dict, Tuple, Annotated

import gradio as gr
import requests
from bs4 import BeautifulSoup
from markdownify import markdownify as md
from readability import Document
from urllib.parse import urljoin, urldefrag, urlparse
from ddgs import DDGS
from PIL import Image
from huggingface_hub import InferenceClient
import time
import tempfile

# Optional imports for Kokoro TTS (loaded lazily)
import numpy as np
try:
    import torch  # type: ignore
except Exception:  # pragma: no cover - optional dependency
    torch = None  # type: ignore
try:
    from kokoro import KModel, KPipeline  # type: ignore
except Exception:  # pragma: no cover - optional dependency
    KModel = None  # type: ignore
    KPipeline = None  # type: ignore


# ==============================
# Fetch: Enhanced HTTP + extraction utils
# ==============================

def _http_get_enhanced(url: str) -> requests.Response:
    """
    Download the page with enhanced headers, timeout handling, and better error recovery.
    """
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
        "Accept-Language": "en-US,en;q=0.9",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
        "Accept-Encoding": "gzip, deflate, br",
        "DNT": "1",
        "Connection": "keep-alive",
        "Upgrade-Insecure-Requests": "1",
    }
    
    # Apply rate limiting
    _fetch_rate_limiter.acquire()
    
    try:
        response = requests.get(
            url, 
            headers=headers, 
            timeout=30,  # Increased timeout
            allow_redirects=True,
            stream=False
        )
        response.raise_for_status()
        return response
    except requests.exceptions.Timeout:
        raise requests.exceptions.RequestException("Request timed out. The webpage took too long to respond.")
    except requests.exceptions.ConnectionError:
        raise requests.exceptions.RequestException("Connection error. Please check the URL and your internet connection.")
    except requests.exceptions.HTTPError as e:
        if response.status_code == 403:
            raise requests.exceptions.RequestException("Access forbidden. The website may be blocking automated requests.")
        elif response.status_code == 404:
            raise requests.exceptions.RequestException("Page not found. Please check the URL.")
        elif response.status_code == 429:
            raise requests.exceptions.RequestException("Rate limited. Please try again in a few minutes.")
        else:
            raise requests.exceptions.RequestException(f"HTTP error {response.status_code}: {str(e)}")

def _normalize_whitespace(text: str) -> str:
    """
    Squeeze extra spaces and blank lines to keep things compact.
    (Layman's terms: tidy up the text so it’s not full of weird spacing.)
    """
    text = re.sub(r"[ \t\u00A0]+", " ", text)
    text = re.sub(r"\n\s*\n\s*\n+", "\n\n", text.strip())
    return text.strip()


def _truncate(text: str, max_chars: int) -> Tuple[str, bool]:
    """
    Cut text if it gets too long; return the text and whether we trimmed.
    (Layman's terms: shorten long text and tell us if we had to cut it.)
    """
    if max_chars is None or max_chars <= 0 or len(text) <= max_chars:
        return text, False
    return text[:max_chars].rstrip() + " …", True


def _shorten(text: str, limit: int) -> str:
    """
    Hard cap a string with an ellipsis to keep tokens small.
    (Layman's terms: force a string to a max length with an ellipsis.)
    """
    if limit <= 0 or len(text) <= limit:
        return text
    return text[: max(0, limit - 1)].rstrip() + "…"


def _domain_of(url: str) -> str:
    """
    Show a friendly site name like "example.com".
    (Layman's terms: pull the website's domain.)
    """
    try:
        return urlparse(url).netloc or ""
    except Exception:
        return ""


def _meta(soup: BeautifulSoup, name: str) -> str | None:
    tag = soup.find("meta", attrs={"name": name})
    return tag.get("content") if tag and tag.has_attr("content") else None


def _og(soup: BeautifulSoup, prop: str) -> str | None:
    tag = soup.find("meta", attrs={"property": prop})
    return tag.get("content") if tag and tag.has_attr("content") else None


def _extract_metadata(soup: BeautifulSoup, final_url: str) -> Dict[str, str]:
    """
    Pull the useful bits: title, description, site name, canonical URL, language, etc.
    (Layman's terms: gather page basics like title/description/address.)
    """
    meta: Dict[str, str] = {}

    # Title preference: <title> > og:title > twitter:title
    title_candidates = [
        (soup.title.string if soup.title and soup.title.string else None),
        _og(soup, "og:title"),
        _meta(soup, "twitter:title"),
    ]
    meta["title"] = next((t.strip() for t in title_candidates if t and t.strip()), "")

    # Description preference: description > og:description > twitter:description
    desc_candidates = [
        _meta(soup, "description"),
        _og(soup, "og:description"),
        _meta(soup, "twitter:description"),
    ]
    meta["description"] = next((d.strip() for d in desc_candidates if d and d.strip()), "")

    # Canonical link (helps dedupe)
    link_canonical = soup.find("link", rel=lambda v: v and "canonical" in v)
    meta["canonical"] = (link_canonical.get("href") or "").strip() if link_canonical else ""

    # Site name + language info if present
    meta["site_name"] = (_og(soup, "og:site_name") or "").strip()
    html_tag = soup.find("html")
    meta["lang"] = (html_tag.get("lang") or "").strip() if html_tag else ""

    # Final URL + domain
    meta["fetched_url"] = final_url
    meta["domain"] = _domain_of(final_url)

    return meta


def _extract_main_text(html: str) -> Tuple[str, BeautifulSoup]:
    """
    Use Readability to isolate the main article and turn it into clean text.
    Returns (clean_text, soup_of_readable_html).
    (Layman's terms: find the real article text and clean it.)
    """
    # Simplified article HTML from Readability
    doc = Document(html)
    readable_html = doc.summary(html_partial=True)

    # Parse simplified HTML
    s = BeautifulSoup(readable_html, "lxml")

    # Remove noisy tags
    for sel in ["script", "style", "noscript", "iframe", "svg"]:
        for tag in s.select(sel):
            tag.decompose()

    # Keep paragraphs, list items, and subheadings for structure without bloat
    text_parts: List[str] = []
    for p in s.find_all(["p", "li", "h2", "h3", "h4", "blockquote"]):
        chunk = p.get_text(" ", strip=True)
        if chunk:
            text_parts.append(chunk)

    clean_text = _normalize_whitespace("\n\n".join(text_parts))
    return clean_text, s


def _fullpage_markdown_from_soup(full_soup: BeautifulSoup, base_url: str) -> str:
    """
    Convert the page's main content (or body fallback) to Markdown, similar to
    web-scraper's Content Scraper tool, but without any file download side-effects.

    Steps:
      - Remove noisy elements (script/style/nav/footer/header/aside)
      - Prefer <main>, <article>, or common content containers; fallback to <body>
      - Convert to Markdown with ATX headings
      - Clean up excessive newlines, empty links, and whitespace
      - Prepend a title header when available
    """
    # Remove unwanted elements globally first
    for element in full_soup.select("script, style, nav, footer, header, aside"):
        element.decompose()

    # Try common main-content containers, then fallback to body
    main = (
        full_soup.find("main")
        or full_soup.find("article")
        or full_soup.find("div", class_=re.compile(r"content|main|post|article", re.I))
        or full_soup.find("body")
    )

    if not main:
        return "No main content found on the webpage."

    # Convert selected HTML to Markdown
    markdown_text = md(str(main), heading_style="ATX")

    # Clean up the markdown similar to web-scraper
    markdown_text = re.sub(r"\n{3,}", "\n\n", markdown_text)
    markdown_text = re.sub(r"\[\s*\]\([^)]*\)", "", markdown_text)  # empty links
    markdown_text = re.sub(r"[ \t]+", " ", markdown_text)
    markdown_text = markdown_text.strip()

    # Add title if present
    title = full_soup.find("title")
    if title and title.get_text(strip=True):
        markdown_text = f"# {title.get_text(strip=True)}\n\n{markdown_text}"

    return markdown_text or "No content could be extracted."


def _truncate_markdown(markdown: str, max_chars: int) -> str:
    """
    Truncate markdown content to a maximum character count while preserving structure.
    Tries to break at paragraph boundaries when possible.
    """
    if len(markdown) <= max_chars:
        return markdown
    
    # Find a good break point near the limit
    truncated = markdown[:max_chars]
    
    # Try to break at the end of a paragraph (double newline)
    last_paragraph = truncated.rfind('\n\n')
    if last_paragraph > max_chars * 0.7:  # If we find a paragraph break in the last 30%
        truncated = truncated[:last_paragraph]
    
    # Try to break at the end of a sentence
    elif '.' in truncated[-100:]:  # Look for a period in the last 100 chars
        last_period = truncated.rfind('.')
        if last_period > max_chars * 0.8:  # If we find a period in the last 20%
            truncated = truncated[:last_period + 1]
    
    return truncated.rstrip() + "\n\n> *[Content truncated for brevity]*"


def Fetch_Webpage(  # <-- MCP tool #1 (Fetch)
    url: Annotated[str, "The absolute URL to fetch (must return HTML)."],
    verbosity: Annotated[str, "Controls output length: 'Brief' (1000 chars), 'Standard' (3000 chars), or 'Full' (complete page)."] = "Standard",
) -> str:
    """
    Fetch a web page and return it converted to Markdown format with configurable length.
    
    This function retrieves a webpage and converts its main content to clean Markdown,
    preserving headings, formatting, and structure. It automatically removes navigation,
    footers, scripts, and other non-content elements to focus on the main article or
    content area.

    Args:
        url (str): The absolute URL to fetch (must return HTML).
        verbosity (str): Controls output length:
            - "Brief": Truncate to 1000 characters for quick summaries
            - "Standard": Truncate to 3000 characters for balanced content
            - "Full": Return complete page content with no length limit

    Returns:
        str: The webpage content converted to Markdown format with:
            - Page title as H1 header
            - Main content converted to clean Markdown
            - Preserved heading hierarchy
            - Clean formatting without navigation/sidebar elements
            - Length controlled by verbosity setting
    """
    if not url or not url.strip():
        return "Please enter a valid URL."

    try:
        resp = _http_get_enhanced(url)
        resp.raise_for_status()
    except requests.exceptions.RequestException as e:
        return f"An error occurred: {e}"

    final_url = str(resp.url)
    ctype = resp.headers.get("Content-Type", "")
    if "html" not in ctype.lower():
        return f"Unsupported content type for extraction: {ctype or 'unknown'}"

    # Decode to text
    resp.encoding = resp.encoding or resp.apparent_encoding
    html = resp.text

    # Parse HTML and convert to full-page Markdown
    full_soup = BeautifulSoup(html, "lxml")
    markdown_content = _fullpage_markdown_from_soup(full_soup, final_url)
    
    # Apply verbosity-based truncation
    if verbosity == "Brief":
        return _truncate_markdown(markdown_content, 1000)
    elif verbosity == "Standard":
        return _truncate_markdown(markdown_content, 3000)
    else:  # "Full"
        return markdown_content


# ============================================
# DuckDuckGo Search: Enhanced with error handling & rate limiting
# ============================================

import asyncio
from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, requests_per_minute: int = 30):
        self.requests_per_minute = requests_per_minute
        self.requests = []

    def acquire(self):
        """Synchronous rate limiting for non-async context"""
        now = datetime.now()
        # Remove requests older than 1 minute
        self.requests = [
            req for req in self.requests if now - req < timedelta(minutes=1)
        ]

        if len(self.requests) >= self.requests_per_minute:
            # Wait until we can make another request
            wait_time = 60 - (now - self.requests[0]).total_seconds()
            if wait_time > 0:
                time.sleep(max(1, wait_time))  # At least 1 second wait
                
        self.requests.append(now)

# Global rate limiters
_search_rate_limiter = RateLimiter(requests_per_minute=20)
_fetch_rate_limiter = RateLimiter(requests_per_minute=25)

def Search_DuckDuckGo(  # <-- MCP tool #2 (DDG Search)
    query: Annotated[str, "The search query (supports operators like site:, quotes, OR)."],
    max_results: Annotated[int, "Number of results to return (1–20)."] = 5,
) -> str:
    """
    Run a DuckDuckGo search with enhanced error handling and readable text output.
    Always returns results in human-friendly format with snippets included.

    Args:
        query (str): The search query string. Supports operators like site:, quotes for exact matching,
               OR for alternatives, and other DuckDuckGo search syntax.
               Examples:
               - Basic search: "Python programming"
               - Site search: "site:example.com"
               - Exact phrase: "artificial intelligence"
               - Exclude terms: "cats -dogs"
        max_results (int): Number of results to return (1–20). Default: 5.

    Returns:
        str: Search results in readable format with titles, URLs, and snippets as a numbered list.
    """
    if not query or not query.strip():
        return "No search query provided. Please enter a search term."

    # Validate max_results
    max_results = max(1, min(20, max_results))
    
    try:
        # Apply rate limiting to avoid being blocked
        _search_rate_limiter.acquire()
        
        # Perform search with timeout handling
        with DDGS() as ddgs:
            raw = ddgs.text(query, max_results=max_results)
            
    except Exception as e:
        error_msg = f"Search failed: {str(e)[:200]}"
        if "blocked" in str(e).lower() or "rate" in str(e).lower():
            error_msg = "Search temporarily blocked due to rate limiting. Please try again in a few minutes."
        elif "timeout" in str(e).lower():
            error_msg = "Search timed out. Please try again with a simpler query."
        elif "network" in str(e).lower() or "connection" in str(e).lower():
            error_msg = "Network connection error. Please check your internet connection and try again."
            
        return f"Error: {error_msg}"

    if not raw:
        return f"No results found for query: {query}"

    results = []

    for r in raw or []:
        title = (r.get("title") or "").strip()
        url = (r.get("href") or r.get("link") or "").strip()
        body = (r.get("body") or r.get("snippet") or "").strip()

        if not url:
            continue

        result_obj = {
            "title": title or _domain_of(url),
            "url": url,
            "snippet": body
        }
            
        results.append(result_obj)

    if not results:
        return f"No valid results found for query: {query}"

    # Format output in readable format
    lines = [f"Found {len(results)} search results for: {query}\n"]
    for i, result in enumerate(results, 1):
        lines.append(f"{i}. {result['title']}")
        lines.append(f"   URL: {result['url']}")
        if result['snippet']:
            lines.append(f"   Summary: {result['snippet']}")
        lines.append("")  # Empty line between results
    return "\n".join(lines)


# ======================================
# Code Execution: Python (MCP tool #3)
# ======================================

def Execute_Python(code: Annotated[str, "Python source code to run; stdout is captured and returned."]) -> str:
    """
    Execute arbitrary Python code and return captured stdout or an error message.

    Args:
        code (str): Python source code to run; stdout is captured and returned.

    Returns:
        str: Combined stdout produced by the code, or the exception text if
        execution failed.
    """
    if code is None:
        return "No code provided."

    old_stdout = sys.stdout
    redirected_output = sys.stdout = StringIO()
    try:
        exec(code)
        return redirected_output.getvalue()
    except Exception as e:
        return str(e)
    finally:
        sys.stdout = old_stdout


# ==========================
# Kokoro TTS (MCP tool #4)
# ==========================

_KOKORO_STATE = {
    "initialized": False,
    "device": "cpu",
    "model": None,
    "pipelines": {},
}


def get_kokoro_voices():
    """Get comprehensive list of available Kokoro voice IDs (54 total)."""
    try:
        from huggingface_hub import list_repo_files
        # Get voice files from the Kokoro repository
        files = list_repo_files('hexgrad/Kokoro-82M')
        voice_files = [f for f in files if f.endswith('.pt') and f.startswith('voices/')]
        voices = [f.replace('voices/', '').replace('.pt', '') for f in voice_files]
        return sorted(voices) if voices else _get_fallback_voices()
    except Exception:
        return _get_fallback_voices()


def _get_fallback_voices():
    """Return comprehensive fallback list of known Kokoro voices (54 total)."""
    return [
        # American Female (11 voices)
        "af_alloy", "af_aoede", "af_bella", "af_heart", "af_jessica", 
        "af_kore", "af_nicole", "af_nova", "af_river", "af_sarah", "af_sky",
        # American Male (9 voices)
        "am_adam", "am_echo", "am_eric", "am_fenrir", "am_liam", 
        "am_michael", "am_onyx", "am_puck", "am_santa",
        # British Female (4 voices)
        "bf_alice", "bf_emma", "bf_isabella", "bf_lily",
        # British Male (4 voices)
        "bm_daniel", "bm_fable", "bm_george", "bm_lewis",
        # European Female/Male (3 voices)
        "ef_dora", "em_alex", "em_santa",
        # French Female (1 voice)
        "ff_siwis",
        # Hindi Female/Male (4 voices)
        "hf_alpha", "hf_beta", "hm_omega", "hm_psi",
        # Italian Female/Male (2 voices)
        "if_sara", "im_nicola",
        # Japanese Female/Male (5 voices)
        "jf_alpha", "jf_gongitsune", "jf_nezumi", "jf_tebukuro", "jm_kumo",
        # Portuguese Female/Male (3 voices)
        "pf_dora", "pm_alex", "pm_santa",
        # Chinese Female/Male (8 voices)
        "zf_xiaobei", "zf_xiaoni", "zf_xiaoxiao", "zf_xiaoyi",
        "zm_yunjian", "zm_yunxi", "zm_yunxia", "zm_yunyang"
    ]


def _init_kokoro() -> None:
    """Lazy-initialize Kokoro model and pipelines on first use.

    Tries CUDA if torch is present and available; falls back to CPU. Keeps a
    minimal English pipeline and custom lexicon tweak for the word "kokoro".
    """
    if _KOKORO_STATE["initialized"]:
        return

    if KModel is None or KPipeline is None:
        raise RuntimeError(
            "Kokoro is not installed. Please install the 'kokoro' package (>=0.9.4)."
        )

    device = "cpu"
    if torch is not None:
        try:
            if torch.cuda.is_available():  # type: ignore[attr-defined]
                device = "cuda"
        except Exception:
            device = "cpu"

    model = KModel().to(device).eval()
    pipelines = {"a": KPipeline(lang_code="a", model=False)}
    # Custom pronunciation
    try:
        pipelines["a"].g2p.lexicon.golds["kokoro"] = "kˈOkəɹO"
    except Exception:
        pass

    _KOKORO_STATE.update(
        {
            "initialized": True,
            "device": device,
            "model": model,
            "pipelines": pipelines,
        }
    )


def List_Kokoro_Voices() -> List[str]:
    """
    Get a list of all available Kokoro voice identifiers.
    
    This MCP tool helps clients discover the 54 available voice options
    for the Generate_Speech tool.
    
    Returns:
        List[str]: A list of voice identifiers (e.g., ["af_heart", "am_adam", "bf_alice", ...])
        
    Voice naming convention:
        - First 2 letters: Language/Region (af=American Female, am=American Male, bf=British Female, etc.)
        - Following letters: Voice name (heart, adam, alice, etc.)
        
    Available categories:
        - American Female/Male (20 voices)
        - British Female/Male (8 voices) 
        - European Female/Male (3 voices)
        - French Female (1 voice)
        - Hindi Female/Male (4 voices)
        - Italian Female/Male (2 voices)
        - Japanese Female/Male (5 voices)
        - Portuguese Female/Male (3 voices)
        - Chinese Female/Male (8 voices)
    """
    return get_kokoro_voices()


def Generate_Speech(  # <-- MCP tool #4 (Generate Speech)
    text: Annotated[str, "The text to synthesize (English)."],
    speed: Annotated[float, "Speech speed multiplier in 0.5–2.0; 1.0 = normal speed."] = 1.25,
    voice: Annotated[str, "Voice identifier from 54 available options. Use List_Kokoro_Voices() to see all choices. Examples: 'af_heart' (US female), 'am_adam' (US male), 'bf_alice' (British female), 'jf_alpha' (Japanese female)."] = "af_heart",
) -> Tuple[int, np.ndarray]:
    """
    Synthesize speech from text using the Kokoro-82M model with 54 voice options.

    This function returns raw audio suitable for a Gradio Audio component and is
    also exposed as an MCP tool. It supports 54 different voices across multiple
    languages and accents including American, British, European, Hindi, Italian,
    Japanese, Portuguese, and Chinese speakers.

    Enhanced for longer audio generation:
        - Processes ALL text segments (not just the first one)
        - Can generate audio of any length based on input text
        - Concatenates multiple segments for seamless longer audio

    Default behavior:
        - Speed defaults to 1.25 (slightly brisk cadence) for clearer, snappier delivery.
        - Voice defaults to "af_heart" (American Female, Heart voice)

    Args:
        text (str): The text to synthesize. Works best with English but supports multiple languages.
        speed (float): Speech speed multiplier in 0.5–2.0; 1.0 = normal speed. Default: 1.25 (slightly brisk).
        voice (str): Voice identifier from 54 available options. Use List_Kokoro_Voices() to see all choices. Default: 'af_heart'.

    Returns:
        A tuple of (sample_rate_hz, audio_waveform) where:
        - sample_rate_hz: int sample rate in Hz (24_000)
        - audio_waveform: numpy.ndarray float32 mono waveform in range [-1, 1]

    Notes:
        - Requires the 'kokoro' package (>=0.9.4). If unavailable, an error is raised.
        - Runs on CUDA if available; otherwise CPU.
        - Supports 54 voices across 9 language/accent categories.
        - Can generate audio of any length - no 30 second limit!
        - Use List_Kokoro_Voices() MCP tool to discover all available voice options.
    """
    if not text or not text.strip():
        raise gr.Error("Please provide non-empty text to synthesize.")

    _init_kokoro()
    model = _KOKORO_STATE["model"]
    pipelines = _KOKORO_STATE["pipelines"]

    pipeline = pipelines.get("a")
    if pipeline is None:
        raise gr.Error("Kokoro English pipeline not initialized.")

    # Process ALL segments for longer audio generation
    audio_segments = []
    pack = pipeline.load_voice(voice)
    
    try:
        # Get all segments first to show progress for long text
        segments = list(pipeline(text, voice, speed))
        total_segments = len(segments)
        
        # Iterate through ALL segments instead of just the first one
        for segment_idx, (text_chunk, ps, _) in enumerate(segments):
            ref_s = pack[len(ps) - 1]
            try:
                audio = model(ps, ref_s, float(speed))
                audio_segments.append(audio.detach().cpu().numpy())
                
                # For very long text (>10 segments), show progress every few segments
                if total_segments > 10 and (segment_idx + 1) % 5 == 0:
                    print(f"Progress: Generated {segment_idx + 1}/{total_segments} segments...")
                    
            except Exception as e:
                raise gr.Error(f"Error generating audio for segment {segment_idx + 1}: {str(e)}")
        
        if not audio_segments:
            raise gr.Error("No audio was generated (empty synthesis result).")
        
        # Concatenate all segments to create the complete audio
        if len(audio_segments) == 1:
            final_audio = audio_segments[0]
        else:
            final_audio = np.concatenate(audio_segments, axis=0)
            # For multi-segment audio, provide completion info
            duration = len(final_audio) / 24_000
            if total_segments > 1:
                print(f"Completed: {total_segments} segments concatenated into {duration:.1f} seconds of audio")
            
        # Return 24 kHz mono waveform
        return 24_000, final_audio
        
    except gr.Error:
        raise  # Re-raise Gradio errors as-is
    except Exception as e:
        raise gr.Error(f"Error during speech generation: {str(e)}")


# ======================
# UI: four-tab interface
# ======================

# --- Fetch tab (compact controllable extraction) ---
fetch_interface = gr.Interface(
    fn=Fetch_Webpage,
    inputs=[
        gr.Textbox(label="URL", placeholder="https://example.com/article"),
        gr.Dropdown(
            label="Verbosity", 
            choices=["Brief", "Standard", "Full"], 
            value="Standard",
            info="Brief: 1000 chars, Standard: 3000 chars, Full: complete page"
        ),
    ],
    outputs=gr.Markdown(label="Extracted Markdown"),
    title="Fetch Webpage",
    description=(
        "<div style=\"text-align:center\">Convert any webpage to clean Markdown format with configurable length, preserving structure and formatting while removing navigation and clutter.</div>"
    ),
    api_description=(
        "Fetch a web page and return it converted to Markdown format with configurable length. "
        "This function retrieves a webpage and converts its main content to clean Markdown, "
        "preserving headings, formatting, and structure while removing navigation, footers, scripts, "
        "and other non-content elements. Parameters: url (str - absolute URL), verbosity (str - "
        "Brief/Standard/Full controlling output length: Brief=1000 chars, Standard=3000 chars, Full=complete page). "
        "Returns clean Markdown with page title as H1 header and preserved content hierarchy."
    ),
    flagging_mode="never",
)

# --- Simplified DDG tab (readable output only) ---
concise_interface = gr.Interface(
    fn=Search_DuckDuckGo,
    inputs=[
        gr.Textbox(label="Query", placeholder="topic OR site:example.com"),
        gr.Slider(minimum=1, maximum=20, value=5, step=1, label="Max results"),
    ],
    outputs=gr.Textbox(label="Search Results", interactive=False),
    title="DuckDuckGo Search",
    description=(
        "<div style=\"text-align:center\">Enhanced web search with readable output format. Always includes snippets for better context and understanding.</div>"
    ),
    api_description=(
        "Run a DuckDuckGo search with enhanced error handling and readable text output. "
        "Always returns results in human-friendly format with snippets included for better context. "
        "Supports advanced search operators: site: for specific domains, quotes for exact phrases, "
        "OR for alternatives, and - to exclude terms. Examples: 'Python programming', 'site:example.com', "
        "'\"artificial intelligence\"', 'cats -dogs', 'Python OR JavaScript'."
    ),
    flagging_mode="never",
    submit_btn="Search",
)

##

# --- Execute Python tab (simple code interpreter) ---
code_interface = gr.Interface(
    fn=Execute_Python,
    inputs=gr.Code(label="Python Code", language="python"),
    outputs=gr.Textbox(label="Output"),
    title="Python Code Executor",
    description=(
        "<div style=\"text-align:center\">Execute Python code and see the output.</div>"
    ),
    api_description=(
        "Execute arbitrary Python code and return captured stdout or an error message. "
        "Supports any valid Python code including imports, variables, functions, loops, and calculations. "
        "Examples: 'print(2+2)', 'import math; print(math.sqrt(16))', 'for i in range(3): print(i)'. "
        "Parameters: code (str - Python source code to execute). "
        "Returns: Combined stdout output or exception text if execution fails."
    ),
    flagging_mode="never",
)

CSS_STYLES = """
    .gradio-container h1 {
        text-align: center;
        /* Ensure main title appears first, then our two subtitle lines */
        display: grid;
        justify-items: center;
    }
    /* Place bold tools list on line 2, normal auth note on line 3 (below title) */
    .gradio-container h1::before {
        grid-row: 2;
        content: "Fetch Webpage | Search DuckDuckGo | Code Interpreter | Kokoro TTS (54 voices) | Image Generation | Video Generation";
        display: block;
        font-size: 1rem;
        font-weight: 700;
        opacity: 0.9;
        margin-top: 6px;
        white-space: pre-wrap;
    }
    .gradio-container h1::after {
        grid-row: 3;
        content: "Authentication is optional but Image/Video Generation require a `HF_READ_TOKEN` in env secrets. They are hidden otherwise.";
        display: block;
        font-size: 1rem;
        font-weight: 400;
        opacity: 0.9;
        margin-top: 2px;
        white-space: pre-wrap;
    }

    /* Remove inside tab panels so it doesn't duplicate under each tool title */
    .gradio-container [role=\"tabpanel\"] h1::before,
    .gradio-container [role=\"tabpanel\"] h1::after {
        content: none !important;
    }
"""

# --- Kokoro TTS tab (text to speech) ---
available_voices = get_kokoro_voices()
kokoro_interface = gr.Interface(
    fn=Generate_Speech,
    inputs=[
        gr.Textbox(label="Text", placeholder="Type text to synthesize…", lines=4),
        gr.Slider(minimum=0.5, maximum=2.0, value=1.25, step=0.1, label="Speed"),
        gr.Dropdown(
            label="Voice", 
            choices=available_voices, 
            value="af_heart", 
            info="Select from 54 available voices across multiple languages and accents"
        ),
    ],
    outputs=gr.Audio(label="Audio", type="numpy", format="wav", show_download_button=True),
    title="Kokoro TTS",
    description=(
        "<div style=\"text-align:center\">Generate speech with Kokoro-82M using 54 different voices. Supports multiple languages and accents. Can generate audio of any length! Runs on CPU or CUDA if available.</div>"
    ),
    api_description=(
        "Synthesize speech from text using Kokoro-82M with 54 voice options. Returns (sample_rate, waveform) suitable for playback. "
        "Supports unlimited text length by processing all segments. Voice examples: 'af_heart' (US female), 'am_adam' (US male), "
        "'bf_alice' (British female), 'bm_daniel' (British male), 'jf_alpha' (Japanese female), 'zf_xiaoni' (Chinese female). "
        "Parameters: text (str), speed (float 0.5–2.0, default 1.25x), voice (str from 54 available options, default 'af_heart'). "
        "Use List_Kokoro_Voices() to see all available voices. "
        "Return the generated media to the user in this format `![Alt text](URL)`"
    ),
    flagging_mode="never",
)

# ==========================
# Image Generation (Serverless)
# ==========================

HF_API_TOKEN = os.getenv("HF_READ_TOKEN")


def Generate_Image(  # <-- MCP tool #5 (Generate Image)
    prompt: Annotated[str, "Text description of the image to generate."],
    model_id: Annotated[str, "Hugging Face model id in the form 'creator/model-name' (e.g., black-forest-labs/FLUX.1-Krea-dev)."] = "black-forest-labs/FLUX.1-Krea-dev",
    negative_prompt: Annotated[str, "What should NOT appear in the image." ] = (
        "(deformed, distorted, disfigured), poorly drawn, bad anatomy, wrong anatomy, extra limb, "
        "missing limb, floating limbs, (mutated hands and fingers), disconnected limbs, mutation, "
        "mutated, ugly, disgusting, blurry, amputation, misspellings, typos"
    ),
    steps: Annotated[int, "Number of denoising steps (1–100). Higher = slower, potentially higher quality."] = 35,
    cfg_scale: Annotated[float, "Classifier-free guidance scale (1–20). Higher = follow the prompt more closely."] = 7.0,
    sampler: Annotated[str, "Sampling method label (UI only). Common options: 'DPM++ 2M Karras', 'DPM++ SDE Karras', 'Euler', 'Euler a', 'Heun', 'DDIM'."] = "DPM++ 2M Karras",
    seed: Annotated[int, "Random seed for reproducibility. Use -1 for a random seed per call."] = -1,
    width: Annotated[int, "Output width in pixels (64–1216, multiple of 32 recommended)."] = 1024,
    height: Annotated[int, "Output height in pixels (64–1216, multiple of 32 recommended)."] = 1024,
) -> Image.Image:
    """
    Generate a single image from a text prompt using a Hugging Face model via
    serverless Inference. Returns a PIL image. By default, the model is
    black-forest-labs/FLUX.1-Krea-dev.

    Notes (MCP):
    - Per the latest Gradio MCP docs, images returned from tools are handled by the server and
      converted to file URLs automatically for MCP clients. Ensure type hints and this docstring
      "Args:" block are present so the tool schema is accurate.

    Args:
        prompt (str): Text description of the image to generate.
        model_id (str): The Hugging Face model id (creator/model-name). Defaults to "black-forest-labs/FLUX.1-Krea-dev".
        negative_prompt (str): What should NOT appear in the image.
        steps (int): Number of denoising steps (1–100). Higher can improve quality.
        cfg_scale (float): Guidance scale (1–20). Higher = follow the prompt more closely.
        sampler (str): Sampling method label for UI; not all providers expose this control.
        seed (int): Random seed. Use -1 to randomize on each call.
        width (int): Output width in pixels (64–1216; multiples of 32 recommended).
        height (int): Output height in pixels (64–1216; multiples of 32 recommended).

    Returns:
        PIL.Image.Image: The generated image.

    Error modes:
        - Raises gr.Error with a user-friendly message on auth/model/load errors.
    """
    if not prompt or not prompt.strip():
        raise gr.Error("Please provide a non-empty prompt.")

    # Slightly enhance prompt for quality (kept consistent with Serverless space)
    enhanced_prompt = f"{prompt} | ultra detail, ultra elaboration, ultra quality, perfect."

    # Try multiple providers for resilience
    providers = ["auto", "replicate", "fal-ai"]
    last_error: Exception | None = None

    for provider in providers:
        try:
            client = InferenceClient(api_key=HF_API_TOKEN, provider=provider)
            image = client.text_to_image(
                prompt=enhanced_prompt,
                negative_prompt=negative_prompt,
                model=model_id,
                width=width,
                height=height,
                num_inference_steps=steps,
                guidance_scale=cfg_scale,
                seed=seed if seed != -1 else random.randint(1, 1_000_000_000),
            )
            return image
        except Exception as e:  # try next provider, transform last one to friendly error
            last_error = e
            continue

    # If we reach here, all providers failed
    msg = str(last_error) if last_error else "Unknown error"
    if "404" in msg:
        raise gr.Error(f"Model not found or unavailable: {model_id}. Check the id and your HF token access.")
    if "503" in msg:
        raise gr.Error("The model is warming up. Please try again shortly.")
    if "401" in msg or "403" in msg:
        raise gr.Error("Authentication failed. Set HF_READ_TOKEN environment variable with access to the model.")
    raise gr.Error(f"Image generation failed: {msg}")


image_generation_interface = gr.Interface(
    fn=Generate_Image,
    inputs=[
        gr.Textbox(label="Prompt", placeholder="Enter a prompt", lines=2),
        gr.Textbox(label="Model", value="black-forest-labs/FLUX.1-Krea-dev", placeholder="creator/model-name"),
        gr.Textbox(
            label="Negative Prompt",
            value=(
                "(deformed, distorted, disfigured), poorly drawn, bad anatomy, wrong anatomy, extra limb, "
                "missing limb, floating limbs, (mutated hands and fingers), disconnected limbs, mutation, "
                "mutated, ugly, disgusting, blurry, amputation, misspellings, typos"
            ),
            lines=2,
        ),
        gr.Slider(minimum=1, maximum=100, value=35, step=1, label="Steps"),
        gr.Slider(minimum=1.0, maximum=20.0, value=7.0, step=0.1, label="CFG Scale"),
        gr.Radio(label="Sampler", value="DPM++ 2M Karras", choices=[
            "DPM++ 2M Karras", "DPM++ SDE Karras", "Euler", "Euler a", "Heun", "DDIM"
        ]),
        gr.Slider(minimum=-1, maximum=1_000_000_000, value=-1, step=1, label="Seed (-1 = random)"),
        gr.Slider(minimum=64, maximum=1216, value=1024, step=32, label="Width"),
        gr.Slider(minimum=64, maximum=1216, value=1024, step=32, label="Height"),
    ],
    outputs=gr.Image(label="Generated Image"),
    title="Image Generation",
    description=(
        "<div style=\"text-align:center\">Generate images via Hugging Face Inference. "
        "Default model is FLUX.1-Krea</div>"
    ),
    api_description=(
        "Generate a single image from a text prompt using a Hugging Face model (serverless Inference). "
        "Supports creative prompts like 'a serene mountain landscape at sunset', 'portrait of a wise owl', "
        "'futuristic city with flying cars'. Default model: FLUX.1-Krea-dev (high quality). "
        "Parameters: prompt (str), model_id (str, creator/model-name), negative_prompt (str), steps (int, 1–100), "
        "cfg_scale (float, 1–20), sampler (str), seed (int, -1=random), width/height (int, 64–1216). "
        "Returns a PIL.Image. Return the generated media to the user in this format `![Alt text](URL)`"
    ),
    flagging_mode="never",
)

# ==========================
# Video Generation (Serverless)
# ==========================

def _write_video_tmp(data_iter_or_bytes: object, suffix: str = ".mp4") -> str:
    """Write video bytes or iterable of bytes to a system temporary file and return its path.

    This avoids polluting the project directory. The file is created in the OS temp
    location; Gradio will handle serving & offering the download button.
    """
    fd, fname = tempfile.mkstemp(suffix=suffix)
    try:
        with os.fdopen(fd, "wb") as f:
            if isinstance(data_iter_or_bytes, (bytes, bytearray)):
                f.write(data_iter_or_bytes)  # type: ignore[arg-type]
            elif hasattr(data_iter_or_bytes, "read"):
                f.write(data_iter_or_bytes.read())  # type: ignore[call-arg]
            elif hasattr(data_iter_or_bytes, "content"):
                f.write(data_iter_or_bytes.content)  # type: ignore[attr-defined]
            elif hasattr(data_iter_or_bytes, "__iter__") and not isinstance(data_iter_or_bytes, (str, dict)):
                for chunk in data_iter_or_bytes:  # type: ignore[assignment]
                    if chunk:
                        f.write(chunk)
            else:
                raise gr.Error("Unsupported video data type returned by provider.")
    except Exception:
        # Clean up if writing failed
        try:
            os.remove(fname)
        except Exception:
            pass
        raise
    return fname


HF_VIDEO_TOKEN = os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")


def Generate_Video(  # <-- MCP tool #6 (Generate Video)
    prompt: Annotated[str, "Text description of the video to generate (e.g., 'a red fox running through a snowy forest at sunrise')."],
    model_id: Annotated[str, "Hugging Face model id in the form 'creator/model-name'. Defaults to Wan-AI/Wan2.2-T2V-A14B."] = "Wan-AI/Wan2.2-T2V-A14B",
    negative_prompt: Annotated[str, "What should NOT appear in the video."] = "",
    steps: Annotated[int, "Number of denoising steps (1–100). Higher can improve quality but is slower."] = 25,
    cfg_scale: Annotated[float, "Guidance scale (1–20). Higher = follow the prompt more closely, lower = more creative."] = 3.5,
    seed: Annotated[int, "Random seed for reproducibility. Use -1 for a random seed per call."] = -1,
    width: Annotated[int, "Output width in pixels (multiples of 8 recommended)."] = 768,
    height: Annotated[int, "Output height in pixels (multiples of 8 recommended)."] = 768,
    fps: Annotated[int, "Frames per second of the output video (e.g., 24)."] = 24,
    duration: Annotated[float, "Target duration in seconds (provider/model dependent, commonly 2–6s)."] = 4.0,
) -> str:
    """
    Generate a short video from a text prompt using Hugging Face Inference Providers (Serverless Inference).

    This tool follows the latest MCP guidance for Gradio-based MCP servers: clear type hints and
    docstrings define the tool schema automatically. The returned file path will be converted to a file URL
    for MCP clients.

    Args:
        prompt (str): Text description of the video to generate.
    model_id (str): The Hugging Face model id (creator/model-name). Defaults to "Wan-AI/Wan2.2-T2V-A14B".
        negative_prompt (str): What should NOT appear in the video.
        steps (int): Number of denoising steps (1–100). Higher can improve quality but is slower.
        cfg_scale (float): Guidance scale (1–20). Higher = follow the prompt more closely.
        seed (int): Random seed. Use -1 to randomize on each call.
        width (int): Output width in pixels.
        height (int): Output height in pixels.
        fps (int): Frames per second.
        duration (float): Target duration in seconds.

    Returns:
        str: Path to an MP4 file on disk (Gradio will serve this file; MCP converts it to a file URL).

    Error modes:
        - Raises gr.Error with a user-friendly message on auth/model/load errors or unsupported parameters.
    """
    if not prompt or not prompt.strip():
        raise gr.Error("Please provide a non-empty prompt.")

    if not HF_VIDEO_TOKEN:
        # Still attempt without a token (public models), but warn earlier if it fails.
        pass

    providers = ["auto", "replicate", "fal-ai"]
    last_error: Exception | None = None

    # Build a common parameters dict. Providers may ignore unsupported keys.
    parameters = {
        "negative_prompt": negative_prompt or None,
        "num_inference_steps": steps,
        "guidance_scale": cfg_scale,
        "seed": seed if seed != -1 else random.randint(1, 1_000_000_000),
        "width": width,
        "height": height,
        "fps": fps,
        # Some providers/models expect num_frames instead of duration; we pass both-friendly value
        # when supported; they may be ignored by the backend.
        "duration": duration,
    }

    for provider in providers:
        try:
            client = InferenceClient(api_key=HF_VIDEO_TOKEN, provider=provider)
            # Use the documented text_to_video API with correct parameters
            if hasattr(client, "text_to_video"):
                # Calculate num_frames from duration and fps if both provided
                num_frames = int(duration * fps) if duration and fps else None
                
                # Build extra_body for provider-specific parameters
                extra_body = {}
                if width:
                    extra_body["width"] = width
                if height:
                    extra_body["height"] = height
                if fps:
                    extra_body["fps"] = fps
                if duration:
                    extra_body["duration"] = duration
                
                result = client.text_to_video(
                    prompt=prompt,
                    model=model_id,
                    guidance_scale=cfg_scale,
                    negative_prompt=[negative_prompt] if negative_prompt else None,
                    num_frames=num_frames,
                    num_inference_steps=steps,
                    seed=parameters["seed"],
                    extra_body=extra_body if extra_body else None,
                )
            else:
                # Generic POST fallback for older versions
                result = client.post(
                    model=model_id,
                    json={
                        "inputs": prompt,
                        "parameters": {k: v for k, v in parameters.items() if v is not None},
                    },
                )

            # Save output to an .mp4
            path = _write_video_tmp(result, suffix=".mp4")
            return path
        except Exception as e:
            last_error = e
            continue

    msg = str(last_error) if last_error else "Unknown error"
    if "404" in msg:
        raise gr.Error(f"Model not found or unavailable: {model_id}. Check the id and HF token access.")
    if "503" in msg:
        raise gr.Error("The model is warming up. Please try again shortly.")
    if "401" in msg or "403" in msg:
        raise gr.Error("Authentication failed or not permitted. Set HF_READ_TOKEN/HF_TOKEN with inference access.")
    raise gr.Error(f"Video generation failed: {msg}")


video_generation_interface = gr.Interface(
    fn=Generate_Video,
    inputs=[
        gr.Textbox(label="Prompt", placeholder="Enter a prompt for the video", lines=2),
    gr.Textbox(label="Model", value="Wan-AI/Wan2.2-T2V-A14B", placeholder="creator/model-name"),
        gr.Textbox(label="Negative Prompt", value="", lines=2),
        gr.Slider(minimum=1, maximum=100, value=25, step=1, label="Steps"),
        gr.Slider(minimum=1.0, maximum=20.0, value=3.5, step=0.1, label="CFG Scale"),
        gr.Slider(minimum=-1, maximum=1_000_000_000, value=-1, step=1, label="Seed (-1 = random)"),
        gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Width"),
        gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Height"),
        gr.Slider(minimum=4, maximum=60, value=24, step=1, label="FPS"),
        gr.Slider(minimum=1.0, maximum=10.0, value=4.0, step=0.5, label="Duration (s)"),
    ],
    outputs=gr.Video(label="Generated Video", show_download_button=True, format="mp4"),
    title="Video Generation",
    description=(
    "<div style=\"text-align:center\">Generate short videos via Hugging Face Inference Providers. "
    "Default model is Wan2.2-T2V-A14B.</div>"
    ),
    api_description=(
        "Generate a short video from a text prompt using a Hugging Face model (Serverless Inference). "
        "Create dynamic scenes like 'a red fox running through a snowy forest at sunrise', 'waves crashing on a rocky shore', "
        "'time-lapse of clouds moving across a blue sky'. Default model: Wan2.2-T2V-A14B (2-6 second videos). "
        "Parameters: prompt (str), model_id (str), negative_prompt (str), steps (int), cfg_scale (float), seed (int), "
        "width/height (int), fps (int), duration (float in seconds). Returns MP4 file path. "
        "Return the generated media to the user in this format `![Alt text](URL)`"
    ),
    flagging_mode="never",
)

# Build tabbed app; disable Image/Video tools if no HF token is present
HAS_HF_TOKEN = bool(HF_API_TOKEN or HF_VIDEO_TOKEN)

_interfaces = [
    fetch_interface,
    concise_interface,
    code_interface,
    kokoro_interface,
]
_tab_names = [
    "Fetch Webpage",
    "DuckDuckGo Search",
    "Python Code Executor",
    "Kokoro TTS",
]

if HAS_HF_TOKEN:
    _interfaces.extend([image_generation_interface, video_generation_interface])
    _tab_names.extend(["Image Generation", "Video Generation"])

demo = gr.TabbedInterface(
    interface_list=_interfaces,
    tab_names=_tab_names,
    title="Tools MCP",
    theme="Nymbo/Nymbo_Theme",
    css=CSS_STYLES,
)

# Launch the UI and expose all functions as MCP tools in one server
if __name__ == "__main__":
    demo.launch(mcp_server=True)