Merge pull request #200 from Cybermaxi7/feat/revenue-sharing-stakeholder-persistence

feat: comprehensive revenue sharing enhancements

Author: BigBen-7

.coverage

@@ -27,4 +27,6 @@ qrcode==7.4.2
 
 # Logging and Metrics
 prometheus-client==0.20.0
+slowapi==0.1.9
+cachetools==5.3.3
 

requirements.txt

@@ -0,0 +1,128 @@
+import json
+import logging
+import uuid
+from typing import Any, Dict, List, Optional
+from sqlalchemy import text
+from src.db import get_engine
+from src.revenue_sharing_models import RevenueCalculationResult, PayoutDistribution
+
+logger = logging.getLogger("veritix.calculation_history_store")
+
+def create_revenue_calculations_table() -> None:
+    """Create the revenue_calculations table if it does not yet exist."""
+    engine = get_engine()
+    if engine is None:
+        logger.info("Skipping revenue_calculations table creation — no DB engine")
+        return
+    
+    with engine.connect() as conn:
+        # Use FLOAT and TEXT for cross-DB compatibility (SQLite/Postgres)
+        conn.execute(text("""
+            CREATE TABLE IF NOT EXISTS revenue_calculations (
+                id                  TEXT PRIMARY KEY,
+                event_id            TEXT NOT NULL,
+                total_gross_sales   FLOAT NOT NULL,
+                total_fees          FLOAT NOT NULL,
+                net_revenue         FLOAT NOT NULL,
+                distributions       TEXT NOT NULL,
+                total_paid_out      FLOAT NOT NULL,
+                remaining_balance   FLOAT NOT NULL,
+                rules_applied       TEXT NOT NULL,
+                calculated_at       TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+        """))
+        conn.commit()
+    logger.info("revenue_calculations table ready")
+
+def save_calculation(result: RevenueCalculationResult) -> str:
+    """Persist a revenue calculation result to the database."""
+    engine = get_engine()
+    if engine is None:
+        return ""
+    
+    calculation_id = str(uuid.uuid4())
+    
+    query = text("""
+        INSERT INTO revenue_calculations (
+            id, event_id, total_gross_sales, total_fees, net_revenue, 
+            distributions, total_paid_out, remaining_balance, rules_applied
+        ) VALUES (
+            :id, :event_id, :total_gross_sales, :total_fees, :net_revenue, 
+            :distributions, :total_paid_out, :remaining_balance, :rules_applied
+        )
+    """)
+    
+    with engine.connect() as conn:
+        conn.execute(query, {
+            "id": calculation_id,
+            "event_id": result.event_id,
+            "total_gross_sales": result.total_gross_sales,
+            "total_fees": result.total_fees,
+            "net_revenue": result.net_revenue,
+            "distributions": json.dumps([d.model_dump() for d in result.distributions]),
+            "total_paid_out": result.total_paid_out,
+            "remaining_balance": result.remaining_balance,
+            "rules_applied": json.dumps(result.rules_applied)
+        })
+        conn.commit()
+    
+    logger.info(f"Saved revenue calculation {calculation_id} for event {result.event_id}")
+    return calculation_id
+
+def get_history_for_event(event_id: str, page: int = 1, limit: int = 20) -> List[Dict[str, Any]]:
+    """Retrieve paginated calculation history for an event."""
+    engine = get_engine()
+    if engine is None:
+        return []
+    
+    offset = (page - 1) * limit
+    
+    query = text("""
+        SELECT id, event_id, total_gross_sales, total_fees, net_revenue, 
+               distributions, total_paid_out, remaining_balance, rules_applied, calculated_at
+        FROM revenue_calculations
+        WHERE event_id = :event_id
+        ORDER BY calculated_at DESC
+        LIMIT :limit OFFSET :offset
+    """)
+    
+    with engine.connect() as conn:
+        result = conn.execute(query, {"event_id": event_id, "limit": limit, "offset": offset})
+        history = []
+        for row in result:
+            history.append(_row_to_dict(row))
+        return history
+
+def get_calculation_by_id(calculation_id: str) -> Optional[Dict[str, Any]]:
+    """Retrieve a specific calculation by its ID."""
+    engine = get_engine()
+    if engine is None:
+        return None
+    
+    query = text("""
+        SELECT id, event_id, total_gross_sales, total_fees, net_revenue, 
+               distributions, total_paid_out, remaining_balance, rules_applied, calculated_at
+        FROM revenue_calculations
+        WHERE id = :id
+    """)
+    
+    with engine.connect() as conn:
+        result = conn.execute(query, {"id": calculation_id}).fetchone()
+        if result:
+            return _row_to_dict(result)
+        return None
+
+def _row_to_dict(row) -> Dict[str, Any]:
+    """Helper to convert a DB row to a dictionary with parsed JSON fields."""
+    return {
+        "id": row[0],
+        "event_id": row[1],
+        "total_gross_sales": row[2],
+        "total_fees": row[3],
+        "net_revenue": row[4],
+        "distributions": json.loads(row[5]),
+        "total_paid_out": row[6],
+        "remaining_balance": row[7],
+        "rules_applied": json.loads(row[8]),
+        "calculated_at": row[9]
+    }

