app.py

# -*- coding: utf-8 -*-
"""
Gradio Screenshot-zu-SugarCRM Lead-Bot.

Set these as environment variables or Hugging Face Spaces Secrets:
    OPENAI_API_KEY
    GLS_NAME
    GLS_PASSWORD

Optional Colab Secrets:
    OPENAI_MODEL        default: gpt-4o
    SUGAR_BASE_URL      default: https://glsps.sugaropencloud.eu/rest/v11_27/
    SUGAR_PLATFORM      default: base
    SUGAR_CLIENT_ID     default: sugar
    SUGAR_CLIENT_SECRET default: empty
    APP_USER / APP_PASS for basic auth on the app
"""

from __future__ import annotations

import base64
import json
import mimetypes
import os
import re
import shutil
import time
import traceback
from datetime import datetime, time as date_time, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import urljoin
from zoneinfo import ZoneInfo

import gradio as gr
import pandas as pd
import requests
import spaces
from openai import OpenAI

try:
    from google.colab import userdata  # type: ignore
except Exception:
    userdata = None


DEFAULT_SUGAR_BASE_URL = "https://glsps.sugaropencloud.eu/rest/v11_27/"
DEFAULT_SUGAR_PLATFORM = "base"
DEFAULT_OPENAI_MODEL = "gpt-4o"
DEFAULT_LEAD_STATUS_LABEL = "Kontaktiert"
DEFAULT_LEAD_STATUS_VALUE = "In Process"
VISIT_TIMEZONE = "Europe/Berlin"
VISIT_START_HOUR = 10
VISIT_START_MINUTE = 0
VISIT_DURATION_MINUTES = 5
VISIT_GAP_MINUTES = 20
SCREENSHOT_DIR = Path("screenshots")
SCREENSHOT_DIR.mkdir(exist_ok=True)

DISPLAY_COLUMNS = [
    "Firma",
    "Strasse",
    "PLZ",
    "Ort",
    "Land",
    "Branche",
    "Empfehlung",
    "Lead-Typ",
    "Status",
    "Telefon",
]

EXTRACTION_SCHEMA = {
    "name": "lead_from_screenshot",
    "schema": {
        "type": "object",
        "additionalProperties": False,
        "properties": {
            "company_name": {"type": "string"},
            "street": {"type": "string"},
            "postal_code": {"type": "string"},
            "city": {"type": "string"},
            "country": {"type": "string"},
            "phone": {"type": "string"},
        },
        "required": [
            "company_name",
            "street",
            "postal_code",
            "city",
            "country",
            "phone",
        ],
    },
    "strict": True,
}


def secret(name: str, default: str = "") -> str:
    if userdata is not None:
        try:
            value = userdata.get(name)
            if value:
                return str(value)
        except Exception:
            pass
    return os.environ.get(name, default)


def normalize_base_url(value: str) -> str:
    value = (value or DEFAULT_SUGAR_BASE_URL).strip()
    return value if value.endswith("/") else value + "/"


def log_line(message: str) -> str:
    return f"[{time.strftime('%H:%M:%S')}] {message}"


def file_to_data_url(path: str) -> str:
    mime = mimetypes.guess_type(path)[0] or "image/png"
    with open(path, "rb") as handle:
        encoded = base64.b64encode(handle.read()).decode("ascii")
    return f"data:{mime};base64,{encoded}"


def clean_text(value: Any) -> str:
    return re.sub(r"\s+", " ", str(value or "")).strip()


def normalize_match_text(value: Any) -> str:
    return re.sub(r"[^a-z0-9]+", "", clean_text(value).casefold())


def normalize_phone(value: Any) -> str:
    return re.sub(r"\D+", "", str(value or ""))


def normalize_lead_status(value: Any) -> str:
    status = clean_text(value)
    if not status:
        return DEFAULT_LEAD_STATUS_VALUE
    return {
        "Neu": "New",
        "New": "New",
        "Kontaktiert": DEFAULT_LEAD_STATUS_VALUE,
        "Contacted": DEFAULT_LEAD_STATUS_VALUE,
        DEFAULT_LEAD_STATUS_VALUE: DEFAULT_LEAD_STATUS_VALUE,
    }.get(status, status)


