app.py

import json
import os
import time
from typing import List, Tuple

import gradio as gr
import requests
from duckduckgo_search import DDGS


DEFAULT_MODEL = "gemma-4-31b-it"
GOOGLE_API_BASE = "https://generativelanguage.googleapis.com/v1beta/models"
DEBUG_LOG_PATH = "debug-1caba1.log"
DEBUG_SESSION_ID = "1caba1"


def debug_log(run_id: str, hypothesis_id: str, location: str, message: str, data: dict) -> None:
    payload = {
        "sessionId": DEBUG_SESSION_ID,
        "runId": run_id,
        "hypothesisId": hypothesis_id,
        "location": location,
        "message": message,
        "data": data,
        "timestamp": int(time.time() * 1000),
    }
    # region agent log
    with open(DEBUG_LOG_PATH, "a", encoding="utf-8") as f:
        f.write(json.dumps(payload, ensure_ascii=True) + "\n")
    # endregion


def normalize_model_id(model_input: str) -> str:
    cleaned = (model_input or "").strip().lower().replace(" ", "-")
    aliases = {
        "gemma-4-31b-it": "gemma-4-31b-it",
        "gemma4-31b-it": "gemma-4-31b-it",
        "gemma-4-31b-instruct": "gemma-4-31b-it",
        "gemma-4-31b": "gemma-4-31b-it",
    }
    return aliases.get(cleaned, cleaned)


def parse_model_response(text: str, option_a: str, option_b: str) -> Tuple[str, str, str]:
    fallback_choice = option_a or option_b or "Expand to Singapore Market"
    fallback_reason = "High expected growth with favorable risk profile."
    fallback_confidence = "76%"

    if not text:
        return fallback_choice, fallback_reason, fallback_confidence

    lines = [line.strip() for line in text.split("\n") if line.strip()]
    choice = ""
    reason = ""
    confidence = ""

    for line in lines:
        lower = line.lower()
        if not choice and lower.startswith("choice:"):
            choice = line[7:].strip()
        elif not reason and lower.startswith("reason:"):
            reason = line[7:].strip()
        elif not confidence and lower.startswith("confidence:"):
            confidence = line[11:].strip()

    if not reason and lines:
        reason = " ".join(lines[:2])

    return (
        choice or fallback_choice,
        reason or fallback_reason,
        confidence or fallback_confidence,
    )


def build_web_context(dilemma: str, option_a: str, option_b: str, max_results: int) -> str:
    query = " ".join(
        [
            dilemma or "",
            option_a or "",
            option_b or "",
            "market trends risk analysis recent data",
        ]
    ).strip()

    if not query:
        return "No web context available."

    rows: List[str] = []
    # region agent log
    debug_log(
        "pre-fix",
        "H1",
        "app.py:build_web_context",
        "DuckDuckGo query prepared",
        {"query_len": len(query), "max_results": int(max_results)},
    )
    # endregion
    with DDGS() as ddgs:
        results = ddgs.text(query, max_results=max_results)
        for idx, item in enumerate(results, start=1):
            title = (item.get("title") or "").strip()
            href = (item.get("href") or "").strip()
            body = (item.get("body") or "").strip()
            if not (title or href or body):
                continue
            rows.append(f"{idx}. {title}\nURL: {href}\nSnippet: {body}")
    # region agent log
    debug_log(
        "pre-fix",
        "H1",
        "app.py:build_web_context",
        "DuckDuckGo query completed",
        {"result_count": len(rows)},
    )
    # endregion

    return "\n\n".join(rows) if rows else "No web results returned by DuckDuckGo."


def call_google_model(
    api_key: str,
    model: str,
    dilemma: str,
    option_a: str,
    option_b: str,
    web_context: str,
) -> str:
    endpoint = f"{GOOGLE_API_BASE}/{model}:generateContent?key={api_key}"
    # region agent log
    debug_log(
        "pre-fix",
        "H3",
        "app.py:call_google_model",
        "Prepared model endpoint",
        {"model": model, "endpoint_without_key": f"{GOOGLE_API_BASE}/{model}:generateContent"},
    )
    # endregion
    prompt = "\n".join(
        [
            "You are a strategic decision assistant.",
            "Given a dilemma and two options, pick the best option.",
            "Return exactly 3 lines in this strict format:",
            "CHOICE: <best option text>",
            "REASON: <single concise reason>",
            "CONFIDENCE: <0-100%>",
            "",
            "Use the web context below as additional evidence when relevant.",
            f"WEB_CONTEXT:\n{web_context or 'No web context provided.'}",
            "",
            f"DILEMMA: {dilemma or 'N/A'}",
            f"OPTION_ALPHA: {option_a or 'N/A'}",
            f"OPTION_BETA: {option_b or 'N/A'}",
        ]
    )

    response = requests.post(
        endpoint,
        headers={"Content-Type": "application/json"},
        data=json.dumps({"contents": [{"parts": [{"text": prompt}]}]}),
        timeout=60,
    )

    data = response.json()
    # region agent log
    debug_log(
        "pre-fix",
        "H4",
        "app.py:call_google_model",
        "Model response received",
        {
            "status_code": int(response.status_code),
            "ok": bool(response.ok),
            "has_candidates": isinstance(data.get("candidates", []), list),
            "error_message": data.get("error", {}).get("message", ""),
        },
    )
    # endregion
    if not response.ok:
        message = data.get("error", {}).get("message", "Google AI Studio request failed.")
        raise RuntimeError(message)

    candidates = data.get("candidates", [])
    if not candidates:
        return ""

    parts = candidates[0].get("content", {}).get("parts", [])
    return "\n".join(part.get("text", "") for part in parts).strip()


