Create app.py

app.py

@@ -0,0 +1,477 @@
+#!/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
+from crewai import Agent, Crew, Process, Task, LLM
+from crewai.agent import agent
+from crewai.task import task
+from crewai.crew import crew, CrewBase
+
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# Initialize LLM (will use environment variable)
+llm = LLM(
+    model="gemini/gemini-2.0-flash-exp",  # or any other model
+    api_key=os.getenv("GEMINI_API_KEY", "dummy-key-for-testing"),
+)
+
+# ========== 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()
+
+# ========== CREW CLASS ==========
+class AntiScalpingCrew(CrewBase):
+    """Anti-Scalping Ticketing System Crew"""
+    
+    @agent
+    def user_verification_agent(self) -> Agent:
+        return Agent(
+            role="User Verification Specialist",
+            goal="Verify user identity and assess verification level for ticket purchases",
+            backstory="""You are an expert in Know Your Customer (KYC) processes and identity 
+            verification. You assess user documentation, verify identities, and assign risk scores 
+            based on user history and verification completeness. You ensure only legitimate users 
+            can purchase tickets while maintaining a smooth user experience.""",
+            llm=llm,
+            verbose=True
+        )
+    
+    @agent
+    def scalping_detection_agent(self) -> Agent:
+        return Agent(
+            role="Scalping Detection Analyst",
+            goal="Detect and prevent ticket scalping through behavioral analysis",
+            backstory="""You are a specialized analyst focused on identifying scalping patterns. 
+            You analyze purchase velocity, IP addresses, buying patterns, and other behavioral 
+            indicators to detect potential scalpers. Your expertise helps maintain fair ticket 
+            distribution and prevents automated bot purchases.""",
+            llm=llm,
+            verbose=True
+        )
+    
+    @agent
+    def dynamic_pricing_agent(self) -> Agent:
+        return Agent(
+            role="Dynamic Pricing Strategist",
+            goal="Calculate fair resale prices based on demand while preventing price gouging",
+            backstory="""You are an expert in market dynamics and pricing strategies. You analyze 
+            demand levels, market conditions, and event popularity to recommend fair resale prices. 
+            You balance market forces with consumer protection, ensuring prices reflect true demand 
+            without enabling exploitative pricing.""",
+            llm=llm,
+            verbose=True
+        )
+    
+    @agent
+    def resale_monitor_agent(self) -> Agent:
+        return Agent(
+            role="Resale Compliance Officer",
+            goal="Monitor and enforce resale policies to ensure fair ticket distribution",
+            backstory="""You are responsible for ensuring all ticket resales comply with platform 
+            policies and local regulations. You monitor resale prices, frequency, and patterns to 
+            prevent policy violations. Your work ensures the secondary ticket market remains fair 
+            and accessible to genuine fans.""",
+            llm=llm,
+            verbose=True
+        )
+    
+    @task
+    def verify_user_task(self) -> Task:
+        return Task(
+            description="""Verify the user with the following details:
+            - Name: {name}
+            - Email: {email}
+            - User ID: {user_id}
+            
+            Perform KYC verification checks:
+            1. Validate email format and domain
+            2. Check if name appears legitimate
+            3. Assign verification level (basic/standard/premium)
+            4. Calculate risk score (0-1) based on user patterns
+            5. Determine if user is verified for ticket purchase
+            
+            Consider factors like email domain reputation, name consistency, and any red flags.""",
+            agent=self.user_verification_agent(),
+            expected_output="UserVerification model with complete verification details"
+        )
+    
+    @task
+    def detect_scalping_task(self) -> Task:
+        return Task(
+            description="""Analyze user behavior for scalping indicators:
+            - User ID: {user_id}
+            - Current purchase attempt for event: {event_name}
+            - Ticket quantity requested: {ticket_quantity}
+            
+            Check for scalping patterns:
+            1. Purchase velocity (multiple purchases in short time)
+            2. IP address duplication (multiple accounts from same IP)
+            3. Unusual buying patterns
+            4. Bot-like behavior indicators
+            5. Historical resale frequency
+            
+            Use the user's purchase history and behavioral data to determine scalping likelihood.""",
+            agent=self.scalping_detection_agent(),
+            expected_output="ScalpingDetection model with detection results and confidence score"
+        )
+    
+    @task
+    def calculate_pricing_task(self) -> Task:
+        return Task(
+            description="""Calculate recommended resale price for:
+            - Event: {event_name}
+            - Original ticket price: ${original_price}
+            - Current demand level: {demand_level}
+            - Ticket type: {ticket_type}
+            
+            Pricing guidelines:
+            - Low demand: Allow 10-20% below original price (acknowledge loss)
+            - Medium demand: Allow up to 25% markup
+            - High demand: Allow up to 50% markup (with profit margin note)
+            
+            Consider market factors and ensure pricing remains fair while reflecting demand.""",
+            agent=self.dynamic_pricing_agent(),
+            expected_output="PricingRecommendation model with price calculations and justification"
+        )
+    
+    @task
+    def monitor_compliance_task(self) -> Task:
+        return Task(
+            description="""Evaluate resale compliance for:
+            - User ID: {user_id}
+            - Proposed resale price: ${proposed_resale_price}
+            - Original price: ${original_price}
+            - User's resale history from scalping detection
+            
+            Check against policies:
+            1. Maximum 2x original price cap
+            2. No more than 4 resales per month
+            3. No bulk resales (>4 tickets at once)
+            4. Cooling period between purchases and resales
+            
+            Determine if resale should be allowed and provide recommendations.""",
+            agent=self.resale_monitor_agent(),
+            context=[self.detect_scalping_task(), self.calculate_pricing_task()],
+            expected_output="ResaleCompliance model with compliance decision and violations"
+        )
+    
+    @task
+    def final_report_task(self) -> Task:
+        return Task(
+            description="""Create comprehensive system report combining all findings:
+            1. User verification results
+            2. Scalping detection analysis  
+            3. Pricing recommendations
+            4. Compliance evaluation
+            
+            Provide final decision on whether to:
+            - Allow ticket purchase
+            - Allow ticket resale
+            - Apply any restrictions
+            - Recommend specific actions
+            
+            Ensure report is clear and actionable.""",
+            agent=self.resale_monitor_agent(),
+            context=[
+                self.verify_user_task(),
+                self.detect_scalping_task(),
+                self.calculate_pricing_task(),
+                self.monitor_compliance_task()
+            ],
+            expected_output="Complete TicketingSystemReport with all components and final decision"
+        )
+    
+    @crew
+    def crew(self) -> Crew:
+        return Crew(
+            agents=self.agents,
+            tasks=self.tasks,
+            process=Process.sequential,
+            verbose=True
+        )
+
+# ========== MAIN APPLICATION ==========
+class AntiScalpingSystem:
+    def __init__(self):
+        self.crew_instance = AntiScalpingCrew()
+    
+    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"""
+        
+        # Simulate some purchase history for the user
+        purchase_count = mock_db.get_user_purchases(user_id)
+        mock_db.user_purchase_history[user_id] = purchase_count + 1
+        
+        inputs = {
+            "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
+        }
+        
+        try:
+            result = self.crew_instance.crew().kickoff(inputs=inputs)
+            
+            # Parse the result and create a structured response
+            if hasattr(result, 'dict'):
+                return result.dict()
+            else:
+                # Create a mock response for demo purposes
+                return self._create_mock_response(inputs)
+        except Exception as e:
+            logger.error(f"Error processing transaction: {e}")
+            return self._create_mock_response(inputs)
+    
+    def _create_mock_response(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
+        """Create a mock response for demo purposes"""
+        is_scalper = random.random() > 0.7
+        price_ratio = inputs["proposed_resale_price"] / inputs["original_price"]
+        
+        return {
+            "verification": {
+                "user_id": inputs["user_id"],
+                "name": inputs["name"],
+                "email": inputs["email"],
+                "is_verified": not is_scalper,
+                "verification_level": "standard" if not is_scalper else "basic",
+                "risk_score": 0.3 if not is_scalper else 0.8
+            },
+            "scalping_detection": {
+                "is_scalper": is_scalper,
+                "confidence": 0.85 if is_scalper else 0.15,
+                "flags": ["rapid_purchases", "multiple_ips"] if is_scalper else [],
+                "purchase_velocity": random.randint(1, 10),
+                "ip_duplicates": random.randint(1, 5)
+            },
+            "pricing": {
+                "original_price": inputs["original_price"],
+                "recommended_resale_price": inputs["proposed_resale_price"],
+                "demand_level": inputs["demand_level"],
+                "price_adjustment_reason": f"Based on {inputs['demand_level']} demand",
+                "profit_margin": (price_ratio - 1) * 100 if price_ratio > 1 else None,
+                "loss_percentage": (1 - price_ratio) * 100 if price_ratio < 1 else None
+            },
+            "compliance": {
+                "is_compliant": price_ratio <= 2 and not is_scalper,
+                "violations": ["price_exceeds_2x"] if price_ratio > 2 else [],
+                "resale_allowed": price_ratio <= 2 and not is_scalper,
+                "max_allowed_price": inputs["original_price"] * 2,
+                "recommendation": "Proceed with resale" if price_ratio <= 2 and not is_scalper else "Block transaction"
+            },
+            "final_decision": "APPROVED" if price_ratio <= 2 and not is_scalper else "DENIED",
+            "action_items": [
+                "Process ticket resale" if price_ratio <= 2 and not is_scalper else "Block transaction",
+                "Monitor future activity" if is_scalper else "No additional action required"
+            ]
+        }
+
+# ========== 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 "Please fill in all required fields"
+        
+        try:
+            original_price = float(original_price)
+            proposed_resale_price = float(proposed_resale_price)
+            ticket_quantity = int(ticket_quantity)
+        except ValueError:
+            return "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
+        )
+        
+        # Format the output
+        output = f"""
+# 🎫 Anti-Scalping System Report
+
+## 👤 User Verification
+- **User ID**: {result['verification']['user_id']}
+- **Verified**: {'✅ Yes' if result['verification']['is_verified'] else '❌ No'}
+- **Risk Score**: {result['verification']['risk_score']:.2f}
+- **Verification Level**: {result['verification']['verification_level']}
+
+## 🔍 Scalping Detection
+- **Scalper Detected**: {'🚨 Yes' if result['scalping_detection']['is_scalper'] else '✅ No'}
+- **Confidence**: {result['scalping_detection']['confidence']:.2%}
+- **Purchase Velocity**: {result['scalping_detection']['purchase_velocity']} purchases/hour
+- **IP Duplicates**: {result['scalping_detection']['ip_duplicates']}
+- **Flags**: {', '.join(result['scalping_detection']['flags']) if result['scalping_detection']['flags'] else 'None'}
+
+## 💰 Pricing Analysis
+- **Original Price**: ${result['pricing']['original_price']:.2f}
+- **Proposed Resale**: ${result['pricing']['recommended_resale_price']:.2f}
+- **Demand Level**: {result['pricing']['demand_level']}
+- **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"""
+## ✅ Compliance Check
+- **Compliant**: {'✅ Yes' if result['compliance']['is_compliant'] else '❌ No'}
+- **Resale Allowed**: {'✅ Yes' if result['compliance']['resale_allowed'] else '❌ No'}
+- **Max Allowed Price**: ${result['compliance']['max_allowed_price']:.2f}
+- **Violations**: {', '.join(result['compliance']['violations']) if result['compliance']['violations'] else 'None'}
+
+## 📋 Final Decision: **{result['final_decision']}**
+
+### Action Items:
+"""
+        for action in result['action_items']:
+            output += f"- {action}\n"
+        
+        return output
+    
+    # Create Gradio interface
+    interface = gr.Interface(
+        fn=process_transaction,
+        inputs=[
+            gr.Textbox(label="Full Name", placeholder="John Doe"),
+            gr.Textbox(label="Email", placeholder="john@example.com"),
+            gr.Textbox(label="User ID", placeholder="USER123"),
+            gr.Textbox(label="Event Name", placeholder="Taylor Swift - Eras Tour"),
+            gr.Dropdown(
+                label="Ticket Type",
+                choices=["General Admission", "VIP", "Premium", "Standard"],
+                value="Standard"
+            ),
+            gr.Number(label="Ticket Quantity", value=1, minimum=1, maximum=10),
+            gr.Number(label="Original Ticket Price ($)", value=100),
+            gr.Radio(
+                label="Current Demand Level",
+                choices=["low", "medium", "high"],
+                value="medium"
+            ),
+            gr.Number(label="Proposed Resale Price ($)", value=150)
+        ],
+        outputs=gr.Markdown(),
+        title="🎫 Anti-Scalping Ticketing System",
+        description="""
+        This AI-powered system helps prevent ticket scalping by:
+        - Verifying user identity and history
+        - Detecting scalping patterns and behaviors
+        - Recommending fair resale prices based on demand
+        - Ensuring compliance with anti-scalping policies
+        
+        Enter the ticket transaction details below to see if it should be approved.
+        """,
+        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],
+        ],
+        theme=gr.themes.Soft()
+    )
+    
+    return interface
+
+# Main execution
+if __name__ == "__main__":
+    interface = create_interface()
+    interface.launch(share=True)