def default_credentials(
    openai_key: str = "",
    sugar_user: str = "",
    sugar_password: str = "",
    sugar_base_url: str = "",
    sugar_platform: str = "",
    sugar_client_id: str = "",
    sugar_client_secret: str = "",
) -> dict[str, str]:
    return {
        "openai_key": clean_text(openai_key) or secret("OPENAI_API_KEY"),
        "openai_model": secret("OPENAI_MODEL", DEFAULT_OPENAI_MODEL),
        "sugar_user": clean_text(sugar_user) or secret("GLS_NAME"),
        "sugar_password": str(sugar_password or "") or secret("GLS_PASSWORD"),
        "sugar_base_url": normalize_base_url(clean_text(sugar_base_url) or secret("SUGAR_BASE_URL", DEFAULT_SUGAR_BASE_URL)),
        "sugar_platform": clean_text(sugar_platform) or secret("SUGAR_PLATFORM", DEFAULT_SUGAR_PLATFORM),
        "sugar_client_id": clean_text(sugar_client_id) or secret("SUGAR_CLIENT_ID", "sugar"),
        "sugar_client_secret": str(sugar_client_secret or "") or secret("SUGAR_CLIENT_SECRET", ""),
    }


def get_crm_session(cfg: dict[str, str]) -> dict[str, Any]:
    payload = {
        "grant_type": "password",
        "client_id": cfg["sugar_client_id"],
        "client_secret": cfg["sugar_client_secret"],
        "username": cfg["sugar_user"],
        "password": cfg["sugar_password"],
        "platform": cfg["sugar_platform"],
    }
    response = requests.post(
        urljoin(cfg["sugar_base_url"], "oauth2/token"),
        json=payload,
        headers={"Content-Type": "application/json", "Cache-Control": "no-cache"},
        timeout=30,
    )
    raise_for_sugar_status(response)
    data = response.json()
    if not data.get("access_token"):
        raise RuntimeError("SugarCRM did not return an access_token.")
    return data


def sugar_headers(access_token: str) -> dict[str, str]:
    return {
        "OAuth-Token": access_token,
        "Content-Type": "application/json",
        "Cache-Control": "no-cache",
    }


def raise_for_sugar_status(response: requests.Response) -> None:
    try:
        response.raise_for_status()
    except requests.HTTPError as exc:
        details = clean_text(response.text)
        if details:
            raise RuntimeError(f"{exc}; SugarCRM response: {details}") from exc
        raise


def validate_sugar_login(*args) -> str:
    cfg = default_credentials(*args)
    missing = [name for name in ("sugar_user", "sugar_password") if not cfg[name]]
    if missing:
        return log_line(f"SugarCRM Zugangsdaten fehlen: {', '.join(missing)}")
    try:
        token = get_crm_session(cfg)
        expires = token.get("expires_in", "unknown")
        return "\n".join([
            log_line("SugarCRM Verbindung OK."),
            log_line(f"REST Basis-URL: {cfg['sugar_base_url']}"),
            log_line(f"Platform: {cfg['sugar_platform']}"),
            log_line(f"Token läuft ab in: {expires} Sekunden"),
        ])
    except Exception as exc:
        return "\n".join([
            log_line("SugarCRM Verbindung fehlgeschlagen."),
            str(exc),
            "",
            "Falls das Konto nur SugarIdentity/SSO nutzt, braucht es API-Zugriff oder einen Integrationsbenutzer.",
        ])


def extract_from_image(path: str, client: OpenAI, model: str) -> dict[str, str]:
    prompt = (
        "Extrahiere aus diesem Screenshot einen möglichen deutschen ParcelShop-Lead. "
        "Lies nur Firmenname, Adresse und Telefonnummer aus. "
        "Website, E-Mail und Notizen nicht extrahieren. "
        "Nutze leere Strings für fehlende Werte. Schreibe den Geschäfts- oder Shopnamen in company_name. "
        "Teile die Adresse in street, postal_code, city und country. Bevorzuge country 'Deutschland' für deutsche Adressen."
    )
    response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {"type": "image_url", "image_url": {"url": file_to_data_url(path), "detail": "high"}},
                ],
            }
        ],
        response_format={"type": "json_schema", "json_schema": EXTRACTION_SCHEMA},
    )
    content = response.choices[0].message.content or "{}"
    data = json.loads(content)
    return {key: clean_text(data.get(key, "")) for key in EXTRACTION_SCHEMA["schema"]["properties"]}


