Spaces:
Running
Running
| from core.domain_classifier import domain_classifier | |
| from core.legal_intelligence import legal_intelligence | |
| from layers.layer_4_logic.smart_money import smart_money | |
| from layers.layer_4_logic.pos_engine import pos_engine | |
| import re | |
| def analyze(user_query: str, context: dict, domain_info: dict): | |
| domain = domain_info['primary_domain'] | |
| jurisdiction = context['country_code'] | |
| # 1. Response Strategy | |
| strategy = domain_classifier.get_response_strategy( | |
| domain, jurisdiction, legal_intelligence | |
| ) | |
| # 2. Financial Logic | |
| analysis = { | |
| "domain": domain, | |
| "strategy": strategy, | |
| "actions_requested": [] | |
| } | |
| # Check for specific actions (POS, payment requests, etc.) | |
| if domain == "PAYMENTS" or domain == "BUSINESS_FINANCE": | |
| sale_record = pos_engine.parse_sale(user_query) | |
| if sale_record: | |
| analysis['actions_requested'].append({ | |
| "type": "record_sale", | |
| "data": sale_record | |
| }) | |
| # Check for SMS receipt/payment request | |
| phone_match = re.search( | |
| r'(?:tuma|send|receipt to|risiti|omba malipo)\s+(?:kwa\s+)?(\+?0?[17]\d{8})', | |
| user_query, | |
| re.IGNORECASE | |
| ) | |
| if phone_match: | |
| phone = phone_match.group(1) | |
| payment_trigger = any( | |
| trigger in user_query.lower() | |
| for trigger in ['charge', 'collect', 'omba malipo', 'request payment'] | |
| ) | |
| if payment_trigger: | |
| analysis['actions_requested'].append({ | |
| "type": "request_payment", | |
| "data": {"phone": phone, "amount": sale_record['revenue']} | |
| }) | |
| else: | |
| analysis['actions_requested'].append({ | |
| "type": "send_receipt", | |
| "data": {"phone": phone, "sale_record": sale_record} | |
| }) | |
| # 3. Calculations | |
| if domain == "TAXATION": | |
| # Simplified: identify if it's PAYE or Turnover | |
| if "paye" in user_query.lower() or "salary" in user_query.lower(): | |
| analysis['calculation_type'] = "PAYE" | |
| elif "turnover" in user_query.lower() or "business" in user_query.lower(): | |
| analysis['calculation_type'] = "TURNOVER_TAX" | |
| return analysis | |
| def run_calculators(calculators: list, query: str, jurisdiction: str, currency: str): | |
| from core.financial_calculator import calculator | |
| # Extract numbers | |
| numbers = [float(n.replace(',', '')) for n in re.findall(r'\b\d+(?:,\d+)*(?:\.\d+)?\b', query)] | |
| val = numbers[0] if numbers else 0 | |
| results = [] | |
| for c in calculators: | |
| if c == 'paye': | |
| res = calculator.run(c, monthly_salary=val, currency=currency) | |
| elif c == 'turnover_tax': | |
| res = calculator.run(c, annual_revenue=val * 12, currency=currency) | |
| elif c == 'fuliza_cost': | |
| res = calculator.run(c, amount=val, days=30, currency=currency) | |
| else: | |
| res = {} | |
| if res and res.get("formatted"): | |
| results.append(res.get("formatted")) | |
| return "\n\n".join(results) if results else None | |
| def check_fraud(transactions, user_hash): | |
| return {"aml_flagged": False, "flags": []} | |
| def apply_legal_rules(rules, domain, jurisdiction): | |
| return {"disclaimer": None, "rules_used": [], "refer_professional": False} | |
| def detect_stress(query): | |
| return False | |
| def query_user_data(query, transactions, profile, language, currency): | |
| from core.data_conversation import data_conversation | |
| data_query = data_conversation.classify_query(query) | |
| if data_query.get('is_data_query'): | |
| result = data_conversation.execute_query( | |
| query_classification=data_query, | |
| transactions=transactions, | |
| user_goals=[], | |
| language=language | |
| ) | |
| if result and 'error' not in result: | |
| return data_conversation.format_response( | |
| result=result, | |
| original_query=query, | |
| language=language, | |
| currency=currency | |
| ) | |
| return None | |
| def assess_risk(profile, query): | |
| return None | |