app.py

# -*- coding: utf-8 -*-
import os
import time
import uuid
import requests
import pandas as pd
import gradio as gr
from bs4 import BeautifulSoup

# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
HEADERS = {
    "User-Agent": "AuditorLocalProCristobal_v1_2026 (cristobal_dev_leads@outlook.com)",
    "Accept-Language": "en"
}
NOMINATIM_EMAIL = os.environ.get("NOMINATIM_EMAIL", "")
CSV_FOLDER = "generated_csv"
os.makedirs(CSV_FOLDER, exist_ok=True)

# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
def _make_csv_path() -> str:
    """Create a unique CSV file path."""
    unique_id = uuid.uuid4().hex
    return os.path.join(CSV_FOLDER, f"leads_{unique_id}.csv")

def _sanitize_url(url: str) -> str:
    """Ensure the URL has a scheme and strip whitespace."""
    url = url.strip()
    if not url:
        return ""
    if not (url.startswith("http://") or url.startswith("https://")):
        url = "http://" + url
    return url

def _df_to_html(df: pd.DataFrame) -> str:
    """Convert a DataFrame to an HTML table with clickable website links."""
    if df.empty:
        return "<p>No data to display.</p>"
    html = ["<table style='width:100%;border-collapse:collapse;'>"]
    # Header
    html.append("<tr>")
    for col in df.columns:
        html.append(f"<th style='border:1px solid #ddd;padding:8px;text-align:left;'>{col}</th>")
    html.append("</tr>")
    # Rows
    for _, row in df.iterrows():
        html.append("<tr>")
        for col in df.columns:
            cell = row[col]
            if col.lower() == "website" and isinstance(cell, str) and cell.startswith("http"):
                cell = f"<a href='{cell}' target='_blank'>{cell}</a>"
            html.append(f"<td style='border:1px solid #ddd;padding:8px;'>{cell}</td>")
        html.append("</tr>")
    html.append("</table>")
    return "\n".join(html)

def _request_with_retry(url: str, params: dict = None, max_retries: int = 3, backoff_factor: float = 1.5):
    """
    Perform a GET request with simple exponential backoff on HTTP 429.
    Returns a requests.Response object or raises the last exception.
    """
    attempt = 0
    last_err = None
    while attempt < max_retries:
        try:
            resp = requests.get(url, headers=HEADERS, params=params, timeout=10)
            if resp.status_code == 429:
                # Too many requests – wait and retry
                wait = backoff_factor ** attempt
                time.sleep(wait)
                attempt += 1
                continue
            return resp
        except Exception as e:
            last_err = e
            time.sleep(backoff_factor)
            attempt += 1
    # If we exit the loop without returning, raise the last error
    if last_err:
        raise last_err
    raise RuntimeError("Request failed after retries without a captured exception.")

