| |
| """ |
| 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 |
| 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.") |
|
|
| |
| 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) |
|
|
