Upload 29 files

app/__init__.py

@@ -0,0 +1,8 @@
+"""
+RivalLens - Competitor Intelligence API
+
+This package provides functionality for analyzing and comparing companies
+in a given market space using various data sources and LLM-powered analysis.
+"""
+
+__version__ = "0.1.0"

app/config.py

@@ -0,0 +1,36 @@
+"""Configuration settings for the RivalLens application."""
+import os
+from pydantic_settings import BaseSettings
+from pydantic import HttpUrl
+from typing import Optional
+
+class Settings(BaseSettings):
+    # API Keys
+    DEEPSEEK_API_KEY: str
+    DEEPSEEK_API_URL: str = "https://api.deepseek.com/v1/chat/completions"
+    DEEPSEEK_ENDPOINT: Optional[str] = None  # For backward compatibility
+    NEWS_API_KEY: Optional[str] = None
+    CRUNCHBASE_API_KEY: Optional[str] = None
+    
+    # Application settings
+    APP_NAME: str = "RivalLens API"
+    DEBUG: bool = False
+    
+    # API settings
+    API_PREFIX: str = "/api/v1"
+    MAX_COMPETITORS: int = 5
+    DEFAULT_CITATION_DEPTH: int = 3
+    
+    class Config:
+        env_file = ".env"
+        env_file_encoding = 'utf-8'
+        case_sensitive = True
+        extra = 'ignore'  # Ignore extra environment variables
+        
+        @classmethod
+        def customise_sources(cls, init_settings, env_settings, file_secret_settings):
+            # This ensures that .env file is loaded with higher priority than environment variables
+            return (env_settings, init_settings, file_secret_settings)
+
+# Create settings instance
+settings = Settings()

app/main.py

