from flask import Flask, request, jsonify, abort from flask_cors import CORS import mysql.connector import os import logging import random from google import genai def initialize_model(client,prompt): chat = client.chats.create( model="gemini-2.0-flash" ) response = chat.send_message(prompt) return chat app = Flask(__name__) CORS(app) logging.basicConfig(level=logging.INFO) client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY")) chats = initialize_model(client,os.getenv("SYSTEM_PROMPT")) # === MySQL Connection === def get_connection(): return mysql.connector.connect( host=os.getenv("Host"), user=os.getenv("User"), password=os.getenv("Pass"), database=os.getenv("DB"), port=os.getenv("Port") ) # === Endpoints === @app.route("/", methods=["GET"]) def root(): return jsonify({"message": "ERP Accounting API is running."}) @app.route("/chat/", methods=["POST"]) def chat(): data = request.get_json() message = data.get("message") response = chats.send_message(message) logging.info(message) return jsonify({"response": response.text}) # -- Users -- @app.route("/users/", methods=["POST"]) def create_user(): data = request.get_json() # Simple validation if not data or not all(k in data for k in ("name", "email", "password_hash")): abort(400, description="Missing required fields: name, email, password_hash") name = data["name"] email = data["email"] password_hash = data["password_hash"] conn = get_connection() cursor = conn.cursor() try: cursor.execute(""" INSERT INTO users (name, email, password_hash) VALUES (%s, %s, %s) """, (name, email, password_hash)) conn.commit() return jsonify({"message": "User created successfully."}), 201 except mysql.connector.Error as e: conn.rollback() return jsonify({"detail": str(e)}), 400 finally: cursor.close() conn.close() # -- Accounts -- @app.route("/accounts/", methods=["POST"]) def create_account(): data = request.get_json() # Simple validation if not data or not all(k in data for k in ("account_name", "account_code", "account_type")): abort(400, description="Missing required fields: account_name, account_code, account_type") account_name = data["account_name"] account_code = data["account_code"] account_type = data["account_type"] description = data.get("description", None) conn = get_connection() cursor = conn.cursor() try: cursor.execute(""" INSERT INTO accounts (account_name, account_code, account_type, description) VALUES (%s, %s, %s, %s) """, (account_name, account_code, account_type, description)) conn.commit() return jsonify({"message": "Account created successfully."}), 201 except mysql.connector.Error as e: conn.rollback() return jsonify({"detail": str(e)}), 400 finally: cursor.close() conn.close() @app.route("/accounts/", methods=["GET"]) def list_accounts(): conn = get_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute("SELECT * FROM accounts") accounts = cursor.fetchall() return jsonify(accounts) finally: cursor.close() conn.close() # -- Journal Entries -- @app.route("/journal/", methods=["POST"]) def create_journal(): data = request.get_json() if not data or "lines" not in data or "created_by" not in data: abort(400, description="Missing required fields: lines, created_by") created_by = data["created_by"] entry_date = data["lines"][0].get("entry_date") description = data.get("description", "") if not entry_date or not isinstance(data["lines"], list): abort(400, description="Invalid or missing entry_date or lines") conn = get_connection() cursor = conn.cursor(dictionary=True) try: # Validate user exists cursor.execute("SELECT user_id FROM users WHERE user_id = %s", (created_by,)) if not cursor.fetchone(): return jsonify({"error": "User ID does not exist"}), 400 # Insert into journal_entries cursor.execute(""" INSERT INTO journal_entries (entry_date, description, created_by) VALUES (%s, %s, %s) """, (entry_date, description, created_by)) journal_id = cursor.lastrowid for line in data["lines"]: account_name = line.get("account_name") is_debit = line.get("is_debit") amount = line.get("amount") entry_date = line.get("entry_date") line_description = line.get("description", "Auto-created") account_type = line.get("account_id") if not account_name or amount is None or is_debit is None: abort(400, description="Missing account_name, amount or is_debit in line") # Check if account exists cursor.execute("SELECT account_id FROM accounts WHERE account_name = %s", (account_name,)) account = cursor.fetchone() if account: account_id = account["account_id"] else: # Create new account with default values generated_code = f"{account_type}{random.randint(100,1000)}" cursor.execute(""" INSERT INTO accounts (account_name, account_code, account_type, description) VALUES (%s, %s, %s, %s) """, (account_name, generated_code, account_type,"Created Via Journal Entry")) account_id = cursor.lastrowid # Insert journal line cursor.execute(""" INSERT INTO journal_lines (journal_id, account_id, is_debit, amount, entry_date, description) VALUES (%s, %s, %s, %s, %s, %s) """, (journal_id, account_id, is_debit, amount, entry_date, description)) # Insert into ledger cursor.execute(""" INSERT INTO ledger (account_id, journal_id, entry_date, debit_amount, credit_amount) VALUES (%s, %s, %s, %s, %s) """, ( account_id, journal_id, entry_date, amount if is_debit else 0, amount if not is_debit else 0 )) conn.commit() return jsonify({"message": "Journal entry created successfully."}), 201 except mysql.connector.Error as e: conn.rollback() return jsonify({"detail": str(e)}), 400 finally: cursor.close() conn.close() @app.route("/journal/", methods=["GET"]) def get_all_journals(): conn = get_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute(""" SELECT je.journal_id, je.entry_date, jl.description AS journal_description, je.created_by, u.name AS created_by_name, jl.line_id, a.account_name, a.account_type, jl.is_debit, jl.amount FROM journal_entries je JOIN journal_lines jl ON je.journal_id = jl.journal_id JOIN accounts a ON jl.account_id = a.account_id LEFT JOIN users u ON je.created_by = u.user_id ORDER BY je.journal_id DESC, jl.line_id ASC """) results = cursor.fetchall() # Group lines under their journal entries journals = {} for row in results: j_id = row["journal_id"] if j_id not in journals: journals[j_id] = { "journal_id": j_id, "entry_date": row["entry_date"], "description": row["journal_description"], "created_by": row["created_by"], "created_by_name": row.get("created_by_name"), "lines": [] } journals[j_id]["lines"].append({ "line_id": row["line_id"], "account_name": row["account_name"], "account_type": row["account_type"], "is_debit": row["is_debit"], "amount": float(row["amount"]), }) return jsonify(list(journals.values())) finally: cursor.close() conn.close() # -- Ledger View -- @app.route("/ledger/", methods=["GET"]) def get_ledger(): conn = get_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute(""" SELECT l.entry_date, a.account_name, a.account_type, l.debit_amount, l.credit_amount FROM ledger l JOIN accounts a ON l.account_id = a.account_id ORDER BY l.entry_date ASC """) ledger = cursor.fetchall() return jsonify(ledger) finally: cursor.close() conn.close() @app.route("/ledger/", methods=["GET"]) def get_ledger_by_account(account_id): conn = get_connection() cursor = conn.cursor(dictionary=True) try: # Check if account exists cursor.execute("SELECT account_name FROM accounts WHERE account_id = %s", (account_id,)) account = cursor.fetchone() if not account: abort(404, description="Account not found.") cursor.execute(""" SELECT l.entry_date, l.debit_amount, l.credit_amount, j.description AS journal_description FROM ledger l JOIN journal_entries j ON l.journal_id = j.journal_id WHERE l.account_id = %s ORDER BY l.entry_date ASC """, (account_id,)) ledger = cursor.fetchall() return jsonify({ "account_id": account_id, "account_name": account["account_name"], "entries": ledger }) finally: cursor.close() conn.close() # -- Trial Balance -- @app.route("/trial-balance/", methods=["GET"]) def get_trial_balance(): conn = get_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute(""" SELECT a.account_name, a.account_type, SUM(l.debit_amount) AS total_debit, SUM(l.credit_amount) AS total_credit FROM ledger l JOIN accounts a ON l.account_id = a.account_id GROUP BY a.account_id """) trial_balance = cursor.fetchall() return jsonify(trial_balance) finally: cursor.close() conn.close() # -- Profit & Loss Statement -- @app.route("/report/pnl/", methods=["GET"]) def get_profit_and_loss(): conn = get_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute(""" SELECT a.account_name, SUM(l.debit_amount) AS total_debit, SUM(l.credit_amount) AS total_credit FROM ledger l JOIN accounts a ON l.account_id = a.account_id WHERE a.account_type IN ('I', 'E') GROUP BY a.account_id """) pnl = cursor.fetchall() return jsonify(pnl) finally: cursor.close() conn.close() # -- Balance Sheet -- @app.route("/report/balance-sheet/", methods=["GET"]) def get_balance_sheet(): conn = get_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute(""" SELECT a.account_name, a.account_type, SUM(l.debit_amount) AS total_debit, SUM(l.credit_amount) AS total_credit FROM ledger l JOIN accounts a ON l.account_id = a.account_id WHERE a.account_type IN ('A', 'L', 'C') GROUP BY a.account_id """) balance_sheet = cursor.fetchall() # Fetch net profit cursor.execute(""" SELECT SUM(CASE WHEN a.account_type = 'I' THEN l.credit_amount ELSE 0 END) - SUM(CASE WHEN a.account_type = 'E' THEN l.debit_amount ELSE 0 END) AS net_profit FROM ledger l JOIN accounts a ON l.account_id = a.account_id """) net_profit = cursor.fetchone()["net_profit"] or 0 return jsonify({"Balance Sheet":balance_sheet, "Net Profit":net_profit}) finally: cursor.close() conn.close() # -- Reset Database -- @app.route("/delete/", methods=["DELETE"]) def clear_data(): conn = get_connection() cursor = conn.cursor() try: queries = [ "delete from ledger;", "ALTER TABLE ledger AUTO_INCREMENT = 1;", "delete from journal_lines;", "ALTER TABLE journal_lines AUTO_INCREMENT = 1;", "delete from journal_entries;", "ALTER TABLE journal_entries AUTO_INCREMENT = 1;", "delete from accounts;", "ALTER TABLE accounts AUTO_INCREMENT = 1;", ] for query in queries: cursor.execute(query) conn.commit() message = "Data Deleted Successfully" return jsonify(message) except Error as e: return jsonify(e) finally: cursor.close() conn.close() @app.route("/journal/", methods=["DELETE"]) def delete_journal_entry(journal_id): conn = get_connection() cursor = conn.cursor() try: cursor.execute("DELETE FROM ledger WHERE journal_id = %s", (journal_id,)) cursor.execute("DELETE FROM journal_lines WHERE journal_id = %s", (journal_id,)) cursor.execute("DELETE FROM journal_entries WHERE journal_id = %s", (journal_id,)) conn.commit() return jsonify({"message": "Journal entry deleted"}), 200 except mysql.connector.Error as e: conn.rollback() return jsonify({"error": str(e)}), 400 finally: cursor.close() conn.close() if __name__ == "__main__": app.run(debug=True)