def prepare_payload(row: dict[str, Any]) -> dict[str, Any]:
    company_name = clean_text(row.get("Firma"))
    if not company_name:
        raise ValueError("Firma ist erforderlich.")

    payload = {
        "salutation": "Comp",
        "first_name": "leer",
        "last_name": "leer",
        "account_name": company_name,
        "status": normalize_lead_status(row.get("Status")),
        "primary_address_street": clean_text(row.get("Strasse")),
        "primary_address_postalcode": clean_text(row.get("PLZ")),
        "primary_address_city": clean_text(row.get("Ort")),
        "primary_address_country": clean_text(row.get("Land")) or "Deutschland",
        "list_industry_belonging_c": clean_text(row.get("Branche")) or "Other",
        "empfehlung_von_c": clean_text(row.get("Empfehlung")) or "Eigenrecherche",
        "lead_type_c": clean_text(row.get("Lead-Typ")) or "ParcelShop",
        "phone_work": clean_text(row.get("Telefon")),
    }

    return {key: value for key, value in payload.items() if value not in ("", None)}


def send_to_crm(payload: dict[str, Any], access_token: str, cfg: dict[str, str]) -> dict[str, Any]:
    return create_crm_record("Leads", payload, access_token, cfg)


def create_crm_record(module: str, payload: dict[str, Any], access_token: str, cfg: dict[str, str]) -> dict[str, Any]:
    response = requests.post(
        urljoin(cfg["sugar_base_url"], module),
        headers=sugar_headers(access_token),
        json=payload,
        timeout=30,
    )
    raise_for_sugar_status(response)
    return response.json()


def update_crm_record(module: str, record_id: str, payload: dict[str, Any], access_token: str, cfg: dict[str, str]) -> dict[str, Any]:
    response = requests.put(
        urljoin(cfg["sugar_base_url"], f"{module}/{record_id}"),
        headers=sugar_headers(access_token),
        json=payload,
        timeout=30,
    )
    raise_for_sugar_status(response)
    return response.json()


def get_current_sugar_user(access_token: str, cfg: dict[str, str]) -> dict[str, str]:
    response = requests.get(
        urljoin(cfg["sugar_base_url"], "me"),
        headers=sugar_headers(access_token),
        timeout=30,
    )
    raise_for_sugar_status(response)
    user = response.json().get("current_user") or {}
    user_id = clean_text(user.get("id"))
    if not user_id:
        raise RuntimeError("SugarCRM konnte den aktuellen Benutzer nicht ermitteln.")
    return {
        "id": user_id,
        "name": clean_text(user.get("full_name")) or clean_text(user.get("user_name")),
    }


def visit_times(slot_index: int) -> tuple[datetime, datetime]:
    timezone = ZoneInfo(VISIT_TIMEZONE)
    today = datetime.now(timezone).date()
    start = datetime.combine(
        today,
        date_time(VISIT_START_HOUR, VISIT_START_MINUTE),
        tzinfo=timezone,
    )
    start += timedelta(minutes=slot_index * (VISIT_DURATION_MINUTES + VISIT_GAP_MINUTES))
    end = start + timedelta(minutes=VISIT_DURATION_MINUTES)
    return start, end


def visit_subject(row: dict[str, Any]) -> str:
    parts = [
        "A",
        clean_text(row.get("Firma")),
        clean_text(row.get("PLZ")),
        clean_text(row.get("Ort")),
    ]
    return "_".join(re.sub(r"\s+", "_", part) for part in parts if part)


