# ===================================================================
# MODULE 10, LESSON 3: DATA MESH ARCHITECTURE
# ===================================================================

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import json
import requests
from flask import Flask, request, jsonify  # Simulating microservices
import warnings
warnings.filterwarnings('ignore')

print("="*70)
print("DATA MESH PROTOTYPE – DOMAIN-ORIENTED DATA PRODUCTS")
print("="*70)

# ----------------------------------------------------------------
# PART A: DOMAIN 1 – CUSTOMER DOMAIN (Customer 360 Data Product)
# ----------------------------------------------------------------
print("\n" + "-"*60)
print("PART A: Domain Data Product – Customer Identity")
print("-"*60)

class CustomerDomain:
    """Simulates a microservice for customer data."""
    def __init__(self):
        # Simulated database of customers
        self.customers = pd.DataFrame({
            'customer_id': ['C001', 'C002', 'C003'],
            'name': ['Alice Johnson', 'Bob Smith', 'Carol Davis'],
            'email': ['alice@email.com', 'bob@email.com', 'carol@email.com'],
            'segment': ['Premium', 'Standard', 'Student'],
            'kyc_status': ['Verified', 'Pending', 'Verified'],
            'joined_date': ['2020-01-15', '2021-06-30', '2022-11-01']
        })
        print("Customer Domain Data Product initialized.")

    def get_customer_profile(self, customer_id):
        """API endpoint: Returns a unified customer profile."""
        record = self.customers[self.customers['customer_id'] == customer_id]
        if record.empty:
            return {"error": "Customer not found"}
        return record.to_dict('records')[0]

    def get_customers_by_segment(self, segment):
        """API endpoint: Returns customers filtered by segment."""
        records = self.customers[self.customers['segment'] == segment]
        return records.to_dict('records')

# Instantiate domain
customer_domain = CustomerDomain()
print(f"\nSample Customer Profile: {customer_domain.get_customer_profile('C001')}")

# ----------------------------------------------------------------
# PART B: DOMAIN 2 – TRANSACTIONS DOMAIN
# ----------------------------------------------------------------
print("\n" + "-"*60)
print("PART B: Domain Data Product – Transaction Ledger")
print("-"*60)

class TransactionDomain:
    """Simulates a microservice for transaction data."""
    def __init__(self):
        np.random.seed(42)
        # Simulated transaction data for the last 30 days
        dates = pd.date_range(start=datetime.now() - timedelta(days=30), end=datetime.now(), freq='D')
        self.transactions = pd.DataFrame({
            'transaction_id': [f'TX{i}' for i in range(100)],
            'customer_id': np.random.choice(['C001', 'C002', 'C003'], 100),
            'amount': np.random.uniform(10, 500, 100).round(2),
            'type': np.random.choice(['Debit', 'Credit', 'Transfer'], 100, p=[0.5, 0.3, 0.2]),
            'status': np.random.choice(['Settled', 'Pending', 'Failed'], 100, p=[0.8, 0.1, 0.1]),
            'date': np.random.choice(dates, 100)
        })
        # Add a daily summary pre-computed (Data Product freshness)
        self.daily_summary = self.transactions.groupby('date').agg(
            total_volume=('amount', 'sum'),
            transaction_count=('transaction_id', 'count')
        ).reset_index()
        print("Transaction Domain Data Product initialized.")

    def get_transactions_for_customer(self, customer_id, days=30):
        """API endpoint: Returns transactions for a customer."""
        cutoff = datetime.now() - timedelta(days=days)
        filtered = self.transactions[
            (self.transactions['customer_id'] == customer_id) & 
            (self.transactions['date'] >= cutoff)
        ]
        return filtered.to_dict('records')

    def get_daily_summary(self, date_from, date_to):
        """API endpoint: Returns aggregate metrics for analytics."""
        mask = (self.daily_summary['date'] >= date_from) & (self.daily_summary['date'] <= date_to)
        return self.daily_summary[mask].to_dict('records')

# Instantiate domain
transaction_domain = TransactionDomain()
print(f"\nSample Daily Summary: {transaction_domain.daily_summary.head(2).to_dict('records')}")

# ----------------------------------------------------------------
# PART C: DOMAIN 3 – FRAUD & RISK DOMAIN
# ----------------------------------------------------------------
print("\n" + "-"*60)
print("PART C: Domain Data Product – Real-time Risk Score")
print("-"*60)

class FraudDomain:
    """Simulates a machine learning domain serving risk scores."""
    def __init__(self):
        # Simulating a trained model
        self.risk_weights = {
            'high_velocity': 0.3,     # >10 transactions per day
            'unusual_location': 0.4,
            'device_fingerprint': 0.3
        }
        print("Fraud Domain Data Product initialized.")

    def calculate_risk_score(self, customer_id, transaction_amount, location, device_id):
        """API endpoint: Returns a dynamic risk score (1-100)."""
        # Simulating feature engineering and model inference
        risk = 10  # Base score
        if transaction_amount > 1000:
            risk += 20
        if location not in ['USA', 'UK']:
            risk += 30  # Unusual location
        if device_id == 'unknown':
            risk += 40
        return {
            'customer_id': customer_id,
            'risk_score': min(risk, 99),
            'decision': 'Approve' if risk < 70 else 'Review'
        }