def analyze_decision(
    dilemma: str,
    option_a: str,
    option_b: str,
    model_input: str,
    api_key_input: str,
    use_web_search: bool,
    max_search_results: int,
) -> Tuple[str, str, str, str, str]:
    api_key = (api_key_input or "").strip() or os.getenv("GOOGLE_API_KEY", "").strip()
    model = normalize_model_id(model_input) or DEFAULT_MODEL
    web_context = "Web search disabled."
    # region agent log
    debug_log(
        "pre-fix",
        "H2",
        "app.py:analyze_decision",
        "Decision analysis started",
        {
            "has_api_key": bool(api_key),
            "model": model,
            "use_web_search": bool(use_web_search),
            "max_search_results": int(max_search_results),
            "has_dilemma": bool((dilemma or "").strip()),
            "has_option_a": bool((option_a or "").strip()),
            "has_option_b": bool((option_b or "").strip()),
        },
    )
    # endregion

    if not api_key:
        return (
            option_a or option_b or "Fallback Recommendation",
            "Missing API key. Add `GOOGLE_API_KEY` in Hugging Face Space secrets or input it in the field.",
            "N/A",
            model,
            web_context,
        )

    try:
        if use_web_search:
            safe_results = max(1, min(int(max_search_results), 8))
            web_context = build_web_context(dilemma, option_a, option_b, safe_results)
        raw = call_google_model(api_key, model, dilemma, option_a, option_b, web_context)
        choice, reason, confidence = parse_model_response(raw, option_a, option_b)
        # region agent log
        debug_log(
            "pre-fix",
            "H5",
            "app.py:analyze_decision",
            "Parsed model output",
            {
                "raw_len": len(raw or ""),
                "choice_len": len(choice or ""),
                "reason_len": len(reason or ""),
                "confidence": confidence,
            },
        )
        # endregion
        return choice, reason, confidence, model, web_context
    except Exception as exc:
        # region agent log
        debug_log(
            "pre-fix",
            "H4",
            "app.py:analyze_decision",
            "Analyze decision failed",
            {"error": str(exc)},
        )
        # endregion
        return (
            option_a or option_b or "Fallback Recommendation",
            f"Model call failed: {exc}",
            "N/A",
            model,
            web_context,
        )


with gr.Blocks(theme=gr.themes.Soft(), title="The Oracle | Decision Engine") as demo:
    gr.Markdown("## The Oracle - Hugging Face Spaces App")
    gr.Markdown(
        "Enter your dilemma and options, then run analysis with Google AI Studio (Gemma). "
        "Default model is `gemma-4-31b-it`."
    )

    with gr.Row():
        dilemma = gr.Textbox(
            label="Core dilemma",
            placeholder="Should we pivot strategy toward AI-integrated logistics by Q3?",
            lines=4,
        )

    with gr.Row():
        option_a = gr.Textbox(label="Option Alpha", placeholder="Immediate full pivot")
        option_b = gr.Textbox(label="Option Beta", placeholder="Incremental integration")

    with gr.Row():
        model_input = gr.Textbox(label="Model ID", value=DEFAULT_MODEL)
        api_key_input = gr.Textbox(
            label="Google API key (optional if set in Space secrets as GOOGLE_API_KEY)",
            type="password",
        )
    with gr.Row():
        use_web_search = gr.Checkbox(label="Enable DuckDuckGo web search context", value=True)
        max_search_results = gr.Slider(
            label="DuckDuckGo results to use",
            minimum=1,
            maximum=8,
            step=1,
            value=4,
        )

    run_btn = gr.Button("Generate Analysis", variant="primary")

    with gr.Row():
        out_choice = gr.Textbox(label="Recommended Choice")
        out_reason = gr.Textbox(label="Reason")
    with gr.Row():
        out_confidence = gr.Textbox(label="Confidence")
        out_model = gr.Textbox(label="Model Used")
    out_web_context = gr.Textbox(label="DuckDuckGo Research Context", lines=12)

    run_btn.click(
        fn=analyze_decision,
        inputs=[
            dilemma,
            option_a,
            option_b,
            model_input,
            api_key_input,
            use_web_search,
            max_search_results,
        ],
        outputs=[out_choice, out_reason, out_confidence, out_model, out_web_context],
    )


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