app.py

#!/usr/bin/env python
import os
import json
import logging
import random
from datetime import datetime, timedelta
from typing import Dict, Any, List, Optional
from pydantic import BaseModel, Field
import gradio as gr

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)

# ========== DATA MODELS ==========
class UserVerification(BaseModel):
    user_id: str = Field(..., description="User ID")
    name: str = Field(..., description="User name")
    email: str = Field(..., description="User email")
    is_verified: bool = Field(..., description="KYC verification status")
    verification_level: str = Field(..., description="Verification level: basic, standard, premium")
    risk_score: float = Field(..., description="Risk score 0-1")

class ScalpingDetection(BaseModel):
    is_scalper: bool = Field(..., description="Whether user is detected as scalper")
    confidence: float = Field(..., description="Detection confidence 0-1")
    flags: List[str] = Field(..., description="Suspicious activity flags")
    purchase_velocity: int = Field(..., description="Number of purchases in last hour")
    ip_duplicates: int = Field(..., description="Number of accounts from same IP")

class PricingRecommendation(BaseModel):
    original_price: float = Field(..., description="Original ticket price")
    recommended_resale_price: float = Field(..., description="Recommended resale price")
    demand_level: str = Field(..., description="Current demand level")
    price_adjustment_reason: str = Field(..., description="Reason for price adjustment")
    profit_margin: Optional[float] = Field(None, description="Profit margin percentage")
    loss_percentage: Optional[float] = Field(None, description="Loss percentage if selling below cost")

class ResaleCompliance(BaseModel):
    is_compliant: bool = Field(..., description="Whether resale is policy compliant")
    violations: List[str] = Field(..., description="List of policy violations")
    resale_allowed: bool = Field(..., description="Whether resale is allowed")
    max_allowed_price: float = Field(..., description="Maximum allowed resale price")
    recommendation: str = Field(..., description="Compliance recommendation")

class TicketingSystemReport(BaseModel):
    verification: UserVerification
    scalping_detection: ScalpingDetection
    pricing: PricingRecommendation
    compliance: ResaleCompliance
    final_decision: str = Field(..., description="Final system decision")
    action_items: List[str] = Field(..., description="Recommended actions")

# ========== MOCK DATA STORE ==========
class MockDatabase:
    """Simulates a database for the MVP"""
    def __init__(self):
        self.user_purchase_history = {}
        self.ip_addresses = {}
        self.resale_history = {}
    
    def get_user_purchases(self, user_id: str) -> int:
        """Get number of recent purchases for a user"""
        return self.user_purchase_history.get(user_id, random.randint(0, 5))
    
    def get_ip_accounts(self, user_id: str) -> int:
        """Get number of accounts from same IP"""
        return random.randint(1, 4)
    
    def get_resale_frequency(self, user_id: str) -> int:
        """Get user's resale frequency"""
        return self.resale_history.get(user_id, random.randint(0, 10))

# Initialize mock database
mock_db = MockDatabase()

# ========== VERIFICATION LOGIC ==========
class UserVerificationEngine:
    @staticmethod
    def verify_user(name: str, email: str, user_id: str) -> UserVerification:
        """Verify user based on provided information"""
        
        # Email validation
        email_valid = "@" in email and "." in email.split("@")[-1]
        
        # Name validation (basic check)
        name_valid = len(name.strip()) >= 2 and not any(char.isdigit() for char in name)
        
        # Risk assessment
        risk_factors = []
        if not email_valid:
            risk_factors.append("invalid_email")
        if not name_valid:
            risk_factors.append("suspicious_name")
        if email.endswith(('.temp', '.fake', '.test')):
            risk_factors.append("temporary_email")
        
        risk_score = min(0.9, len(risk_factors) * 0.3 + random.uniform(0.1, 0.3))
        is_verified = risk_score < 0.5 and email_valid and name_valid
        
        verification_level = "premium" if risk_score < 0.2 else "standard" if risk_score < 0.5 else "basic"
        
        return UserVerification(
            user_id=user_id,
            name=name,
            email=email,
            is_verified=is_verified,
            verification_level=verification_level,
            risk_score=risk_score
        )

