Update app.py

app.py

@@ -1,171 +1,43 @@
+# app.py
 import gradio as gr
-import torch
-from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
-from threading import Thread
-import os
-import pytesseract
-from pdf2image import convert_from_path
-from PIL import Image
-import trafilatura
-import datetime
-import json
-import requests
-import re
-import random
+import ai_engine
+import zoho_client
+import utils
 
-# ==========================================
-# 1. CREDENTIALS
-# ==========================================
-CLIENT_ID = "1000.SIMKGAO5719K0TQ0QZQ31ZU57RLFNQ"
-CLIENT_SECRET = "60b329b4fe51930abee900cba6524ec7332cd67e06"
-
-# *** USE YOUR 'FULL ACCESS' REFRESH TOKEN HERE ***
-REFRESH_TOKEN = "1000.fcbd3eb6544c48591ceb462eefe439e1.e6dacb8875fec5fa150596fa44eb9326"
-ORGANIZATION_ID = "60058878536" # Your Zoho Books Free Org
-
-# CORRECT API FOR YOUR ACCOUNT (Books, not Invoice)
-AUTH_URL = "https://accounts.zoho.in/oauth/v2/token"
-API_BASE = "https://www.zohoapis.in/books/v3" 
-MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct"
-
-# ==========================================
-# 2. ZOHO CONNECTIVITY
-# ==========================================
-def get_zoho_headers():
-    try:
-        print("🔄 Refreshing Token...")
-        params = {
-            "refresh_token": REFRESH_TOKEN,
-            "client_id": CLIENT_ID,
-            "client_secret": CLIENT_SECRET,
-            "grant_type": "refresh_token",
-            "redirect_uri": "http://www.google.com"
-        }
-        resp = requests.post(AUTH_URL, data=params)
-        if resp.status_code != 200: return None, f"Auth Failed: {resp.text}"
-        return {"Authorization": f"Zoho-oauthtoken {resp.json().get('access_token')}"}, None
-    except Exception as e:
-        return None, str(e)
-
-def find_or_create_customer(name, headers):
-    """Creates a CUSTOMER (Required for Sales Invoices)"""
-    if not name or len(str(name)) < 2: name = "MCP Customer"
-    
-    # 1. Search
-    search_url = f"{API_BASE}/contacts"
-    try:
-        params = {
-            "organization_id": ORGANIZATION_ID, 
-            "contact_name_contains": name,
-            "contact_type": "customer" # <--- IMPORTANT: Customer, not Vendor
-        }
-        resp = requests.get(search_url, headers=headers, params=params)
-        contacts = resp.json().get("contacts", [])
-        if contacts: 
-            print(f"✅ Found Customer: {contacts[0]['contact_name']}")
-            return contacts[0]["contact_id"], "Existing"
-    except: pass
-
-    # 2. Create
-    print(f"🆕 Creating Customer: {name}")
-    create_payload = {
-        "contact_name": name,
-        "contact_type": "customer"
-    }
-    create_resp = requests.post(search_url, headers=headers, params={"organization_id": ORGANIZATION_ID}, json=create_payload)
-    
-    if create_resp.status_code == 201:
-        return create_resp.json().get("contact", {}).get("contact_id"), "Created New"
-    else:
-        return None, f"Contact Error: {create_resp.text}"
-
-def push_invoice_to_zoho(payload):
-    headers, error = get_zoho_headers()
-    if error: return error
-    
-    # 1. Get Customer
-    cid, c_status = find_or_create_customer(payload.get("vendor_name_raw"), headers)
-    if not cid: return f"❌ Customer Logic Failed: {c_status}"
-    
-    # 2. Construct Sales Invoice
-    final_json = {
-        "customer_id": cid,
-        "date": payload.get("date", datetime.date.today().strftime("%Y-%m-%d")),
-        "line_items": payload.get("line_items", []),
-        "status": "draft"
-    }
-    
-    # 3. Push
-    print(f"🚀 Creating Sales Invoice for {cid}...")
-    url = f"{API_BASE}/invoices?organization_id={ORGANIZATION_ID}"
-    resp = requests.post(url, headers=headers, json=final_json)
+def process_pipeline(file):
+    # 1. OCR
+    text, img = ai_engine.perform_ocr(file)
+    if not text: return None, "❌ OCR Failed. Upload a valid image/PDF."
     
