import gradio as gr
import pandas as pd
import requests
import json
import pdfplumber
import re
import spaces
from processor import process_invoices_backend
from google_sheet_sync import (
fetch_all_from_sheet, push_rules_to_sheet, push_template_file_to_sheet,
load_template_bytes_from_sheet
)
from pdf_engine import extract_header_value
WEB_APP_URL = "https://script.google.com/macros/s/AKfycbxxKu82K0V3VFn7tRlj3ddzB7Wy-iPyrupAV9tZzjBdhcXZPWFIXodJv6D03M-X_KSn/exec"
SPREADSHEET_ID = "1GJlIl_ANRPs4lID1aZ4WZiWVg_m5AQd7cuQSx6LCnxw"
def fetch_shipper_names():
try:
data = fetch_all_from_sheet()
if isinstance(data, dict) and "shippers" in data:
names = list(data["shippers"].keys())
if names:
return gr.update(choices=names, value=None)
json_db_url = f"https://docs.google.com/spreadsheets/d/{SPREADSHEET_ID}/gviz/tq?tqx=out:csv&sheet=Shipper_JSON_Database"
df_json = pd.read_csv(json_db_url)
if not df_json.empty and "ShipperName" in df_json.columns:
names = df_json["ShipperName"].dropna().tolist()
if names:
return gr.update(choices=names, value=None)
except Exception:
pass
return gr.update(choices=[], value=None)
@spaces.GPU
def process_invoice_action(shipper, main_inv, gst_inv, deec_decl):
if not shipper:
return None, "⚠️ Kripya pehle shipper chunein!"
sheet_data = fetch_all_from_sheet()
shippers_dict = sheet_data.get("shippers", {}) if sheet_data else {}
shipper_info = shippers_dict.get(shipper, {})
file_bytes, filename_or_err = process_invoices_backend(shipper, shipper_info, main_inv, gst_inv, deec_decl)
if file_bytes is None:
return None, f"❌ Error: {filename_or_err}"
import tempfile
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".xlsx")
tmp.write(file_bytes)
tmp.close()
return tmp.name, f"🎉 Success! File '{filename_or_err}' taiyar hai."
@spaces.GPU
def process_rate_pdf(file_obj):
if file_obj is None:
return "Kripya PDF file upload karein!"
try:
with pdfplumber.open(file_obj.name) as pdf:
text = ""
for page in pdf.pages:
t = page.extract_text()
if t: text += t + "\n"
date_match = re.search(r"w\.e\.f[\s\.:]*([\d]{2}[\-\/][\d]{2}[\-\/][\d]{4})", text, re.IGNORECASE)
rate_date = date_match.group(1).strip() if date_match else "N/A"
return f"🎉 Rate PDF Successfully Parsed! w.e.f: {rate_date}"
except Exception as e:
return f"Error: {str(e)}"
def verify_password(pwd):
if pwd == "CKJ":
return gr.update(visible=False), gr.update(visible=True)
else:
return gr.update(visible=True), gr.update(visible=False)
def verify_admin(pwd):
if pwd == "TEST":
return gr.update(visible=False), gr.update(visible=True)
else:
return gr.update(visible=True), gr.update(visible=False)
def on_shipper_select(shipper):
if shipper and shipper.strip():
return gr.update(visible=True)
return gr.update(visible=False)
def on_file_upload(file_obj):
if file_obj is not None:
return gr.update(visible=True)
return gr.update(visible=False)
custom_css = """
.creator-card {
background: linear-gradient(135deg, #1e3c72 0%, #2a5298 100%);
padding: 10px;
border-radius: 8px;
color: white;
text-align: center;
box-shadow: 0 4px 6px rgba(0,0,0,0.15);
margin-bottom: 8px;
margin-top: 6px;
max-width: 220px;
margin-left: auto;
margin-right: auto;
}
.creator-name {
font-size: 15px;
font-weight: 700;
margin-top: 4px;
margin-bottom: 2px;
}
.creator-title {
font-size: 10px;
color: #d1d8e0;
letter-spacing: 1px;
text-transform: uppercase;
font-weight: 600;
}
"""
with gr.Blocks(theme=gr.themes.Soft(), css=custom_css) as demo:
# 🔒 1. Global Password Lock Screen
with gr.Column(visible=True) as login_screen:
with gr.Row():
gr.Column(scale=1)
with gr.Column(scale=2):
gr.Markdown("
🚢 CK Export Invoice Processor Pro
")
gr.Markdown("Kripya aage badhne ke liye app ka password darj karein.
")
pass_input = gr.Textbox(label="Password darj karein:", type="password")
unlock_btn = gr.Button("Unlock App", variant="primary")
gr.Column(scale=1)
# 🚀 2. Main Application Layout
with gr.Column(visible=False) as main_app:
with gr.Row():
# 📌 Left Sidebar
with gr.Column(scale=1, min_width=240):
try:
gr.Image("ck_photo.jpg", show_label=False, interactive=False, width=210)
except:
gr.Markdown("*(Chetan Joshi Photo)*")
gr.HTML("""
Chetan Joshi
📞 +91 98253 06898
CK Export Invoice Pro v2.0
Enterprise Automation & Precision.
""")
with gr.Row():
gr.Number(value=109.8, label="EUR", interactive=False)
gr.Number(value=128.15, label="GBP", interactive=False)
gr.Number(value=94.8, label="USD", interactive=False)
gr.HTML("📅 w.e.f: N/A
")
gr.Markdown("---")
gr.Markdown("##### 💱 Customs Exchange Rates")
pdf_rate_input = gr.File(label="Upload Rate PDF", file_types=[".pdf"])
rate_output = gr.Textbox(label="Status", interactive=False)
pdf_rate_input.change(fn=process_rate_pdf, inputs=pdf_rate_input, outputs=rate_output)
# 📌 Right Main Dashboard
with gr.Column(scale=3):
with gr.Column(visible=True) as user_dashboard:
gr.Markdown("# 🚢 CK Export Invoice Processor")
gr.Markdown("---")
gr.Markdown("### 📥 Invoice Processing Zone (Multi-Document)")
shipper_dropdown = gr.Dropdown(
choices=[],
label="किस शिपर का इनवॉइस प्रोसेस करना है?",
interactive=True,
value=None
)
with gr.Column(visible=False) as upload_section:
gr.Markdown("### 📄 Upload Invoices & Supporting Documents")
with gr.Row():
main_inv_file = gr.File(label="मुख्य इनवॉइस (PDF / Excel) #1")
gst_inv_file = gr.File(label="GST Invoice #1 (PDF/Excel)")
deec_decl_file = gr.File(label="DEEC Decl. #1 (PDF/Excel)")
with gr.Column(visible=False) as process_section:
process_btn = gr.Button("🚀 Process & Generate Excel (ZeroGPU)", variant="primary")
output_file = gr.File(label="📥 Download Generated Excel")
output_status = gr.Textbox(label="Processing Status")
main_inv_file.change(fn=on_file_upload, inputs=main_inv_file, outputs=process_section)
process_btn.click(
fn=process_invoice_action,
inputs=[shipper_dropdown, main_inv_file, gst_inv_file, deec_decl_file],
outputs=[output_file, output_status]
)
shipper_dropdown.change(fn=on_shipper_select, inputs=shipper_dropdown, outputs=upload_section)
gr.Markdown("---")
with gr.Accordion("🛠️ Admin Settings Access", open=False):
admin_pwd_input = gr.Textbox(label="Admin password darj karein:", type="password")
admin_login_btn = gr.Button("Login Karein")
# Admin Mode View with Live Google Sheet Binding
with gr.Column(visible=False) as admin_dashboard:
gr.Markdown("# 🛠️ CK Export Processor - Admin Mode")
admin_back_btn = gr.Button("🚪 Log Out Admin", variant="primary")
admin_tabs = gr.Radio(
["i. 🏢 Add Shipper Name & Setup", "iii. 🌍 Global Masters & Common Dictionaries"],
label="📋 Admin settings chunein:",
value="i. 🏢 Add Shipper Name & Setup"
)
with gr.Column() as admin_content_box:
gr.Markdown("### 🏢 Add Shipper Name & No-Code Visual Mapping Builder")
admin_shipper_dropdown = gr.Dropdown(choices=[], label="1. कॉन्फ़िगर करने के लिए शिपर चुनें:", interactive=True)
gr.Markdown("---")
gr.Markdown("### 📁 2. टेम्पलेट फ़ाइल अपलोड (Full Job Excel Template)")
with gr.Row():
tpl_file_input = gr.File(label="Blank Full Job Excel Format File (Template) चुनें", file_types=[".xlsx", ".xls"])
tpl_save_btn = gr.Button("🚀 Save Template to Google Sheet", variant="primary")
tpl_status_box = gr.Textbox(label="Template Status", interactive=False)
gr.Markdown("---")
gr.Markdown("### 🧪 3. Sample PDF Upload & Text Viewer")
sample_pdf_input = gr.File(label="टेस्ट करने के लिए सैंपल इनवॉइस PDF अपलोड करें", file_types=[".pdf"])
sample_text_output = gr.Textbox(label="Extracted PDF Text Preview", lines=6, interactive=False)
def load_sample_pdf_text(file_obj):
if file_obj is None:
return "कोई फाइल अपलोड नहीं की गई।"
try:
with pdfplumber.open(file_obj.name) as pdf:
txt = "".join([p.extract_text() or "" for p in pdf.pages])
return txt[:3000] + "\n...(Text truncated)..." if len(txt) > 3000 else txt
except Exception as e:
return f"Error reading PDF: {str(e)}"
sample_pdf_input.change(fn=load_sample_pdf_text, inputs=sample_pdf_input, outputs=sample_text_output)
gr.Markdown("---")
gr.Markdown("### ⚡ 4. Smart Test & Save Generator (Box & Position)")
with gr.Row():
test_target_input = gr.Textbox(label="1. टारगेट वैल्यू / फील्ड नाम:")
test_kw_input = gr.Textbox(label="2. मुख्य कीवर्ड:")
test_pos_dropdown = gr.Dropdown(["Right (आगे)", "📦 Extract Inside Box (डब्बे के अंदर का टेक्स्ट)", "Below (नीचे)"], label="3. दिशा / तरीका:", value="Right (आगे)")
test_index_number = gr.Number(value=1, label="4. Index:")
test_run_btn = gr.Button("🚀 Run Live Single Field Inspection", variant="primary")
test_result_output = gr.Textbox(label="Inspection Result", interactive=False)
def run_live_inspection(shipper_name, sample_file, target_field, keyword, direction, index_val):
if not sample_file:
return "⚠️ कृपया पहले सैंपल PDF अपलोड करें!"
if not keyword:
return "⚠️ कृपया मुख्य कीवर्ड दर्ज करें!"
try:
with pdfplumber.open(sample_file.name) as pdf:
pdf_text = "".join([p.extract_text() or "" for p in pdf.pages])
pdf_lines = pdf_text.split("\n")
# Call core extraction engine
res = extract_header_value(pdf_lines, pdf_text, keyword, direction, "Exact Word", "", "None", field_label=target_field)
return f"🎯 Result Found: {res}" if res else "❌ No value found for this keyword/direction."
except Exception as e:
return f"Error during inspection: {str(e)}"
test_run_btn.click(
fn=run_live_inspection,
inputs=[admin_shipper_dropdown, sample_pdf_input, test_target_input, test_kw_input, test_pos_dropdown, test_index_number],
outputs=test_result_output
)
gr.Markdown("---")
gr.Markdown("### 🛠️ 5. Header Fields Mapping & Regex Rules")
header_rules_table = gr.Dataframe(
headers=["Field Name", "Source Doc", "Keyword", "Cell", "Prompt", "Result Ex"],
datatype=["str", "str", "str", "str", "str", "str"],
row_count=5,
col_count=6,
label="Header Rules Configuration",
interactive=True
)
gr.Markdown("---")
gr.Markdown("### 📋 6. Dynamic Item Table Rules & Mapping")
parser_selector = gr.Dropdown(["parser_welspun", "parser_polycab", "parser_bkt", "parser_vapi_welspun"], label="इस शिपर के लिए आइटम पार्सर चुनें:", value="parser_welspun")
item_rules_table = gr.Dataframe(
headers=["Item Field Name", "Excel Col", "Source Type", "Extraction Rule / Keyword", "Result Example"],
datatype=["str", "str", "str", "str", "str"],
row_count=5,
col_count=5,
label="Item Table Rules Configuration",
interactive=True
)
gr.Markdown("---")
gr.Markdown("### ⚙️ 7. IGST & Lut Configuration")
with gr.Row():
lut_keywords_input = gr.Textbox(label="LUT Keywords (कॉमा से अलग करें):", value="LUT, UNDER LUT, UNDER BOND")
paid_keywords_input = gr.Textbox(label="Paid Keywords (कॉमा से अलग करें):", value="SUPPLY MEANT FOR EXPORT ON PAYMENT OF IGST.")
save_all_rules_btn = gr.Button("💾 Save All Rules & Sync to Google Sheet", variant="primary", size="lg")
save_status_output = gr.Textbox(label="Sync Status", interactive=False)
# 🔄 Live Data Populator when Shipper is selected in Admin
def on_admin_shipper_change(shipper_name):
if not shipper_name:
return "", [], [], "parser_welspun", "LUT, UNDER LUT, UNDER BOND", "SUPPLY MEANT FOR EXPORT ON PAYMENT OF IGST."
sheet_data = fetch_all_from_sheet()
shippers_dict = sheet_data.get("shippers", {}) if sheet_data else {}
s_info = shippers_dict.get(shipper_name, {})
# 1. Template Status
has_tpl = load_template_bytes_from_sheet(shipper_name) is not None
tpl_msg = "✅ Template Excel File is Uploaded & Saved in Google Sheet." if has_tpl else "❌ No template uploaded for this shipper."
# 2. Header Rules Table Data
mapping_rules = s_info.get("mapping_rules", {})
h_rows = []
for f_name, f_val in mapping_rules.items():
h_rows.append([
f_name,
f_val.get("logic", "Main Invoice"),
f_val.get("keyword", ""),
f_val.get("cell", ""),
f_val.get("ai_prompt", ""),
f_val.get("result_example", "")
])
if not h_rows:
h_rows = [["", "", "", "", "", ""]]
# 3. Item Table Rules Data
item_rules = s_info.get("item_table_rules", {})
i_rows = []
for i_name, i_val in item_rules.items():
i_rows.append([
i_name,
i_val.get("col", ""),
i_val.get("type", "PDF Row Item"),
i_val.get("rule", ""),
i_val.get("result_example", "")
])
if not i_rows:
i_rows = [["", "", "", "", ""]]
# 4. Parser Name
p_name = s_info.get("item_table_rule_name", "parser_welspun")
# 5. IGST Config
igst_cfg = s_info.get("igst_config", {})
lut_kw = igst_cfg.get("lut_keywords", "LUT, UNDER LUT, UNDER BOND")
paid_kw = igst_cfg.get("paid_keywords", "SUPPLY MEANT FOR EXPORT ON PAYMENT OF IGST.")
return tpl_msg, h_rows, i_rows, p_name, lut_kw, paid_kw
admin_shipper_dropdown.change(
fn=on_admin_shipper_change,
inputs=admin_shipper_dropdown,
outputs=[tpl_status_box, header_rules_table, item_rules_table, parser_selector, lut_keywords_input, paid_keywords_input]
)
# Event Handlers & Initial Data Load
def update_admin_shippers():
names_update = fetch_shipper_names()
return names_update, names_update
demo.load(fn=update_admin_shippers, outputs=[shipper_dropdown, admin_shipper_dropdown])
unlock_btn.click(fn=verify_password, inputs=pass_input, outputs=[login_screen, main_app])
pass_input.submit(fn=verify_password, inputs=pass_input, outputs=[login_screen, main_app])
admin_login_btn.click(fn=verify_admin, inputs=admin_pwd_input, outputs=[user_dashboard, admin_dashboard])
admin_back_btn.click(fn=lambda: (gr.update(visible=True), gr.update(visible=False)), outputs=[user_dashboard, admin_dashboard])
if __name__ == "__main__":
demo.launch()