def prepare_visit_payload(
    row: dict[str, Any],
    lead_record: dict[str, Any],
    assigned_user: dict[str, str],
    slot_index: int,
) -> dict[str, Any]:
    lead_id = clean_text(lead_record.get("id"))
    if not lead_id:
        raise ValueError("Lead wurde erstellt, aber SugarCRM hat keine Lead-ID zurueckgegeben.")

    start, end = visit_times(slot_index)
    parent_name = clean_text(lead_record.get("name")) or clean_text(row.get("Firma"))
    user_name = clean_text(assigned_user.get("name"))

    payload = {
        "name": visit_subject(row),
        "status": "Held",
        "date_start": start.isoformat(),
        "date_end": end.isoformat(),
        "duration_hours": 0,
        "duration_minutes": VISIT_DURATION_MINUTES,
        "parent_type": "Leads",
        "parent_id": lead_id,
        "parent_name": parent_name,
        "assigned_user_id": assigned_user["id"],
        "assigned_user_name": user_name,
        "contact_type_c": "in_person",
        "aa_select_visit_type_c": "A",
        "aa_visit_reason_all_c": "Kaltakquise",
        "description": "Visitenkarte & Flyer hinterlassen",
        "aa_abwicklung_radio_c": "it_usage",
        "aa_abwicklung_auswahl_c": ["it_usage"],
        "operative_probleme_c": "Nein",
        "technik_funktionsfaehig_c": "Ja",
        "sonstige_probleme_c": "Nein",
    }
    return {key: value for key, value in payload.items() if value not in ("", None)}


def create_visit_report(
    row: dict[str, Any],
    lead_record: dict[str, Any],
    assigned_user: dict[str, str],
    slot_index: int,
    access_token: str,
    cfg: dict[str, str],
) -> dict[str, Any]:
    payload = prepare_visit_payload(row, lead_record, assigned_user, slot_index)
    created = create_crm_record("Meetings", payload, access_token, cfg)
    meeting_id = clean_text(created.get("id"))
    if not meeting_id:
        raise RuntimeError("SugarCRM hat keine Besuchsbericht-ID zurueckgegeben.")

    # A second save lets Sugar recalculate relationship-based subject fields.
    return update_crm_record("Meetings", meeting_id, payload, access_token, cfg)


def find_existing_leads(payload: dict[str, Any], access_token: str, cfg: dict[str, str]) -> list[dict[str, Any]]:
    clauses: list[dict[str, Any]] = []
    account_name = clean_text(payload.get("account_name"))
    street = clean_text(payload.get("primary_address_street"))
    postal_code = clean_text(payload.get("primary_address_postalcode"))
    phone = clean_text(payload.get("phone_work"))

    if account_name:
        clauses.append({"account_name": {"$contains": account_name}})
    if street and postal_code:
        clauses.append({
            "$and": [
                {"primary_address_street": {"$contains": street}},
                {"primary_address_postalcode": {"$equals": postal_code}},
            ]
        })
    if phone:
        clauses.append({"phone_work": {"$contains": phone}})

    if not clauses:
        return []

    body = {
        "filter": [{"$or": clauses}],
        "max_num": 10,
        "offset": 0,
        "fields": "id,account_name,primary_address_street,primary_address_postalcode,primary_address_city,phone_work,status,date_modified",
        "order_by": "date_modified:DESC",
        "favorites": False,
        "my_items": False,
    }
    response = requests.post(
        urljoin(cfg["sugar_base_url"], "Leads/filter"),
        headers=sugar_headers(access_token),
        json=body,
        timeout=30,
    )
    raise_for_sugar_status(response)
    return response.json().get("records", [])


def is_duplicate_lead(payload: dict[str, Any], existing: dict[str, Any]) -> bool:
    payload_name = normalize_match_text(payload.get("account_name"))
    existing_name = normalize_match_text(existing.get("account_name"))
    payload_street = normalize_match_text(payload.get("primary_address_street"))
    existing_street = normalize_match_text(existing.get("primary_address_street"))
    payload_plz = normalize_match_text(payload.get("primary_address_postalcode"))
    existing_plz = normalize_match_text(existing.get("primary_address_postalcode"))
    payload_city = normalize_match_text(payload.get("primary_address_city"))
    existing_city = normalize_match_text(existing.get("primary_address_city"))
    payload_phone = normalize_phone(payload.get("phone_work"))
    existing_phone = normalize_phone(existing.get("phone_work"))

    if payload_phone and existing_phone and len(payload_phone) >= 7:
        if payload_phone[-7:] == existing_phone[-7:] or payload_phone == existing_phone:
            return True

    same_name = payload_name and existing_name and payload_name == existing_name
    same_address = (
        payload_street
        and existing_street
        and payload_plz
        and existing_plz
        and payload_street == existing_street
        and payload_plz == existing_plz
    )
    same_city = payload_city and existing_city and payload_city == existing_city

    if same_name and (same_address or same_city):
        return True
    if same_address:
        return True
    return False