@@ -0,0 +1,304 @@
+"""
+FastAPI application for RivalLens - Competitor Intelligence API
+"""
+import os
+import uuid
+import asyncio
+import logging
+from datetime import datetime
+from typing import List, Optional
+
+from fastapi import FastAPI, HTTPException, Query, BackgroundTasks
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, StreamingResponse
+
+# Configure logging before other imports to ensure all modules use it
+from logging_config import configure_logging
+logger = configure_logging()
+
+# Now import other modules
+from app.models.schemas import UserPayload, ReportResponse, CompanyData, CompetitorInsight
+from app.services.llm_client import llm
+from app.services.search import search_adapter
+from app.utils import pdf_generator, charts
+from app.models.enums import InfoCategory
+from app.config import settings
+
+# Log configuration status
+logger.info(f"Starting {settings.APP_NAME}")
+logger.debug(f"Debug mode: {settings.DEBUG}")
+logger.debug(f"Using API URL: {settings.DEEPSEEK_API_URL}")
+logger.debug(f"API Key configured: {'Yes' if settings.DEEPSEEK_API_KEY else 'No'}")
+
+# Initialize FastAPI app
+app = FastAPI(
+    title=settings.APP_NAME,
+    description="API for generating competitive intelligence reports",
+    version="1.0.0",
+    debug=settings.DEBUG
+)
+
+# Log application startup
+logger.info(f"{settings.APP_NAME} v1.0.0 starting up...")
+logger.info(f"Environment: {'development' if settings.DEBUG else 'production'}")
+logger.info(f"API Key: {'Configured' if settings.DEEPSEEK_API_KEY else 'Not configured'}")
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Helper functions
+async def build_system_prompt(company_name: str, insight_selection: List[str], deep_dive: Optional[List[str]] = None) -> str:
+    """Build a robust system prompt for the LLM based on user payload."""
+    categories = ", ".join(insight_selection)
+    deep_dive_text = ""
+    if deep_dive:
+        deep_dive_text = f"\nFor a deeper analysis, focus on: {', '.join(deep_dive)}."
+    
+    return (
+        f"You are a competitive intelligence analyst for business strategy. "
+        f"Analyze the company '{company_name}' and its competitors. "
+        f"Focus specifically on these categories: {categories}.{deep_dive_text}\n"
+        f"Provide actionable insights, highlight trends, gaps, and opportunities. "
+        f"Be concise, professional, and data-driven. "
+        f"Include an executive summary, detailed competitor insights, and a side-by-side comparison where possible. "
+        f"Use real metrics and notes to synthesize meaningful analysis. "
+        f"Do not use generic phrases like 'leading company' or 'based on available data'."
+    )
+
+
+async def generate_insights(company: CompanyData, categories: list, business_name: str = "your business") -> CompetitorInsight:
+    """Generate insights for a single company using the LLM.
+    
+    Args:
+        company: The company data to analyze
+        categories: List of categories to focus the analysis on
+        business_name: Name of the business being analyzed (for context in the prompt)
+        
+    Returns:
+        CompetitorInsight: Detailed insights about the company
+        
+    Raises:
+        ValueError: If the API key is not configured
+        httpx.HTTPStatusError: If the API request fails
+        Exception: For any other unexpected errors
+    """
+    logger = logging.getLogger(__name__)
+    
+    try:
+        # Build a detailed prompt for the LLM
+        system_prompt = (
+            f"You are a competitive intelligence analyst for business strategy. "
+            f"Analyze the company '{company.name}' as a competitor to '{business_name}'. "
+            f"Focus specifically on these categories: {', '.join(categories)}. "
+            f"Provide actionable insights, highlight trends, gaps, and opportunities. "
+            f"Be concise, professional, and data-driven. "
+            f"Use the company's metrics and notes to provide specific, meaningful analysis. "
+            f"Do not use generic phrases like 'leading company' or 'based on available data'."
+        )
+        
+        user_prompt = (
+            f"Company: {company.name}\n"
+            f"Description: {company.description or 'No description available'}\n"
+            f"Metrics: {company.metrics or 'No metrics available'}\n"
+            f"Notes: {', '.join(company.notes) if company.notes else 'No notes available'}\n\n"
+            f"Please provide a detailed analysis including: "
+            f"1. A comprehensive summary of {company.name}'s competitive position\n"
+            f"2. Analysis for each of these categories: {', '.join(categories)}\n"
+            f"3. Key strengths and weaknesses compared to {business_name}"
+        )
+        
+        logger.info(f"Generating insights for {company.name}...")
+        llm_response = await llm.summarize(system_prompt, user_prompt)
+        
+        # Process the LLM response
+        if not llm_response:
+            raise ValueError("Empty response received from LLM")
+            
+        # Use the full response as the summary
+        summary = llm_response
+        
+        # Create a category breakdown that includes the full analysis for each category
+        category_breakdown = {}
+        lines = [line.strip() for line in llm_response.split('\n') if line.strip()]
+        
+        # If we have categories, try to find sections for each one
+        if categories:
+            for category in categories:
+                # Find all lines that start with the category name or a heading marker
+                category_lines = []
+                in_category = False
+                
+                for line in lines:
+                    # Check if this line starts a new category section
+                    if (line.lower().startswith(f"{category.lower()}:") or 
+                        line.lower().startswith(f"**{category.lower()}**") or
+                        line.lower().startswith(f"### {category}")):
+                        in_category = True
+                        category_lines.append(line)
+                    # If we're in a category section, add lines until we hit another category
+                    elif in_category and any(line.lower().startswith(f"{cat.lower()}: ") for cat in categories if cat != category):
+                        in_category = False
+                        break
+                    elif in_category:
+                        category_lines.append(line)
+                
+                # If we found lines for this category, join them. Otherwise, use the full response.
+                if category_lines:
+                    category_breakdown[category] = '\n'.join(category_lines)
+                else:
+                    category_breakdown[category] = llm_response
+        else:
+            # If no specific categories, include the full response for a default category
+            category_breakdown["analysis"] = llm_response
+        
+        logger.info(f"Successfully generated insights for {company.name}")
+        return CompetitorInsight(
+            company=company,
+            summary=summary,
+            confidence="high",  # Since we're using real LLM now
+            category_breakdown=category_breakdown,
+            sources=company.sources[:3]  # Limit to top 3 sources
+        )
+        
+    except Exception as e:
+        error_msg = f"Failed to generate insights for {company.name}: {str(e)}"
+        logger.error(error_msg, exc_info=True)
+        raise
+
+# API Endpoints
+@app.get("/")
+async def root():
+    """Root endpoint with basic API information."""
+    return {
+        "app": settings.APP_NAME,
+        "status": "running",
+        "version": "1.0.0"
+    }
+
+@app.post("/api/v1/analyze", response_model=ReportResponse)
+async def analyze_competitors(
+    payload: UserPayload,
+    background_tasks: BackgroundTasks
+):
+    """
+    Main endpoint for competitor analysis.
+    """
+    # Generate a unique request ID
+    request_id = str(uuid.uuid4())
+    
+    # Get company info
+    company_name = payload.company_info.name
+    company_website = payload.company_info.website or ""
+    
+    # Get competitors
+    competitors = []
+    if payload.competitor_choice and payload.competitor_choice.competitors:
+        competitors = payload.competitor_choice.competitors
+    else:
+        # Auto-discover competitors if none provided
+        competitors = await search_adapter.discover_competitors(
+            company_name,
+            payload.business_category,
+            None,  # geography can be added later
+            settings.MAX_COMPETITORS
+        )
+    
+    # Check if we have competitors to analyze
+    if not competitors:
+        raise HTTPException(
+            status_code=400,
+            detail="No competitors found or provided for analysis"
+        )
+    
+    # Generate insights for each competitor
+    tasks = []
+    for competitor in competitors:
+        company_data = await search_adapter.enrich_company(
+            competitor,
+            citation_depth=3,  # Default citation depth
+            geography=None  # Can be updated if needed
+        )
+        task = generate_insights(
+            company=company_data,
+            categories=payload.insight_selection,
+            business_name=company_name
+        )
+        tasks.append(task)
+    
+    # Run all tasks concurrently
+    insights = await asyncio.gather(*tasks, return_exceptions=True)
+    
+    # Handle any errors
+    valid_insights = []
+    for i, insight in enumerate(insights):
+        if isinstance(insight, Exception):
+            logger.error(f"Error processing {competitors[i]}: {str(insight)}")
+        else:
+            valid_insights.append(insight)
+    
+    if not valid_insights:
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to generate insights for any competitors"
+        )
+    
+    # Prepare response with valid insights
+    report = ReportResponse(
+        request_id=request_id,
+        executive_summary=f"Analysis of {len(valid_insights)} competitors for {company_name}.",
+        top_insights=[
+            f"{insight.company.name}: {insight.summary.splitlines()[0] if insight.summary else 'No summary available'}"
+            for insight in valid_insights
+        ],
+        detailed={
+            insight.company.name: {
+                "company": insight.company.dict(),
+                "summary": insight.summary,
+                "confidence": insight.confidence,
+                "category_breakdown": insight.category_breakdown,
+                "sources": [
+                    source.dict() if hasattr(source, 'dict') else source 
+                    for source in insight.sources
+                ]
+            }
+            for insight in valid_insights
+        },
+        comparison_table=[],  # This would be populated in a real implementation
+        generated_at=datetime.utcnow(),
+        sources=[],  # This would aggregate sources in a real implementation
+    )
+    
+    # Handle export if requested
+    if payload.preferences and payload.preferences.export_format:
+        export_format = payload.preferences.export_format.lower()
+        chart_tasks = []
+        
+        # For now, we'll just log the export request
+        # In a real implementation, you would generate the appropriate export
+        logger.info(f"Export requested in format: {export_format}")
+        
+        if export_format == 'pdf':
+            # In a real implementation, you would generate charts and PDF here
+            # For now, we'll just add a placeholder
+            report.pdf_url = f"/api/v1/exports/{request_id}.pdf"
+    
+    return report
+
+async def generate_pdf_export(request_id: str, report_data: dict, charts: list):
+    """Background task to generate and store PDF report."""
+    # In a real implementation, you would:
+    # 1. Generate the PDF
+    # 2. Store it in a persistent storage (S3, filesystem, etc.)
+    # 3. Update the report status in your database
+    pass
+
+# Example usage
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

