app.py

"""Transactable Gradio App for Hugging Face Spaces."""

import asyncio
import os

import gradio as gr
import httpx

# API configuration
API_BASE_URL = os.getenv("API_BASE_URL", "https://transactable-api-4uhaa7eqfq-uc.a.run.app")


def get_headers(api_key: str):
    headers = {"Content-Type": "application/json"}
    if api_key:
        headers["X-API-Key"] = api_key
    return headers


# =============================================================================
# API FUNCTIONS
# =============================================================================


async def upload_files(files, api_key: str, progress=gr.Progress()):
    """Upload files with progress tracking."""
    if not files:
        return "Please select files to upload.", ""

    if not isinstance(files, list):
        files = [files]

    total = len(files)
    file_ids = []
    failed = 0

    semaphore = asyncio.Semaphore(10)

    async def upload_one(file):
        nonlocal failed
        async with semaphore:
            try:
                async with httpx.AsyncClient(timeout=60.0) as client:
                    headers = {"X-API-Key": api_key} if api_key else {}
                    with open(file.name, "rb") as f:
                        response = await client.post(
                            f"{API_BASE_URL}/api/v1/files/upload",
                            files={"file": (file.name.split("/")[-1], f)},
                            headers=headers,
                        )
                    if response.status_code == 200:
                        return response.json().get("id")
            except Exception:
                pass
            failed += 1
            return None

    progress(0, desc="Uploading...")

    batch_size = 50
    for i in range(0, total, batch_size):
        batch = files[i : i + batch_size]
        results = await asyncio.gather(*[upload_one(f) for f in batch])
        file_ids.extend([r for r in results if r])
        progress((i + len(batch)) / total)

    return (
        f"**Uploaded:** {len(file_ids)} | **Failed:** {failed} | **Total:** {total}",
        ",".join(file_ids),
    )


async def analyze_files(file_ids: str, analysis_type: str, api_key: str, progress=gr.Progress()):
    """Analyze uploaded files."""
    if not file_ids:
        return "Upload files first."

    ids = [fid.strip() for fid in file_ids.split(",") if fid.strip()]
    total = len(ids)

    if total == 0:
        return "No file IDs."

    results = {"success": 0, "failed": 0, "spending": 0, "categories": {}}
    semaphore = asyncio.Semaphore(5)

    async def analyze_one(file_id):
        async with semaphore:
            try:
                async with httpx.AsyncClient(timeout=180.0) as client:
                    response = await client.post(
                        f"{API_BASE_URL}/api/v1/files/{file_id}/analyze",
                        json={"analysis_type": analysis_type},
                        headers=get_headers(api_key),
                    )
                    if response.status_code == 200:
                        return response.json()
            except Exception:
                pass
            return None

    progress(0, desc="Analyzing...")

    batch_size = 20
    for i in range(0, total, batch_size):
        batch = ids[i : i + batch_size]
        batch_results = await asyncio.gather(*[analyze_one(fid) for fid in batch])

        for r in batch_results:
            if r:
                results["success"] += 1
                results["spending"] += r.get("total_spending", 0) or 0
                for cat, amt in (r.get("categories") or {}).items():
                    results["categories"][cat] = results["categories"].get(cat, 0) + (amt or 0)
            else:
                results["failed"] += 1

        progress((i + len(batch)) / total)

    output = f"""## Analysis Complete

**✓ Success:** {results['success']} | **✗ Failed:** {results['failed']}
**Total Spending:** ${results['spending']:,.2f}

### Top Categories
"""
    for cat, amt in sorted(results["categories"].items(), key=lambda x: -x[1])[:10]:
        output += f"- {cat}: ${amt:,.2f}\n"

    return output


async def ask_question(question: str, conversation_id: str, api_key: str):
    """Ask a question about documents."""
    if not question.strip():
        return "Enter a question.", conversation_id

    async with httpx.AsyncClient(timeout=60.0) as client:
        payload = {"question": question}
        if conversation_id:
            payload["conversation_id"] = conversation_id

        response = await client.post(
            f"{API_BASE_URL}/api/v1/ask",
            json=payload,
            headers=get_headers(api_key),
        )

        if response.status_code == 200:
            data = response.json()
            return data.get("answer", "No answer."), data.get("conversation_id", conversation_id)
        return f"Error: {response.status_code}", conversation_id


# =============================================================================
# GRADIO UI
# =============================================================================

with gr.Blocks(title="Transactable", theme=gr.themes.Soft()) as app:
    gr.Markdown(
        """
    # 💰 Transactable
    Upload financial documents, analyze spending, and ask questions.
    """
    )

    with gr.Row():
        api_key = gr.Textbox(
            label="API Key",
            placeholder="Enter your API key",
            type="password",
            scale=3,
        )

    with gr.Tabs():
        # Upload Tab
        with gr.TabItem("📤 Upload & Analyze"):
            with gr.Row():
                with gr.Column():
                    files = gr.File(
                        label="Documents (PDF, PNG, JPG)",
                        file_count="multiple",
                        file_types=[".pdf", ".png", ".jpg", ".jpeg"],
                    )
                    upload_btn = gr.Button("⬆️ Upload", variant="primary")

                with gr.Column():
                    upload_status = gr.Markdown("Select files and click Upload.")
                    file_ids = gr.Textbox(label="File IDs", lines=2)

            upload_btn.click(upload_files, [files, api_key], [upload_status, file_ids])

            gr.Markdown("---")

            with gr.Row():
                analysis_type = gr.Dropdown(
                    ["spending", "income", "general"],
                    value="spending",
                    label="Analysis Type",
                )
                analyze_btn = gr.Button("🔍 Analyze All", variant="primary")

            analysis_result = gr.Markdown()
            analyze_btn.click(analyze_files, [file_ids, analysis_type, api_key], analysis_result)

        # Q&A Tab
        with gr.TabItem("💬 Ask"):
            conversation = gr.State(value=None)
            question = gr.Textbox(label="Question", placeholder="What was my total spending?")
            ask_btn = gr.Button("Ask", variant="primary")
            answer = gr.Markdown()

            ask_btn.click(ask_question, [question, conversation, api_key], [answer, conversation])
            question.submit(ask_question, [question, conversation, api_key], [answer, conversation])

    gr.Markdown("---\n*Powered by [Transactable API](https://github.com/transactable)*")

if __name__ == "__main__":
    app.launch()