-    if resp.status_code == 201:
-        inv = resp.json().get("invoice", {})
-        link = f"https://books.zoho.in/app/{ORGANIZATION_ID}#/invoices/{inv.get('invoice_id')}"
-        return f"✅ SUCCESS! Invoice Created.\nNumber: {inv.get('invoice_number')}\nCustomer: {c_status}\nLink: {link}"
-    else:
-        return f"❌ Zoho API Error: {resp.text}"
-
-# ==========================================
-# 3. AI & ORCHESTRATOR
-# ==========================================
-print(">>> Loading AI...")
-try:
-    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
-    model = AutoModelForCausalLM.from_pretrained(MODEL_ID, device_map="cpu", torch_dtype=torch.float32, low_cpu_mem_usage=True)
-except: model = None
-
-def perform_ocr(file_obj):
-    if file_obj is None: return "", 0, None
-    try:
-        filename = os.path.basename(file_obj)
-        if filename.lower().endswith(".pdf"):
-            image = convert_from_path(file_obj, first_page=1, last_page=1)[0]
-        else:
-            image = Image.open(file_obj).convert("RGB")
-        return pytesseract.image_to_string(image), 0.9, image
-    except: return "", 0, None
-
-def query_ai(text):
-    if not model: return {"vendor_name": "Unknown", "total": 0}
+    # 2. AI Extraction
+    data = ai_engine.extract_json(text)
     
-    prompt = f"<|im_start|>user\nExtract JSON: vendor_name, invoice_date (YYYY-MM-DD), total, item_desc\nText:\n{text[:1000]}<|im_end|>\n<|im_start|>assistant\n```json"
-    inputs = tokenizer(prompt, return_tensors="pt")
-    out = model.generate(**inputs, max_new_tokens=200)
-    try:
-        json_str = tokenizer.decode(out[0]).split("```json")[1].split("```")[0].strip()
-        data = json.loads(json_str)
-        return data[0] if isinstance(data, list) else data
-    except:
-        return {"vendor_name": "Unknown", "total": 0}
-
-def run(file):
-    text, score, img = perform_ocr(file)
-    data = query_ai(text)
+    # 3. Data Cleaning (Using utils)
+    rate = utils.clean_amount(data.get("total"))
     
-    try: rate = float(str(data.get("total","0")).replace(",","").replace("INR","").strip())
-    except: rate = 0
-        
     payload = {
-        "vendor_name_raw": data.get("vendor_name"),
+        "vendor_name": data.get("vendor_name"),
         "date": data.get("invoice_date"),
         "line_items": [{"name": data.get("item_desc", "Service"), "rate": rate}]
     }
     
-    return img, push_invoice_to_zoho(payload)
+    # 4. Push to Zoho
+    result = zoho_client.push_invoice(payload)
+    
+    return img, result
 
-# ==========================================
-# 4. UI
-# ==========================================
-with gr.Blocks(title="Zoho Books Invoice Agent") as demo:
-    gr.Markdown("## 🧾 MCP Agent -> Zoho Books (Sales Invoice)")
+# UI Setup
+with gr.Blocks(title="Modular Zoho Agent") as demo:
+    gr.Markdown("## 🧾 Zoho Books Agent (Modular)")
+    
     with gr.Row():
-        f_in = gr.File(label="Upload Document")
-        btn = gr.Button("Create Invoice", variant="primary")
-    out_img = gr.Image(label="View", height=300)
-    out_res = gr.Textbox(label="Status")
-    btn.click(run, f_in, [out_img, out_res])
+        f_in = gr.File(label="Upload Invoice")
+        btn = gr.Button("Process", variant="primary")
+        
+    out_img = gr.Image(label="Preview", height=300)
+    out_res = gr.Textbox(label="Result")
+    
+    btn.click(process_pipeline, f_in, [out_img, out_res])
 
 if __name__ == "__main__":
     demo.launch()