Create app.py

app.py

@@ -0,0 +1,179 @@
+import torch
+from transformers import BertTokenizer, BertModel
+import spacy
+from typing import Dict, List, Tuple
+import sqlite3
+from datetime import datetime
+import uuid
+
+class AccountingNLP:
+    def __init__(self):
+        # Initialize BERT tokenizer and model
+        self.tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
+        self.model = BertModel.from_pretrained('bert-base-uncased')
+        # Initialize spaCy for entity extraction
+        self.nlp = spacy.load("en_core_web_sm")
+        # Initialize accounting knowledge base
+        self.accounting_rules = self.load_accounting_rules()
+        self.connection = sqlite3.connect('accounting_db.sqlite')
+        self.create_tables()
+
+    def load_accounting_rules(self) -> Dict:
+        """Load GAAP/IFRS rules and chart of accounts"""
+        return {
+            'accounts': {
+                'cash': {'type': 'Asset', 'normal_balance': 'Debit'},
+                'office_supplies': {'type': 'Asset', 'normal_balance': 'Debit'},
+                'accounts_payable': {'type': 'Liability', 'normal_balance': 'Credit'}
+            },
+            'rules': {
+                'purchase': {
+                    'debit': ['office_supplies'],
+                    'credit': ['cash', 'accounts_payable']
+                }
+            }
+        }
+
+    def create_tables(self):
+        """Create database tables for general ledger and journal entries"""
+        cursor = self.connection.cursor()
+        cursor.execute('''
+            CREATE TABLE IF NOT EXISTS journal_entries (
+                entry_id TEXT PRIMARY KEY,
+                date TEXT,
+                account TEXT,
+                debit REAL,
+                credit REAL,
+                description TEXT
+            )
+        ''')
+        cursor.execute('''
+            CREATE TABLE IF NOT EXISTS general_ledger (
+                account TEXT,
+                balance REAL,
+                last_updated TEXT
+            )
+        ''')
+        self.connection.commit()
+
+    def process_input(self, text: str) -> Dict:
+        """Process natural language input and extract intent and entities"""
+        doc = self.nlp(text)
+        entities = {
+            'amount': None,
+            'account': None,
+            'date': None,
+            'payment_method': None
+        }
+
+        # Extract entities
+        for ent in doc.ents:
+            if ent.label_ == "MONEY":
+                entities['amount'] = float(ent.text.replace('$', ''))
+            elif ent.label_ == "DATE":
+                entities['date'] = ent.text
+            elif ent.text.lower() in self.accounting_rules['accounts']:
+                entities['account'] = ent.text.lower()
+
+        # Simple intent classification
+        intent = 'record_transaction' if 'record' in text.lower() else 'query'
+
+        return {'intent': intent, 'entities': entities, 'raw_text': text}
+
+    def generate_journal_entry(self, processed_input: Dict) -> List[Dict]:
+        """Generate double-entry journal entries"""
+        if processed_input['intent'] != 'record_transaction':
+            return []
+
+        entities = processed_input['entities']
+        entry_id = str(uuid.uuid4())
+        date = entities['date'] or datetime.now().strftime('%Y-%m-%d')
+        entries = []
+
+        if entities['account'] in self.accounting_rules['rules']['purchase']['debit']:
+            # Debit entry
+            entries.append({
+                'entry_id': entry_id,
+                'date': date,
+                'account': entities['account'],
+                'debit': entities['amount'],
+                'credit': 0.0,
+                'description': processed_input['raw_text']
+            })
+            # Credit entry (assuming cash payment for simplicity)
+            entries.append({
+                'entry_id': entry_id,
+                'date': date,
+                'account': 'cash',
+                'debit': 0.0,
+                'credit': entities['amount'],
+                'description': processed_input['raw_text']
+            })
+
+        return entries
+
+    def validate_transaction(self, entries: List[Dict]) -> Tuple[bool, str]:
+        """Validate journal entries for double-entry compliance"""
+        total_debit = sum(entry['debit'] for entry in entries)
+        total_credit = sum(entry['credit'] for entry in entries)
+
+        if total_debit != total_credit:
+            return False, "Debits and credits must balance"
+        if not entries:
+            return False, "No valid entries generated"
+        return True, "Valid transaction"
+
+    def update_ledger(self, entries: List[Dict]):
+        """Update general ledger with validated entries"""
+        cursor = self.connection.cursor()
+        for entry in entries:
+            cursor.execute('''
+                INSERT INTO journal_entries (entry_id, date, account, debit, credit, description)
+                VALUES (?, ?, ?, ?, ?, ?)
+            ''', (
+                entry['entry_id'],
+                entry['date'],
+                entry['account'],
+                entry['debit'],
+                entry['credit'],
+                entry['description']
+            ))
+
+            # Update general ledger balance
+            cursor.execute('''
+                INSERT OR REPLACE INTO general_ledger (account, balance, last_updated)
+                VALUES (?, 
+                    (SELECT COALESCE((SELECT balance FROM general_ledger WHERE account = ?), 0) 
+                     + ? - ?),
+                    ?)
+            ''', (
+                entry['account'],
+                entry['account'],
+                entry['debit'],
+                entry['credit'],
+                entry['date']
+            ))
+
+        self.connection.commit()
+
+    def generate_response(self, processed_input: Dict, entries: List[Dict]) -> str:
+        """Generate natural language response"""
+        if processed_input['intent'] == 'record_transaction':
+            is_valid, message = self.validate_transaction(entries)
+            if is_valid:
+                self.update_ledger(entries)
+                return f"Successfully recorded transaction: {processed_input['raw_text']}"
+            return f"Error: {message}"
+        return "Query processing not implemented"
+
+    def process(self, text: str) -> str:
+        """Main processing pipeline"""
+        processed_input = self.process_input(text)
+        entries = self.generate_journal_entry(processed_input)
+        return self.generate_response(processed_input, entries)
+
+# Example usage
+if __name__ == "__main__":
+    accounting_ai = AccountingNLP()
+    result = accounting_ai.process("Record a $500 office supplies purchase paid by check")
+    print(result)