Web Form v2.0 API Integration Guide

POST /api/v2/analyze-content

{
  "content": "string - file content or manual input",
  "filename": "string - original filename",
  "file_type": "string - MIME type",
  "metadata": {
    "upload_source": "upload|manual",
    "user_session": "string",
    "timestamp": "ISO datetime"
  }
}

{
  "category": {
    "detected_category": "traditional_team_sports|individual_competitive|emerging_sports|recreational_lifestyle|extreme_action|other_uncategorized",
    "confidence_score": 0.95,
    "category_name": "Traditional Team Sports",
    "indicators_found": ["basketball", "professional", "teams", "fans"],
    "alternative_categories": [
      {"category": "emerging_sports", "confidence": 0.15}
    ]
  },
  "processing_time_ms": 1250,
  "status": "success"
}

POST /api/v2/detect-archetype

{
  "category": "traditional_team_sports",
  "content_analysis": {
    "keywords": ["professional", "teams", "revenue", "fans"],
    "metrics": {
      "team_count": 12,
      "estimated_fans": 500000,
      "revenue_indicators": ["sponsors", "media"]
    }
  },
  "additional_data": {}
}

{
  "archetype": {
    "type": "professional_regional",
    "name": "Regional Professional League",
    "characteristics": ["High revenue", "Regional presence", "Media partnerships"],
    "market_potential": "High",
    "investment_priority": "Medium-High",
    "typical_metrics": {
      "revenue": "5-50M",
      "fans": "100K-1M",
      "teams": "8-20"
    }
  },
  "confidence": 0.87,
  "status": "success"
}

POST /api/v2/evaluate-partnership

{
  "league_data": {
    "archetype": "professional_regional",
    "category": "traditional_team_sports",
    "content_analysis": {},
    "manual_inputs": {}
  },
  "evaluation_criteria": {
    "use_latest_criteria": true,
    "custom_weights": null
  }
}

{
  "evaluation": {
    "social_presence": {
      "score": 0.75,
      "factors": {
        "follower_count": 0.8,
        "engagement_rate": 0.7,
        "content_quality": 0.6,
        "platform_diversity": 0.9
      }
    },
    "market_position": {
      "score": 0.80,
      "factors": {
        "market_size": 0.7,
        "competitive_position": 0.85,
        "geographic_reach": 0.8,
        "media_coverage": 0.75
      }
    },
    "growth_potential": {
      "score": 0.85,
      "factors": {
        "revenue_growth": 0.8,
        "fan_base_growth": 0.9,
        "demographic_trends": 0.85,
        "digital_transformation": 0.8,
        "market_expansion": 0.75
      }
    },
    "strategic_partnerships": {
      "score": 0.75,
      "factors": {
        "sponsor_quality": 0.8,
        "partnership_diversity": 0.7,
        "collaboration_history": 0.75,
        "strategic_alignment": 0.75
      }
    },
    "overall_score": 0.7875,
    "priority_level": "medium_high_priority",
    "recommendation": "Strong partnership potential with development opportunities"
  },
  "next_steps": {
    "actions": [
      "Schedule detailed partnership consultation within 2 weeks",
      "Conduct comprehensive market analysis",
      "Prepare standard partnership package"
    ],
    "timeline": "4-8 weeks to partnership agreement",
    "resource_allocation": "Medium-High - experienced partnership manager"
  },
  "status": "success"
}

GET /api/v2/categorization-examples?category={category}&limit={limit}

{
  "examples": [
    {
      "name": "Metro Professional Basketball League",
      "sport": "Basketball", 
      "description": "12-team professional basketball league...",
      "input_indicators": ["basketball", "professional", "teams"],
      "bucket_reason": "Professional basketball league with traditional structure",
      "archetype": "professional_regional",
      "partnership_potential": 0.85
    }
  ],
  "total_examples": 18,
  "category_info": {
    "name": "Traditional Team Sports",
    "description": "Established team sports with standardized rules"
  }
}

# In pages/35_📋_Web_Form_v2.py
 
import requests
import streamlit as st
 