app/models/enums.py

@@ -0,0 +1,15 @@
+"""Enums and constants for the RivalLens application."""
+from enum import Enum, auto
+
+class InfoCategory(str, Enum):
+    """Allowed parent categories for competitor information."""
+    MARKET_PRESENCE = "Market Presence"
+    FINANCIAL_HEALTH = "Financial Health"
+    PRODUCTS = "Products & Offerings"
+    MARKETING = "Marketing & Branding"
+    TECH = "Technology & Innovation"
+    CUSTOMER_SENTIMENT = "Customer Sentiment"
+    HIRING = "Hiring & Organization"
+
+# Set of all allowed categories for validation
+ALLOWED_CATEGORIES = {category.value for category in InfoCategory}

app/models/schemas.py

@@ -0,0 +1,71 @@
+"""Pydantic models for request/response schemas."""
+from datetime import datetime
+from typing import List, Dict, Any, Optional
+from pydantic import BaseModel, Field, HttpUrl
+from .enums import InfoCategory
+
+class CompanyInfo(BaseModel):
+    """Company information model."""
+    name: str = Field(..., description="Name of the company")
+    website: Optional[str] = Field(None, description="Company website URL")
+
+class CompetitorChoice(BaseModel):
+    """Competitor selection model."""
+    competitors: List[str] = Field(..., description="List of competitor names")
+
+class Preferences(BaseModel):
+    """User preferences for the analysis."""
+    export_format: Optional[str] = Field(
+        None, 
+        description="Export format: PDF, Slide Deck, or None",
+        example="PDF"
+    )
+
+class UserPayload(BaseModel):
+    """Request payload from the user."""
+    business_category: str = Field(..., description="Business category/industry")
+    company_info: CompanyInfo = Field(..., description="Information about the company")
+    competitor_choice: Optional[CompetitorChoice] = Field(
+        None,
+        description="Competitor selection (required for manual competitor selection)"
+    )
+    insight_selection: List[str] = Field(
+        ...,
+        description="List of insights to include in the report"
+    )
+    deep_dive: Optional[List[str]] = Field(
+        None,
+        description="List of areas for deeper analysis"
+    )
+    preferences: Optional[Preferences] = Field(
+        None,
+        description="User preferences for the analysis"
+    )
+
+class CompanyData(BaseModel):
+    """Data structure for company information."""
+    name: str
+    website: Optional[str] = None
+    description: Optional[str] = None
+    metrics: Dict[str, Any] = {}
+    notes: List[str] = []
+    sources: List[Dict[str, Any]] = []
+
+class CompetitorInsight(BaseModel):
+    """Detailed insights for a single competitor."""
+    company: CompanyData
+    summary: str
+    confidence: str
+    category_breakdown: Dict[str, str] = {}
+    sources: List[Dict[str, Any]] = []
+
+class ReportResponse(BaseModel):
+    """Response model for the analysis report."""
+    request_id: str
+    executive_summary: str
+    top_insights: List[str]
+    detailed: Dict[str, CompetitorInsight]
+    comparison_table: List[Dict[str, Any]]
+    generated_at: datetime
+    sources: List[Dict[str, Any]]
+    pdf_url: Optional[str] = None

