import gradio as gr from graph_manager import user_input_handler from nodes import session_form_data, get_form_info, set_form_info import ast import uuid import logging import sqlite3 import pandas as pd import os logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def generate_session_id(): session_id = str(uuid.uuid4()) return session_id, f"Session ID: {session_id}" DB_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'database', 'hello_earth_data_2.db') def connect_db(): return sqlite3.connect(DB_PATH) def fetch_df(query, params=None): with connect_db() as conn: return pd.read_sql_query(query, conn, params=params or ()) def get_deliverable_data(): return fetch_df("SELECT * FROM Deliverable") def get_deliverable_titles(): df = get_deliverable_data() return ['None'] + df['title'].dropna().tolist() def save_expense_to_db(form_info_dict): deliverable_df = get_deliverable_data() title_to_id = dict(zip(deliverable_df['title'], deliverable_df['deliverable_id'])) associated_title = form_info_dict.get("Associated Deliverable", "None") associated_id = title_to_id.get(associated_title, None) if associated_id is None: logger.warning(f"No matching deliverable ID found for title: {associated_title}") return False expense_id = str(uuid.uuid4()) seller_name = form_info_dict.get("Seller Name") seller_address = form_info_dict.get("Seller Address") seller_phone_number = form_info_dict.get("Seller Phone Number") buyer_name = form_info_dict.get("Buyer Name") buyer_address = form_info_dict.get("Buyer Address") raw_date = form_info_dict.get("Transaction Date") transaction_date = pd.to_datetime(raw_date).date().isoformat() if raw_date else None total_payment_amount = form_info_dict.get("Total Payment Amount") expense_description = form_info_dict.get("Expense Description") status = "approved" insert_query = """ INSERT INTO Expense ( expense_id, associated_deliverable_id, seller_name, seller_address, seller_phone_number, buyer_name, buyer_address, transaction_date, total_payment_amount, expense_description, status ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ with connect_db() as conn: cursor = conn.cursor() cursor.execute(insert_query, ( expense_id, associated_id, seller_name, seller_address, seller_phone_number, buyer_name, buyer_address, transaction_date, total_payment_amount, expense_description, status )) conn.commit() logger.info(f"Expense saved: {expense_id}") return True def is_form_complete(form_info_dict): required_keys = [ "Associated Deliverable", "Seller Name", "Seller Address", "Seller Phone Number", "Buyer Name", "Buyer Address", "Transaction Date", "Total Payment Amount", "Expense Description" ] for key in required_keys: value = form_info_dict.get(key) if not value or value == "None": return gr.update(interactive=False) return gr.update(interactive=True) def update_field_input(key): def inner(session_id, value): set_form_info(session_id, key, value) return get_form_info(session_id) return inner def update_form_information_handler(form_information): valid_titles = get_deliverable_titles() target_deliverable = form_information.get("Associated Deliverable") target_deliverable = target_deliverable if target_deliverable in valid_titles else "None" def clean(val): return None if val == "None" else val seller_name_input = clean(form_information.get("Seller Name")) seller_address_input = clean(form_information.get("Seller Address")) seller_phone_number_input = clean(form_information.get("Seller Phone Number")) buyer_name_input = clean(form_information.get("Buyer Name")) buyer_address_input = clean(form_information.get("Buyer Address")) transaction_date_input = clean(form_information.get("Transaction Date")) total_payment_amount_input = form_information.get("Total Payment Amount") if total_payment_amount_input and total_payment_amount_input != "None": total_payment_amount_input = float(total_payment_amount_input) expense_description_input = clean(form_information.get("Expense Description")) return ( target_deliverable, seller_name_input, seller_address_input, seller_phone_number_input, buyer_name_input, buyer_address_input, transaction_date_input, total_payment_amount_input, expense_description_input ) def handle_submit(session_id): form_data = get_form_info(session_id) required_fields = [ "Associated Deliverable", "Seller Name", "Seller Address", "Seller Phone Number", "Buyer Name", "Buyer Address", "Transaction Date", "Total Payment Amount", "Expense Description" ] for field in required_fields: if not form_data.get(field) or form_data.get(field) == "None": return gr.update(value="❌ กรุณากรอกข้อมูลในทุกช่องให้ครบถ้วน", visible=True) save_expense_to_db(form_data) return gr.update(value="✅ บันทึกค่าใช้จ่ายเรียบร้อยแล้ว", visible=True) _chat_prefill = [ {"role": "assistant", "content": "สวัสดีค่ะ ดิฉันชื่อ เอมมี่ ค่ะ ยินดีช่วยคุณเรื่องการนำส่งรายการค่าใช้จ่ายค่ะ คุณสามารถเริ่มต้นด้วยการอัพโหลดรูปภาพของบิล 1 รูปค่ะ"} ] def chat_handler(message,history,session_id): # global CURRENT_SESSION_ID logger.info(f"Received message: {message}") if len(message['files']) > 0: response = user_input_handler(str(message['files'][0]), session_id) elif message['text'] != "": try: response = user_input_handler(message['text'], session_id) except Exception as e: raise e form_information = get_form_info(session_id) update_form_information_handler(form_information) return response, form_information def create_expense_register_page(): with gr.Blocks() as page: session_id = gr.State("") form_information = gr.State({}) with gr.Row(): session_id_display = gr.Markdown("Session ID: Loading...") new_btn = gr.Button("เริ่มใหม่", variant="primary") with gr.Tabs(): with gr.Tab("แชท"): chat_interface = gr.ChatInterface( fn=chat_handler, type="messages", chatbot=gr.Chatbot(value=_chat_prefill, type="messages", label="กล่องข้อความ"), multimodal=True, submit_btn="ส่งข้อความ", autofocus=True, examples=None, additional_inputs=[session_id], additional_outputs=[form_information] ) with gr.Tab("ฟอร์ม"): gr.Markdown("**การส่งมอบงาน**") deliverable_dropdown = gr.Dropdown(choices=get_deliverable_titles(), filterable=False, container=False) gr.Markdown("**ชื่อผู้ขาย**") seller_name_input = gr.Textbox(value=ast.literal_eval(session_form_data[session_id]["Seller Name"]), label="", container=False) gr.Markdown("**ที่อยู่ผู้ขาย**") seller_address_input = gr.Textbox(value=ast.literal_eval(session_form_data[session_id]["Seller Address"]), label="", container=False) gr.Markdown("**เบอร์โทรศัพท์ผู้ขาย**") seller_phone_number_input = gr.Textbox(value=ast.literal_eval(session_form_data[session_id]["Seller Phone Number"]), label="", container=False) gr.Markdown("**ชื่อผู้ซื้อ**") buyer_name_input = gr.Textbox(value=ast.literal_eval(session_form_data[session_id]["Buyer Name"]), label="", container=False) gr.Markdown("**ที่อยู่ผู้ซื้อ**") buyer_address_input = gr.Textbox(value=ast.literal_eval(session_form_data[session_id]["Buyer Address"]), label="", container=False) gr.Markdown("**วันที่ทำรายการ**") transaction_date_input = gr.DateTime(include_time=False, type="datetime", show_label=False) gr.Markdown("**ยอดชำระเงินทั้งหมด**") with gr.Row(): total_payment_amount_input = gr.Number(value=ast.literal_eval(session_form_data[session_id]['Total Payment Amount']), label="", container=False) gr.Markdown("บาท") gr.Markdown("**คำอธิบายค่าใช้จ่าย**") expense_description_input = gr.Textbox(value=ast.literal_eval(session_form_data[session_id]["Expense Description"]), label="", container=False, lines=4) feedback_message = gr.Markdown("", visible=False) submit_btn = gr.Button("บันทึกค่าใช้จ่าย", variant="primary") deliverable_dropdown.change(update_field_input("Associated Deliverable"), inputs=[session_id, deliverable_dropdown], outputs=[form_information],queue=False) seller_name_input.change(update_field_input("Seller Name"), inputs=[session_id, seller_name_input], outputs=[form_information],queue=False) seller_address_input.change(update_field_input("Seller Address"), inputs=[session_id, seller_address_input], outputs=[form_information],queue=False) seller_phone_number_input.change(update_field_input("Seller Phone Number"), inputs=[session_id, seller_phone_number_input], outputs=[form_information],queue=False) buyer_name_input.change(update_field_input("Buyer Name"), inputs=[session_id, buyer_name_input], outputs=[form_information],queue=False) buyer_address_input.change(update_field_input("Buyer Address"), inputs=[session_id, buyer_address_input], outputs=[form_information],queue=False) transaction_date_input.change(update_field_input("Transaction Date"), inputs=[session_id, transaction_date_input], outputs=[form_information],queue=False) total_payment_amount_input.change(update_field_input("Total Payment Amount"), inputs=[session_id, total_payment_amount_input], outputs=[form_information],queue=False) expense_description_input.change(update_field_input("Expense Description"), inputs=[session_id, expense_description_input], outputs=[form_information],queue=False) form_information.change( update_form_information_handler, inputs=[form_information], outputs=[ deliverable_dropdown, seller_name_input, seller_address_input, seller_phone_number_input, buyer_name_input, buyer_address_input, transaction_date_input, total_payment_amount_input, expense_description_input, ], queue=False, ) submit_btn.click(fn=handle_submit, inputs=[session_id], outputs=[feedback_message]) def reset_chat(): new_session_id, new_session_id_display = generate_session_id() logger.info(f"Chat reset with new session ID: {new_session_id}") # Reset the state dictionary for the new session session_form_data[new_session_id] = { "Associated Deliverable": "None", "Seller Name": "", "Seller Address": "", "Seller Phone Number": "", "Buyer Name": "", "Buyer Address": "", "Transaction Date": None, "Total Payment Amount": None, "Expense Description": "", } return ( new_session_id, new_session_id_display, _chat_prefill, session_form_data[new_session_id], # return the new form state "None", # deliverable_dropdown "", # seller_name_input "", # seller_address_input "", # seller_phone_number_input "", # buyer_name_input "", # buyer_address_input None, # transaction_date_input None, # total_payment_amount_input "", # expense_description_input gr.update(visible=False, value="") # feedback_message ) new_btn.click( fn=reset_chat, inputs=[], outputs=[ session_id, session_id_display, chat_interface.chatbot_value, form_information, deliverable_dropdown, seller_name_input, seller_address_input, seller_phone_number_input, buyer_name_input, buyer_address_input, transaction_date_input, total_payment_amount_input, expense_description_input, feedback_message ], ) page.load(generate_session_id, outputs=[session_id, session_id_display]) return page if __name__ == "__main__": page = create_expense_register_page() page.launch()