def duplicate_summary(record: dict[str, Any]) -> str:
    parts = [
        clean_text(record.get("account_name")),
        clean_text(record.get("primary_address_street")),
        clean_text(record.get("primary_address_postalcode")),
        clean_text(record.get("primary_address_city")),
    ]
    label = ", ".join(part for part in parts if part)
    record_id = clean_text(record.get("id")) or "unbekannte ID"
    return f"{label or 'bestehender Lead'} ({record_id})"


def dataframe_from_rows(rows: list[dict[str, Any]]) -> pd.DataFrame:
    df = pd.DataFrame(rows)
    for column in DISPLAY_COLUMNS:
        if column not in df.columns:
            df[column] = ""
    return df[DISPLAY_COLUMNS]


@spaces.GPU(duration=60)
def on_run(files, *credential_args):
    terminal: list[str] = []
    if not files:
        return log_line("Keine Dateien hochgeladen. Bitte zuerst Screenshots hochladen."), dataframe_from_rows([])

    cfg = default_credentials(*credential_args)
    if not cfg["openai_key"]:
        return log_line("OPENAI_API_KEY fehlt. Bitte als Secret eintragen."), dataframe_from_rows([])

    client = OpenAI(api_key=cfg["openai_key"])
    rows: list[dict[str, Any]] = []
    terminal.append(log_line(f"Verarbeite {len(files)} Screenshot(s)."))

    for index, file_obj in enumerate(files, start=1):
        path = getattr(file_obj, "name", file_obj)
        terminal.append(log_line(f"Extrahiere {index}/{len(files)}: {Path(path).name}"))
        try:
            extracted = extract_from_image(path, client, cfg["openai_model"])
            row = {
                "Firma": extracted.get("company_name", ""),
                "Strasse": extracted.get("street", ""),
                "PLZ": extracted.get("postal_code", ""),
                "Ort": extracted.get("city", ""),
                "Land": extracted.get("country", "") or "Deutschland",
                "Branche": "Other",
                "Empfehlung": "Eigenrecherche",
                "Lead-Typ": "ParcelShop",
                "Status": DEFAULT_LEAD_STATUS_LABEL,
                "Telefon": extracted.get("phone", ""),
            }
            rows.append(row)
            shutil.copy2(path, SCREENSHOT_DIR / Path(path).name)
            terminal.append(log_line(f"Erkannt: {row.get('Firma') or '(kein Firmenname)'}"))
        except Exception as exc:
            terminal.append(log_line(f"Extraktion fehlgeschlagen für {Path(path).name}: {exc}"))
            terminal.append(traceback.format_exc())

    if rows:
        terminal.append(log_line("Bitte Tabelle prüfen und bei Bedarf korrigieren. Es wurde noch nichts in SugarCRM gespeichert."))
    else:
        terminal.append(log_line("Keine verwertbaren Einträge erkannt."))

    return "\n".join(terminal), dataframe_from_rows(rows)


def coerce_records(table_value: Any) -> list[dict[str, Any]]:
    if table_value is None:
        return []
    if isinstance(table_value, pd.DataFrame):
        return table_value.fillna("").to_dict(orient="records")
    if isinstance(table_value, list):
        return pd.DataFrame(table_value, columns=DISPLAY_COLUMNS).fillna("").to_dict(orient="records")
    return pd.DataFrame(table_value).fillna("").to_dict(orient="records")


