app.py

import os
import tempfile
from pathlib import Path
from typing import Any
from urllib.parse import urlparse, unquote
from urllib.request import Request, urlopen

import gradio as gr

from parsers import pdf_to_json, image_to_json, xlsx_to_json


IMAGE_EXTS = {"png", "jpg", "jpeg", "webp", "bmp", "tiff", "gif"}
XLSX_EXTS = {"xlsx", "xlsm"}


def _get_path(file_obj: Any) -> str:
    if file_obj is None:
        return None
    if isinstance(file_obj, str):
        return file_obj
    # gr.File may pass a dict or a file-like object
    if hasattr(file_obj, "name"):
        return getattr(file_obj, "name")
    if isinstance(file_obj, dict):
        return file_obj.get("name") or file_obj.get("tmp_path") or file_obj.get("tempfile")
    return None


def _dispatch(path: str) -> dict:
    """Route a local file path to the matching parser based on its extension."""
    suffix = Path(path).suffix.lower().lstrip(".")
    if suffix == "pdf":
        return pdf_to_json(path)
    if suffix in IMAGE_EXTS:
        return image_to_json(path)
    if suffix in XLSX_EXTS:
        return xlsx_to_json(path)
    return {"error": f"unsupported file type: .{suffix}"}


def _download_to_temp(url: str) -> str:
    """Fetch a URL to a temp file, preserving the extension for dispatch."""
    req = Request(url, headers={"User-Agent": "file-to-json"})
    with urlopen(req, timeout=30) as resp:
        data = resp.read()
    suffix = Path(unquote(urlparse(url).path)).suffix
    fd, tmp = tempfile.mkstemp(suffix=suffix)
    with os.fdopen(fd, "wb") as f:
        f.write(data)
    return tmp


def parse_file_url(file_url: str) -> dict:
    """Parse a PDF, image, or Excel file into JSON.

    Args:
        file_url: An http(s) URL pointing to a .pdf, image, or .xlsx/.xlsm file.

    Returns:
        A JSON object containing the file's parsed contents.
    """
    if not file_url:
        return {"error": "no file_url provided"}
    try:
        path = _download_to_temp(file_url)
    except Exception as e:
        return {"error": f"could not fetch url: {e}"}
    try:
        return _dispatch(path)
    except Exception as e:
        return {"error": str(e)}
    finally:
        try:
            os.remove(path)
        except OSError:
            pass


def process_file(file_obj: Any) -> dict:
    """Web-UI handler: parse a file uploaded through the Gradio interface."""
    path = _get_path(file_obj)
    if not path:
        return {"error": "no file provided"}
    try:
        return _dispatch(path)
    except Exception as e:
        return {"error": str(e)}


def main():
    with gr.Blocks() as demo:
        gr.Markdown("# MCP: PDF, Image & Excel → JSON\nUpload a PDF, image, or .xlsx file to parse into JSON.")
        file_in = gr.File(label="Upload PDF, Image, or Excel (.xlsx)")
        out = gr.JSON(label="Parsed JSON")
        # Web-UI only: hidden from the API/MCP so no file-upload tool is exposed.
        file_in.change(fn=process_file, inputs=file_in, outputs=out, api_name=False)

        # The MCP tool: takes a URL string (no file-input upload instructions).
        gr.api(parse_file_url, api_name="parse_file_url")

    demo.launch(mcp_server=True)


if __name__ == '__main__':
    main()