BACKEND_URL = os.getenv('BACKEND_URL', 'http://localhost:8080')
 
def analyze_content_with_api(content: str, filename: str, file_type: str) -> Dict[str, Any]:
    """Call backend API for content analysis"""
    try:
        response = requests.post(
            f"{BACKEND_URL}/api/v2/analyze-content",
            json={
                "content": content,
                "filename": filename, 
                "file_type": file_type,
                "metadata": {
                    "upload_source": "upload" if filename != "manual_input.txt" else "manual",
                    "user_session": st.session_state.get('session_id', 'anonymous'),
                    "timestamp": datetime.now().isoformat()
                }
            },
            timeout=10
        )
        return response.json()
    except Exception as e:
        st.error(f"Analysis failed: {str(e)}")
        return None
 
def detect_archetype_with_api(category: str, content_data: Dict[str, Any]) -> Dict[str, Any]:
    """Call backend API for archetype detection"""
    try:
        response = requests.post(
            f"{BACKEND_URL}/api/v2/detect-archetype",
            json={
                "category": category,
                "content_analysis": content_data,
                "additional_data": st.session_state.get('onboarding_data', {})
            },
            timeout=10
        )
        return response.json()
    except Exception as e:
        st.error(f"Archetype detection failed: {str(e)}")
        return None
 
def evaluate_partnership_with_api(league_data: Dict[str, Any]) -> Dict[str, Any]:
    """Call backend API for partnership evaluation"""
    try:
        response = requests.post(
            f"{BACKEND_URL}/api/v2/evaluate-partnership",
            json={
                "league_data": league_data,
                "evaluation_criteria": {
                    "use_latest_criteria": True,
                    "custom_weights": None
                }
            },
            timeout=15
        )
        return response.json()
    except Exception as e:
        st.error(f"Partnership evaluation failed: {str(e)}")
        return None
 
# Modified detect_category_from_upload function
def detect_category_from_upload(content: str, filename: str) -> str:
    """Enhanced category detection using backend API"""
    
    # Try API first
    api_result = analyze_content_with_api(content, filename, "text/plain")
    if api_result and api_result.get('status') == 'success':
        return api_result['category']['detected_category']
    
    # Fallback to local keyword analysis
    # ... existing keyword logic ...
    return 'other_uncategorized'

# In pages/36_🔮_Web_Form_Popup_Modal.py
 
def quick_analyze_with_api(data: Dict[str, Any]) -> Dict[str, Any]:
    """Quick analysis for modal version"""
    try:
        # Use lightweight analysis endpoint
        response = requests.post(
            f"{BACKEND_URL}/api/v2/quick-analyze",
            json=data,
            timeout=5  # Shorter timeout for quick results
        )
        return response.json()
    except Exception as e:
        # Graceful fallback to local processing
        return quick_evaluate_local(data)
 
def quick_evaluate_local(data: Dict[str, Any]) -> Dict[str, Any]:
    """Local fallback for quick evaluation"""
    # ... existing quick_evaluate logic ...
    pass