# ----------------------------------------------------------------------
# Core functions
# ----------------------------------------------------------------------
def auditar_negocios_space(query: str):
    """
    Search OpenStreetMap for places matching the query, audit their websites,
    and return a CSV file path, a Dataframe for preview, and the CSV content as text.
    """
    query = query.strip()
    if not query:
        error_msg = "Please provide a non-empty search term."
        empty_df = pd.DataFrame({"Error": [error_msg]})
        return None, empty_df, "", empty_df

    # 1. Search OSM via Nominatim
    params = {
        "q": query,
        "format": "json",
        "extratags": "1",
        "limit": "30"
    }
    if NOMINATIM_EMAIL:
        params["email"] = NOMINATIM_EMAIL

    nominatim_url = "https://nominatim.openstreetmap.org/search"
    try:
        resp = _request_with_retry(nominatim_url, params=params)
        if resp.status_code != 200:
            error_msg = f"Map service returned status {resp.status_code}. Please try again later."
            empty_df = pd.DataFrame({"Error": [error_msg]})
            return None, empty_df, "", empty_df
        results = resp.json()
    except Exception as e:
        error_msg = f"Map request error: {str(e)}"
        empty_df = pd.DataFrame({"Error": [error_msg]})
        return None, empty_df, "", empty_df

    if not results:
        error_msg = "No results found for the given query."
        empty_df = pd.DataFrame({"Error": [error_msg]})
        return None, empty_df, "", empty_df

    # 2. Build list of businesses
    businesses = []
    for item in results:
        extratags = item.get("extratags") or {}
        website = (
            extratags.get("website")
            or extratags.get("contact:website")
            or "No website"
        )
        phone = (
            extratags.get("phone")
            or extratags.get("contact:phone")
            or "No phone"
        )
        full_name = item.get("display_name", "")
        short_name = full_name.split(",")[0] if full_name else "Local Business"

        businesses.append({
            "Name": short_name,
            "Phone": phone,
            "Website": website,
            "Address": full_name
        })

    # 3. Audit each website (polite pause + progress)
    for b in businesses:
        web = b["Website"]
        if web == "No website":
            b["Web Audit"] = "NO WEBSITE"
        else:
            try:
                site_resp = requests.get(web, headers=HEADERS, timeout=5)
                if site_resp.status_code == 200:
                    b["Web Audit"] = "ACTIVE"
                else:
                    b["Web Audit"] = f"HTTP {site_resp.status_code}"
            except Exception:
                b["Web Audit"] = "UNREACHABLE"
        time.sleep(0.2)

    # 4. Create DataFrame and CSV file
    df = pd.DataFrame(businesses)
    csv_path = _make_csv_path()
    df.to_csv(csv_path, index=False, encoding="utf-8")
    csv_text = df.to_csv(index=False, encoding="utf-8")

    return csv_path, df, csv_text, df  # df is stored as hidden state

def analyze_website(url: str):
    """
    Perform a simple analysis of the given website URL.
    Returns a multiline string with status, response time, title and basic issues.
    """
    url = _sanitize_url(url)
    if not url:
        return "Please provide a URL to analyze."

    try:
        start = time.time()
        resp = requests.get(url, headers=HEADERS, timeout=10)
        elapsed = time.time() - start
        status = resp.status_code
        size_kb = len(resp.content) / 1024

        soup = BeautifulSoup(resp.text, "html.parser")
        title_tag = (
            soup.title.string.strip()
            if soup.title and soup.title.string
            else "No title found"
        )

        issues = []
        if status != 200:
            issues.append(f"Unexpected HTTP status: {status}")
        if size_kb < 10:
            issues.append("Page size is unusually small (<10KB)")
        if not soup.title:
            issues.append("Missing <title> tag")

        issues_text = "\n".join(issues) if issues else "No obvious issues detected."

        report = (
            f"URL: {url}\n"
            f"Status: {status}\n"
            f"Response time: {elapsed:.2f} seconds\n"
            f"Size: {size_kb:.1f} KB\n"
            f"Title: {title_tag}\n"
            f"Issues:\n{issues_text}"
        )
        return report
    except Exception as e:
        return f"Error accessing the site: {str(e)}"

def generate_prompts(df: pd.DataFrame):
    """
    Generate a simple website creation prompt for each business that lacks a website.
    Returns a markdown string with one prompt per missing entry.
    """
    if df is None or df.empty:
        return "No data available. Run a search first."

    prompts = []
    for _, row in df.iterrows():
        if row.get("Website", "").lower() == "no website":
            name = row.get("Name", "Business")
            address = row.get("Address", "Address not provided")
            phone = row.get("Phone", "Phone not provided")
            prompt = (
                f"Create a clean one-page website for **{name}**.\n"
                f"- Location: {address}\n"
                f"- Contact phone: {phone}\n"
                f"- Suggested sections: About, Services, Contact.\n"
                f"- Use a modern, responsive design with placeholder text.\n"
                f"- Include a simple contact form.\n"
            )
            prompts.append(prompt)

    if not prompts:
        return "All businesses already have a website."

    return "\n---\n".join(prompts)