def on_send(table_value, *credential_args):
    rows = coerce_records(table_value)
    rows = [row for row in rows if clean_text(row.get("Firma"))]
    if not rows:
        return log_line("Keine Zeilen mit Firma zum Senden vorhanden.")

    cfg = default_credentials(*credential_args)
    missing = [name for name in ("sugar_user", "sugar_password") if not cfg[name]]
    if missing:
        return log_line(f"SugarCRM Zugangsdaten fehlen: {', '.join(missing)}")

    terminal = [log_line(f"Melde bei SugarCRM an: {cfg['sugar_base_url']}")]
    try:
        token = get_crm_session(cfg)
        access_token = token["access_token"]
    except Exception as exc:
        return "\n".join([
            *terminal,
            log_line(f"SugarCRM Anmeldung fehlgeschlagen: {exc}"),
            "Es wurden keine Einträge gesendet.",
        ])

    try:
        assigned_user = get_current_sugar_user(access_token, cfg)
        terminal.append(log_line(f"Besuchsberichte werden zugewiesen an: {assigned_user['name']}"))
    except Exception as exc:
        return "\n".join([
            *terminal,
            log_line(f"Aktueller SugarCRM Benutzer konnte nicht gelesen werden: {exc}"),
            "Es wurden keine Eintraege gesendet.",
        ])

    sent = 0
    skipped = 0
    visits_created = 0
    visits_failed = 0
    visit_slot = 0
    for index, row in enumerate(rows, start=1):
        try:
            payload = prepare_payload(row)
            terminal.append(log_line(f"Prüfe Duplikate {index}/{len(rows)}: {payload['account_name']}"))
            matches = find_existing_leads(payload, access_token, cfg)
            duplicate = next((record for record in matches if is_duplicate_lead(payload, record)), None)
            if duplicate:
                terminal.append(log_line(f"Übersprungen, bereits vorhanden: {duplicate_summary(duplicate)}"))
                skipped += 1
                continue

            terminal.append(log_line(f"Sende {index}/{len(rows)}: {payload['account_name']}"))
            response = send_to_crm(payload, access_token, cfg)
            terminal.append(log_line(f"Lead erstellt, ID: {response.get('id', 'unbekannt')}"))
            sent += 1

            try:
                start, end = visit_times(visit_slot)
                visit = create_visit_report(row, response, assigned_user, visit_slot, access_token, cfg)
                terminal.append(log_line(
                    "Besuchsbericht erstellt "
                    f"({start.strftime('%H:%M')}-{end.strftime('%H:%M')}), "
                    f"ID: {visit.get('id', 'unbekannt')}"
                ))
                visits_created += 1
            except Exception as exc:
                visits_failed += 1
                terminal.append(log_line(f"Besuchsbericht konnte nicht erstellt werden: {exc}"))
            finally:
                visit_slot += 1
        except Exception as exc:
            terminal.append(log_line(f"Fehler in Zeile {index}: {exc}"))

    terminal.append(log_line(
        "Fertig. "
        f"Leads erstellt: {sent}, "
        f"Besuchsberichte erstellt: {visits_created}, "
        f"Besuchsberichte fehlgeschlagen: {visits_failed}, "
        f"übersprungen: {skipped}, "
        f"geprüft: {len(rows)}."
    ))
    return "\n".join(terminal)


with gr.Blocks() as demo:
    gr.Markdown("## GLS Lead Bot")

    openai_key = gr.State("")
    sugar_user = gr.State("")
    sugar_password = gr.State("")
    sugar_base_url = gr.State("")
    sugar_platform = gr.State("")
    sugar_client_id = gr.State("")
    sugar_client_secret = gr.State("")

    screenshots = gr.File(label="Screenshots hochladen", file_count="multiple", file_types=["image"])
    extract_btn = gr.Button("Screenshots auslesen")
    send_btn = gr.Button("Leads und Besuchsberichte erstellen")
    records_table = gr.Dataframe(headers=DISPLAY_COLUMNS, interactive=True)
    terminal = gr.Textbox(label="Protokoll", lines=20, value=log_line("Bereit. Screenshots hochladen, auslesen, Tabelle prüfen und senden."))

    credential_inputs = [
        openai_key,
        sugar_user,
        sugar_password,
        sugar_base_url,
        sugar_platform,
        sugar_client_id,
        sugar_client_secret,
    ]

    extract_btn.click(on_run, inputs=[screenshots, *credential_inputs], outputs=[terminal, records_table])
    send_btn.click(on_send, inputs=[records_table, *credential_inputs], outputs=terminal)


if __name__ == "__main__":
    app_user = secret("APP_USER")
    app_pass = secret("APP_PASS")
    if os.environ.get("SPACE_ID") and not (app_user and app_pass):
        raise RuntimeError("Set APP_USER and APP_PASS as Hugging Face Space Secrets before running this public Space.")
    auth = (app_user, app_pass) if app_user and app_pass else None
    demo.launch(auth=auth)