class ScalpingDetectionEngine:
    @staticmethod
    def detect_scalping(user_id: str, ticket_quantity: int, event_name: str) -> ScalpingDetection:
        """Detect potential scalping behavior"""
        
        flags = []
        purchase_velocity = mock_db.get_user_purchases(user_id)
        ip_duplicates = mock_db.get_ip_accounts(user_id)
        resale_frequency = mock_db.get_resale_frequency(user_id)
        
        # Check for scalping indicators
        if purchase_velocity > 3:
            flags.append("rapid_purchases")
        if ip_duplicates > 2:
            flags.append("multiple_ips")
        if ticket_quantity > 4:
            flags.append("bulk_purchase")
        if resale_frequency > 5:
            flags.append("frequent_reseller")
        
        # Calculate scalping probability
        scalping_score = (
            (purchase_velocity / 10) * 0.3 +
            (ip_duplicates / 5) * 0.3 +
            (ticket_quantity / 10) * 0.2 +
            (resale_frequency / 20) * 0.2
        )
        
        is_scalper = scalping_score > 0.5
        confidence = min(0.95, scalping_score + random.uniform(0.1, 0.2))
        
        return ScalpingDetection(
            is_scalper=is_scalper,
            confidence=confidence,
            flags=flags,
            purchase_velocity=purchase_velocity,
            ip_duplicates=ip_duplicates
        )

class DynamicPricingEngine:
    @staticmethod
    def calculate_pricing(original_price: float, demand_level: str, proposed_price: float) -> PricingRecommendation:
        """Calculate recommended pricing based on demand"""
        
        demand_multipliers = {
            "low": (0.8, 1.1),      # Allow 20% below to 10% above
            "medium": (0.9, 1.25),  # Allow 10% below to 25% above
            "high": (1.0, 1.5)      # Allow original price to 50% above
        }
        
        min_mult, max_mult = demand_multipliers.get(demand_level, (0.9, 1.25))
        
        min_price = original_price * min_mult
        max_price = original_price * max_mult
        
        # Recommend price within acceptable range
        recommended_price = max(min_price, min(proposed_price, max_price))
        
        price_ratio = recommended_price / original_price
        
        profit_margin = None
        loss_percentage = None
        
        if price_ratio > 1.0:
            profit_margin = (price_ratio - 1) * 100
        elif price_ratio < 1.0:
            loss_percentage = (1 - price_ratio) * 100
        
        reason = f"Adjusted for {demand_level} demand market conditions"
        if recommended_price != proposed_price:
            reason += f" (modified from ${proposed_price:.2f})"
        
        return PricingRecommendation(
            original_price=original_price,
            recommended_resale_price=recommended_price,
            demand_level=demand_level,
            price_adjustment_reason=reason,
            profit_margin=profit_margin,
            loss_percentage=loss_percentage
        )

class ComplianceEngine:
    @staticmethod
    def check_compliance(
        user_id: str, 
        proposed_price: float, 
        original_price: float,
        scalping_detection: ScalpingDetection
    ) -> ResaleCompliance:
        """Check resale compliance against policies"""
        
        violations = []
        price_ratio = proposed_price / original_price
        
        # Policy checks
        if price_ratio > 2.0:
            violations.append("price_exceeds_2x")
        
        if scalping_detection.is_scalper:
            violations.append("suspected_scalper")
        
        if scalping_detection.purchase_velocity > 5:
            violations.append("excessive_purchase_velocity")
        
        resale_frequency = mock_db.get_resale_frequency(user_id)
        if resale_frequency > 4:
            violations.append("monthly_resale_limit_exceeded")
        
        is_compliant = len(violations) == 0
        resale_allowed = is_compliant and not scalping_detection.is_scalper
        
        max_allowed_price = original_price * 2.0
        
        if resale_allowed:
            recommendation = "Transaction approved - complies with all policies"
        else:
            recommendation = f"Transaction blocked due to: {', '.join(violations)}"
        
        return ResaleCompliance(
            is_compliant=is_compliant,
            violations=violations,
            resale_allowed=resale_allowed,
            max_allowed_price=max_allowed_price,
            recommendation=recommendation
        )

