# -*- coding: utf-8 -*- """ Text2Receipt — HF Space (final clean version) ================================================ Pipeline: free-text Hebrew income note → gemma-2-2b fine-tuned parser (extracts client / items) → fiscal decision layer (VAT cues, doc-type, allocation) → t2r_core.complete() (deterministic arithmetic) → Gradio UI (render document + FAISS recommendations) """ import json, random, re as _re import numpy as np import gradio as gr import t2r_core as core import datetime as _dt # ───────────────────────────────────────────────────────────────────────────── # 1. Lazy model + FAISS state # ───────────────────────────────────────────────────────────────────────────── BASE_MODEL = "unsloth/gemma-2-2b-it" MODEL_REPO = "yonilev/Text2Receipt-parser" _STATE = dict(ready=False, tok=None, model=None, enc=None, index=None, store=None, e5_family=False, rng=random.Random(42), device="cpu") def _lazy_init(): if _STATE["ready"]: return import torch from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig from peft import PeftModel from sentence_transformers import SentenceTransformer import faiss, pandas as pd device = "cuda" if torch.cuda.is_available() else "cpu" _STATE["device"] = device _STATE["rng"] = random.Random(42) tok = AutoTokenizer.from_pretrained(BASE_MODEL) if tok.pad_token is None: tok.pad_token = tok.eos_token tok.padding_side = "left" if device == "cuda": bnb = BitsAndBytesConfig(load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.float16, bnb_4bit_use_double_quant=True) try: base = AutoModelForCausalLM.from_pretrained( BASE_MODEL, quantization_config=bnb, device_map="auto", torch_dtype=torch.float16, attn_implementation="eager") model = PeftModel.from_pretrained(base, MODEL_REPO) print("✅ GPU: fine-tuned adapter loaded") except Exception as e: print(f"⚠ adapter failed ({e}); base model (GPU)") model = AutoModelForCausalLM.from_pretrained( BASE_MODEL, quantization_config=bnb, device_map="auto", torch_dtype=torch.float16, attn_implementation="eager") else: try: base = AutoModelForCausalLM.from_pretrained( BASE_MODEL, device_map="cpu", torch_dtype=torch.float32, attn_implementation="eager") model = PeftModel.from_pretrained(base, MODEL_REPO) print("✅ CPU: fine-tuned adapter loaded") except Exception as e: print(f"⚠ adapter failed ({e}); base model (CPU)") model = AutoModelForCausalLM.from_pretrained( BASE_MODEL, device_map="cpu", torch_dtype=torch.float32, attn_implementation="eager") model.eval() _STATE["tok"] = tok _STATE["model"] = model emb = np.load("receipts_embeddings.npy").astype("float32") store = pd.read_parquet("receipts_store.parquet") with open("embeddings_manifest.json") as f: manifest = json.load(f) faiss.normalize_L2(emb) index = faiss.IndexFlatIP(emb.shape[1]) index.add(emb) enc = SentenceTransformer(manifest["embed_model"], device=device) _STATE.update(enc=enc, index=index, store=store, e5_family=manifest.get("e5_family", False), ready=True) print(f"✅ FAISS ready — {emb.shape[0]} vectors") # ───────────────────────────────────────────────────────────────────────────── # 2. Parser + post-processing cleaner # ───────────────────────────────────────────────────────────────────────────── INSTRUCTION = ( "אתה ממיר הערת הכנסה חופשית בעברית למבנה JSON. " "חלץ אך ורק את מה שכתוב בהערה: שם הלקוח (client_name), " "האם הלקוח עסק (client_is_business), ורשימת פריטים (items) " "כאשר לכל פריט תיאור (description), מחיר ליחידה (unit_price) וכמות (quantity). " "החזר JSON תקין בלבד, ללא טקסט נוסף." ) _VAT_NOISE = [ "לפני מע\"מ", "כולל מע\"מ", "לפני מע״מ", "כולל מע״מ", "בתוספת מע\"מ", "+ מע\"מ", "+מע\"מ", "מע\"מ", "מע״מ", "מעמ", ] _NOISE_SET = {"לפני", "כולל", "נטו", "ברוטו", "מס", "+", "פלוס", "מע\"מ", "מע״מ", "מעמ"} def _build_prompt(raw_text): return f"{INSTRUCTION}\n\nהערה: {raw_text}\n\nJSON:" def _extract_json(text): s = text.find("{") if s < 0: return None depth = 0 for i in range(s, len(text)): if text[i] == "{": depth += 1 elif text[i] == "}": depth -= 1 if depth == 0: try: return json.loads(text[s:i+1]) except: return None return None def _clean_name(name): """Strip VAT-related noise the model may accidentally grab as a client name.""" name = (name or "").strip() for tok in _VAT_NOISE: name = name.replace(tok, "").strip(" ,.-+") return "" if (name in _NOISE_SET or len(name) <= 1) else name def _clean_parse(parse): """Post-process raw model output: remove noise, validate structure.""" if not isinstance(parse, dict): return parse # Clean client name parse["client_name"] = _clean_name(parse.get("client_name", "")) # Clean items clean_items = [] for it in (parse.get("items") or []): desc = (it.get("description") or "").strip() for tok in _VAT_NOISE: desc = desc.replace(tok, "").strip(" ,.-+") if desc in _NOISE_SET or len(desc) <= 0: desc = "שירות" it = dict(it) it["description"] = desc if it.get("unit_price"): clean_items.append(it) parse["items"] = clean_items return parse def model_parse(raw_text): tok, model = _STATE["tok"], _STATE["model"] msgs = [{"role": "user", "content": _build_prompt(raw_text)}] prompt = tok.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True) enc = tok(prompt, return_tensors="pt", add_special_tokens=False).to(model.device) import torch with torch.no_grad(): out = model.generate(**enc, max_new_tokens=128, do_sample=False, pad_token_id=tok.pad_token_id) decoded = tok.decode(out[0, enc["input_ids"].shape[1]:], skip_special_tokens=True) raw = _extract_json(decoded) return _clean_parse(raw) if raw else None def _missing_fields(parse): if parse is None: return ["parse_failed"] m = [] if not parse.get("client_name", "").strip(): m.append("client_name") if not parse.get("items"): m.append("items") return m # ───────────────────────────────────────────────────────────────────────────── # 3. FAISS recommender # ───────────────────────────────────────────────────────────────────────────── def recommend(query_text, k=3): pref = "query: " if _STATE["e5_family"] else "" q_emb = _STATE["enc"].encode([pref + query_text], normalize_embeddings=True).astype("float32") _, I = _STATE["index"].search(q_emb, k + 1) out = [] for idx in I[0][:k]: row = _STATE["store"].iloc[int(idx)] out.append({"raw_text": row["raw_text"], "category": row["category"], "doc_type_he": row["doc_type_he"], "total": float(row["total"]), "client_name": row["client_name"]}) return out # ───────────────────────────────────────────────────────────────────────────── # 4. Fiscal decision layer (app-level, never touches frozen t2r_core.py) # ───────────────────────────────────────────────────────────────────────────── DEMO_ISSUER_BASE = {"name": "מים שקטים", "tax_id": "962569844"} _GROSS_PATTERNS = [r"כולל\s*מע[\"״]?מ", r"כולל\s*מס", r"גרוס", r"ברוטו", r"מחיר\s*כולל"] _NET_PATTERNS = [r"לפני\s*מע[\"״]?מ", r"\+\s*מע[\"״]?מ", r"בתוספת\s*מע[\"״]?מ", r"פלוס\s*מע[\"״]?מ", r"נטו", r"לא\s*כולל\s*מע[\"״]?מ", r"בלי\s*מע[\"״]?מ", r"לפני\s*מס"] def _text_has(text, patterns): t = text.lower() return any(_re.search(p, t) for p in patterns) def _detect_date(text): for pat, order in [(r"(\d{1,2})[/\-\.](\d{1,2})[/\-\.](\d{2,4})", "dmy"), (r"(\d{4})[/\-\.](\d{1,2})[/\-\.](\d{1,2})", "ymd")]: m = _re.search(pat, text) if m: try: g = [int(x) for x in m.groups()] d, mo, y = (g[0], g[1], g[2]) if order == "dmy" else (g[2], g[1], g[0]) if y < 100: y += 2000 return f"{y:04d}-{mo:02d}-{d:02d}" except Exception: pass return None def _today_str(): return _dt.date.today().isoformat() def decide_fiscal(note, issuer_status, parse, doc_radio, clarify_vat=None, clarify_tax_id=None, clarify_name=None): """ Returns (issuer_dict, final_parse, questions_dict). questions_dict empty → ready to generate. """ # ── 1. Issuer required — early return if missing ────────────────────── if not issuer_status: return None, None, {"issuer_status": "בחר סטטוס מנפיק למעלה (עוסק פטור / עוסק מורשה)."} questions = {} exempt = (issuer_status == "פטור") issuer = {**DEMO_ISSUER_BASE, "status": "exempt_dealer" if exempt else "authorized_dealer", "is_company": False} # ── 2. Mandatory content ────────────────────────────────────────────── if clarify_name and clarify_name.strip(): parse["client_name"] = clarify_name.strip() if not parse.get("client_name", "").strip(): questions["client_name"] = "שם הלקוח / המשלם?" if not parse.get("items"): questions["items"] = "מה השירות/המוצר ובאיזה סכום? (למשל: ייעוץ ₪500)" # ── 3. Date: detect from note → today ──────────────────────────────── parse = dict(parse) parse["date"] = _detect_date(note) or _today_str() # ── 4. Doc type + VAT basis ─────────────────────────────────────────── is_gross = _text_has(note, _GROSS_PATTERNS) or clarify_vat == "כולל מע״מ" is_net = _text_has(note, _NET_PATTERNS) or clarify_vat == "לפני מע״מ" has_vat_cue = is_gross or is_net if exempt: doc_type, amount_basis = "receipt", "net" elif doc_radio and doc_radio != "זיהוי אוטומטי": doc_type = {"קבלה": "receipt", "חשבונית מס": "tax_invoice", "חשבונית מס וקבלה": "tax_invoice_receipt"}.get(doc_radio, "receipt") if doc_type != "receipt": if is_gross: amount_basis = "gross" elif is_net: amount_basis = "net" elif clarify_vat in (None, "לא צוין"): total = sum(i.get("unit_price", 0) * i.get("quantity", 1) for i in parse.get("items", [])) questions["vat_basis"] = f"הסכום (₪{total:,.0f}) — כולל מע״מ או לפני מע״מ?" amount_basis = "net" else: amount_basis = "net" else: amount_basis = "net" else: # Auto: VAT cue in note → tax_invoice_receipt; otherwise → receipt (no VAT line) if has_vat_cue: doc_type, amount_basis = "tax_invoice_receipt", "gross" if is_gross else "net" else: doc_type, amount_basis = "receipt", "net" # ── 5. Allocation: only בע"מ / חברה + amount ≥ threshold ───────────── if doc_type in ("tax_invoice", "tax_invoice_receipt") and not exempt: subtotal = sum(it.get("unit_price", 0) * it.get("quantity", 1) for it in parse.get("items", [])) threshold = core.allocation_threshold_for_date( _dt.date.fromisoformat(parse["date"])) name = parse.get("client_name", "") is_corp = any(kw in name for kw in ['בע"מ', "בעמ", "חברה", "בע'מ", "בע.מ"]) if is_corp and subtotal >= threshold: if clarify_tax_id and clarify_tax_id.strip(): parse["client_tax_id"] = clarify_tax_id.strip() elif not parse.get("client_tax_id"): questions["client_tax_id"] = f"ח.פ. הלקוח (נדרש להקצאה — עסקה ≥ ₪{threshold:,})" # ── 6. Finalise parse ───────────────────────────────────────────────── parse.update({"doc_type": doc_type, "amount_basis": amount_basis}) parse.setdefault("payment_method", "bank_transfer") parse.setdefault("currency", "ILS") parse.setdefault("client_tax_id", None) parse.setdefault("client_is_business", False) parse.setdefault("client_name", "—") parse.setdefault("items", []) return issuer, parse, questions def _build_clarification_message(questions): icons = {"issuer_status": "👤", "client_name": "👤", "items": "📦", "vat_basis": "💰", "client_tax_id": "🔖"} lines = "\n".join(f"{icons.get(k,'•')} {v}" for k, v in questions.items()) return f"❓ **כדי להפיק את המסמך, חסר מידע:**\n\n{lines}\n\nהוסף את הפרטים לפתק ולחץ **הפק מסמך** שוב." # ───────────────────────────────────────────────────────────────────────────── # 5. HTML renderers # ───────────────────────────────────────────────────────────────────────────── def render_parse_panel(parse, completed, show_vat): items = parse.get("items", []) item_rows = "".join( f'
{it.get("description","")}' f'{it.get("quantity",1)} × ₪{it.get("unit_price",0):,.0f}
' for it in items) chips = ( f'לקוח: {parse.get("client_name","—")} ' f'' f'{"עסק" if parse.get("client_is_business") else "פרטי"} ' f'{len(items)} פריטים') alloc_line = ( '
' 'מספר הקצאה ✓ נדרש
' if completed.get("allocation_required") else "") vat_line = ( f'
' f'מע"מ {int(round(completed.get("vat_rate",0)*100))}% · ₪{completed.get("vat_amount",0):,.2f}
' if show_vat else '
ללא פירוק מע"מ (קבלה)
') return f"""
🧠 המודל חילץ
{chips}
{item_rows}
gemma-2-2b + LoRA · חוזה רק parse
⚙️ המנוע חישב
{vat_line}
סה"כ ₪{completed.get("total",0):,.2f}
מס׳ מסמך {completed.get("serial_number","")}
{alloc_line}
דטרמיניסטי · אפס הזיות בחשבון
""" def render_document(c, show_vat=True): issuer = c.get("issuer", {}) client = c.get("client", {}) lines = c.get("lines", []) vat_pct = int(round(c.get("vat_rate", 0) * 100)) alloc = (f'
' f'🔖 מספר הקצאה: {c.get("allocation_number","—")}
' if c.get("allocation_required") else "") rows = "".join( f'' f'{ln["description"]}' f'{ln["quantity"]}' f'₪{ln["unit_price"]:,.2f}' f'₪{ln["line_total"]:,.2f}' f'' for ln in lines) vat_block = "" if not show_vat else f"""
סכום לפני מע"מ ₪{c.get("subtotal",0):,.2f}
מע"מ {vat_pct}% ₪{c.get("vat_amount",0):,.2f}
""" return f"""
{c.get("doc_type_he","מסמך")}
מס׳ {c.get("serial_number","")}  ·  {c.get("issue_date","")}
{issuer.get("name","")}
ח.פ. {issuer.get("tax_id","")}
{"עוסק מורשה" if issuer.get("status")=="authorized_dealer" else "עוסק פטור"}
{(client.get("name","?") or "?")[0]}
{client.get("name","")}
{"עסק" if client.get("is_business") else "לקוח פרטי"} {f" · ח.פ. {client.get('tax_id','')}" if client.get("tax_id") else ""}
{alloc} {rows}
פריט כמות מחיר סה"כ
{vat_block}
סה"כ לתשלום ₪{c.get("total",0):,.2f}
{core.PAYMENT_HE.get(c.get("payment_method",""),"—")}
""" def render_recommendations(recs): if not recs: return "" cards = "".join(f"""
{r["doc_type_he"]} · {r["category"]}
{r["raw_text"][:72]}{"…" if len(r["raw_text"])>72 else ""}
{r["client_name"]} ₪{r["total"]:,.0f}
""" for r in recs) return f"""
מסמכים דומים מהמאגר
{cards}
""" # ───────────────────────────────────────────────────────────────────────────── # 6. Main pipeline # ───────────────────────────────────────────────────────────────────────────── def generate(note, issuer_radio, doc_choice, clarify_vat, clarify_tax_id, clarify_name): """ Stateless single handler. Returns: (status_md, parse_html, doc_html, recs_html, name_vis, vat_vis, tax_vis) """ note = (note or "").strip() if not note: return "✏️ כתוב פתק הכנסה ולחץ הפק מסמך.", "", "", "", False, False, False issuer_status = issuer_radio if issuer_radio in ("פטור", "מורשה") else None if not issuer_status: return "👤 בחר תחילה: **עוסק פטור** או **עוסק מורשה** למעלה.", "", "", "", False, False, False _lazy_init() parse = model_parse(note) if parse is None or _missing_fields(parse) == ["parse_failed"]: return ("🤔 לא הצלחתי לחלץ פרטים. נסח מחדש — לדוגמה: 'קיבלתי 500₪ ממשה על ייעוץ'.", "", "", "", False, False, False) cvat = clarify_vat if clarify_vat not in (None, "לא צוין") else None ctax = clarify_tax_id.strip() if clarify_tax_id and clarify_tax_id.strip() else None cname = clarify_name.strip() if clarify_name and clarify_name.strip() else None issuer, final, questions = decide_fiscal( note, issuer_status, parse, doc_choice, cvat, ctax, cname) if questions: show_name_q = "client_name" in questions show_vat_q = "vat_basis" in questions show_taxid_q = "client_tax_id" in questions return (_build_clarification_message(questions), "", "", "", show_name_q, show_vat_q, show_taxid_q) try: completed = core.complete(issuer, final, _STATE["rng"]) except Exception as e: return f"⚠️ שגיאה בעיבוד: {e}", "", "", "", False, False, False show_vat = (final["doc_type"] != "receipt") return ("✅ המסמך הופק בהצלחה.", render_parse_panel(final, completed, show_vat), render_document(completed, show_vat), render_recommendations(recommend(note)), False, False, False) # ───────────────────────────────────────────────────────────────────────────── # 7. Gradio UI # ───────────────────────────────────────────────────────────────────────────── CSS = """ .gradio-container { max-width: 880px !important; margin: 0 auto !important; } footer { display: none !important; } .qs-btn { font-size: 13px !important; background: #f1f5f9 !important; border: 1px solid #e2e8f0 !important; color: #0f172a !important; } .qs-btn:hover { background: #e2e8f0 !important; } """ QUICK_STARTERS = [ ("💼 עוסק מורשה + לקוח עסקי", 'חברת מעגל בע"מ שילמה 36,500 כולל מע"מ על פרויקט אתר. העברה בנקאית.'), ("🧾 עוסק פטור — שיעור פרטי", "קיבלתי 220 שקל מדביר על שיעור שחייה פרטי, מזומן"), ("📦 כמה פריטים", 'נועה כהן שילמה 2×350 ש"ח ייעוץ עסקי + 180 ש"ח הכנת מצגת, לפני מע"מ'), ] with gr.Blocks(title="Text2Receipt", theme=gr.themes.Soft( primary_hue="teal", secondary_hue="slate")) as demo: gr.HTML("""
🧾
Text2Receipt
פתק הכנסה חופשי בעברית  →  מסמך פיסקלי ישראלי תקין
""") issuer_radio = gr.Radio( ["פטור", "מורשה"], value=None, label="מי אני (סטטוס מנפיק)", info='עוסק פטור = ללא מע"מ · עוסק מורשה = עם מע"מ', ) note_input = gr.Textbox( label="פתק / תוכן ההכנסה", lines=2, rtl=True, autofocus=True, placeholder='לדוגמה: קיבלתי 1,200 ש"ח ממשה כהן על ייעוץ עסקי', ) doc_choice = gr.Radio( ["זיהוי אוטומטי", "קבלה", "חשבונית מס", "חשבונית מס וקבלה"], value="זיהוי אוטומטי", label="סוג מסמך", info="חשבונית מס = דרישת תשלום (טרם שולם) · חשבונית מס וקבלה = שניים באחד (שולם)", ) with gr.Row(): submit_btn = gr.Button("הפק מסמך ⚡", variant="primary", scale=2) clear_btn = gr.Button("נקה 🗑", scale=1) status_output = gr.Markdown("") # Clarification fields — appear only when needed with gr.Group(visible=False) as clarify_name_group: gr.HTML('
👤 שם לקוח נדרש
') clarify_name = gr.Textbox( label="שם הלקוח / המשלם", placeholder="למשל: נועה כהן", value="", ) with gr.Group(visible=False) as clarify_vat_group: gr.HTML('
💰 בסיס מע״מ נדרש
') clarify_vat = gr.Radio( ["כולל מע״מ", "לפני מע״מ", "לא צוין"], value="לא צוין", label="הסכום המצוין בפתק", ) with gr.Group(visible=False) as clarify_tax_group: gr.HTML('
🔖 ח.פ. לקוח נדרש (הקצאה)
') clarify_tax_id = gr.Textbox( label="ח.פ. / ת.ז. לקוח", placeholder="למשל: 514123458", value="", ) gr.HTML('
דוגמאות מהירות
') with gr.Row(): for lbl, ex in QUICK_STARTERS: gr.Button(lbl, elem_classes=["qs-btn"], size="sm").click( fn=lambda e=ex: e, outputs=note_input) parse_output = gr.HTML() doc_output = gr.HTML() recs_output = gr.HTML() # Handlers def _on_generate(note, issuer_r, doc_c, cvat, ctax, cname): status, phtml, dhtml, rhtml, show_name_q, show_vat_q, show_taxid_q = generate( note, issuer_r, doc_c, cvat, ctax, cname) return (status, phtml, dhtml, rhtml, gr.update(visible=show_name_q), gr.update(visible=show_vat_q), gr.update(visible=show_taxid_q)) submit_btn.click( fn=_on_generate, inputs=[note_input, issuer_radio, doc_choice, clarify_vat, clarify_tax_id, clarify_name], outputs=[status_output, parse_output, doc_output, recs_output, clarify_name_group, clarify_vat_group, clarify_tax_group], api_name="generate", ) def _on_clear(): return ("", None, "זיהוי אוטומטי", "לא צוין", "", "", "", "", "", "", gr.update(visible=False), gr.update(visible=False), gr.update(visible=False)) clear_btn.click( fn=_on_clear, outputs=[note_input, issuer_radio, doc_choice, clarify_vat, clarify_tax_id, clarify_name, status_output, parse_output, doc_output, recs_output, clarify_name_group, clarify_vat_group, clarify_tax_group], ) # ───────────────────────────────────────────────────────────────────────────── # 8. Launch # ───────────────────────────────────────────────────────────────────────────── if __name__ == "__main__": demo.launch(css=CSS)