Create app.py

app.py

@@ -0,0 +1,343 @@
+import sqlite3
+import json
+import uuid
+import datetime
+import logging
+import re
+import gradio as gr
+
+# Setup logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+# Disable model loading on free tier due to memory constraints
+TRANSFORMERS_AVAILABLE = False
+AutoModelForCausalLM = None
+AutoTokenizer = None
+tokenizer = None
+model = None
+
+# Database setup
+conn = sqlite3.connect("erp.db", check_same_thread=False)
+cursor = conn.cursor()
+
+# Create tables
+cursor.execute("""
+CREATE TABLE IF NOT EXISTS chart_of_accounts (
+    account_id TEXT PRIMARY KEY,
+    account_name TEXT NOT NULL,
+    account_type TEXT NOT NULL,
+    parent_id TEXT,
+    allow_budgeting BOOLEAN,
+    allow_posting BOOLEAN,
+    FOREIGN KEY (parent_id) REFERENCES chart_of_accounts(account_id)
+)
+""")
+cursor.execute("""
+CREATE TABLE IF NOT EXISTS journal_entries (
+    entry_id TEXT PRIMARY KEY,
+    date TEXT NOT NULL,
+    debit_account_id TEXT NOT NULL,
+    credit_account_id TEXT NOT NULL,
+    amount REAL NOT NULL,
+    description TEXT,
+    FOREIGN KEY (debit_account_id) REFERENCES chart_of_accounts(account_id),
+    FOREIGN KEY (credit_account_id) REFERENCES chart_of_accounts(account_id)
+)
+""")
+conn.commit()
+
+# Debit/Credit rules
+ACCOUNT_RULES = {
+    "Asset": {"increase": "Debit", "decrease": "Credit"},
+    "Liability": {"increase": "Credit", "decrease": "Debit"},
+    "Equity": {"increase": "Credit", "decrease": "Debit"},
+    "Revenue": {"increase": "Credit", "decrease": "Debit"},
+    "Expense": {"increase": "Debit", "decrease": "Credit"}
+}
+
+# Initialize chart of accounts
+def initialize_chart_of_accounts():
+    accounts = [
+        ("1", "Assets", "Asset", None, True, False),
+        ("1.1", "Fixed Assets", "Asset", "1", True, False),
+        ("1.1.1", "Plant", "Asset", "1.1", True, True),
+        ("1.1.2", "Machinery", "Asset", "1.1", True, True),
+        ("1.1.3", "Building", "Asset", "1.1", True, True),
+        ("1.2", "Current Assets", "Asset", "1", True, False),
+        ("1.2.1", "Cash", "Asset", "1.2", True, True),
+        ("1.2.2", "Laptop", "Asset", "1.2", True, True),
+        ("1.2.3", "Inventory", "Asset", "1.2", True, True),
+        ("1.2.4", "Accounts Receivable", "Asset", "1.2", True, True),
+        ("1.2.5", "Bank", "Asset", "1.2", True, True),
+        ("2", "Liabilities", "Liability", None, True, False),
+        ("2.1", "Accounts Payable", "Liability", "2", True, True),
+        ("2.2", "Loan Payable", "Liability", "2", True, True),
+        ("3", "Equity", "Equity", None, True, False),
+        ("3.1", "Owner's Equity", "Equity", "3", True, True),
+        ("3.2", "Drawings", "Equity", "3", True, True),
+        ("4", "Revenue", "Revenue", None, True, False),
+        ("4.1", "Sales Revenue", "Revenue", "4", True, True),
+        ("5", "Expenses", "Expense", None, True, False),
+        ("5.1", "Rent Expense", "Expense", "5", True, True),
+        ("5.2", "Salary Expense", "Expense", "5", True, True),
+        ("5.3", "Office Supplies", "Expense", "5", True, True)
+    ]
+    cursor.executemany("""
+    INSERT OR REPLACE INTO chart_of_accounts
+    (account_id, account_name, account_type, parent_id, allow_budgeting, allow_posting)
+    VALUES (?, ?, ?, ?, ?, ?)
+    """, accounts)
+    conn.commit()
+    logging.info("Chart of accounts initialized.")
+
+# Enhanced flexible parser
+def parse_prompt(prompt, state):
+    logging.info(f"Parsing prompt: {prompt}")
+    if model and tokenizer:
+        try:
+            input_text = f"""
+            Parse the following accounting prompt into a JSON object with:
+            - debit: {{account, type, amount}}
+            - credit: {{account, type, amount}}
+            - payment_method: 'cash', 'credit', 'bank', or null
+            Prompt: {prompt}
+            """
+            inputs = tokenizer(input_text, return_tensors="pt")
+            outputs = model.generate(**inputs, max_length=300)
+            response = tokenizer.decode(outputs[0], skip_special_tokens=True)
+            return json.loads(response), state
+        except Exception as e:
+            logging.error(f"Model parsing failed: {e}")
+
+    prompt_lower = prompt.lower().strip()
+    amount = None
+    match = re.search(r'\$?(\d{1,3}(?:,\d{3})*|\d+)(?:\.\d{2})?', prompt_lower)
+    if match:
+        try:
+            amount = float(match.group().replace(',', '').replace('$', ''))
+        except ValueError:
+            return {"error": "Invalid amount format detected. Please use a valid number (e.g., $200 or 200)."}, state
+
+    if not amount and not state.get("pending_prompt"):
+        return {"error": "No amount found in prompt. Please include an amount (e.g., $200)."}, state
+
+    account_mappings = {
+        "laptop": ("Laptop", "Asset"),
+        "inventory": ("Inventory", "Asset"),
+        "machinery": ("Machinery", "Asset"),
+        "building": ("Building", "Asset"),
+        "plant": ("Plant", "Asset"),
+        "office supplies": ("Office Supplies", "Expense"),
+        "cash": ("Cash", "Asset"),
+        "bank": ("Bank", "Asset"),
+        "receivable": ("Accounts Receivable", "Asset"),
+        "payable": ("Accounts Payable", "Liability"),
+        "loan": ("Loan Payable", "Liability"),
+        "equity": ("Owner's Equity", "Equity"),
+        "drawings": ("Drawings", "Equity"),
+        "sales": ("Sales Revenue", "Revenue"),
+        "rent": ("Rent Expense", "Expense"),
+        "salary": ("Salary Expense", "Expense"),
+        "wages": ("Salary Expense", "Expense")
+    }
+
+    debit_account = None
+    debit_type = None
+    credit_account = None
+    credit_type = None
+    payment_method = None
+
+    if state.get("pending_prompt"):
+        follow_up = prompt_lower
+        if follow_up in ["cash", "credit", "bank"]:
+            payment_method = follow_up
+            parsed = state["pending_parsed"]
+            debit_account = parsed["debit"]["account"]
+            debit_type = parsed["debit"]["type"]
+            amount = parsed["debit"]["amount"]
+            if payment_method == "cash":
+                credit_account, credit_type = "Cash", "Asset"
+            elif payment_method == "bank":
+                credit_account, credit_type = "Bank", "Asset"
+            elif payment_method == "credit":
+                credit_account, credit_type = "Accounts Payable", "Liability"
+            state = {}  # Clear state after successful follow-up
+        else:
+            return {"error": "Please specify 'cash', 'credit', or 'bank'."}, state
+    else:
+        # Detect transaction type and accounts
+        for keyword, (account, acc_type) in account_mappings.items():
+            if keyword in prompt_lower:
+                if keyword in ["bought", "purchased"]:
+                    for asset in ["laptop", "inventory", "machinery", "building", "plant", "office supplies"]:
+                        if asset in prompt_lower:
+                            debit_account, debit_type = account_mappings[asset]
+                            break
+                    if not debit_account:
+                        debit_account, debit_type = "Inventory", "Asset"  # Default asset
+                elif keyword in ["paid", "spent"]:
+                    for expense in ["rent", "salary", "wages", "office supplies"]:
+                        if expense in prompt_lower:
+                            debit_account, debit_type = account_mappings[expense]
+                            break
+                    if not debit_account:
+                        debit_account, debit_type = "Expense", "Expense"  # Generic expense
+                    # Detect payment method in initial prompt
+                    if "cash" in prompt_lower:
+                        credit_account, credit_type = "Cash", "Asset"
+                        payment_method = "cash"
+                    elif "bank" in prompt_lower:
+                        credit_account, credit_type = "Bank", "Asset"
+                        payment_method = "bank"
+                    elif "credit" in prompt_lower:
+                        credit_account, credit_type = "Accounts Payable", "Liability"
+                        payment_method = "credit"
+                elif keyword in ["sold", "sales"]:
+                    debit_account, debit_type = "Accounts Receivable", "Asset"
+                    credit_account, credit_type = account_mappings["sales"]
+                elif keyword == "drawings":
+                    debit_account, debit_type = "Drawings", "Equity"
+                    credit_account, credit_type = "Cash", "Asset"
+                elif keyword == "loan":
+                    debit_account, debit_type = "Cash", "Asset"
+                    credit_account, credit_type = "Loan Payable", "Liability"
+                break
+
+        if debit_account and not credit_account and not payment_method:
+            return {
+                "status": "clarify",
+                "message": "Was this transaction made with cash, credit, or bank?",
+                "pending_parsed": {"debit": {"account": debit_account, "type": debit_type, "amount": amount}}
+            }, {"pending_prompt": prompt, "pending_parsed": {"debit": {"account": debit_account, "type": debit_type, "amount": amount}}}
+
+    if debit_account and credit_account:
+        return {
+            "debit": {"account": debit_account, "type": debit_type, "amount": amount},
+            "credit": {"account": credit_account, "type": credit_type, "amount": amount},
+            "payment_method": payment_method
+        }, state
+    return {"error": "Couldn’t interpret the transaction. Try 'Paid rent $400' or 'Bought a laptop for $200'."}, state
+
+# Generate journal entry
+def generate_journal_entry(parsed, state):
+    logging.info(f"Generating journal entry with parsed: {parsed}")
+    if "error" in parsed:
+        return parsed["error"], state
+    if parsed.get("status") == "clarify":
+        return parsed["message"], state
+
+    debit_account = parsed["debit"]["account"]
+    amount = parsed["debit"]["amount"]
+    credit_account = parsed["credit"]["account"]
+    credit_type = parsed["credit"]["type"]
+
+    cursor.execute("SELECT account_id, account_type, allow_posting FROM chart_of_accounts WHERE account_name = ?", (debit_account,))
+    debit_result = cursor.fetchone()
+    cursor.execute("SELECT account_id, account_type, allow_posting FROM chart_of_accounts WHERE account_name = ?", (credit_account,))
+    credit_result = cursor.fetchone()
+
+    if not debit_result or not credit_result:
+        return "One or both accounts not found.", state
+    if not debit_result[2] or not credit_result[2]:
+        return "Posting not allowed for one or both accounts.", state
+    if debit_result[1] != parsed["debit"]["type"] or credit_result[1] != credit_type:
+        return "Account type mismatch.", state
+
+    entry_id = str(uuid.uuid4())
+    date = datetime.datetime.now().isoformat()
+    description = state.get("pending_prompt", "Transaction")
+    try:
+        cursor.execute("""
+        INSERT INTO journal_entries (entry_id, date, debit_account_id, credit_account_id, amount, description)
+        VALUES (?, ?, ?, ?, ?, ?)
+        """, (entry_id, date, debit_result[0], credit_result[0], amount, description))
+        conn.commit()
+        logging.info(f"Journal entry created: Debit {debit_account} ${amount}, Credit {credit_account} ${amount}")
+    except sqlite3.Error as e:
+        logging.error(f"Database error: {e}")
+        return "Database error occurred.", state
+
+    return f"Journal Entry Recorded:\nDebit: {debit_account} ${amount}\nCredit: {credit_account} ${amount}", state
+
+# Generate T-account
+def generate_t_account(account_name):
+    cursor.execute("SELECT account_id FROM chart_of_accounts WHERE account_name = ?", (account_name,))
+    account_id = cursor.fetchone()
+    if not account_id:
+        logging.error(f"Account {account_name} not found.")
+        return "Account not found."
+    
+    account_id = account_id[0]
+    try:
+        cursor.execute("""
+        SELECT date, amount, description, 'Debit' as type FROM journal_entries WHERE debit_account_id = ?
+        UNION
+        SELECT date, amount, description, 'Credit' as type FROM journal_entries WHERE credit_account_id = ?
+        ORDER BY date
+        """, (account_id, account_id))
+        entries = cursor.fetchall()
+        logging.info(f"Retrieved {len(entries)} entries for T-account: {account_name}")
+    except sqlite3.Error as e:
+        logging.error(f"SQL error in generate_t_account: {e}")
+        return "Error retrieving T-account data."
+    
+    t_account = f"T-Account for {account_name}\n{'='*50}\n{'Debit':<20} | {'Credit':<20} | Description\n{'-'*50}\n"
+    debit_total = 0
+    credit_total = 0
+    for date, amount, desc, entry_type in entries:
+        if entry_type == "Debit":
+            t_account += f"${amount:<19} | {'':<20} | {desc}\n"
+            debit_total += amount
+        else:
+            t_account += f"{'':<20} | ${amount:<19} | {desc}\n"
+            credit_total += amount
+    t_account += f"{'-'*50}\nTotal Debit: ${debit_total:<10} | Total Credit: ${credit_total}\n"
+    
+    return t_account
+
+# Gradio chat function
+def chat_function(message, history, state=None):
+    if state is None:
+        state = {}
+        initialize_chart_of_accounts()
+        logging.info("Initialized state and chart of accounts")
+    
+    logging.info(f"Received message: {message}")
+    
+    if message.lower().startswith("t-account "):
+        account_name = message[10:].strip()
+        if account_name:
+            response = generate_t_account(account_name)
+        else:
+            response = "Please provide an account name (e.g., 't-account Cash')."
+    else:
+        parsed, state = parse_prompt(message, state)
+        response, state = generate_journal_entry(parsed, state)
+    
+    if history is not None:
+        history.append({"role": "user", "content": message})
+        history.append({"role": "assistant", "content": response})
+    else:
+        history = [
+            {"role": "user", "content": message},
+            {"role": "assistant", "content": response}
+        ]
+    
+    return history, state, ""  # Clear the textbox
+
+# Gradio interface
+with gr.Blocks() as demo:
+    gr.Markdown("# AI ERP System")
+    gr.Markdown("Enter accounting prompts naturally (e.g., 'Paid rent $400' or 't-account Cash'). I’ll guide you like an experienced accountant if clarification is needed.")
+    chatbot = gr.Chatbot(type="messages")
+    msg = gr.Textbox(placeholder="Type your prompt here...", lines=1, submit_btn=None)
+    clear = gr.Button("Clear")
+    
+    state = gr.State({})
+    
+    msg.submit(chat_function, [msg, chatbot, state], [chatbot, state, msg])
+    clear.click(lambda: ([], {}, ""), None, [chatbot, state, msg], queue=False)
+
+# Launch Gradio
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)