# ========== MAIN APPLICATION ==========
class AntiScalpingSystem:
    def __init__(self):
        self.verification_engine = UserVerificationEngine()
        self.scalping_engine = ScalpingDetectionEngine()
        self.pricing_engine = DynamicPricingEngine()
        self.compliance_engine = ComplianceEngine()
    
    def process_ticket_transaction(
        self,
        name: str,
        email: str, 
        user_id: str,
        event_name: str,
        ticket_type: str,
        ticket_quantity: int,
        original_price: float,
        demand_level: str,
        proposed_resale_price: float
    ) -> Dict[str, Any]:
        """Process a ticket transaction through the anti-scalping system"""
        
        try:
            # Step 1: User Verification
            verification = self.verification_engine.verify_user(name, email, user_id)
            
            # Step 2: Scalping Detection
            scalping_detection = self.scalping_engine.detect_scalping(
                user_id, ticket_quantity, event_name
            )
            
            # Step 3: Dynamic Pricing
            pricing = self.pricing_engine.calculate_pricing(
                original_price, demand_level, proposed_resale_price
            )
            
            # Step 4: Compliance Check
            compliance = self.compliance_engine.check_compliance(
                user_id, proposed_resale_price, original_price, scalping_detection
            )
            
            # Step 5: Final Decision
            final_decision = "APPROVED" if (
                verification.is_verified and 
                compliance.resale_allowed and 
                not scalping_detection.is_scalper
            ) else "DENIED"
            
            # Action items
            action_items = []
            if final_decision == "APPROVED":
                action_items.append("✅ Process ticket resale")
                action_items.append("📊 Monitor user activity")
            else:
                action_items.append("❌ Block transaction")
                if scalping_detection.is_scalper:
                    action_items.append("🚨 Flag user account for review")
                if not verification.is_verified:
                    action_items.append("📋 Require additional verification")
            
            # Create report
            report = TicketingSystemReport(
                verification=verification,
                scalping_detection=scalping_detection,
                pricing=pricing,
                compliance=compliance,
                final_decision=final_decision,
                action_items=action_items
            )
            
            return report.dict()
            
        except Exception as e:
            logger.error(f"Error processing transaction: {e}")
            return self._create_error_response(str(e))
    
    def _create_error_response(self, error_msg: str) -> Dict[str, Any]:
        """Create error response"""
        return {
            "error": True,
            "message": f"System error: {error_msg}",
            "final_decision": "ERROR",
            "action_items": ["🔧 Contact system administrator"]
        }