app/services/llm_client.py

@@ -0,0 +1,180 @@
+"""LLM client for interacting with Deepseek API."""
+import httpx
+import time
+import json
+import logging
+from typing import Optional, Dict, Any
+from app.config import settings
+
+# Get the root logger
+logger = logging.getLogger()
+
+class LLMClient:
+    """Client for interacting with LLM services."""
+    
+    def __init__(self, 
+                 api_url: Optional[str] = settings.DEEPSEEK_API_URL,
+                 api_key: Optional[str] = settings.DEEPSEEK_API_KEY):
+        self.api_url = api_url
+        self.api_key = api_key
+
+    async def summarize(self, system_prompt: str, user_prompt: str, max_tokens: int = 600) -> str:
+        """
+        Call LLM to produce a summary. Returns text.
+        Raises:
+            ValueError: If API key is not configured
+            httpx.HTTPStatusError: If the API request fails
+        """
+        # Log API key status (masking the actual key for security)
+        if not self.api_key:
+            error_msg = "Deepseek API key is not configured. Please set DEEPSEEK_API_KEY in your environment variables."
+            logger.error(error_msg)
+            raise ValueError(error_msg)
+            
+        api_key_display = f"{self.api_key[:4]}...{self.api_key[-4:]}"
+        
+        # Log request details
+        logger.info("=== LLM API Request ===")
+        # logger.info(f"API Endpoint: {self.api_url}")
+        # logger.info(f"API Key: {api_key_display}")
+        # logger.info("System Prompt:")
+        # logger.info(system_prompt)
+        logger.info("\nUser Prompt:")
+        # logger.info(user_prompt)
+        # logger.info(f"Max Tokens: {max_tokens}")
+        
+        # Record start time for performance tracking
+        start_time = time.time()
+
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {self.api_key}",
+        }
+        payload = {
+            "model": "deepseek-chat",  
+            "messages": [
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt}
+            ],
+            "max_tokens": max_tokens,
+            "temperature": 0.7
+        }
+        
+        # Prepare request data
+        request_data = {
+            "model": "deepseek-chat",
+            "messages": [
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt}
+            ],
+            "max_tokens": max_tokens,
+            "temperature": 0.7,
+            "top_p": 1.0,
+            "frequency_penalty": 0.0,
+            "presence_penalty": 0.0
+        }
+        
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {self.api_key}",
+            "Accept": "application/json"
+        }
+        
+        # Log full request payload
+        logger.info("\nRequest Payload:")
+        logger.info(json.dumps(request_data, indent=2, ensure_ascii=False))
+        logger.info("\nSending request...")
+        
+        try:
+            async with httpx.AsyncClient(timeout=60.0) as client:
+                # Make the API request
+                resp = await client.post(
+                    self.api_url,
+                    json=request_data,
+                    headers=headers,
+                    timeout=60.0
+                )
+                
+                # Calculate request duration
+                duration = time.time() - start_time
+                
+                # Log response status and timing
+                logger.info(f"\n=== LLM API Response ===")
+                logger.info(f"Status Code: {resp.status_code}")
+                logger.info(f"Response Time: {duration:.2f} seconds")
+                
+                # Parse response
+                resp.raise_for_status()
+                data = resp.json()
+                
+                # Log full response
+                logger.info("\nResponse Headers:")
+                for header, value in resp.headers.items():
+                    logger.info(f"  {header}: {value}")
+                
+                logger.info("\nResponse Body:")
+                logger.info(json.dumps(data, indent=2, ensure_ascii=False))
+                
+                # Extract and log content
+                content = None
+                if "choices" in data and len(data["choices"]) > 0:
+                    content = data["choices"][0]["message"]["content"]
+                    logger.info("\nGenerated Content:")
+                    logger.info(content)
+                elif "text" in data:
+                    content = data["text"]
+                elif "output" in data:
+                    content = data["output"]
+                else:
+                    # If we get here, the response format is unexpected
+                    error_msg = f"Unexpected API response format: {data}"
+                    logger.error(error_msg)
+                    raise ValueError(error_msg)
+                
+                # Log token usage if available
+                # if "usage" in data:
+                #     usage = data["usage"]
+                #     logger.info("\nToken Usage:")
+                #     logger.info(f"Prompt Tokens: {usage.get('prompt_tokens', 'N/A')}")
+                #     logger.info(f"Completion Tokens: {usage.get('completion_tokens', 'N/A')}")
+                #     logger.info(f"Total Tokens: {usage.get('total_tokens', 'N/A')}")
+                
+                logger.info("=" * 50)  # End of request/response log
+                
+                if content is None:
+                    raise ValueError("No content found in the response")
+                    
+                return content
+                
+        except httpx.HTTPStatusError as e:
+            duration = time.time() - start_time
+            error_msg = f"API request failed with status {e.response.status_code} after {duration:.2f}s"
+            logger.error(error_msg)
+            try:
+                error_data = e.response.json()
+                logger.error("Error details: %s", json.dumps(error_data, indent=2))
+            except:
+                logger.error("Response text: %s", e.response.text)
+            raise
+            
+        except httpx.RequestError as e:
+            duration = time.time() - start_time
+            error_msg = f"Failed to connect to the API after {duration:.2f}s: {str(e)}"
+            logger.error(error_msg, exc_info=True)
+            raise ConnectionError(error_msg) from e
+            
+        except json.JSONDecodeError as e:
+            duration = time.time() - start_time
+            error_msg = f"Failed to parse API response after {duration:.2f}s: {str(e)}"
+            logger.error(error_msg)
+            logger.error("Response text: %s", getattr(resp, 'text', 'No response content'))
+            raise ValueError("Invalid JSON response from API") from e
+            
+        except Exception as e:
+            duration = time.time() - start_time
+            error_msg = f"Unexpected error after {duration:.2f}s during API call: {str(e)}"
+            logger.error(error_msg, exc_info=True)
+            raise
+
+# Singleton instance
+llm = LLMClient()