src/calculation_history_store.py

@@ -47,6 +47,7 @@ def validate_api_keys(cls, v: str, info) -> str:
                 f"{info.field_name} must be at least 32 characters and must not use a default value"
             )
         return v
+    SUPPORTED_CURRENCIES: str = "USD,NGN,GBP,EUR,KES"
     KNOWN_LOCATIONS: str = (
         "lagos,abuja,port harcourt,kano,ibadan,benin,kaduna,jos,enugu,calabar,"
         "owerri,warri,uyo,akure,ilorin,sokoto,zaria,maiduguri,asaba,nnewi"

src/config.py

@@ -0,0 +1,65 @@
+import logging
+import requests
+from typing import Dict, List, Optional
+from cachetools import TTLCache, cached
+from fastapi import HTTPException, status
+from src.config import get_settings
+
+logger = logging.getLogger("veritix.currency_service")
+
+class ServiceUnavailableException(HTTPException):
+    def __init__(self, detail: str = "Currency exchange service is currently unavailable"):
+        super().__init__(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=detail)
+
+# Cache rates for 10 minutes (600 seconds)
+rates_cache = TTLCache(maxsize=1, ttl=600)
+
+@cached(cache=rates_cache)
+def get_exchange_rates() -> Dict[str, float]:
+    """Fetch latest exchange rates from external API relative to USD."""
+    url = "https://api.exchangerate-api.com/v4/latest/USD"
+    try:
+        response = requests.get(url, timeout=10)
+        response.raise_for_status()
+        data = response.json()
+        return data.get("rates", {})
+    except Exception as exc:
+        logger.error(f"Failed to fetch exchange rates: {exc}")
+        raise ServiceUnavailableException()
+
+def get_exchange_rate(from_currency: str, to_currency: str = "USD") -> float:
+    """
+    Get the exchange rate between two currencies.
+    
+    If from_currency is same as to_currency, returns 1.0.
+    """
+    if from_currency == to_currency:
+        return 1.0
+        
+    supported = get_settings().SUPPORTED_CURRENCIES.split(",")
+    if from_currency not in supported or to_currency not in supported:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"Currency not supported. Supported: {supported}"
+        )
+        
+    rates = get_exchange_rates()
+    
+    if from_currency not in rates or to_currency not in rates:
+        logger.error(f"Currency {from_currency} or {to_currency} not found in FX rates")
+        raise ServiceUnavailableException(detail="Exchange rate for requested currency not available")
+        
+    # Standardize to USD then to target
+    # rate_to_usd = 1 / rates[from_currency] if from_currency != "USD" else 1.0
+    # target_rate = rate_to_usd * rates[to_currency]
+    
+    # Simpler since API is based on USD:
+    # 1 USD = rates[NGN] NGN
+    # 1 USD = rates[GBP] GBP
+    # 1 NGN = (1 / rates[NGN]) USD
+    # 1 NGN = (1 / rates[NGN]) * rates[GBP] GBP
+    
+    usd_rate_for_from = rates[from_currency]
+    usd_rate_for_to = rates[to_currency]
+    
+    return usd_rate_for_to / usd_rate_for_from

src/currency_service.py

@@ -9,7 +9,7 @@
 from typing import Annotated, Any, Dict, List, Optional
 
 import numpy as np