# Instantiate domain
fraud_domain = FraudDomain()
print(f"\nSample Risk Assessment: {fraud_domain.calculate_risk_score('C001', 1500, 'Nigeria', 'unknown')}")

# ----------------------------------------------------------------
# PART D: FEDERATED CONSUMPTION (The Analytics Layer)
# ----------------------------------------------------------------
print("\n" + "-"*60)
print("PART D: Consuming Data Products – Federated Query Engine")
print("-"*60)

class AnalyticsConsumer:
    """
    Simulates a central analytics team consuming multiple domains.
    This is the "Data Mesh" in action – no single database, just API calls.
    """
    def __init__(self, customer_api, transaction_api, fraud_api):
        self.customer_api = customer_api
        self.transaction_api = transaction_api
        self.fraud_api = fraud_api
        print("Analytics Consumer initialized. Ready to query across domains.")

    def generate_customer_risk_report(self, customer_id):
        """Federated query: Joins data from three domains without a central DB."""
        print(f"\n--- Generating Composite Report for {customer_id} ---")
        
        # 1. Fetch from Customer Domain
        profile = self.customer_api.get_customer_profile(customer_id)
        if 'error' in profile:
            return "Customer not found."
        
        # 2. Fetch from Transaction Domain
        txn_data = self.transaction_api.get_transactions_for_customer(customer_id, days=7)
        total_spend = sum([t['amount'] for t in txn_data]) if txn_data else 0
        avg_txn = total_spend / len(txn_data) if txn_data else 0
        
        # 3. Fetch from Fraud Domain (Simulate a new transaction)
        risk = self.fraud_api.calculate_risk_score(
            customer_id, 
            transaction_amount=avg_txn, 
            location='USA',
            device_id='device_123'
        )
        
        # 4. Compile the report (Data Product assembly)
        report = {
            "report_generated": datetime.now().isoformat(),
            "customer_profile": profile,
            "transaction_summary": {
                "last_7_days_count": len(txn_data),
                "total_spend": total_spend,
                "average_transaction": avg_txn
            },
            "risk_summary": risk,
            "data_lineage": ["Customer-Domain", "Transaction-Domain", "Fraud-Domain"]
        }
        return report

# Instantiate the consumer with all domains
analytics = AnalyticsConsumer(customer_domain, transaction_domain, fraud_domain)

# Generate a composite report for a specific customer
report = analytics.generate_customer_risk_report('C002')
print(json.dumps(report, indent=2, default=str))

# ----------------------------------------------------------------
# PART E: FEDERATED GOVERNANCE (Data Quality Monitoring)
# ----------------------------------------------------------------
print("\n" + "-"*60)
print("PART E: Federated Governance – Automated Data Quality Checks")
print("-"*60)

class DataQualityMonitor:
    """
    Simulates automated governance policies applied across domains.
    """
    def __init__(self, domains):
        self.domains = domains  # List of domain objects
        self.quality_checks = []
        
    def run_quality_checks(self):
        print("Running automated data quality checks...")
        results = []
        
        # Check 1: Customer Domain completeness
        customers = self.domains[0].customers
        missing_emails = customers['email'].isna().sum()
        if missing_emails == 0:
            results.append({"domain": "Customer", "check": "Email completeness", "status": "PASS"})
        else:
            results.append({"domain": "Customer", "check": "Email completeness", "status": "FAIL", "count": missing_emails})
        
        # Check 2: Transaction Domain freshness (last update)
        latest_txn_date = self.domains[1].transactions['date'].max()
        hours_since_update = (datetime.now() - latest_txn_date).total_seconds() / 3600
        if hours_since_update < 24:
            results.append({"domain": "Transactions", "check": "Data freshness (< 24h)", "status": "PASS"})
        else:
            results.append({"domain": "Transactions", "check": "Data freshness (< 24h)", "status": "FAIL", "hours": hours_since_update})
        
        # Check 3: Fraud Domain - risk score range
        # (We can't easily check without running inference, so we mock a pass)
        results.append({"domain": "Fraud", "check": "Risk score output range (0-100)", "status": "PASS"})
        
        return pd.DataFrame(results)

# Run the monitor
monitor = DataQualityMonitor([customer_domain, transaction_domain, fraud_domain])
quality_report = monitor.run_quality_checks()
print(quality_report.to_string(index=False))

# ----------------------------------------------------------------
# SECTION 6: SUMMARY FOR THE DATA PRACTITIONER
# ----------------------------------------------------------------
print("\n" + "="*70)
print("LESSON 3 SUMMARY FOR THE DATA PRACTITIONER")
print("="*70)

print("""
1. Centralized Data Lakes create bottlenecks. Data Mesh distributes ownership to domains.
2. Four Principles: Domain Ownership, Data as a Product, Self-Service Infrastructure, Federated Governance.
3. We simulated three domains (Customer, Transactions, Fraud) exposing APIs.
4. Federated Queries (like in Part D) allow analytics without ETL into a central warehouse.
5. Data Quality must be automated at the source, not fixed downstream.
6. Action: Map your organization's business domains and prototype a 'Data Product' API for one domain.
""")

print("="*70)
print("END OF LESSON 3 – MODULE 10")
print("="*70)