app/services/search.py

@@ -0,0 +1,64 @@
+"""Search functionality for company data and competitor discovery."""
+from typing import List, Optional, Dict, Any
+from app.models.schemas import CompanyData
+from app.config import settings
+
+class SearchAdapter:
+    """Provides company discovery and enrichment functions."""
+    
+    def __init__(self, 
+                 news_api_key: Optional[str] = settings.NEWS_API_KEY, 
+                 crunchbase_key: Optional[str] = settings.CRUNCHBASE_API_KEY):
+        self.news_api_key = news_api_key
+        self.crunchbase_key = crunchbase_key
+
+    async def discover_competitors(self, 
+                                 business_name: str, 
+                                 business_desc: str, 
+                                 geography: Optional[str] = None, 
+                                 limit: int = 5) -> List[str]:
+        """Discover potential competitors for a business."""
+        if self.crunchbase_key:
+            # Placeholder for actual Crunchbase API integration
+            pass
+
+        # Fallback mock strategy
+        keywords = business_desc.lower()
+        if "hr" in keywords or "human resources" in keywords or "payroll" in keywords:
+            candidates = ["BambooHR", "Gusto", "Rippling", "Zoho People", "UKG"]
+        elif "saas" in keywords and "analytics" in keywords:
+            candidates = ["Mixpanel", "Amplitude", "Heap", "Pendo", "Looker"]
+        elif "ecommerce" in keywords or "shop" in keywords:
+            candidates = ["Shopify", "BigCommerce", "Magento", "Wix eCommerce", "WooCommerce"]
+        else:
+            candidates = [f"Competitor {chr(65 + i)}" for i in range(5)]
+
+        return candidates[:limit]
+
+    async def enrich_company(self, 
+                           company_name: str, 
+                           citation_depth: int = 3, 
+                           geography: Optional[str] = None) -> CompanyData:
+        """Gather structured and unstructured info for a company."""
+        # Mock implementation - replace with actual API calls
+        return CompanyData(
+            name=company_name,
+            website=f"https://{company_name.lower().replace(' ', '')}.example.com",
+            description=f"A leading company in their industry, {company_name} provides excellent services.",
+            metrics={
+                "employees": 1000,
+                "revenue": "$10M - $50M",
+                "founded": 2010
+            },
+            notes=[
+                f"{company_name} recently expanded to new markets.",
+                "Strong social media presence with growing engagement."
+            ],
+            sources=[
+                {"type": "web", "url": f"https://{company_name.lower().replace(' ', '')}.com/about"},
+                {"type": "news", "title": f"{company_name} announces new product line"}
+            ][:citation_depth]
+        )
+
+# Singleton instance
+search_adapter = SearchAdapter()

