# -*- 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)