app.py

"""
Shopping Assistant – Main Gradio Application
=============================================

Architecture overview (for ANLY 656 students):

    This file orchestrates a **two-pass LLM pipeline** backed by real
    product data from Google Shopping.

    ┌──────────────────────────────────────────────────────────────────┐
    │  User types a plain-English shopping request                    │
    │         ↓                                                       │
    │  PASS 1 – NLP Parser  (nlp_parser.py)                          │
    │      • Sends the request to an LLM (Qwen 2.5, via HF API).    │
    │      • Returns structured JSON: search queries + requirements.  │
    │         ↓                                                       │
    │  SerpApi – Google Shopping  (search_engine.py)                  │
    │      • Calls the Google Shopping API through SerpApi.           │
    │      • Returns product titles, prices, thumbnails, links.      │
    │         ↓                                                       │
    │  SerpApi – Immersive Product Detail  (search_engine.py)        │
    │      • For the top-N candidates, fetches full descriptions     │
    │        and specification sheets via a second SerpApi call.      │
    │         ↓                                                       │
    │  PASS 2 – Product Evaluator  (product_evaluator.py)            │
    │      • Sends each product's description + specs to the LLM.    │
    │      • LLM checks every user requirement → met / not_met /     │
    │        unknown.                                                 │
    │         ↓                                                       │
    │  Gradio UI  (ui_components.py + this file)                     │
    │      • Renders product cards, comparison table, and a plain-   │
    │        language summary of how the system interpreted the       │
    │        request.                                                 │
    └──────────────────────────────────────────────────────────────────┘

Key APIs used:
    • Hugging Face Inference API  – free-tier chat completions
    • SerpApi                     – Google Shopping + Immersive Product

Dependencies:
    gradio, huggingface-hub, google-search-results (serpapi),
    python-dotenv, pandas
"""

# ── Standard-library imports ────────────────────────────────────────
import logging
import time
from typing import Any

# ── Third-party imports ─────────────────────────────────────────────
import gradio as gr                # Web-UI framework (renders in browser)
import pandas as pd                # Tabular data manipulation

# ── Project-local imports ───────────────────────────────────────────
# Each module encapsulates one responsibility:
from config import MAX_DETAIL_LOOKUPS          # Caps SerpApi detail calls
from nlp_parser import parse_user_request      # LLM Pass 1 – intent parsing
from product_evaluator import evaluate_batch   # LLM Pass 2 – compliance check
from product_parser import (                   # Data normalisation helpers
    apply_filters,
    attach_details,
    parse_results,
)
from search_engine import (                    # SerpApi wrappers
    get_product_details,
    search_shopping,
)
from ui_components import (                    # HTML / DataFrame builders
    build_comparison_table,
    build_product_cards,
)

# ── Logging configuration ──────────────────────────────────────────
# Best practice: use the `logging` module instead of bare `print()`.
# This lets you control verbosity (DEBUG / INFO / WARNING / ERROR)
# from one place, and the output is timestamped automatically.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-8s  %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)


# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
#  Helper Functions
#  ────────────────
#  Breaking a large function into small helpers is a Python best
#  practice.  Each helper does one thing, is easy to test, and
#  can be re-used independently.
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━