app/utils/charts.py

@@ -0,0 +1,100 @@
+"""Chart generation utilities."""
+import io
+from typing import List, Dict, Any, Optional
+import matplotlib.pyplot as plt
+import numpy as np
+from app.models.schemas import CompanyData
+
+def generate_bar_chart(
+    data: Dict[str, float],
+    title: str,
+    x_label: str,
+    y_label: str
+) -> bytes:
+    """Generate a bar chart and return as PNG bytes."""
+    plt.figure(figsize=(10, 6))
+    
+    # Prepare data
+    labels = list(data.keys())
+    values = list(data.values())
+    
+    # Create bar chart
+    bars = plt.bar(labels, values, color='skyblue')
+    
+    # Add value labels on top of each bar
+    for bar in bars:
+        height = bar.get_height()
+        plt.text(bar.get_x() + bar.get_width()/2., height,
+                f'{height:,.0f}',
+                ha='center', va='bottom')
+    
+    # Customize the chart
+    plt.title(title, fontsize=14, pad=20)
+    plt.xlabel(x_label, fontsize=12)
+    plt.ylabel(y_label, fontsize=12)
+    plt.xticks(rotation=45, ha='right')
+    plt.tight_layout()
+    
+    # Save to bytes
+    buffer = io.BytesIO()
+    plt.savefig(buffer, format='png', dpi=100, bbox_inches='tight')
+    plt.close()
+    
+    buffer.seek(0)
+    return buffer.getvalue()
+
+async def generate_company_metrics_charts(
+    companies: List[CompanyData],
+    metrics: List[str]
+) -> List[bytes]:
+    """Generate charts for company metrics."""
+    charts = []
+    
+    # Example: Employee count comparison
+    employee_data = {}
+    revenue_data = {}
+    
+    for company in companies:
+        if 'employees' in company.metrics:
+            try:
+                employee_data[company.name] = float(company.metrics['employees'])
+            except (ValueError, TypeError):
+                pass
+                
+        if 'revenue' in company.metrics and isinstance(company.metrics['revenue'], str):
+            # Simple revenue parsing (in a real app, use a proper currency parser)
+            rev_str = company.metrics['revenue'].replace('$', '').replace(',', '').replace(' ', '')
+            if '-' in rev_str:
+                rev_avg = sum(float(x) for x in rev_str.split('-')) / 2
+                revenue_data[company.name] = rev_avg
+            else:
+                try:
+                    revenue_data[company.name] = float(rev_str)
+                except (ValueError, TypeError):
+                    pass
+    
+    # Generate employee chart if we have data
+    if employee_data:
+        charts.append(
+            generate_bar_chart(
+                employee_data,
+                "Employee Count Comparison",
+                "Company",
+                "Number of Employees"
+            )
+        )
+    
+    # Generate revenue chart if we have data
+    if revenue_data:
+        # Convert to millions for better readability
+        revenue_millions = {k: v / 1_000_000 for k, v in revenue_data.items()}
+        charts.append(
+            generate_bar_chart(
+                revenue_millions,
+                "Estimated Annual Revenue (Millions)",
+                "Company",
+                "Revenue ($M)"
+            )
+        )
+    
+    return charts