# ========== GRADIO INTERFACE ==========
def create_interface():
    system = AntiScalpingSystem()
    
    def process_transaction(
        name, email, user_id, event_name, ticket_type,
        ticket_quantity, original_price, demand_level, proposed_resale_price
    ):
        """Process the transaction and return formatted results"""
        
        # Validate inputs
        if not all([name, email, user_id, event_name]):
            return "❌ **Error**: Please fill in all required fields"
        
        try:
            original_price = float(original_price) if original_price else 0
            proposed_resale_price = float(proposed_resale_price) if proposed_resale_price else 0
            ticket_quantity = int(ticket_quantity) if ticket_quantity else 1
            
            if original_price <= 0 or proposed_resale_price <= 0:
                return "❌ **Error**: Prices must be greater than 0"
                
        except (ValueError, TypeError):
            return "❌ **Error**: Please enter valid numbers for prices and quantity"
        
        # Process through the system
        result = system.process_ticket_transaction(
            name=name,
            email=email,
            user_id=user_id,
            event_name=event_name,
            ticket_type=ticket_type,
            ticket_quantity=ticket_quantity,
            original_price=original_price,
            demand_level=demand_level,
            proposed_resale_price=proposed_resale_price
        )
        
        # Handle errors
        if result.get("error"):
            return f"❌ **System Error**: {result.get('message', 'Unknown error occurred')}"
        
        # Format the output
        decision_emoji = "✅" if result['final_decision'] == "APPROVED" else "❌"
        
        output = f"""
# 🎫 Anti-Scalping System Analysis Report

## {decision_emoji} Final Decision: **{result['final_decision']}**

---

## 👤 User Verification
- **User ID**: `{result['verification']['user_id']}`
- **Name**: {result['verification']['name']}
- **Email**: {result['verification']['email']}
- **Verified**: {'✅ Yes' if result['verification']['is_verified'] else '❌ No'}
- **Risk Score**: {result['verification']['risk_score']:.1%} {'🟢' if result['verification']['risk_score'] < 0.3 else '🟡' if result['verification']['risk_score'] < 0.6 else '🔴'}
- **Verification Level**: {result['verification']['verification_level'].title()}

## 🔍 Scalping Detection Analysis
- **Scalper Detected**: {'🚨 YES' if result['scalping_detection']['is_scalper'] else '✅ NO'}
- **Detection Confidence**: {result['scalping_detection']['confidence']:.1%}
- **Purchase Velocity**: {result['scalping_detection']['purchase_velocity']} purchases/hour
- **IP Address Duplicates**: {result['scalping_detection']['ip_duplicates']} accounts
- **Red Flags**: {', '.join(result['scalping_detection']['flags']) if result['scalping_detection']['flags'] else '✅ None detected'}

## 💰 Dynamic Pricing Analysis
- **Original Price**: ${result['pricing']['original_price']:.2f}
- **Proposed Resale**: ${proposed_resale_price:.2f}
- **Recommended Price**: ${result['pricing']['recommended_resale_price']:.2f}
- **Demand Level**: {result['pricing']['demand_level'].title()} 📈
- **Price Ratio**: {result['pricing']['recommended_resale_price']/result['pricing']['original_price']:.2f}x
"""
        
        if result['pricing'].get('profit_margin'):
            output += f"- **Profit Margin**: {result['pricing']['profit_margin']:.1f}% 📈\n"
        elif result['pricing'].get('loss_percentage'):
            output += f"- **Loss**: -{result['pricing']['loss_percentage']:.1f}% 📉\n"
        
        output += f"""- **Reason**: {result['pricing']['price_adjustment_reason']}

## ✅ Compliance Evaluation
- **Policy Compliant**: {'✅ Yes' if result['compliance']['is_compliant'] else '❌ No'}
- **Resale Permitted**: {'✅ Yes' if result['compliance']['resale_allowed'] else '❌ No'}
- **Maximum Allowed**: ${result['compliance']['max_allowed_price']:.2f}
- **Policy Violations**: {', '.join(result['compliance']['violations']) if result['compliance']['violations'] else '✅ None'}
- **Recommendation**: {result['compliance']['recommendation']}

---

## 📋 Required Actions
"""
        for i, action in enumerate(result['action_items'], 1):
            output += f"{i}. {action}\n"
        
        # Add summary box
        if result['final_decision'] == "APPROVED":
            output += """
> ✅ **TRANSACTION APPROVED** - All checks passed. Resale can proceed as recommended.
"""
        else:
            output += """
> ❌ **TRANSACTION BLOCKED** - Policy violations detected. Review required before proceeding.
"""
        
        return output
    
    # Create Gradio interface with better styling
    with gr.Blocks(
        title="🎫 Anti-Scalping Ticketing System",
        theme=gr.themes.Soft(
            primary_hue="blue",
            secondary_hue="slate",
        )
    ) as interface:
        
        gr.Markdown("""
        # 🎫 AI-Powered Anti-Scalping Ticketing System
        
        This intelligent system prevents ticket scalping by analyzing user behavior, verifying identities, 
        and ensuring fair pricing compliance. Enter transaction details below for real-time analysis.
        
        ---
        """)
        
        with gr.Row():
            with gr.Column(scale=1):
                gr.Markdown("### 👤 User Information")
                name = gr.Textbox(label="Full Name", placeholder="John Doe", value="")
                email = gr.Textbox(label="Email Address", placeholder="john@example.com", value="")
                user_id = gr.Textbox(label="User ID", placeholder="USER123", value="")
                
                gr.Markdown("### 🎟️ Event Details")
                event_name = gr.Textbox(label="Event Name", placeholder="Taylor Swift - Eras Tour", value="")
                ticket_type = gr.Dropdown(
                    label="Ticket Type",
                    choices=["General Admission", "VIP", "Premium", "Standard", "Balcony", "Floor"],
                    value="Standard"
                )
                ticket_quantity = gr.Number(label="Number of Tickets", value=1, minimum=1, maximum=10, precision=0)
                
                gr.Markdown("### 💲 Pricing Information")
                original_price = gr.Number(label="Original Ticket Price ($)", value=100, minimum=1)
                demand_level = gr.Radio(
                    label="Current Market Demand",
                    choices=["low", "medium", "high"],
                    value="medium"
                )
                proposed_resale_price = gr.Number(label="Proposed Resale Price ($)", value=150, minimum=1)
                
                submit_btn = gr.Button("🔍 Analyze Transaction", variant="primary", size="lg")
            
            with gr.Column(scale=2):
                output = gr.Markdown(value="👆 Fill in the form and click 'Analyze Transaction' to get started!")
        
        # Event handlers
        submit_btn.click(
            fn=process_transaction,
            inputs=[
                name, email, user_id, event_name, ticket_type,
                ticket_quantity, original_price, demand_level, proposed_resale_price
            ],
            outputs=output
        )
        
        # Examples section
        gr.Markdown("---")
        gr.Markdown("### 📋 Example Scenarios")
        
        examples = gr.Examples(
            examples=[
                ["John Smith", "john@gmail.com", "USER001", "Taylor Swift - Eras Tour", "VIP", 2, 500, "high", 750],
                ["Jane Doe", "jane@company.com", "USER002", "NBA Finals Game 7", "Premium", 4, 300, "high", 1200],
                ["Bob Wilson", "bob@email.com", "USER003", "Local Concert", "General Admission", 1, 50, "low", 40],
                ["Scalper Bot", "temp@fake.com", "BOT999", "Popular Event", "Standard", 8, 100, "high", 300],
            ],
            inputs=[
                name, email, user_id, event_name, ticket_type,
                ticket_quantity, original_price, demand_level, proposed_resale_price
            ]
        )
    
    return interface

# Main execution
if __name__ == "__main__":
    interface = create_interface()
    interface.launch(
        share=True,
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True
    )