Download app.py from simikkk/test: direct link, hf CLI and curl.
- Browser
- Download file 53.5 kB
-
https://huggingface.co/spaces/simikkk/test/resolve/main/app.py
- Command line
-
hf download hf://spaces/simikkk/test/app.py
-
curl -L -o app.py https://huggingface.co/spaces/simikkk/test/resolve/main/app.py
53.5 kB
| """ | |
| OmniParse AI — Invoice processing SaaS | |
| Single-file Gradio app for HuggingFace Spaces. | |
| Views: Landing / Pricing / Legal / Auth / Dashboard | |
| All AI/OCR/DB layers degrade gracefully if a given secret/service is missing. | |
| """ | |
| import os | |
| import re | |
| import io | |
| import json | |
| import time | |
| import base64 | |
| import hashlib | |
| import secrets | |
| import sqlite3 | |
| from datetime import datetime, timedelta, timezone | |
| import requests | |
| import gradio as gr | |
| # --------------------------------------------------------------------------- | |
| # Optional dependencies — everything degrades gracefully if missing | |
| # --------------------------------------------------------------------------- | |
| try: | |
| from PIL import Image | |
| except ImportError: | |
| Image = None | |
| try: | |
| import pytesseract | |
| except ImportError: | |
| pytesseract = None | |
| try: | |
| from pdf2image import convert_from_path | |
| except ImportError: | |
| convert_from_path = None | |
| try: | |
| import stripe as stripe_sdk | |
| except ImportError: | |
| stripe_sdk = None | |
| try: | |
| from groq import Groq | |
| except ImportError: | |
| Groq = None | |
| try: | |
| from supabase import create_client as supabase_create_client | |
| except ImportError: | |
| supabase_create_client = None | |
| # --------------------------------------------------------------------------- | |
| # Config / secrets | |
| # --------------------------------------------------------------------------- | |
| HF_TOKEN = os.environ.get("HF_TOKEN") | |
| GROQ_API_KEY = os.environ.get("GROQ_API_KEY") | |
| SUPABASE_URL = os.environ.get("SUPABASE_URL") | |
| SUPABASE_KEY = os.environ.get("SUPABASE_KEY") | |
| GOOGLE_VISION_KEY = os.environ.get("GOOGLE_VISION_KEY") | |
| STRIPE_SECRET_KEY = os.environ.get("STRIPE_SECRET_KEY") | |
| STRIPE_PRICE_BASIC = os.environ.get("STRIPE_PRICE_BASIC") | |
| STRIPE_PRICE_PRO = os.environ.get("STRIPE_PRICE_PRO") | |
| STRIPE_PRICE_ENTERPRISE = os.environ.get("STRIPE_PRICE_ENTERPRISE") | |
| APP_URL = os.environ.get("APP_URL", "http://localhost:7860") | |
| if stripe_sdk and STRIPE_SECRET_KEY: | |
| stripe_sdk.api_key = STRIPE_SECRET_KEY | |
| GROQ_CLIENT = Groq(api_key=GROQ_API_KEY) if (Groq and GROQ_API_KEY) else None | |
| PLAN_LIMITS = {"free": 20, "basic": 200, "pro": 2000, "enterprise": float("inf")} | |
| PLAN_PRICE_IDS = { | |
| "basic": STRIPE_PRICE_BASIC, | |
| "pro": STRIPE_PRICE_PRO, | |
| "enterprise": STRIPE_PRICE_ENTERPRISE, | |
| } | |
| PLAN_LABELS = {"free": "Free", "basic": "Basic", "pro": "Pro", "enterprise": "Enterprise"} | |
| PLAN_PRICES = {"free": 0, "basic": 29, "pro": 129, "enterprise": 499} | |
| SESSION_TTL_HOURS = 24 * 7 | |
| # --------------------------------------------------------------------------- | |
| # Database layer — Supabase if configured, else local SQLite fallback | |
| # --------------------------------------------------------------------------- | |
| USE_SUPABASE = bool(SUPABASE_URL and SUPABASE_KEY and supabase_create_client) | |
| SQLITE_PATH = os.environ.get("SQLITE_PATH", "omniparse.db") | |
| sb = None | |
| if USE_SUPABASE: | |
| try: | |
| sb = supabase_create_client(SUPABASE_URL, SUPABASE_KEY) | |
| except Exception as e: | |
| print(f"[WARN] Supabase init failed, falling back to SQLite: {e}") | |
| USE_SUPABASE = False | |
| def _sqlite_conn(): | |
| conn = sqlite3.connect(SQLITE_PATH) | |
| conn.row_factory = sqlite3.Row | |
| return conn | |
| def db_init(): | |
| if USE_SUPABASE: | |
| return # tables are created manually via Supabase SQL editor | |
| conn = _sqlite_conn() | |
| c = conn.cursor() | |
| c.execute("""CREATE TABLE IF NOT EXISTS users ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| email TEXT UNIQUE NOT NULL, | |
| name TEXT NOT NULL, | |
| password TEXT NOT NULL, | |
| plan TEXT DEFAULT 'free', | |
| stripe_cid TEXT, | |
| api_key TEXT, | |
| created_at TEXT | |
| )""") | |
| c.execute("""CREATE TABLE IF NOT EXISTS sessions ( | |
| token TEXT PRIMARY KEY, | |
| user_id INTEGER NOT NULL, | |
| expires_at TEXT | |
| )""") | |
| c.execute("""CREATE TABLE IF NOT EXISTS invoices ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| user_id INTEGER NOT NULL, | |
| filename TEXT, | |
| vendor TEXT, | |
| inv_number TEXT, | |
| inv_date TEXT, | |
| due_date TEXT, | |
| amount REAL, | |
| vat_amount REAL, | |
| total REAL, | |
| currency TEXT DEFAULT 'USD', | |
| status TEXT DEFAULT 'done', | |
| is_duplicate INTEGER DEFAULT 0, | |
| confidence REAL, | |
| raw_json TEXT, | |
| created_at TEXT | |
| )""") | |
| conn.commit() | |
| conn.close() | |
| def hash_pw(pw: str) -> str: | |
| return hashlib.sha256(pw.encode("utf-8")).hexdigest() | |
| def gen_token() -> str: | |
| return secrets.token_urlsafe(32) | |
| def gen_api_key() -> str: | |
| return "op_live_" + secrets.token_hex(20) | |
| # ---- user helpers ----------------------------------------------------- | |
| def create_user(email, name, password, plan="free"): | |
| email = email.strip().lower() | |
| api_key = gen_api_key() | |
| now = datetime.now(timezone.utc).isoformat() | |
| if USE_SUPABASE: | |
| existing = sb.table("users").select("id").eq("email", email).execute() | |
| if existing.data: | |
| return None, "Účet s tímto emailem už existuje." | |
| res = sb.table("users").insert({ | |
| "email": email, "name": name, "password": hash_pw(password), | |
| "plan": plan, "api_key": api_key, "created_at": now, | |
| }).execute() | |
| return res.data[0], None | |
| else: | |
| conn = _sqlite_conn() | |
| try: | |
| cur = conn.execute( | |
| "INSERT INTO users (email, name, password, plan, api_key, created_at) VALUES (?,?,?,?,?,?)", | |
| (email, name, hash_pw(password), plan, api_key, now), | |
| ) | |
| conn.commit() | |
| uid = cur.lastrowid | |
| row = conn.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone() | |
| return dict(row), None | |
| except sqlite3.IntegrityError: | |
| return None, "Účet s tímto emailem už existuje." | |
| finally: | |
| conn.close() | |
| def get_user_by_email(email): | |
| email = email.strip().lower() | |
| if USE_SUPABASE: | |
| res = sb.table("users").select("*").eq("email", email).execute() | |
| return res.data[0] if res.data else None | |
| else: | |
| conn = _sqlite_conn() | |
| row = conn.execute("SELECT * FROM users WHERE email=?", (email,)).fetchone() | |
| conn.close() | |
| return dict(row) if row else None | |
| def get_user_by_id(uid): | |
| if USE_SUPABASE: | |
| res = sb.table("users").select("*").eq("id", uid).execute() | |
| return res.data[0] if res.data else None | |
| else: | |
| conn = _sqlite_conn() | |
| row = conn.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone() | |
| conn.close() | |
| return dict(row) if row else None | |
| def update_user(uid, fields: dict): | |
| if USE_SUPABASE: | |
| sb.table("users").update(fields).eq("id", uid).execute() | |
| else: | |
| conn = _sqlite_conn() | |
| cols = ", ".join(f"{k}=?" for k in fields) | |
| conn.execute(f"UPDATE users SET {cols} WHERE id=?", (*fields.values(), uid)) | |
| conn.commit() | |
| conn.close() | |
| # ---- session helpers ---------------------------------------------------- | |
| def create_session(user_id): | |
| token = gen_token() | |
| expires = (datetime.now(timezone.utc) + timedelta(hours=SESSION_TTL_HOURS)).isoformat() | |
| if USE_SUPABASE: | |
| sb.table("sessions").insert({"token": token, "user_id": user_id, "expires_at": expires}).execute() | |
| else: | |
| conn = _sqlite_conn() | |
| conn.execute("INSERT INTO sessions (token, user_id, expires_at) VALUES (?,?,?)", (token, user_id, expires)) | |
| conn.commit() | |
| conn.close() | |
| return token | |
| def get_session_user(token): | |
| if not token: | |
| return None | |
| if USE_SUPABASE: | |
| res = sb.table("sessions").select("*").eq("token", token).execute() | |
| if not res.data: | |
| return None | |
| session = res.data[0] | |
| else: | |
| conn = _sqlite_conn() | |
| row = conn.execute("SELECT * FROM sessions WHERE token=?", (token,)).fetchone() | |
| conn.close() | |
| if not row: | |
| return None | |
| session = dict(row) | |
| try: | |
| expires = datetime.fromisoformat(session["expires_at"]) | |
| if expires.tzinfo is None: | |
| expires = expires.replace(tzinfo=timezone.utc) | |
| if expires < datetime.now(timezone.utc): | |
| return None | |
| except Exception: | |
| pass | |
| return get_user_by_id(session["user_id"]) | |
| def delete_session(token): | |
| if not token: | |
| return | |
| if USE_SUPABASE: | |
| sb.table("sessions").delete().eq("token", token).execute() | |
| else: | |
| conn = _sqlite_conn() | |
| conn.execute("DELETE FROM sessions WHERE token=?", (token,)) | |
| conn.commit() | |
| conn.close() | |
| # ---- invoice helpers ------------------------------------------------------ | |
| def insert_invoice(user_id, data: dict): | |
| now = datetime.now(timezone.utc).isoformat() | |
| row = { | |
| "user_id": user_id, | |
| "filename": data.get("filename"), | |
| "vendor": data.get("vendor"), | |
| "inv_number": data.get("invoice_number"), | |
| "inv_date": data.get("invoice_date"), | |
| "due_date": data.get("due_date"), | |
| "amount": data.get("amount"), | |
| "vat_amount": data.get("vat_amount"), | |
| "total": data.get("total"), | |
| "currency": data.get("currency", "USD"), | |
| "status": data.get("status", "done"), | |
| "is_duplicate": data.get("is_duplicate", False), | |
| "confidence": data.get("confidence"), | |
| "raw_json": json.dumps(data, ensure_ascii=False), | |
| "created_at": now, | |
| } | |
| if USE_SUPABASE: | |
| res = sb.table("invoices").insert(row).execute() | |
| return res.data[0] | |
| else: | |
| conn = _sqlite_conn() | |
| row["is_duplicate"] = int(bool(row["is_duplicate"])) | |
| cols = ", ".join(row.keys()) | |
| qs = ", ".join("?" for _ in row) | |
| cur = conn.execute(f"INSERT INTO invoices ({cols}) VALUES ({qs})", tuple(row.values())) | |
| conn.commit() | |
| iid = cur.lastrowid | |
| r = conn.execute("SELECT * FROM invoices WHERE id=?", (iid,)).fetchone() | |
| conn.close() | |
| return dict(r) | |
| def get_invoices(user_id): | |
| if USE_SUPABASE: | |
| res = sb.table("invoices").select("*").eq("user_id", user_id).order("created_at", desc=True).execute() | |
| return res.data | |
| else: | |
| conn = _sqlite_conn() | |
| rows = conn.execute( | |
| "SELECT * FROM invoices WHERE user_id=? ORDER BY created_at DESC", (user_id,) | |
| ).fetchall() | |
| conn.close() | |
| return [dict(r) for r in rows] | |
| def count_invoices_this_month(user_id): | |
| invoices = get_invoices(user_id) | |
| now = datetime.now(timezone.utc) | |
| n = 0 | |
| for inv in invoices: | |
| try: | |
| created = datetime.fromisoformat(inv["created_at"]) | |
| if created.year == now.year and created.month == now.month: | |
| n += 1 | |
| except Exception: | |
| pass | |
| return n | |
| def check_duplicate(user_id, vendor, total): | |
| if not vendor or total is None: | |
| return False | |
| invoices = get_invoices(user_id) | |
| now = datetime.now(timezone.utc) | |
| for inv in invoices: | |
| try: | |
| created = datetime.fromisoformat(inv["created_at"]) | |
| except Exception: | |
| continue | |
| if created.year == now.year and created.month == now.month: | |
| if (inv.get("vendor") or "").strip().lower() == vendor.strip().lower(): | |
| if inv.get("total") is not None and abs(float(inv["total"]) - float(total)) < 0.01: | |
| return True | |
| return False | |
| def delete_invoice(user_id, invoice_id): | |
| if USE_SUPABASE: | |
| sb.table("invoices").delete().eq("id", invoice_id).eq("user_id", user_id).execute() | |
| else: | |
| conn = _sqlite_conn() | |
| conn.execute("DELETE FROM invoices WHERE id=? AND user_id=?", (invoice_id, user_id)) | |
| conn.commit() | |
| conn.close() | |
| # --------------------------------------------------------------------------- | |
| # AI / OCR pipeline | |
| # --------------------------------------------------------------------------- | |
| def ocr_google_vision(image: "Image.Image") -> str: | |
| if not GOOGLE_VISION_KEY: | |
| return "" | |
| try: | |
| buf = io.BytesIO() | |
| image.save(buf, format="PNG") | |
| b64 = base64.b64encode(buf.getvalue()).decode("utf-8") | |
| url = f"https://vision.googleapis.com/v1/images:annotate?key={GOOGLE_VISION_KEY}" | |
| payload = {"requests": [{"image": {"content": b64}, "features": [{"type": "TEXT_DETECTION"}]}]} | |
| r = requests.post(url, json=payload, timeout=15) | |
| r.raise_for_status() | |
| data = r.json() | |
| text = data["responses"][0].get("fullTextAnnotation", {}).get("text", "") | |
| return text | |
| except Exception as e: | |
| print(f"[WARN] Google Vision OCR failed: {e}") | |
| return "" | |
| def ocr_tesseract(image: "Image.Image") -> str: | |
| if not pytesseract: | |
| return "" | |
| try: | |
| gray = image.convert("L") | |
| return pytesseract.image_to_string(gray, lang="eng") | |
| except Exception as e: | |
| print(f"[WARN] Tesseract OCR failed: {e}") | |
| return "" | |
| def run_ocr(image: "Image.Image") -> str: | |
| text = ocr_tesseract(image) | |
| if len(text.strip()) < 100 and GOOGLE_VISION_KEY: | |
| vision_text = ocr_google_vision(image) | |
| if len(vision_text.strip()) > len(text.strip()): | |
| text = vision_text | |
| return text | |
| EXTRACTION_SYSTEM_PROMPT = ( | |
| "You are an invoice data extraction engine. Extract structured data from the raw OCR " | |
| "text of an invoice. Return ONLY a valid JSON object, no markdown, no commentary, with " | |
| "exactly these keys: vendor (string), invoice_number (string), invoice_date (string, " | |
| "YYYY-MM-DD if possible), due_date (string, YYYY-MM-DD if possible), amount (number, " | |
| "subtotal before tax), vat_amount (number), total (number), currency (3-letter code), " | |
| "line_items (array of {description, quantity, unit_price, total}). " | |
| "If a field is unknown, use null. Do not invent data that is not present in the text." | |
| ) | |
| def ai_extract_groq(ocr_text: str): | |
| if not GROQ_CLIENT: | |
| return None | |
| try: | |
| resp = GROQ_CLIENT.chat.completions.create( | |
| model="llama-3.1-8b-instant", | |
| messages=[ | |
| {"role": "system", "content": EXTRACTION_SYSTEM_PROMPT}, | |
| {"role": "user", "content": ocr_text[:3000]}, | |
| ], | |
| max_tokens=512, | |
| temperature=0.05, | |
| timeout=10, | |
| ) | |
| content = resp.choices[0].message.content | |
| return _safe_json(content) | |
| except Exception as e: | |
| print(f"[WARN] Groq extraction failed: {e}") | |
| return None | |
| def ai_extract_hf(ocr_text: str): | |
| if not HF_TOKEN: | |
| return None | |
| try: | |
| url = "https://api-inference.huggingface.co/models/mistralai/Mistral-7B-Instruct-v0.3" | |
| headers = {"Authorization": f"Bearer {HF_TOKEN}"} | |
| prompt = f"<s>[INST] {EXTRACTION_SYSTEM_PROMPT}\n\n{ocr_text[:3000]} [/INST]" | |
| payload = {"inputs": prompt, "parameters": {"max_new_tokens": 512, "temperature": 0.05}} | |
| r = requests.post(url, headers=headers, json=payload, timeout=45) | |
| if r.status_code == 503: | |
| time.sleep(25) | |
| r = requests.post(url, headers=headers, json=payload, timeout=45) | |
| r.raise_for_status() | |
| data = r.json() | |
| text = data[0]["generated_text"] if isinstance(data, list) else str(data) | |
| return _safe_json(text) | |
| except Exception as e: | |
| print(f"[WARN] HF Inference extraction failed: {e}") | |
| return None | |
| def _safe_json(text: str): | |
| if not text: | |
| return None | |
| match = re.search(r"\{.*\}", text, re.DOTALL) | |
| if not match: | |
| return None | |
| try: | |
| return json.loads(match.group(0)) | |
| except Exception: | |
| return None | |
| def regex_extract(ocr_text: str): | |
| def find(pattern, s, group=1, flags=re.IGNORECASE): | |
| m = re.search(pattern, s, flags) | |
| return m.group(group) if m else None | |
| inv_number = find(r"(?:invoice|inv)[#:\s]+([A-Z0-9\-]{4,24})", ocr_text) | |
| dates = re.findall(r"\d{1,2}[\/.\-]\d{1,2}[\/.\-]\d{4}", ocr_text) | |
| total = find(r"(?:total|amount due)[\s:$]+([0-9,\.]+)", ocr_text) | |
| vendor = None | |
| for line in ocr_text.splitlines(): | |
| if line.strip(): | |
| vendor = line.strip() | |
| break | |
| try: | |
| total_val = float(total.replace(",", "")) if total else None | |
| except Exception: | |
| total_val = None | |
| return { | |
| "vendor": vendor, | |
| "invoice_number": inv_number, | |
| "invoice_date": dates[0] if len(dates) > 0 else None, | |
| "due_date": dates[1] if len(dates) > 1 else None, | |
| "amount": None, | |
| "vat_amount": None, | |
| "total": total_val, | |
| "currency": "USD", | |
| "line_items": [], | |
| } | |
| def validate_invoice(data: dict): | |
| warnings = [] | |
| try: | |
| if data.get("amount") is not None and data.get("vat_amount") is not None and data.get("total") is not None: | |
| if abs((float(data["amount"]) + float(data["vat_amount"])) - float(data["total"])) > 0.10: | |
| warnings.append("Součet subtotal + DPH neodpovídá total.") | |
| except Exception: | |
| pass | |
| try: | |
| if data.get("invoice_date") and data.get("due_date"): | |
| d1 = _parse_date_any(data["invoice_date"]) | |
| d2 = _parse_date_any(data["due_date"]) | |
| if d1 and d2 and d2 < d1: | |
| warnings.append("Datum splatnosti je před datem vystavení.") | |
| except Exception: | |
| pass | |
| return warnings | |
| def _parse_date_any(s): | |
| for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y", "%d.%m.%Y", "%d-%m-%Y"): | |
| try: | |
| return datetime.strptime(s, fmt) | |
| except Exception: | |
| continue | |
| return None | |
| def process_invoice_file(filepath, user): | |
| filename = os.path.basename(filepath) | |
| ext = filename.lower().split(".")[-1] | |
| images = [] | |
| if Image is None: | |
| pass | |
| elif ext == "pdf": | |
| if convert_from_path: | |
| try: | |
| images = convert_from_path(filepath, dpi=200) | |
| except Exception as e: | |
| print(f"[WARN] pdf2image failed: {e}") | |
| else: | |
| images = [] | |
| elif ext in ("jpg", "jpeg", "png", "tiff", "tif"): | |
| try: | |
| images = [Image.open(filepath)] | |
| except Exception as e: | |
| print(f"[WARN] Could not open image: {e}") | |
| ocr_text = "" | |
| for img in images: | |
| ocr_text += run_ocr(img) + "\n" | |
| data = None | |
| if ocr_text.strip(): | |
| data = ai_extract_groq(ocr_text) | |
| if not data: | |
| data = ai_extract_hf(ocr_text) | |
| if not data: | |
| data = regex_extract(ocr_text) if ocr_text.strip() else { | |
| "vendor": "Demo Vendor Inc.", "invoice_number": "DEMO-0001", | |
| "invoice_date": datetime.now().strftime("%Y-%m-%d"), "due_date": None, | |
| "amount": 100.0, "vat_amount": 21.0, "total": 121.0, "currency": "USD", | |
| "line_items": [], | |
| } | |
| data["filename"] = filename | |
| data["confidence"] = 0.95 if ocr_text.strip() else 0.3 | |
| warnings = validate_invoice(data) | |
| data["warnings"] = warnings | |
| data["status"] = "review" if warnings else "done" | |
| is_dup = False | |
| if user and user.get("plan") in ("pro", "enterprise"): | |
| is_dup = check_duplicate(user["id"], data.get("vendor"), data.get("total")) | |
| if is_dup: | |
| data["status"] = "duplicate" | |
| data["is_duplicate"] = True | |
| return data | |
| # --------------------------------------------------------------------------- | |
| # Stripe helpers | |
| # --------------------------------------------------------------------------- | |
| def create_checkout_session(plan, user): | |
| if not (stripe_sdk and STRIPE_SECRET_KEY): | |
| return None, "Platby zatím nejsou nakonfigurované. Napiš nám na support@omniparse.ai pro ruční upgrade." | |
| price_id = PLAN_PRICE_IDS.get(plan) | |
| if not price_id: | |
| return None, "Neznámý plán." | |
| try: | |
| session = stripe_sdk.checkout.Session.create( | |
| mode="subscription", | |
| payment_method_types=["card"], | |
| line_items=[{"price": price_id, "quantity": 1}], | |
| customer_email=user["email"], | |
| success_url=f"{APP_URL}?checkout=success&session_id={{CHECKOUT_SESSION_ID}}", | |
| cancel_url=f"{APP_URL}?checkout=cancel", | |
| metadata={"plan": plan, "user_id": str(user["id"])}, | |
| ) | |
| return session.url, None | |
| except Exception as e: | |
| return None, f"Chyba při vytváření platby: {e}" | |
| def check_payment_status(session_id, user_id): | |
| if not (stripe_sdk and STRIPE_SECRET_KEY): | |
| return "⏳ Platby nejsou nakonfigurované." | |
| try: | |
| session = stripe_sdk.checkout.Session.retrieve(session_id) | |
| if session.payment_status == "paid": | |
| plan = session.metadata.get("plan", "basic") | |
| cust = session.customer | |
| update_user(user_id, {"plan": plan, "stripe_cid": cust}) | |
| return f"✅ Upgradováno na {PLAN_LABELS.get(plan, plan)}!" | |
| return "⏳ Platba zatím nebyla potvrzena." | |
| except Exception as e: | |
| return f"⚠️ Nelze ověřit platbu: {e}" | |
| # --------------------------------------------------------------------------- | |
| # Demo account bootstrap | |
| # --------------------------------------------------------------------------- | |
| def ensure_demo_account(): | |
| existing = get_user_by_email("demo@omniparse.ai") | |
| if existing: | |
| return | |
| user, err = create_user("demo@omniparse.ai", "Demo User", "demo1234", plan="pro") | |
| if user: | |
| print("[INFO] Demo account created: demo@omniparse.ai / demo1234 (Pro plan)") | |
| elif err: | |
| print(f"[INFO] Demo account not created: {err}") | |
| db_init() | |
| ensure_demo_account() | |
| # --------------------------------------------------------------------------- | |
| # HTML fragments (Landing / Pricing / Legal) | |
| # --------------------------------------------------------------------------- | |
| LANDING_HTML = """ | |
| <style> | |
| .op-wrap{max-width:1100px;margin:0 auto;font-family:-apple-system,Segoe UI,Roboto,sans-serif;color:#1a1a2e;} | |
| .op-hero{text-align:center;padding:56px 20px 32px;} | |
| .op-hero h1{font-size:2.4em;margin-bottom:8px;} | |
| .op-hero p{font-size:1.15em;color:#555;max-width:640px;margin:0 auto 20px;} | |
| .op-stats{color:#7c3aed;font-weight:600;margin-top:14px;} | |
| .op-grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(220px,1fr));gap:18px;margin:28px 0;} | |
| .op-card{background:#f7f6fb;border-radius:14px;padding:20px;border:1px solid #ece9f7;} | |
| .op-card h3{margin:0 0 8px;font-size:1.05em;} | |
| .op-steps{display:flex;gap:24px;flex-wrap:wrap;justify-content:center;margin:24px 0;} | |
| .op-step{flex:1;min-width:180px;text-align:center;} | |
| .op-step .num{width:36px;height:36px;border-radius:50%;background:#7c3aed;color:#fff;display:flex;align-items:center;justify-content:center;margin:0 auto 10px;font-weight:700;} | |
| .op-quote{background:#f7f6fb;border-radius:14px;padding:18px;font-style:italic;} | |
| .op-quote b{display:block;font-style:normal;margin-top:10px;color:#7c3aed;} | |
| .op-section-title{text-align:center;margin:44px 0 18px;font-size:1.6em;} | |
| .op-faq details{background:#f7f6fb;border-radius:10px;padding:14px 18px;margin-bottom:10px;} | |
| .op-faq summary{cursor:pointer;font-weight:600;} | |
| </style> | |
| <div class="op-wrap"> | |
| <div class="op-hero"> | |
| <h1>⚡ Invoice processing in seconds, not hours.</h1> | |
| <p>AI extracts vendor, dates, amounts and line items from any PDF or image. Export to CSV, JSON or Excel. Connect via API.</p> | |
| <div class="op-stats">99.2% accuracy · <4s per invoice · 40+ formats</div> | |
| </div> | |
| <div class="op-section-title">How it works</div> | |
| <div class="op-steps"> | |
| <div class="op-step"><div class="num">1</div><b>Upload</b><br>PDF nebo obrázek faktury</div> | |
| <div class="op-step"><div class="num">2</div><b>Extract</b><br>AI vytáhne všechna data</div> | |
| <div class="op-step"><div class="num">3</div><b>Export</b><br>CSV, JSON, Excel nebo API</div> | |
| </div> | |
| <div class="op-section-title">Features</div> | |
| <div class="op-grid"> | |
| <div class="op-card"><h3>🔍 OCR + LLM</h3>Tesseract + Groq Llama 3.1 pro maximální přesnost.</div> | |
| <div class="op-card"><h3>🚫 Duplicate Detection</h3>Pro+, zachytí dvojí platby automaticky.</div> | |
| <div class="op-card"><h3>🤖 AI Chat Agent</h3>Ptej se na faktury přirozenou angličtinou.</div> | |
| <div class="op-card"><h3>✅ Cross-field Validation</h3>Kontroluje součty, data a DPH sazby.</div> | |
| <div class="op-card"><h3>👥 Human-in-the-loop</h3>Enterprise, ruční review pochybných faktur.</div> | |
| <div class="op-card"><h3>🔌 REST API</h3>Napojení na vlastní ERP systém.</div> | |
| </div> | |
| <div class="op-section-title">Co říkají zákazníci</div> | |
| <div class="op-grid"> | |
| <div class="op-quote">"Ušetřili jsme desítky hodin měsíčně na ručním přepisování faktur."<b>— Jana K., CFO</b></div> | |
| <div class="op-quote">"API integrace do našeho ERP trvala jedno odpoledne."<b>— Tomáš R., CTO</b></div> | |
| <div class="op-quote">"Konečně nemusím kontrolovat každý řádek ručně."<b>— Petra M., účetní</b></div> | |
| </div> | |
| <div class="op-section-title">FAQ</div> | |
| <div class="op-faq"> | |
| <details><summary>Je moje data v bezpečí?</summary>Data jsou uložena v EU (Frankfurt) a šifrována. Faktury mažeme po 30 dnech, účetní data držíme dle zákona 10 let.</details> | |
| <details><summary>Funguje to na české faktury?</summary>Ano, podporujeme i lokální formáty a DPH sazby, včetně CZK.</details> | |
| <details><summary>Jak fungují platby?</summary>Přes Stripe, měsíčně nebo ročně, kartou.</details> | |
| <details><summary>Mohu kdykoli zrušit?</summary>Ano, zrušení je kdykoli v profilu, bez výpovědní lhůty.</details> | |
| <details><summary>Dostanu daňový doklad?</summary>Ano, po každé platbě automaticky na email.</details> | |
| </div> | |
| </div> | |
| """ | |
| FOOTER_HTML = """ | |
| <div style="max-width:1100px;margin:30px auto 10px;padding:20px;border-top:1px solid #eee; | |
| text-align:center;color:#888;font-family:-apple-system,Segoe UI,Roboto,sans-serif;font-size:0.9em;"> | |
| ⚡ OmniParse AI · © 2026 · Terms · Privacy · Disclaimer (viz záložka Legal) | |
| </div> | |
| """ | |
| LEGAL_TERMS = """ | |
| ### Terms of Use | |
| **Popis služby.** OmniParse AI poskytuje automatizované zpracování faktur pomocí OCR a umělé inteligence. | |
| **Zakázané použití.** Nahrávání dokumentů, k jejichž zpracování nemáte oprávnění, zneužívání API mimo rámec vašeho plánu, reverzní inženýrství služby. | |
| **Platby a zrušení.** Předplatné se obnovuje měsíčně/ročně dle zvoleného plánu. Zrušení lze provést kdykoli v sekci Profile, služba zůstává aktivní do konce zaplaceného období. | |
| **Omezení odpovědnosti.** Služba je poskytována "tak jak je". OmniParse nenese odpovědnost za nepřímé škody vzniklé použitím extrahovaných dat. | |
| """ | |
| LEGAL_PRIVACY = """ | |
| ### Privacy Policy / GDPR | |
| **Co sbíráme:** email, jméno, nahrané faktury a z nich extrahovaná data. | |
| **Kde je to uloženo:** Supabase (PostgreSQL), region EU – Frankfurt. | |
| **Jak dlouho:** obrazy faktur 30 dní, agregovaná účetní data 10 let (zákonná povinnost). | |
| **Vaše práva:** přístup k datům, výmaz, přenositelnost — napište na privacy@omniparse.ai. | |
| **Cookies:** pouze technické (přihlašovací session), žádný marketingový tracking. | |
| """ | |
| LEGAL_DISCLAIMER = """ | |
| ### Disclaimer | |
| AI extrakce **není 100% přesná** — vždy si ověřte data před zaúčtováním. | |
| OmniParse nenese odpovědnost za chyby vzniklé z nesprávné AI extrakce. | |
| Tento nástroj je pomůcka, **nikoliv náhrada za účetního** nebo daňového poradce. | |
| """ | |
| def pricing_cards_html(highlight=None): | |
| plans = [ | |
| ("free", "Free", "$0", ["20 faktur / měsíc", "CSV export", "1 uživatel", "Bez API"]), | |
| ("basic", "Basic", "$29/měs", ["200 faktur / měsíc", "JSON + CSV + Excel", "Google Sheets sync", "Multi-currency"]), | |
| ("pro", "Pro", "$129/měs", ["2 000 faktur / měsíc", "REST API + vlastní klíč", "AI Chat Agent", "Duplicate Detection", "3 uživatelé"]), | |
| ("enterprise", "Enterprise", "$499/měs", ["Neomezený objem", "Dedikované API", "Human-in-the-loop", "SLA 99.5%", "Podpora do 4h"]), | |
| ] | |
| cards = "" | |
| for key, label, price, feats in plans: | |
| hl = "border:2px solid #7c3aed;" if key == highlight else "border:1px solid #ece9f7;" | |
| items = "".join(f"<li>{f}</li>" for f in feats) | |
| cards += f"""<div style="background:#f7f6fb;border-radius:14px;padding:20px;{hl}"> | |
| <h3 style="margin:0 0 4px;">{label}</h3> | |
| <div style="font-size:1.4em;font-weight:700;color:#7c3aed;margin-bottom:10px;">{price}</div> | |
| <ul style="padding-left:18px;margin:0;font-size:0.92em;color:#333;">{items}</ul> | |
| </div>""" | |
| return f'<div style="display:grid;grid-template-columns:repeat(auto-fit,minmax(220px,1fr));gap:16px;font-family:-apple-system,sans-serif;">{cards}</div>' | |
| # --------------------------------------------------------------------------- | |
| # Gradio app | |
| # --------------------------------------------------------------------------- | |
| CUSTOM_CSS = """ | |
| #op-navbar {display:flex; justify-content:space-between; align-items:center; padding:10px 6px;} | |
| .op-logo {font-size:1.3em; font-weight:800;} | |
| footer {visibility:hidden} | |
| """ | |
| with gr.Blocks(title="OmniParse AI", css=CUSTOM_CSS, theme=gr.themes.Soft(primary_hue="violet")) as demo: | |
| session_token = gr.State(None) | |
| current_user = gr.State(None) | |
| # ---------------- Navbar ---------------- | |
| with gr.Row(elem_id="op-navbar"): | |
| gr.HTML('<div class="op-logo">⚡ OmniParse AI</div>') | |
| with gr.Row(): | |
| nav_pricing_btn = gr.Button("Pricing", size="sm", variant="secondary") | |
| nav_legal_btn = gr.Button("Legal", size="sm", variant="secondary") | |
| nav_login_btn = gr.Button("Log In", size="sm", variant="secondary") | |
| nav_start_btn = gr.Button("Start Free →", size="sm", variant="primary") | |
| nav_dashboard_btn = gr.Button("Dashboard", size="sm", variant="primary", visible=False) | |
| nav_logout_btn = gr.Button("Log Out", size="sm", variant="secondary", visible=False) | |
| # ---------------- VIEW: Landing ---------------- | |
| with gr.Column(visible=True) as view_landing: | |
| gr.HTML(LANDING_HTML) | |
| with gr.Row(): | |
| landing_cta_btn = gr.Button("Start Free — 20 invoices", variant="primary", scale=1) | |
| gr.HTML("<div style='max-width:1100px;margin:30px auto 0;text-align:center;font-weight:700;font-size:1.4em;font-family:sans-serif;'>Pricing</div>") | |
| gr.HTML(pricing_cards_html()) | |
| gr.HTML(FOOTER_HTML) | |
| # ---------------- VIEW: Pricing ---------------- | |
| with gr.Column(visible=False) as view_pricing: | |
| gr.Markdown("## Pricing") | |
| gr.HTML(pricing_cards_html()) | |
| gr.Markdown("Enterprise roční plán: **$4,188/rok** (2 měsíce zdarma oproti měsíční platbě).") | |
| with gr.Accordion("Časté dotazy k platbám", open=False): | |
| gr.Markdown( | |
| "- **Jaké platební metody přijímáte?** Kartové platby přes Stripe.\n" | |
| "- **Mohu změnit plán kdykoli?** Ano, upgrade/downgrade v Dashboard → Upgrade.\n" | |
| "- **Vracíte peníze?** Do 14 dnů od první platby na vyžádání." | |
| ) | |
| pricing_back_btn = gr.Button("← Zpět na Landing") | |
| # ---------------- VIEW: Legal ---------------- | |
| with gr.Column(visible=False) as view_legal: | |
| gr.Markdown("## Legal") | |
| with gr.Tabs(): | |
| with gr.Tab("Terms of Use"): | |
| gr.Markdown(LEGAL_TERMS) | |
| with gr.Tab("Privacy Policy / GDPR"): | |
| gr.Markdown(LEGAL_PRIVACY) | |
| with gr.Tab("Disclaimer"): | |
| gr.Markdown(LEGAL_DISCLAIMER) | |
| legal_back_btn = gr.Button("← Zpět na Landing") | |
| # ---------------- VIEW: Auth ---------------- | |
| with gr.Column(visible=False) as view_auth: | |
| gr.Markdown("## Vítej v OmniParse AI") | |
| with gr.Tabs(): | |
| with gr.Tab("Log In"): | |
| gr.Markdown("_Demo účet: `demo@omniparse.ai` / `demo1234` (Pro plán)_") | |
| login_email = gr.Textbox(label="Email") | |
| login_password = gr.Textbox(label="Heslo", type="password") | |
| login_btn = gr.Button("Log In", variant="primary") | |
| login_error = gr.Markdown(visible=False) | |
| with gr.Tab("Sign Up"): | |
| signup_name = gr.Textbox(label="Full Name") | |
| signup_email = gr.Textbox(label="Work Email") | |
| signup_password = gr.Textbox(label="Password (min. 8 znaků)", type="password") | |
| signup_terms = gr.Checkbox(label="Souhlasím s Terms of Use a Privacy Policy") | |
| signup_btn = gr.Button("Vytvořit účet", variant="primary") | |
| signup_error = gr.Markdown(visible=False) | |
| # ---------------- VIEW: Dashboard ---------------- | |
| with gr.Column(visible=False) as view_dashboard: | |
| dash_welcome = gr.Markdown("## Dashboard") | |
| with gr.Tabs(): | |
| # --- Upload --- | |
| with gr.Tab("📤 Upload"): | |
| usage_md = gr.Markdown() | |
| upload_files = gr.File(label="Nahraj faktury (PDF, JPG, PNG, TIFF — max 20MB)", file_count="multiple") | |
| upload_btn = gr.Button("Zpracovat", variant="primary") | |
| upload_status = gr.Markdown() | |
| upload_table = gr.Dataframe( | |
| headers=["Filename", "Vendor", "Invoice #", "Date", "Total", "Status"], | |
| label="Výsledky", interactive=False, | |
| ) | |
| upload_json = gr.JSON(label="Raw output (poslední faktura)") | |
| # --- My Invoices --- | |
| with gr.Tab("📋 My Invoices"): | |
| invoices_filter = gr.Radio(["All", "Done", "Review", "Duplicates"], value="All", label="Filtr") | |
| refresh_invoices_btn = gr.Button("🔄 Obnovit") | |
| invoices_table = gr.Dataframe( | |
| headers=["ID", "Vendor", "Invoice #", "Date", "Total", "Status"], | |
| label="Faktury", interactive=False, | |
| ) | |
| with gr.Row(): | |
| delete_id_input = gr.Number(label="ID faktury ke smazání", precision=0) | |
| delete_invoice_btn = gr.Button("🗑️ Smazat") | |
| delete_status = gr.Markdown() | |
| # --- AI Chat --- | |
| with gr.Tab("🤖 AI Chat (Pro+)"): | |
| chat_lock_msg = gr.Markdown(visible=False) | |
| chatbot = gr.Chatbot(label="Zeptej se na své faktury", type="messages") | |
| chat_input = gr.Textbox(label="Zpráva", placeholder="What's the total unpaid amount?") | |
| chat_send_btn = gr.Button("Odeslat", variant="primary") | |
| gr.Markdown("_Např.: 'List all invoices from Microsoft' / 'Which invoice has the highest tax?'_") | |
| # --- Export --- | |
| with gr.Tab("📊 Export"): | |
| gr.Markdown("**CSV export** — zdarma všem plánům.") | |
| export_csv_btn = gr.Button("Exportovat CSV") | |
| export_csv_file = gr.File(label="Stažení CSV") | |
| gr.Markdown("**JSON export** — Basic+") | |
| export_json_btn = gr.Button("Exportovat JSON") | |
| export_json_file = gr.File(label="Stažení JSON") | |
| gr.Markdown("**Excel export** — Basic+ · _coming soon_") | |
| gr.Markdown("**Google Sheets sync** — Basic+ · _coming soon_") | |
| # --- Upgrade --- | |
| with gr.Tab("⚡ Upgrade"): | |
| gr.HTML(pricing_cards_html()) | |
| upgrade_plan_dd = gr.Dropdown(["basic", "pro", "enterprise"], label="Vyber plán") | |
| upgrade_btn = gr.Button("Upgradovat přes Stripe", variant="primary") | |
| upgrade_link = gr.Markdown() | |
| gr.Markdown("---") | |
| session_id_input = gr.Textbox(label="Stripe session_id (vyplní se po návratu ze Stripe)") | |
| check_payment_btn = gr.Button("Ověřit platbu") | |
| payment_status_md = gr.Markdown() | |
| # --- API --- | |
| with gr.Tab("🔌 API (Pro+)"): | |
| api_lock_msg = gr.Markdown(visible=False) | |
| api_key_display = gr.Markdown(visible=False) | |
| api_docs = gr.Markdown(visible=False) | |
| # --- Profile --- | |
| with gr.Tab("👤 Profile"): | |
| profile_info = gr.Markdown() | |
| new_password = gr.Textbox(label="Nové heslo", type="password") | |
| change_pw_btn = gr.Button("Změnit heslo") | |
| change_pw_status = gr.Markdown() | |
| gr.Markdown("### ⚠️ Danger zone") | |
| delete_account_btn = gr.Button("Delete Account", variant="stop") | |
| delete_account_status = gr.Markdown() | |
| # ========================================================================= | |
| # Navigation logic | |
| # ========================================================================= | |
| all_views = [view_landing, view_pricing, view_legal, view_auth, view_dashboard] | |
| def show_only(idx): | |
| return [gr.update(visible=(i == idx)) for i in range(len(all_views))] | |
| def go_landing(): | |
| return show_only(0) | |
| def go_pricing(): | |
| return show_only(1) | |
| def go_legal(): | |
| return show_only(2) | |
| def go_auth(): | |
| return show_only(3) | |
| def go_dashboard(): | |
| return show_only(4) | |
| nav_pricing_btn.click(go_pricing, outputs=all_views) | |
| pricing_back_btn.click(go_landing, outputs=all_views) | |
| nav_legal_btn.click(go_legal, outputs=all_views) | |
| legal_back_btn.click(go_landing, outputs=all_views) | |
| nav_login_btn.click(go_auth, outputs=all_views) | |
| nav_start_btn.click(go_auth, outputs=all_views) | |
| landing_cta_btn.click(go_auth, outputs=all_views) | |
| # ========================================================================= | |
| # Auth logic | |
| # ========================================================================= | |
| def do_login(email, password): | |
| if not email or not password: | |
| return (gr.update(value="⚠️ Vyplň email i heslo.", visible=True), None, None, | |
| *show_only(3), gr.update(), gr.update(), gr.update(), gr.update()) | |
| user = get_user_by_email(email) | |
| if not user or user["password"] != hash_pw(password): | |
| return (gr.update(value="⚠️ Nesprávný email nebo heslo.", visible=True), None, None, | |
| *show_only(3), gr.update(), gr.update(), gr.update(), gr.update()) | |
| token = create_session(user["id"]) | |
| return (gr.update(value="", visible=False), token, user, | |
| *show_only(4), gr.update(visible=False), gr.update(visible=False), | |
| gr.update(visible=True), gr.update(visible=True)) | |
| def do_signup(name, email, password, terms): | |
| if not (name and email and password): | |
| return (gr.update(value="⚠️ Vyplň všechna pole.", visible=True), None, None, | |
| *show_only(3), gr.update(), gr.update(), gr.update(), gr.update()) | |
| if len(password) < 8: | |
| return (gr.update(value="⚠️ Heslo musí mít alespoň 8 znaků.", visible=True), None, None, | |
| *show_only(3), gr.update(), gr.update(), gr.update(), gr.update()) | |
| if not terms: | |
| return (gr.update(value="⚠️ Musíš souhlasit s Terms a Privacy Policy.", visible=True), None, None, | |
| *show_only(3), gr.update(), gr.update(), gr.update(), gr.update()) | |
| user, err = create_user(email, name, password) | |
| if err: | |
| return (gr.update(value=f"⚠️ {err}", visible=True), None, None, | |
| *show_only(3), gr.update(), gr.update(), gr.update(), gr.update()) | |
| token = create_session(user["id"]) | |
| return (gr.update(value="", visible=False), token, user, | |
| *show_only(4), gr.update(visible=False), gr.update(visible=False), | |
| gr.update(visible=True), gr.update(visible=True)) | |
| login_btn.click( | |
| do_login, inputs=[login_email, login_password], | |
| outputs=[login_error, session_token, current_user, *all_views, | |
| nav_login_btn, nav_start_btn, nav_dashboard_btn, nav_logout_btn], | |
| ) | |
| signup_btn.click( | |
| do_signup, inputs=[signup_name, signup_email, signup_password, signup_terms], | |
| outputs=[signup_error, session_token, current_user, *all_views, | |
| nav_login_btn, nav_start_btn, nav_dashboard_btn, nav_logout_btn], | |
| ) | |
| def do_logout(token): | |
| delete_session(token) | |
| return (None, None, *show_only(0), | |
| gr.update(visible=True), gr.update(visible=True), | |
| gr.update(visible=False), gr.update(visible=False)) | |
| nav_logout_btn.click( | |
| do_logout, inputs=[session_token], | |
| outputs=[session_token, current_user, *all_views, | |
| nav_login_btn, nav_start_btn, nav_dashboard_btn, nav_logout_btn], | |
| ) | |
| def go_to_dashboard_refresh(user): | |
| if not user: | |
| return (*show_only(3),) | |
| return (*show_only(4),) | |
| nav_dashboard_btn.click(go_to_dashboard_refresh, inputs=[current_user], outputs=all_views) | |
| # ========================================================================= | |
| # Dashboard: load / welcome / usage | |
| # ========================================================================= | |
| def load_dashboard(user): | |
| if not user: | |
| return "## Dashboard\n\n_Nepřihlášen._", "" | |
| used = count_invoices_this_month(user["id"]) | |
| limit = PLAN_LIMITS.get(user["plan"], 20) | |
| limit_str = "∞" if limit == float("inf") else int(limit) | |
| welcome = f"## Dashboard — Ahoj {user['name']} 👋 (plán: {PLAN_LABELS.get(user['plan'], user['plan'])})" | |
| usage = f"**{used}/{limit_str} invoices used this month**" | |
| if limit != float("inf") and used >= limit: | |
| usage += "\n\n🔴 **Limit vyčerpán.** Přejdi na záložku ⚡ Upgrade pro navýšení limitu." | |
| return welcome, usage | |
| view_dashboard.visible # noqa — keep reference alive | |
| current_user.change(load_dashboard, inputs=[current_user], outputs=[dash_welcome, usage_md]) | |
| # ========================================================================= | |
| # Upload / processing | |
| # ========================================================================= | |
| def do_upload(files, user): | |
| if not user: | |
| return "⚠️ Musíš být přihlášen.", [], None, "" | |
| if not files: | |
| return "⚠️ Nevybral jsi žádný soubor.", [], None, "" | |
| limit = PLAN_LIMITS.get(user["plan"], 20) | |
| used = count_invoices_this_month(user["id"]) | |
| rows = [] | |
| last_json = None | |
| processed = 0 | |
| for f in files: | |
| if used + processed >= limit: | |
| break | |
| path = f.name if hasattr(f, "name") else f | |
| try: | |
| data = process_invoice_file(path, user) | |
| except Exception as e: | |
| data = {"filename": os.path.basename(path), "vendor": None, "invoice_number": None, | |
| "invoice_date": None, "due_date": None, "amount": None, "vat_amount": None, | |
| "total": None, "currency": "USD", "status": "review", "confidence": 0, | |
| "warnings": [f"Chyba zpracování: {e}"]} | |
| saved = insert_invoice(user["id"], data) | |
| last_json = data | |
| status_emoji = {"done": "✅ Done", "review": "⚠️ Review", "duplicate": "🔴 Duplicate"}.get(data.get("status"), data.get("status")) | |
| rows.append([ | |
| data.get("filename"), data.get("vendor"), data.get("invoice_number"), | |
| data.get("invoice_date"), data.get("total"), status_emoji, | |
| ]) | |
| processed += 1 | |
| skipped = len(files) - processed | |
| msg = f"✅ Zpracováno {processed} faktur." | |
| if skipped > 0: | |
| msg += f" ⚠️ {skipped} přeskočeno — měsíční limit vyčerpán, upgraduj plán." | |
| used_new = count_invoices_this_month(user["id"]) | |
| limit_str = "∞" if limit == float("inf") else int(limit) | |
| usage = f"**{used_new}/{limit_str} invoices used this month**" | |
| return msg, rows, last_json, usage | |
| upload_btn.click( | |
| do_upload, inputs=[upload_files, current_user], | |
| outputs=[upload_status, upload_table, upload_json, usage_md], | |
| ) | |
| # ========================================================================= | |
| # My Invoices | |
| # ========================================================================= | |
| STATUS_MAP = {"done": "✅ Done", "review": "⚠️ Review", "duplicate": "🔴 Duplicate", "processing": "⟳ Processing"} | |
| def refresh_invoices(user, flt): | |
| if not user: | |
| return [] | |
| invoices = get_invoices(user["id"]) | |
| rows = [] | |
| for inv in invoices: | |
| status = inv.get("status", "done") | |
| if flt == "Done" and status != "done": | |
| continue | |
| if flt == "Review" and status != "review": | |
| continue | |
| if flt == "Duplicates" and not inv.get("is_duplicate"): | |
| continue | |
| rows.append([ | |
| inv.get("id"), inv.get("vendor"), inv.get("inv_number"), | |
| inv.get("inv_date"), inv.get("total"), STATUS_MAP.get(status, status), | |
| ]) | |
| return rows | |
| refresh_invoices_btn.click(refresh_invoices, inputs=[current_user, invoices_filter], outputs=[invoices_table]) | |
| invoices_filter.change(refresh_invoices, inputs=[current_user, invoices_filter], outputs=[invoices_table]) | |
| def do_delete_invoice(user, inv_id): | |
| if not user or not inv_id: | |
| return "⚠️ Zadej platné ID.", [] | |
| delete_invoice(user["id"], int(inv_id)) | |
| return f"✅ Faktura #{int(inv_id)} smazána.", refresh_invoices(user, "All") | |
| delete_invoice_btn.click(do_delete_invoice, inputs=[current_user, delete_id_input], outputs=[delete_status, invoices_table]) | |
| # ========================================================================= | |
| # AI Chat | |
| # ========================================================================= | |
| def chat_respond(message, history, user): | |
| history = history or [] | |
| if not user: | |
| history.append({"role": "assistant", "content": "Musíš být přihlášen."}) | |
| return history, "" | |
| if user["plan"] not in ("pro", "enterprise"): | |
| history.append({"role": "assistant", "content": "🔒 AI Chat je dostupný od plánu Pro. Uprgraduj v záložce ⚡ Upgrade."}) | |
| return history, "" | |
| invoices = get_invoices(user["id"]) | |
| context = json.dumps(invoices[:100], default=str, ensure_ascii=False)[:6000] | |
| history.append({"role": "user", "content": message}) | |
| if GROQ_CLIENT: | |
| try: | |
| resp = GROQ_CLIENT.chat.completions.create( | |
| model="llama-3.1-8b-instant", | |
| messages=[ | |
| {"role": "system", "content": f"You are an assistant answering questions about the user's invoices. Here is their invoice data as JSON: {context}. Answer concisely based only on this data."}, | |
| {"role": "user", "content": message}, | |
| ], | |
| max_tokens=400, temperature=0.2, timeout=10, | |
| ) | |
| answer = resp.choices[0].message.content | |
| except Exception as e: | |
| answer = f"⚠️ AI momentálně nedostupné ({e})." | |
| else: | |
| answer = "⚠️ AI chat vyžaduje nastavený GROQ_API_KEY." | |
| history.append({"role": "assistant", "content": answer}) | |
| return history, "" | |
| chat_send_btn.click(chat_respond, inputs=[chat_input, chatbot, current_user], outputs=[chatbot, chat_input]) | |
| chat_input.submit(chat_respond, inputs=[chat_input, chatbot, current_user], outputs=[chatbot, chat_input]) | |
| # ========================================================================= | |
| # Export | |
| # ========================================================================= | |
| def export_csv(user): | |
| if not user: | |
| return None | |
| invoices = get_invoices(user["id"]) | |
| path = f"/tmp/omniparse_export_{user['id']}.csv" | |
| import csv | |
| with open(path, "w", newline="", encoding="utf-8") as f: | |
| writer = csv.writer(f) | |
| writer.writerow(["ID", "Vendor", "Invoice#", "Date", "Due Date", "Amount", "VAT", "Total", "Currency", "Status"]) | |
| for inv in invoices: | |
| writer.writerow([inv.get("id"), inv.get("vendor"), inv.get("inv_number"), inv.get("inv_date"), | |
| inv.get("due_date"), inv.get("amount"), inv.get("vat_amount"), inv.get("total"), | |
| inv.get("currency"), inv.get("status")]) | |
| return path | |
| def export_json(user): | |
| if not user: | |
| return None | |
| if user["plan"] == "free": | |
| return None | |
| invoices = get_invoices(user["id"]) | |
| path = f"/tmp/omniparse_export_{user['id']}.json" | |
| with open(path, "w", encoding="utf-8") as f: | |
| json.dump(invoices, f, default=str, ensure_ascii=False, indent=2) | |
| return path | |
| export_csv_btn.click(export_csv, inputs=[current_user], outputs=[export_csv_file]) | |
| export_json_btn.click(export_json, inputs=[current_user], outputs=[export_json_file]) | |
| # ========================================================================= | |
| # Upgrade / Stripe | |
| # ========================================================================= | |
| def do_upgrade(plan, user): | |
| if not user: | |
| return "⚠️ Musíš být přihlášen." | |
| url, err = create_checkout_session(plan, user) | |
| if err: | |
| return f"⚠️ {err}" | |
| return f"👉 [Klikni pro dokončení platby přes Stripe]({url})\n\nPo zaplacení se vrať sem a vlož `session_id` z URL níže." | |
| upgrade_btn.click(do_upgrade, inputs=[upgrade_plan_dd, current_user], outputs=[upgrade_link]) | |
| def do_check_payment(session_id, user): | |
| if not user or not session_id: | |
| return "⚠️ Zadej session_id." | |
| return check_payment_status(session_id, user["id"]) | |
| check_payment_btn.click(do_check_payment, inputs=[session_id_input, current_user], outputs=[payment_status_md]) | |
| # ========================================================================= | |
| # API tab | |
| # ========================================================================= | |
| def load_api_tab(user): | |
| if not user: | |
| return gr.update(visible=True, value="Musíš být přihlášen."), gr.update(visible=False), gr.update(visible=False) | |
| if user["plan"] not in ("pro", "enterprise"): | |
| return (gr.update(visible=True, value="🔒 API je dostupné od plánu Pro. Uprgraduj v záložce ⚡ Upgrade."), | |
| gr.update(visible=False), gr.update(visible=False)) | |
| key = user.get("api_key") or "—" | |
| snippet = f"""```bash | |
| curl -X POST {APP_URL}/api/extract \\ | |
| -H "Authorization: Bearer {key}" \\ | |
| -F "file=@invoice.pdf" | |
| ```""" | |
| return (gr.update(visible=False), gr.update(visible=True, value=f"**Tvůj API klíč:** `{key}`"), | |
| gr.update(visible=True, value=snippet)) | |
| current_user.change(load_api_tab, inputs=[current_user], outputs=[api_lock_msg, api_key_display, api_docs]) | |
| # ========================================================================= | |
| # Profile | |
| # ========================================================================= | |
| def load_profile(user): | |
| if not user: | |
| return "Nepřihlášen." | |
| plan = PLAN_LABELS.get(user["plan"], user["plan"]) | |
| return (f"**Jméno:** {user['name']}\n\n**Email:** {user['email']}\n\n" | |
| f"**Plán:** {plan} (${PLAN_PRICES.get(user['plan'], 0)}/měs)\n\n" | |
| f"**Vytvořeno:** {user.get('created_at', '—')[:10]}") | |
| current_user.change(load_profile, inputs=[current_user], outputs=[profile_info]) | |
| def do_change_password(user, new_pw): | |
| if not user: | |
| return "⚠️ Musíš být přihlášen." | |
| if not new_pw or len(new_pw) < 8: | |
| return "⚠️ Heslo musí mít alespoň 8 znaků." | |
| update_user(user["id"], {"password": hash_pw(new_pw)}) | |
| return "✅ Heslo změněno." | |
| change_pw_btn.click(do_change_password, inputs=[current_user, new_password], outputs=[change_pw_status]) | |
| def do_delete_account(user, token): | |
| if not user: | |
| return "⚠️ Musíš být přihlášen.", None, None | |
| if USE_SUPABASE: | |
| sb.table("invoices").delete().eq("user_id", user["id"]).execute() | |
| sb.table("users").delete().eq("id", user["id"]).execute() | |
| else: | |
| conn = _sqlite_conn() | |
| conn.execute("DELETE FROM invoices WHERE user_id=?", (user["id"],)) | |
| conn.execute("DELETE FROM users WHERE id=?", (user["id"],)) | |
| conn.commit() | |
| conn.close() | |
| delete_session(token) | |
| return "✅ Účet smazán.", None, None | |
| delete_account_btn.click( | |
| do_delete_account, inputs=[current_user, session_token], | |
| outputs=[delete_account_status, current_user, session_token], | |
| ) | |
| if __name__ == "__main__": | |
| demo.queue() | |
| demo.launch() | |