app/utils/pdf_generator.py

@@ -0,0 +1,50 @@
+"""PDF generation utilities for reports."""
+import io
+from datetime import datetime
+from typing import List, Optional
+from reportlab.lib.pagesizes import letter
+from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image, Table
+from reportlab.lib.styles import getSampleStyleSheet
+from reportlab.lib import colors
+
+async def generate_pdf_report(
+    request_id: str,
+    report_data: dict,
+    charts: Optional[List[bytes]] = None
+) -> bytes:
+    """Generate a PDF report from the analysis results."""
+    buffer = io.BytesIO()
+    doc = SimpleDocTemplate(buffer, pagesize=letter)
+    styles = getSampleStyleSheet()
+    elements = []
+    
+    # Title
+    title = Paragraph("Competitor Analysis Report", styles['Title'])
+    elements.append(title)
+    
+    # Report metadata
+    elements.append(Paragraph(f"Report ID: {request_id}", styles['Normal']))
+    elements.append(Paragraph(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", 
+                           styles['Normal']))
+    elements.append(Spacer(1, 20))
+    
+    # Executive Summary
+    elements.append(Paragraph("Executive Summary", styles['Heading1']))
+    elements.append(Paragraph(report_data.get('executive_summary', ''), styles['Normal']))
+    
+    # Add charts if available
+    if charts:
+        elements.append(Spacer(1, 20))
+        elements.append(Paragraph("Key Metrics", styles['Heading2']))
+        for chart in charts:
+            try:
+                img = Image(io.BytesIO(chart), width=400, height=300)
+                elements.append(img)
+                elements.append(Spacer(1, 10))
+            except:
+                continue
+    
+    # Build PDF
+    doc.build(elements)
+    buffer.seek(0)
+    return buffer.getvalue()

logging_config.py

@@ -0,0 +1,62 @@
+"""Logging configuration for RivalLens."""
+import logging
+import sys
+from pathlib import Path
+
+# Create logs directory if it doesn't exist
+log_dir = Path("logs")
+log_dir.mkdir(exist_ok=True)
+
+# Logging configuration
+LOGGING_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "formatters": {
+        "standard": {
+            "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+            "datefmt": "%Y-%m-%d %H:%M:%S"
+        },
+    },
+    "handlers": {
+        "console": {
+            "level": "INFO",
+            "formatter": "standard",
+            "class": "logging.StreamHandler",
+            "stream": sys.stdout,
+        },
+        "file": {
+            "level": "DEBUG",
+            "formatter": "standard",
+            "class": "logging.handlers.RotatingFileHandler",
+            "filename": log_dir / "rival_lens.log",
+            "maxBytes": 10485760,  # 10MB
+            "backupCount": 5,
+            "encoding": "utf8"
+        },
+    },
+    "loggers": {
+        "": {  # root logger
+            "handlers": ["console", "file"],
+            "level": "DEBUG",
+            "propagate": True
+        },
+        "app": {
+            "handlers": ["console", "file"],
+            "level": "DEBUG",
+            "propagate": False
+        },
+        "__main__": {
+            "handlers": ["console", "file"],
+            "level": "DEBUG",
+            "propagate": False
+        },
+    }
+}
+
+def configure_logging():
+    """Configure logging for the application."""
+    import logging.config
+    logging.config.dictConfig(LOGGING_CONFIG)
+    logger = logging.getLogger(__name__)
+    logger.info("Logging configured")
+    return logger