# ----------------------------------------------------------------------
# Gradio Interface
# ----------------------------------------------------------------------
css = """
.gradio-container {font-family: Arial, Helvetica, sans-serif;}
"""

with gr.Blocks() as demo:
    gr.Markdown("# Extractor and Web Auditor for Local Businesses")

    # Hidden state to keep the latest DataFrame
    df_state = gr.State()

    with gr.Tabs():
        # ------------------------------------------------------------------
        # Tab 1: Search & Audit
        # ------------------------------------------------------------------
        with gr.TabItem("Search & Audit"):
            gr.Markdown(
                "Enter a search term (e.g. \"dentist in Madrid\" or \"bakery in Barcelona\") "
                "to generate a list of potential leads instantly."
            )
            with gr.Row():
                query_input = gr.Textbox(
                    label="Search term",
                    placeholder="e.g. coffee shops in Valencia",
                    lines=1
                )
                search_btn = gr.Button("Search and Audit", variant="primary")
                clear_search_btn = gr.Button("Clear", variant="secondary")
            with gr.Row():
                file_output = gr.File(label="Download CSV")
            df_output = gr.Dataframe(
                label="Preview of results (click a row to select)",
                interactive=True,
                headers=None
            )
            html_output = gr.HTML(label="Clickable website links")
            copy_code = gr.Code(
                label="Copy CSV content",
                language="markdown"
            )
            # Search action
            search_btn.click(
                fn=auditar_negocios_space,
                inputs=query_input,
                outputs=[file_output, df_output, copy_code, df_state]
            )
            # Update HTML when Dataframe changes
            df_output.change(
                fn=lambda df: _df_to_html(pd.DataFrame(df) if isinstance(df, list) else df),
                inputs=df_output,
                outputs=html_output
            )
            # Clear action
            clear_search_btn.click(
                fn=lambda: (None, pd.DataFrame(), "", pd.DataFrame()),
                inputs=None,
                outputs=[file_output, df_output, copy_code, df_state]
            )
            clear_search_btn.click(
                fn=lambda: "",
                inputs=None,
                outputs=html_output
            )

        # ------------------------------------------------------------------
        # Tab 2: Website Analysis
        # ------------------------------------------------------------------
        with gr.TabItem("Website Analysis"):
            gr.Markdown(
                "Paste a website URL to get a quick health report. "
                "The analysis checks HTTP status, response time, page size, title tag "
                "and reports simple issues."
            )
            with gr.Row():
                url_input = gr.Textbox(
                    label="Website URL",
                    placeholder="e.g. https://example.com",
                    lines=1
                )
                analyze_btn = gr.Button("Analyze", variant="primary")
                clear_analyze_btn = gr.Button("Clear", variant="secondary")
            analysis_output = gr.Textbox(
                label="Analysis Report",
                lines=10,
                interactive=False
            )
            # Analyze action
            analyze_btn.click(
                fn=analyze_website,
                inputs=url_input,
                outputs=analysis_output
            )
            # Clear action
            clear_analyze_btn.click(
                fn=lambda: "",
                inputs=None,
                outputs=analysis_output
            )

        # ------------------------------------------------------------------
        # Tab 3: Prompt Generator for Missing Websites
        # ------------------------------------------------------------------
        with gr.TabItem("Prompt Generator"):
            gr.Markdown(
                "Generate a ready-to-use prompt to create a simple website for each business "
                "that currently has no website."
            )
            with gr.Row():
                gen_prompt_btn = gr.Button("Generate Prompts", variant="primary")
                clear_prompt_btn = gr.Button("Clear", variant="secondary")
            prompt_output = gr.Code(
                label="Generated Prompts",
                language="markdown"
            )
            # Generate prompts action
            gen_prompt_btn.click(
                fn=generate_prompts,
                inputs=df_state,
                outputs=prompt_output
            )
            # Clear prompts action
            clear_prompt_btn.click(
                fn=lambda: "",
                inputs=None,
                outputs=prompt_output
            )

if __name__ == "__main__":
    demo.launch(theme=gr.themes.Soft(), css=css)