-- League assessments table
CREATE TABLE league_assessments (
    id SERIAL PRIMARY KEY,
    session_id VARCHAR(255),
    league_name VARCHAR(255),
    uploaded_filename VARCHAR(255),
    detected_category VARCHAR(100),
    category_confidence DECIMAL(3,2),
    detected_archetype VARCHAR(100),
    archetype_confidence DECIMAL(3,2),
    social_score DECIMAL(3,2),
    market_score DECIMAL(3,2), 
    growth_score DECIMAL(3,2),
    partnership_score DECIMAL(3,2),
    overall_score DECIMAL(3,2),
    priority_level VARCHAR(50),
    recommendation TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
 
-- Assessment actions table
CREATE TABLE assessment_actions (
    id SERIAL PRIMARY KEY,
    assessment_id INTEGER REFERENCES league_assessments(id),
    action_type VARCHAR(100),
    action_description TEXT,
    timeline VARCHAR(100),
    status VARCHAR(50) DEFAULT 'pending',
    assigned_to VARCHAR(255),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
 
-- Categorization feedback table
CREATE TABLE categorization_feedback (
    id SERIAL PRIMARY KEY,
    assessment_id INTEGER REFERENCES league_assessments(id),
    original_category VARCHAR(100),
    corrected_category VARCHAR(100),
    feedback_reason TEXT,
    user_feedback_score INTEGER CHECK (user_feedback_score >= 1 AND user_feedback_score <= 5),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

# In your backend API (app/routers/assessment_v2.py)
 
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import Dict, Any, List, Optional
import json
import os
 
router = APIRouter(prefix="/api/v2", tags=["assessment"])
 
class ContentAnalysisRequest(BaseModel):
    content: str
    filename: str
    file_type: str
    metadata: Dict[str, Any]
 
class ContentAnalysisResponse(BaseModel):
    category: Dict[str, Any]
    processing_time_ms: int
    status: str
 
@router.post("/analyze-content", response_model=ContentAnalysisResponse)
async def analyze_content(request: ContentAnalysisRequest):
    """Analyze uploaded content for sport categorization"""
    
    start_time = time.time()
    
    try:
        # Load categorization algorithm
        with open('config/company_evaluation_criteria.json', 'r') as f:
            criteria = json.load(f)
        
        algorithm = criteria.get('categorization_algorithm', {})
        
        # Perform analysis
        category_result = perform_content_analysis(
            request.content, 
            request.filename,
            algorithm
        )
        
        processing_time = int((time.time() - start_time) * 1000)
        
        return ContentAnalysisResponse(
            category=category_result,
            processing_time_ms=processing_time,
            status="success"
        )
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
 
class ArchetypeDetectionRequest(BaseModel):
    category: str
    content_analysis: Dict[str, Any]
    additional_data: Dict[str, Any]
 
@router.post("/detect-archetype")
async def detect_archetype(request: ArchetypeDetectionRequest):
    """Determine league archetype based on analysis"""
    
    try:
        # Load archetype definitions
        with open('database/categorization_examples.json', 'r') as f:
            examples_data = json.load(f)
        
        # Determine archetype using algorithm
        archetype_result = calculate_archetype(
            request.category,
            request.content_analysis,
            examples_data
        )
        
        return {"archetype": archetype_result, "confidence": 0.87, "status": "success"}
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
 
class PartnershipEvaluationRequest(BaseModel):
    league_data: Dict[str, Any]
    evaluation_criteria: Dict[str, Any]
 
@router.post("/evaluate-partnership")
async def evaluate_partnership(request: PartnershipEvaluationRequest):
    """Calculate comprehensive partnership potential"""
    
    try:
        # Load evaluation criteria
        with open('config/company_evaluation_criteria.json', 'r') as f:
            criteria = json.load(f)
        
        # Calculate scores
        evaluation_result = calculate_partnership_scores(
            request.league_data,
            criteria['evaluation_criteria']
        )
        
        # Generate recommendations
        recommendations = generate_recommendations(
            evaluation_result,
            criteria['decision_matrix']
        )
        
        return {
            "evaluation": evaluation_result,
            "next_steps": recommendations,
            "status": "success"
        }
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
 
@router.get("/categorization-examples")
async def get_categorization_examples(category: Optional[str] = None, limit: int = 10):
    """Retrieve categorization examples for user education"""
    
    try:
        with open('database/categorization_examples.json', 'r') as f:
            examples_data = json.load(f)
        
        if category:
            category_data = examples_data['categorization_examples'].get(category, {})
            examples = category_data.get('examples', [])[:limit]
        else:
            # Return examples from all categories
            examples = []
            for cat_data in examples_data['categorization_examples'].values():
                examples.extend(cat_data.get('examples', []))
            examples = examples[:limit]
        
        return {
            "examples": examples,
            "total_examples": len(examples),
            "category_info": category_data if category else None
        }
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# app/services/content_analyzer.py
 
import re
import json
from typing import Dict, List, Any
from collections import defaultdict
 
class ContentAnalyzer:
    def __init__(self, criteria_file: str):
        with open(criteria_file, 'r') as f:
            self.criteria = json.load(f)
        
        self.algorithm = self.criteria.get('categorization_algorithm', {})
        self.primary_keywords = self.algorithm.get('primary_keywords', {})
        self.context_indicators = self.algorithm.get('context_indicators', {})
    
    def analyze_content(self, content: str, filename: str) -> Dict[str, Any]:
        """Perform comprehensive content analysis"""
        
        content_lower = content.lower()
        filename_lower = filename.lower()
        combined_text = f"{content_lower} {filename_lower}"
        
        # Calculate category scores
        category_scores = {}
        for category, keywords in self.primary_keywords.items():
            score = sum(1 for keyword in keywords if keyword in combined_text)
            # Normalize by keyword count
            normalized_score = score / len(keywords) if keywords else 0
            category_scores[category] = normalized_score
        
        # Find best category
        best_category = max(category_scores, key=category_scores.get)
        best_score = category_scores[best_category]
        
        # Get alternative categories
        sorted_categories = sorted(
            category_scores.items(), 
            key=lambda x: x[1], 
            reverse=True
        )[1:3]  # Top 2 alternatives
        
        # Extract specific indicators found
        indicators_found = []
        for keyword in self.primary_keywords.get(best_category, []):
            if keyword in combined_text:
                indicators_found.append(keyword)
        
        return {
            "detected_category": best_category,
            "confidence_score": min(best_score * 2, 1.0),  # Scale to 0-1
            "category_name": self._get_category_name(best_category),
            "indicators_found": indicators_found,
            "alternative_categories": [
                {"category": cat, "confidence": score} 
                for cat, score in sorted_categories
            ]
        }
    
    def _get_category_name(self, category_key: str) -> str:
        """Convert category key to display name"""
        return category_key.replace('_', ' ').title()
 
class ArchetypeDetector:
    def __init__(self, examples_file: str):
        with open(examples_file, 'r') as f:
            self.examples_data = json.load(f)
    
    def detect_archetype(self, category: str, content_analysis: Dict[str, Any]) -> Dict[str, Any]:
        """Detect league archetype based on category and content"""
        
        # Simplified archetype detection logic
        # In production, use more sophisticated ML model
        
        content_str = str(content_analysis).lower()
        
        # Look for archetype indicators
        professional_indicators = ['professional', 'pro', 'salary', 'contract', 'broadcast', 'media']
        revenue_indicators = ['million', 'revenue', 'budget', 'sponsors', 'ticket sales']
        scale_indicators = ['national', 'international', 'regional', 'teams', 'fans']
        
        pro_score = sum(1 for indicator in professional_indicators if indicator in content_str)
        revenue_score = sum(1 for indicator in revenue_indicators if indicator in content_str) 
        scale_score = sum(1 for indicator in scale_indicators if indicator in content_str)
        
        total_score = pro_score + revenue_score + scale_score
        
        # Map to archetype
        if total_score >= 8 and category in ['traditional_team_sports']:
            archetype_key = 'professional_major'
        elif total_score >= 6:
            archetype_key = 'professional_regional'
        elif total_score >= 4:
            archetype_key = 'semi_pro_emerging'
        elif category in ['recreational_lifestyle']:
            archetype_key = 'amateur_organized'
        elif category in ['emerging_sports']:
            archetype_key = 'semi_pro_emerging'
        else:
            archetype_key = 'niche_specialty'
        
        # Load archetype data (you'll need to add this to your database)
        archetype_definitions = {
            'professional_major': {
                'name': 'Major Professional League',
                'characteristics': ['High revenue', 'National/International reach', 'Media partnerships'],
                'market_potential': 'Premium',
                'investment_priority': 'High'
            },
            'professional_regional': {
                'name': 'Regional Professional League', 
                'characteristics': ['Moderate revenue', 'Regional presence', 'Local media'],
                'market_potential': 'High',
                'investment_priority': 'Medium-High'
            }
            # ... other archetypes
        }
        
        return {
            'type': archetype_key,
            **archetype_definitions.get(archetype_key, {})
        }
 
class PartnershipEvaluator:
    def __init__(self, criteria_file: str):
        with open(criteria_file, 'r') as f:
            self.criteria = json.load(f)
        
        self.evaluation_criteria = self.criteria.get('evaluation_criteria', {})
        self.archetype_modifiers = self.criteria.get('archetype_specific_modifiers', {})
    
    def evaluate_partnership(self, league_data: Dict[str, Any]) -> Dict[str, Any]:
        """Calculate comprehensive partnership evaluation"""
        
        archetype = league_data.get('archetype', {}).get('type', 'amateur_organized')
        category = league_data.get('category', 'other_uncategorized')
        content_analysis = league_data.get('content_analysis', {})
        
        # Calculate individual dimension scores
        scores = {}
        
        # Social presence score
        scores['social_presence'] = self._calculate_social_score(content_analysis, archetype)
        
        # Market position score 
        scores['market_position'] = self._calculate_market_score(content_analysis, archetype)
        
        # Growth potential score
        scores['growth_potential'] = self._calculate_growth_score(content_analysis, archetype)
        
        # Strategic partnerships score
        scores['strategic_partnerships'] = self._calculate_partnership_score(content_analysis, archetype)
        
        # Calculate weighted overall score
        weights = {dim: data['weight'] for dim, data in self.evaluation_criteria.items()}
        overall_score = sum(scores[dim] * weights.get(dim, 0.25) for dim in scores)
        
        # Apply archetype modifiers
        if archetype in self.archetype_modifiers:
            modifiers = self.archetype_modifiers[archetype]
            for dim in scores:
                modifier_key = f"{dim.split('_')[0]}_modifier"
                if modifier_key in modifiers:
                    scores[dim] *= modifiers[modifier_key]
        
        # Recalculate overall with modifiers
        overall_score = sum(scores[dim] * weights.get(dim, 0.25) for dim in scores)
        
        return {
            **{dim: {"score": score} for dim, score in scores.items()},
            "overall_score": min(overall_score, 1.0),
            "priority_level": self._get_priority_level(overall_score),
            "recommendation": self._generate_recommendation(overall_score, archetype)
        }
    
    def _calculate_social_score(self, content_analysis: Dict[str, Any], archetype: str) -> float:
        """Calculate social presence score based on content indicators"""
        base_score = 0.5
        content_str = str(content_analysis).lower()
        
        # Look for social indicators
        social_indicators = ['social', 'facebook', 'instagram', 'twitter', 'youtube', 'followers', 'digital']
        found_indicators = sum(1 for indicator in social_indicators if indicator in content_str)
        
        # Scale based on indicators found
        score_boost = min(found_indicators * 0.1, 0.4)
        return min(base_score + score_boost, 1.0)
    
    # ... implement other scoring methods
    
    def _get_priority_level(self, overall_score: float) -> str:
        """Map overall score to priority level"""
        if overall_score >= 0.8:
            return "high_priority"
        elif overall_score >= 0.65:
            return "medium_high_priority"
        elif overall_score >= 0.5:
            return "medium_priority"
        elif overall_score >= 0.35:
            return "low_priority"
        else:
            return "no_priority"

# Add new routers to main FastAPI app
# app/main.py
from routers.assessment_v2 import router as assessment_router
app.include_router(assessment_router)
 
# Deploy services
python -m uvicorn app.main:app --host 0.0.0.0 --port 8080

# Navigate to Streamlit client directory
cd clients/streamlit-001-league-questionnaire-to-contract-uses-endpoint
 
# Install any new dependencies
pip install -r requirements.txt
 
# Run Streamlit with new pages
streamlit run app.py

# Run schema migrations
python manage.py migrate
 
# Load initial data
python manage.py loaddata categorization_examples.json

# Set environment variables
export BACKEND_URL=http://localhost:8080
export ENABLE_V2_FEATURES=true
export EVALUATION_MODE=production  # or development
 
# Update config files
cp config/company_evaluation_criteria.json /path/to/backend/config/