def _build_parsed_summary(
    searches: list[dict[str, Any]],
    requirements: list[str],
) -> str:
    """
    Convert the NLP-parsed output into a human-readable Markdown summary.

    This is displayed in the "What I Understood" panel so the user can
    verify that the LLM correctly interpreted their intent.

    Parameters
    ----------
    searches : list[dict]
        Each dict represents one product search the user requested.
        Keys: query, category, brand, min_price, max_price, store.
    requirements : list[str]
        Specific product features / criteria the user mentioned
        (e.g. "vibration pump", "made in Italy").

    Returns
    -------
    str
        Markdown-formatted summary text.
    """
    parts: list[str] = ["## 📋 What I Understood:\n"]

    # ── Summarise each search query ─────────────────────────────────
    if searches:
        parts.append("### Products to Search:")
        for idx, search in enumerate(searches, start=1):
            query = search.get("query", "")
            parts.append(f"**{idx}. {query}**")

            # Collect optional parameters recognised by the LLM.
            params: list[str] = []
            if search.get("category"):
                params.append(f"  • Category: {search['category']}")
            if search.get("brand"):
                params.append(f"  • Brand: {search['brand']}")

            # Price range – show in a tidy "$X – $Y" format.
            low = search.get("min_price")
            high = search.get("max_price")
            if low or high:
                low_str = f"${low}" if low else "$0"
                high_str = f"${high}" if high else "∞"
                params.append(f"  • Price Range: {low_str} – {high_str}")

            if search.get("store"):
                params.append(f"  • Store: {search['store']}")

            parts.extend(params)
    else:
        parts.append("⚠️ Could not parse any search queries")

    # ── Summarise requirements ──────────────────────────────────────
    if requirements:
        parts.append("\n### Specific Requirements (I'll verify these):")
        for idx, req in enumerate(requirements, start=1):
            parts.append(f"**{idx}.** {req}")
    else:
        parts.append("\n### ℹ️ No specific requirements")
        parts.append(
            "_You didn't mention specific features to verify "
            "(like 'made in Italy', 'vibration pump', etc.)_"
        )

    return "\n".join(parts)


def _fetch_details_for_candidates(
    df: pd.DataFrame,
) -> list[tuple[str, dict]]:
    """
    Fetch full product descriptions and specs for the top-N candidates.

    Why a separate function?
    • It isolates the SerpApi "Immersive Product" calls, which are the
      most expensive part of the pipeline (each one costs 1 SerpApi
      credit).
    • We cap the number of lookups with ``MAX_DETAIL_LOOKUPS`` so a
      single user query never exhausts the API quota.

    Parameters
    ----------
    df : pd.DataFrame
        The search-results DataFrame (must contain an
        ``immersive_token`` column from the initial Shopping search).

    Returns
    -------
    list[tuple[str, dict]]
        Each tuple is (immersive_token, detail_dict).
    """
    top_products = df.head(MAX_DETAIL_LOOKUPS)
    logger.info(
        "Fetching product details for top %d candidates …",
        len(top_products),
    )

    details: list[tuple[str, dict]] = []
    for _, row in top_products.iterrows():
        token: str = row.get("immersive_token", "")
        if token:
            # ── SerpApi call #2: google_immersive_product ───────────
            # This retrieves the "about the product" section that
            # Google surfaces on its immersive product pages,
            # including a prose description and a structured
            # features / specs list.
            detail = get_product_details(token)
            details.append((token, detail))

    return details


def _evaluate_products(
    df: pd.DataFrame,
    requirements: list[str],
) -> list[dict[str, Any]]:
    """
    Run LLM Pass 2 – check each candidate against the user's
    requirements.

    This is the "AI evaluation" step that makes the assistant smarter
    than a keyword search.  The LLM reads the product description and
    spec sheet, then decides for **every** requirement whether it is
    met, not met, or unknown (data missing).

    Parameters
    ----------
    df : pd.DataFrame
        Products with descriptions & specs already attached.
    requirements : list[str]
        The specific criteria extracted by LLM Pass 1.

    Returns
    -------
    list[dict]
        One evaluation dict per row in *df*.  Keys:
        ``verdict`` ("match" | "possible" | "no_match"),
        ``notes``, ``requirements_met``.
    """
    if not requirements:
        # Nothing to check → every product is a potential match.
        return [
            {"verdict": "match", "notes": "", "requirements_met": []}
        ] * len(df)

    # Evaluate the top-N products that have full detail data.
    evaluated_count = min(MAX_DETAIL_LOOKUPS, len(df))
    products_to_eval = df.head(evaluated_count).to_dict("records")

    logger.info(
        "Evaluating %d products against %d requirements …",
        len(products_to_eval),
        len(requirements),
    )

    evaluations_top = evaluate_batch(products_to_eval, requirements)

    # Products beyond top-N were not looked up → mark as "possible"
    # (we lack information, so we cannot rule them out).
    remaining = len(df) - len(evaluations_top)
    evaluations_rest = [
        {"verdict": "possible", "notes": "", "requirements_met": []}
    ] * remaining

    return evaluations_top + evaluations_rest


