""" 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"[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 = """

⚡ Invoice processing in seconds, not hours.

AI extracts vendor, dates, amounts and line items from any PDF or image. Export to CSV, JSON or Excel. Connect via API.

99.2% accuracy · <4s per invoice · 40+ formats
How it works
1
Upload
PDF nebo obrázek faktury
2
Extract
AI vytáhne všechna data
3
Export
CSV, JSON, Excel nebo API
Features

🔍 OCR + LLM

Tesseract + Groq Llama 3.1 pro maximální přesnost.

🚫 Duplicate Detection

Pro+, zachytí dvojí platby automaticky.

🤖 AI Chat Agent

Ptej se na faktury přirozenou angličtinou.

✅ Cross-field Validation

Kontroluje součty, data a DPH sazby.

👥 Human-in-the-loop

Enterprise, ruční review pochybných faktur.

🔌 REST API

Napojení na vlastní ERP systém.
Co říkají zákazníci
"Ušetřili jsme desítky hodin měsíčně na ručním přepisování faktur."— Jana K., CFO
"API integrace do našeho ERP trvala jedno odpoledne."— Tomáš R., CTO
"Konečně nemusím kontrolovat každý řádek ručně."— Petra M., účetní
FAQ
Je moje data v bezpečí?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.
Funguje to na české faktury?Ano, podporujeme i lokální formáty a DPH sazby, včetně CZK.
Jak fungují platby?Přes Stripe, měsíčně nebo ročně, kartou.
Mohu kdykoli zrušit?Ano, zrušení je kdykoli v profilu, bez výpovědní lhůty.
Dostanu daňový doklad?Ano, po každé platbě automaticky na email.
""" FOOTER_HTML = """
⚡ OmniParse AI  ·  © 2026  ·  Terms · Privacy · Disclaimer (viz záložka Legal)
""" 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"
  • {f}
    • " for f in feats) cards += f"""

    {label}

    {price}
    """ return f'
    {cards}
    ' # --------------------------------------------------------------------------- # 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('') 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("
    Pricing
    ") 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()