-from fastapi import Depends, FastAPI, HTTPException, Query, Request, WebSocket, WebSocketDisconnect
+from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException, Query, Request, WebSocket, WebSocketDisconnect
 from fastapi.responses import FileResponse, JSONResponse, PlainTextResponse
 from fastapi.staticfiles import StaticFiles
 from slowapi.errors import RateLimitExceeded
@@ -57,8 +57,14 @@
     EventRevenueInput,
     RevenueCalculationResult,
     RevenueShareConfig,
+    Stakeholder,
 )
 from src.revenue_sharing_service import revenue_sharing_service
+from src import (
+    calculation_history_store,
+    currency_service,
+    stakeholder_store,
+)
 from src.recommender import (
     build_item_similarity_matrix,
     get_item_recommendations,
@@ -182,6 +188,8 @@ def on_startup() -> None:
     global model_pipeline, etl_scheduler
     settings = get_settings()
     create_generated_reports_table()
+    stakeholder_store.create_stakeholders_table()
+    calculation_history_store.create_revenue_calculations_table()
     if not settings.SKIP_MODEL_TRAINING:
         model_pipeline = train_logistic_regression_pipeline()
 
@@ -586,7 +594,10 @@ def recommend_events(payload: RecommendRequest) -> RecommendResponse:
 # ---------------------------------------------------------------------------
 
 @app.post("/calculate-revenue-share", response_model=RevenueCalculationResult)
-def calculate_revenue_share(input_data: EventRevenueInput) -> RevenueCalculationResult:
+def calculate_revenue_share(
+    input_data: EventRevenueInput,
+    background_tasks: BackgroundTasks,
+) -> RevenueCalculationResult:
     """Calculate revenue shares for stakeholders based on event sales."""
     log_info("Revenue share calculation requested", {
         "event_id": input_data.event_id,
@@ -599,30 +610,42 @@ def calculate_revenue_share(input_data: EventRevenueInput) -> RevenueCalculation
         raise HTTPException(status_code=400, detail={"errors": errors})
     try:
         result = revenue_sharing_service.calculate_revenue_shares(input_data)
+        background_tasks.add_task(calculation_history_store.save_calculation, result)
         return result
+    except HTTPException:
+        raise
     except Exception as exc:
         log_error("Revenue share calculation failed", {"error": str(exc)})
         raise HTTPException(status_code=500, detail=f"Revenue calculation failed: {exc}")
 
 
 @app.post("/calculate-revenue-share/batch", response_model=List[RevenueCalculationResult])
 def calculate_revenue_share_batch(
-    inputs: List[EventRevenueInput],
+    inputs: List[Dict[str, Any]],
+    background_tasks: BackgroundTasks,
 ) -> List[RevenueCalculationResult]:
-    """Calculate revenue shares for multiple events."""
+    """Calculate revenue shares for multiple events, skipping invalid ones."""
     log_info("Batch revenue share calculation requested", {"event_count": len(inputs)})
     results: List[RevenueCalculationResult] = []
-    for input_data in inputs:
+    for item in inputs:
         try:
+            # Validate model first
+            input_data = EventRevenueInput.model_validate(item)
+            # Run service-level validation
             is_valid, errors = revenue_sharing_service.validate_input(input_data)
             if not is_valid:
+                log_info("Skipping invalid event in batch", {"event_id": item.get("event_id"), "errors": errors})
                 continue
-            results.append(revenue_sharing_service.calculate_revenue_shares(input_data))
+            
+            result = revenue_sharing_service.calculate_revenue_shares(input_data)
+            background_tasks.add_task(calculation_history_store.save_calculation, result)
+            results.append(result)
         except Exception as exc:
-            log_error("Batch revenue calculation failed", {
-                "event_id": input_data.event_id,
+            log_error("Batch revenue calculation partially failed", {
+                "event_id": item.get("event_id"),
                 "error": str(exc),
             })
+            continue
     log_info("Batch calculation completed", {
         "processed_count": len(results),
         "requested_count": len(inputs),
@@ -637,6 +660,17 @@ def get_revenue_share_config() -> RevenueShareConfig:
     return revenue_sharing_service.config
 
 
+@app.get("/revenue-share/exchange-rates")
+def get_exchange_rates() -> Dict[str, float]:
+    """Return the current currency exchange rates relative to USD."""
+    log_info("Currency exchange rates requested")
+    try:
+        return currency_service.get_exchange_rates()
+    except Exception as exc:
+        log_error("Failed to fetch exchange rates", {"error": str(exc)})
+        raise HTTPException(status_code=503, detail="Exchange rate service unavailable")
+
+
 @app.get("/revenue-share/example", response_model=EventRevenueInput)
 def get_example_revenue_input() -> EventRevenueInput:
     """Return an example revenue calculation input."""
@@ -650,6 +684,73 @@ def get_example_revenue_input() -> EventRevenueInput:
     )
 
 
+@app.post("/revenue-share/stakeholders/{event_id}")
+def save_stakeholders(
+    event_id: str,
+    stakeholders: List[Stakeholder],
+    _: str = Depends(require_admin_key),
+) -> Dict[str, Any]:
+    """Save custom stakeholder configuration for an event (ADMIN)."""
+    log_info("Saving custom stakeholders", {"event_id": event_id, "count": len(stakeholders)})
+    try:
+        stakeholder_store.save_stakeholders_for_event(event_id, stakeholders)
+        return {"success": True, "message": f"Saved {len(stakeholders)} stakeholders for event {event_id}"}
+    except Exception as exc:
+        log_error("Failed to save stakeholders", {"event_id": event_id, "error": str(exc)})
+        raise HTTPException(status_code=500, detail=f"Failed to save stakeholders: {exc}")
+
+
+@app.get("/revenue-share/stakeholders/{event_id}", response_model=List[Stakeholder])
+def get_stakeholders(event_id: str) -> List[Stakeholder]:
+    """Return the stakeholder configuration for an event (custom or default)."""
+    log_info("Retrieving stakeholders", {"event_id": event_id})
+    try:
+        stakeholders = stakeholder_store.get_stakeholders_for_event(event_id)
+        if not stakeholders:
+            # Fallback logic duplicated here for the GET endpoint
+            stakeholders = revenue_sharing_service._get_default_stakeholders(event_id)
+        return stakeholders
+    except Exception as exc:
+        log_error("Failed to retrieve stakeholders", {"event_id": event_id, "error": str(exc)})
+        raise HTTPException(status_code=500, detail=f"Failed to retrieve stakeholders: {exc}")
+
+
+@app.get("/revenue-share/history/{event_id}")
+def get_revenue_history(
+    event_id: str,
+    page: int = Query(1, ge=1),
+    limit: int = Query(20, ge=1, le=100),
+    _: str = Depends(require_admin_key),
+) -> List[Dict[str, Any]]:
+    """Retrieve paginated calculation history for an event (ADMIN)."""
+    log_info("Retrieving revenue history", {"event_id": event_id, "page": page, "limit": limit})
+    try:
+        return calculation_history_store.get_history_for_event(event_id, page, limit)
+    except Exception as exc:
+        log_error("Failed to retrieve history", {"event_id": event_id, "error": str(exc)})
+        raise HTTPException(status_code=500, detail=f"Failed to retrieve history: {exc}")
+
+
+@app.get("/revenue-share/history/{event_id}/{calculation_id}")
+def get_calculation_detail(
+    event_id: str,
+    calculation_id: str,
+    _: str = Depends(require_admin_key),
+) -> Dict[str, Any]:
+    """Retrieve a specific stored calculation by ID (ADMIN)."""
+    log_info("Retrieving calculation detail", {"event_id": event_id, "calculation_id": calculation_id})
+    try:
+        detail = calculation_history_store.get_calculation_by_id(calculation_id)
+        if not detail or detail["event_id"] != event_id:
+            raise HTTPException(status_code=404, detail="Calculation not found for this event")
+        return detail
+    except HTTPException:
+        raise
+    except Exception as exc:
+        log_error("Failed to retrieve calculation detail", {"calculation_id": calculation_id, "error": str(exc)})
+        raise HTTPException(status_code=500, detail=f"Failed to retrieve detail: {exc}")
+
+
 # ---------------------------------------------------------------------------
 # Daily report
 # ---------------------------------------------------------------------------