def _build_status_message(
    total_products: int,
    evaluations: list[dict[str, Any]],
    requirements: list[str],
    searches: list[dict[str, Any]],
    elapsed_seconds: float,
) -> str:
    """
    Compose the one-line status bar shown beneath the product grid.

    Parameters
    ----------
    total_products : int
        Number of products returned by SerpApi.
    evaluations : list[dict]
        The evaluation results (same length as the DataFrame).
    requirements : list[str]
        User requirements (may be empty).
    searches : list[dict]
        All parsed search objects (to note if multi-search was used).
    elapsed_seconds : float
        Wall-clock time for the entire pipeline.

    Returns
    -------
    str
        A concise, human-readable status string.
    """
    parts: list[str] = [
        f"✅ Evaluated {total_products} products in {elapsed_seconds:.1f}s"
    ]

    if requirements:
        matches = sum(1 for e in evaluations if e["verdict"] == "match")
        possibles = sum(1 for e in evaluations if e["verdict"] == "possible")
        no_matches = sum(1 for e in evaluations if e["verdict"] == "no_match")
        parts.append(
            f"Results: {matches} match, {possibles} maybe, {no_matches} no match"
        )
    else:
        parts.append("(No requirements to check – all products shown)")

    if len(searches) > 1:
        parts.append("(Showing results for first search only)")

    return "\n".join(parts)


# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
#  Main Pipeline
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━


def search_products(
    user_text: str,
) -> tuple[str, str, str, pd.DataFrame]:
    """
    End-to-end pipeline:  Parse → Search → Detail → Evaluate → Render.

    This is the single callback wired to Gradio's "Shop" button.
    Gradio calls it with the contents of the text box and expects
    four return values (one per UI output component).

    Parameters
    ----------
    user_text : str
        The shopper's free-form request typed into the Gradio text box.

    Returns
    -------
    tuple
        parsed_summary : str          – Markdown for "What I Understood"
        cards_html     : str          – HTML product cards
        status_msg     : str          – One-line status bar
        comparison_df  : pd.DataFrame – Comparison table
    """
    start_time = time.time()

    # Guard clause – early return on empty input.
    if not user_text.strip():
        return (
            "No request entered.",
            "",
            "Please enter what you're looking for.",
            pd.DataFrame(),
        )

    # ── STEP 1  ·  LLM Pass 1 – Parse the natural-language request ─
    #
    #   API call: Hugging Face Inference → chat completion
    #   Purpose:  Convert free-form text into structured JSON so we
    #             can build an API query and know what to verify later.
    #
    parsed: dict[str, Any] = parse_user_request(user_text)

    searches: list[dict[str, Any]] = parsed.get("searches", [])
    requirements: list[str] = parsed.get("requirements", [])

    # Show the user exactly how the LLM interpreted their request.
    parsed_summary = _build_parsed_summary(searches, requirements)

    if not searches:
        return (
            parsed_summary,
            "",
            "No valid searches found. Try rephrasing.",
            pd.DataFrame(),
        )

    # ── STEP 2  ·  SerpApi – Google Shopping search ─────────────────
    #
    #   API call: SerpApi → engine "google_shopping"
    #   Purpose:  Retrieve candidate products (titles, prices,
    #             thumbnails) from Google Shopping.
    #
    #   Note: We do NOT pass price filters to SerpApi. SerpApi's price
    #   filtering is unreliable and limits results unpredictably.
    #   Instead, we fetch a broad result set and filter client-side
    #   in Step 3 for more accurate price-based filtering.
    #
    # NOTE: We currently handle only the *first* search object.
    # Multi-product requests (e.g. "machine AND grinder") could be
    # implemented by iterating over `searches` and merging results.
    first_search = searches[0]
    query: str = first_search.get("query", "")

    if not query:
        return parsed_summary, "", "Empty search query.", pd.DataFrame()

    raw_results = search_shopping(query)

    # ── STEP 3  ·  Normalise and filter ─────────────────────────────
    #
    #   No API call here – pure Pandas work.
    #   parse_results() flattens the nested SerpApi JSON into a tidy
    #   DataFrame; apply_filters() trims by price / sort order.
    #
    df: pd.DataFrame = parse_results(raw_results)

    if df.empty:
        return (
            parsed_summary,
            "",
            f"No products found for '{query}'. Try a different search.",
            pd.DataFrame(),
        )

    df = apply_filters(
        df,
        min_price=first_search.get("min_price"),
        max_price=first_search.get("max_price"),
        sort_by=first_search.get("sort_by", "relevance"),
    )

    # ── STEP 4  ·  Fetch detailed product info (specs / description)─
    #
    #   API call: SerpApi → engine "google_immersive_product"
    #   Purpose:  Get the full product description and feature list so
    #             the evaluator LLM has enough context to verify the
    #             user's requirements.
    #
    #   We only do this when there are requirements to check, and we
    #   cap the number of calls at MAX_DETAIL_LOOKUPS (default 10)
    #   to keep SerpApi usage predictable.
    #
    if requirements:
        details_list = _fetch_details_for_candidates(df)
        df = attach_details(df, details_list)

    # ── STEP 5  ·  LLM Pass 2 – Evaluate candidates ────────────────
    #
    #   API call: Hugging Face Inference → chat completion (one per
    #             product).
    #   Purpose:  For every top-N product, the LLM reads the
    #             description/specs and checks each requirement:
    #             met / not_met / unknown.
    #
    evaluations = _evaluate_products(df, requirements)

    # ── STEP 6  ·  Render the UI ───────────────────────────────────
    #
    #   No API calls – this step builds HTML cards and a Pandas
    #   DataFrame that Gradio renders in the browser.
    #
    cards_html: str = build_product_cards(df, evaluations)
    comparison_df: pd.DataFrame = build_comparison_table(df, evaluations)

    elapsed = time.time() - start_time
    status_msg = _build_status_message(
        total_products=len(df),
        evaluations=evaluations,
        requirements=requirements,
        searches=searches,
        elapsed_seconds=elapsed,
    )

    return parsed_summary, cards_html, status_msg, comparison_df


# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
#  Gradio User Interface
#  ─────────────────────
#  Gradio is a Python library that turns a function into a web app.
#  `gr.Blocks` gives full layout control (rows, columns, tabs).
#  Each UI widget (Textbox, Button, HTML, …) is an *input* or
#  *output* that Gradio wires to a Python callback with `.click()`.
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

with gr.Blocks(title="Shopping Assistant") as demo:

    # ── Header ──────────────────────────────────────────────────────
    gr.Markdown("# 🛒 Shopping Assistant")
    gr.Markdown(
        "Tell me what you're looking for in plain English. "
        "I'll understand your requirements and search for matching products."
    )

    # ── Input area ──────────────────────────────────────────────────
    with gr.Row():
        query_input = gr.Textbox(
            label="What are you looking for?",
            placeholder=(
                'Example: "I want a prosumer espresso machine from '
                "Italy or Spain. I also want a grinder for the beans. "
                "The machine should use a vibration pump and have a "
                'water reservoir that holds at least one liter."'
            ),
            lines=5,
        )

    search_button = gr.Button("Shop", variant="primary", size="lg")

    # ── Output components ───────────────────────────────────────────
    # Gradio renders each component in the order declared here.
    products_html = gr.HTML(label="Product Results")
    status_output = gr.Textbox(label="Status", interactive=False)
    results_output = gr.Dataframe(
        label="Comparison Table",
        wrap=True,
        interactive=False,
    )
    parsed_output = gr.Markdown(label="What I Understood", value="")

    # ── Wire the button to the pipeline ─────────────────────────────
    #
    #   Gradio's `.click()` connects a UI event (button press) to a
    #   Python function.  `inputs` lists the widgets whose current
    #   values are passed as arguments; `outputs` lists the widgets
    #   that receive the function's return values (in order).
    #
    search_button.click(
        fn=search_products,
        inputs=[query_input],
        outputs=[parsed_output, products_html, status_output, results_output],
    )

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
#  Entry Point
#  ───────────
#  The `if __name__` guard is a Python best practice.  It ensures the
#  server only starts when you run `python app.py` directly — not
#  when another module imports this file.
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

if __name__ == "__main__":
    # launch() starts a local web server (default http://127.0.0.1:7860).
    # On Hugging Face Spaces, Gradio auto-detects the environment and
    # binds to the correct port without extra configuration.
    demo.launch()