Deployment from Replit

hf_database.py

@@ -10,7 +10,12 @@ from sqlalchemy.orm import sessionmaker
 from sqlalchemy.pool import StaticPool
 from src.models.base import Base
 from src.models.user import User
-from src.api.security import get_password_hash
+try:
+    # Try to import from src.api.security first (for local development)
+    from src.api.security import get_password_hash
+except ImportError:
+    # Fall back to simplified security module for HF (copied during deployment)
+    from security_hf import get_password_hash
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)

security_hf.py

@@ -0,0 +1,16 @@
+"""
+Simplified security module for Hugging Face deployment.
+Contains only the essential functions needed for HF deployment.
+"""
+from passlib.context import CryptContext
+
+# Set up password hashing
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+def get_password_hash(password: str) -> str:
+    """Hash a password for storage"""
+    return pwd_context.hash(password)
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """Verify a password against a hash"""
+    return pwd_context.verify(plain_password, hashed_password)

src/__init__.py

@@ -0,0 +1,3 @@
+"""
+Package initialization for src.
+"""

src/api/__init__.py

@@ -0,0 +1,3 @@
+"""
+Package initialization for API.
+"""

src/api/auth.py

@@ -0,0 +1,105 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy.ext.asyncio import AsyncSession
+from jose import JWTError, jwt
+from datetime import datetime, timedelta
+from typing import Optional, Dict, Any
+import os
+import logging
+
+from src.api.database import get_db
+from src.api.schemas import TokenData, UserInDB
+from src.api.services.user_service import get_user_by_username
+
+# Configure logger
+logger = logging.getLogger(__name__)
+
+# Constants for JWT
+SECRET_KEY = os.getenv("JWT_SECRET_KEY", "your-secret-key-for-jwt-please-change-in-production")
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+# OAuth2PasswordBearer for token extraction
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/v1/auth/token")
+
+def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
+    """
+    Create a JWT access token.
+    
+    Args:
+        data: Dictionary of data to encode in the token
+        expires_delta: Optional expiration time delta
+        
+    Returns:
+        str: JWT token
+    """
+    to_encode = data.copy()
+    
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+        
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    
+    return encoded_jwt
+
+async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)) -> UserInDB:
+    """
+    Get the current authenticated user based on the JWT token.
+    
+    Args:
+        token: JWT token
+        db: Database session
+        
+    Returns:
+        UserInDB: User data
+        
+    Raises:
+        HTTPException: If authentication fails
+    """
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    
+    try:
+        # Decode JWT
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        
+        if username is None:
+            raise credentials_exception
+            
+        token_data = TokenData(username=username)
+    except JWTError as e:
+        logger.error(f"JWT error: {e}")
+        raise credentials_exception
+        
+    # Get user from database
+    user = await get_user_by_username(db, username=token_data.username)
+    
+    if user is None:
+        raise credentials_exception
+        
+    return user
+
+async def get_current_active_user(current_user: UserInDB = Depends(get_current_user)) -> UserInDB:
+    """
+    Get the current active user.
+    
+    Args:
+        current_user: Current authenticated user
+        
+    Returns:
+        UserInDB: User data
+        
+    Raises:
+        HTTPException: If user is inactive
+    """
+    if not current_user.is_active:
+        raise HTTPException(status_code=400, detail="Inactive user")
+        
+    return current_user

src/api/database.py

@@ -0,0 +1,73 @@
+"""
+Database configuration and setup for API.
+"""
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+import os
+from typing import AsyncGenerator
+import logging
+
+# Configure logger
+logger = logging.getLogger(__name__)
+
+# Get database URL from environment (convert synchronous to async URL)
+db_url = os.getenv("DATABASE_URL", "")
+if db_url.startswith("postgresql://"):
+    # Remove sslmode parameter if present which causes issues with asyncpg
+    if "?" in db_url:
+        base_url, params = db_url.split("?", 1)
+        param_list = params.split("&")
+        filtered_params = [p for p in param_list if not p.startswith("sslmode=")]
+        if filtered_params:
+            db_url = f"{base_url}?{'&'.join(filtered_params)}"
+        else:
+            db_url = base_url
+    
+    ASYNC_DATABASE_URL = db_url.replace("postgresql://", "postgresql+asyncpg://", 1)
+else:
+    ASYNC_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres"
+
+# Create async engine
+engine = create_async_engine(
+    ASYNC_DATABASE_URL,
+    echo=False,  # Set to True for debugging
+    future=True,
+    pool_size=5,
+    max_overflow=10
+)
+
+# Create async session factory
+async_session = sessionmaker(
+    engine, 
+    class_=AsyncSession, 
+    expire_on_commit=False
+)
+
+async def get_db() -> AsyncGenerator[AsyncSession, None]:
+    """
+    Get database session generator.
+    
+    Yields:
+        AsyncSession: Database session
+    """
+    session = async_session()
+    try:
+        yield session
+        await session.commit()
+    except Exception as e:
+        await session.rollback()
+        logger.error(f"Database error: {e}")
+        raise
+    finally:
+        await session.close()
+
+# Dependency for getting DB session
+async def get_db_session() -> AsyncSession:
+    """
+    Get database session.
+    
+    Returns:
+        AsyncSession: Database session
+    """
+    async with async_session() as session:
+        return session

src/api/main.py

@@ -0,0 +1,73 @@
+from fastapi import FastAPI, Depends, HTTPException, status
+from fastapi.middleware.cors import CORSMiddleware
+import logging
+from typing import List
+
+from src.api.database import get_db
+from src.api.auth import get_current_user
+from src.api.routers import threats_router, indicators_router, auth_router, admin_router
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    handlers=[
+        logging.StreamHandler(),
+        logging.FileHandler("app.log")
+    ]
+)
+logger = logging.getLogger(__name__)
+
+# Create FastAPI app
+app = FastAPI(
+    title="CyberForge OSINT API",
+    description="API for Dark Web OSINT platform",
+    version="1.0.0"
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Update for production
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Include routers for different endpoints
+app.include_router(auth_router.router, prefix="/api/v1")
+app.include_router(
+    threats_router.router, 
+    prefix="/api/v1/threats",
+    tags=["threats"],
+    dependencies=[Depends(get_current_user)]
+)
+app.include_router(
+    indicators_router.router, 
+    prefix="/api/v1/indicators",
+    tags=["indicators"],
+    dependencies=[Depends(get_current_user)]
+)
+app.include_router(
+    admin_router.router, 
+    prefix="/api/v1/admin",
+    tags=["admin"],
+    dependencies=[Depends(get_current_user)]
+)
+
+@app.get("/api/health")
+async def health_check():
+    """Health check endpoint for monitoring."""
+    return {"status": "healthy", "version": "1.0.0"}
+
+@app.on_event("startup")
+async def startup_event():
+    """Event handler for application startup."""
+    logger.info("Starting the CyberForge OSINT API")
+    # Add any startup tasks here (database connection, cache warming, etc.)
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    """Event handler for application shutdown."""
+    logger.info("Shutting down the CyberForge OSINT API")
+    # Add any cleanup tasks here (close connections, save state, etc.)

src/api/routers/auth_router.py

@@ -0,0 +1,74 @@
+"""
+Authentication router.
+
+This module provides authentication endpoints for the API.
+"""
+from datetime import datetime, timedelta
+from typing import Any, Dict, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.api.database import get_db
+from src.api.security import (
+    ACCESS_TOKEN_EXPIRE_MINUTES,
+    Token,
+    UserInDB,
+    authenticate_user,
+    create_access_token,
+    get_current_active_user,
+)
+
+router = APIRouter(tags=["authentication"])
+
+@router.post("/token", response_model=Token)
+async def login_for_access_token(
+    form_data: OAuth2PasswordRequestForm = Depends(),
+    db: AsyncSession = Depends(get_db)
+) -> Dict[str, Any]:
+    """
+    OAuth2 compatible token login, get an access token for future requests.
+    """
+    user = await authenticate_user(db, form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+        data={"sub": user.username, "scopes": user.scopes},
+        expires_delta=access_token_expires
+    )
+    
+    expires_at = datetime.utcnow() + access_token_expires
+    
+    return {
+        "access_token": access_token,
+        "token_type": "bearer",
+        "expires_at": expires_at
+    }
+
+@router.get("/users/me", response_model=UserInDB)
+async def read_users_me(
+    current_user: UserInDB = Depends(get_current_active_user)
+) -> UserInDB:
+    """
+    Get current user.
+    """
+    return current_user
+
+@router.get("/users/me/scopes")
+async def read_own_scopes(
+    current_user: UserInDB = Depends(get_current_active_user)
+) -> Dict[str, Any]:
+    """
+    Get current user's scopes.
+    """
+    return {
+        "username": current_user.username,
+        "scopes": current_user.scopes
+    }

src/api/routers/scraping_router.py

@@ -0,0 +1,161 @@
+from fastapi import APIRouter, Depends, HTTPException, status, Body, BackgroundTasks
+from sqlalchemy.ext.asyncio import AsyncSession
+from typing import List, Optional, Dict, Any
+import logging
+from datetime import datetime
+
+from src.api.database import get_db
+from src.api.auth import get_current_user
+from src.api.schemas import User, CrawlRequest, CrawlResult
+from src.services.scraper import WebScraper, ScraperError
+from src.services.tor_proxy import TorProxyService, TorProxyError
+
+# Configure logger
+logger = logging.getLogger(__name__)
+
+router = APIRouter(
+    prefix="/scraping",
+    tags=["scraping"],
+    responses={404: {"description": "Not found"}}
+)
+
+# Initialize services
+scraper = WebScraper()
+
+@router.post("/test-tor", response_model=Dict[str, Any])
+async def test_tor_connection(
+    current_user: User = Depends(get_current_user)
+):
+    """
+    Test Tor connection.
+    
+    Args:
+        current_user: Current authenticated user
+        
+    Returns:
+        Dict[str, Any]: Connection status
+    """
+    try:
+        tor_proxy = TorProxyService()
+        is_connected = await tor_proxy.check_connection()
+        
+        return {
+            "status": "success",
+            "is_connected": is_connected,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+    except TorProxyError as e:
+        logger.error(f"Tor proxy error: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Tor proxy error: {str(e)}"
+        )
+    except Exception as e:
+        logger.error(f"Error testing Tor connection: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"An error occurred: {str(e)}"
+        )
+
+@router.post("/scrape", response_model=Dict[str, Any])
+async def scrape_page(
+    url: str,
+    use_tor: bool = Body(False),
+    current_user: User = Depends(get_current_user)
+):
+    """
+    Scrape a single page.
+    
+    Args:
+        url: URL to scrape
+        use_tor: Whether to use Tor proxy
+        current_user: Current authenticated user
+        
+    Returns:
+        Dict[str, Any]: Scraped content
+    """
+    try:
+        result = await scraper.extract_content(url, use_tor=use_tor)
+        
+        return {
+            "status": "success",
+            "data": result,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+    except ScraperError as e:
+        logger.error(f"Scraper error: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Scraper error: {str(e)}"
+        )
+    except Exception as e:
+        logger.error(f"Error scraping page: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"An error occurred: {str(e)}"
+        )
+
+@router.post("/crawl", response_model=Dict[str, Any])
+async def crawl_site(
+    crawl_request: CrawlRequest,
+    background_tasks: BackgroundTasks,
+    current_user: User = Depends(get_current_user)
+):
+    """
+    Crawl a site.
+    
+    Args:
+        crawl_request: Crawl request data
+        background_tasks: Background tasks
+        current_user: Current authenticated user
+        
+    Returns:
+        Dict[str, Any]: Crawl status
+    """
+    # For longer crawls, we add them as background tasks
+    # This prevents timeouts on the API request
+    
+    # Start crawl in background
+    if crawl_request.max_depth > 1 or '.onion' in crawl_request.url:
+        # Add to background tasks
+        background_tasks.add_task(
+            scraper.crawl,
+            crawl_request.url,
+            max_depth=crawl_request.max_depth,
+            max_pages=50,
+            keyword_filter=crawl_request.keywords
+        )
+        
+        return {
+            "status": "started",
+            "message": "Crawl started in background",
+            "timestamp": datetime.utcnow().isoformat()
+        }
+    else:
+        # For simple crawls, we perform them synchronously
+        try:
+            results = await scraper.crawl(
+                crawl_request.url,
+                max_depth=crawl_request.max_depth,
+                max_pages=10,
+                keyword_filter=crawl_request.keywords
+            )
+            
+            return {
+                "status": "completed",
+                "results": results,
+                "count": len(results),
+                "timestamp": datetime.utcnow().isoformat()
+            }
+        except ScraperError as e:
+            logger.error(f"Scraper error: {e}")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail=f"Scraper error: {str(e)}"
+            )
+        except Exception as e:
+            logger.error(f"Error crawling site: {e}")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail=f"An error occurred: {str(e)}"
+            )

src/api/routers/threats_router.py

@@ -0,0 +1,217 @@
+from fastapi import APIRouter, Depends, HTTPException, status, Body, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from typing import List, Optional
+import logging
+
+from src.api.database import get_db
+from src.api.auth import get_current_user
+from src.api.schemas import (
+    Threat, ThreatCreate, ThreatUpdate, ThreatFilter, 
+    PaginationParams, User
+)
+from src.api.services.threat_service import (
+    create_threat, get_threat_by_id, update_threat, 
+    delete_threat, get_threats, get_threat_statistics
+)
+
+# Configure logger
+logger = logging.getLogger(__name__)
+
+router = APIRouter(
+    tags=["threats"],
+    responses={404: {"description": "Not found"}}
+)
+
+@router.post("/", response_model=Threat, status_code=status.HTTP_201_CREATED)
+async def create_threat_endpoint(
+    threat_data: ThreatCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user)
+):
+    """
+    Create a new threat.
+    
+    Args:
+        threat_data: Threat data
+        db: Database session
+        current_user: Current authenticated user
+        
+    Returns:
+        Threat: Created threat
+    """
+    try:
+        threat = await create_threat(db, threat_data)
+        
+        if not threat:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Failed to create threat"
+            )
+            
+        return threat
+    except Exception as e:
+        logger.error(f"Error creating threat: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"An error occurred: {str(e)}"
+        )
+
+@router.get("/{threat_id}", response_model=Threat)
+async def get_threat_endpoint(
+    threat_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user)
+):
+    """
+    Get threat by ID.
+    
+    Args:
+        threat_id: Threat ID
+        db: Database session
+        current_user: Current authenticated user
+        
+    Returns:
+        Threat: Threat data
+    """
+    threat = await get_threat_by_id(db, threat_id)
+    
+    if not threat:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Threat with ID {threat_id} not found"
+        )
+        
+    return threat
+
+@router.put("/{threat_id}", response_model=Threat)
+async def update_threat_endpoint(
+    threat_id: int,
+    threat_data: ThreatUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user)
+):
+    """
+    Update threat.
+    
+    Args:
+        threat_id: Threat ID
+        threat_data: Threat data
+        db: Database session
+        current_user: Current authenticated user
+        
+    Returns:
+        Threat: Updated threat
+    """
+    # Check if threat exists
+    threat = await get_threat_by_id(db, threat_id)
+    
+    if not threat:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Threat with ID {threat_id} not found"
+        )
+    
+    # Update threat
+    updated_threat = await update_threat(db, threat_id, threat_data)
+    
+    if not updated_threat:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Failed to update threat"
+        )
+        
+    return updated_threat
+
+@router.delete("/{threat_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_threat_endpoint(
+    threat_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user)
+):
+    """
+    Delete threat.
+    
+    Args:
+        threat_id: Threat ID
+        db: Database session
+        current_user: Current authenticated user
+    """
+    # Check if threat exists
+    threat = await get_threat_by_id(db, threat_id)
+    
+    if not threat:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Threat with ID {threat_id} not found"
+        )
+    
+    # Delete threat
+    deleted = await delete_threat(db, threat_id)
+    
+    if not deleted:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Failed to delete threat"
+        )
+
+@router.get("/", response_model=List[Threat])
+async def get_threats_endpoint(
+    pagination: PaginationParams = Depends(),
+    severity: Optional[List[str]] = Query(None),
+    status: Optional[List[str]] = Query(None),
+    category: Optional[List[str]] = Query(None),
+    search: Optional[str] = Query(None),
+    from_date: Optional[str] = Query(None),
+    to_date: Optional[str] = Query(None),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user)
+):
+    """
+    Get threats with filtering and pagination.
+    
+    Args:
+        pagination: Pagination parameters
+        severity: Filter by severity
+        status: Filter by status
+        category: Filter by category
+        search: Search in title and description
+        from_date: Filter from date
+        to_date: Filter to date
+        db: Database session
+        current_user: Current authenticated user
+        
+    Returns:
+        List[Threat]: List of threats
+    """
+    # Create filter params
+    filter_params = ThreatFilter(
+        severity=severity,
+        status=status,
+        category=category,
+        search=search,
+        from_date=from_date,
+        to_date=to_date
+    )
+    
+    # Get threats
+    threats, total = await get_threats(db, filter_params, pagination)
+    
+    return threats
+
+@router.get("/statistics", response_model=dict)
+async def get_threat_statistics_endpoint(
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user)
+):
+    """
+    Get threat statistics.
+    
+    Args:
+        db: Database session
+        current_user: Current authenticated user
+        
+    Returns:
+        dict: Threat statistics
+    """
+    statistics = await get_threat_statistics(db)
+    return statistics

src/api/schemas.py

@@ -0,0 +1,310 @@
+"""
+API schemas for data validation and serialization.
+"""
+from pydantic import BaseModel, Field, validator, EmailStr
+from typing import Optional, List, Dict, Any, Union
+from datetime import datetime
+from enum import Enum
+
+# Pagination
+class PaginationParams(BaseModel):
+    """Pagination parameters."""
+    page: int = Field(1, ge=1, description="Page number")
+    size: int = Field(10, ge=1, le=100, description="Items per page")
+
+# User schemas
+class UserBase(BaseModel):
+    """Base user schema."""
+    username: str
+    email: EmailStr
+    full_name: Optional[str] = None
+    is_active: bool = True
+
+class UserCreate(UserBase):
+    """User creation schema."""
+    password: str
+
+class UserUpdate(BaseModel):
+    """User update schema."""
+    username: Optional[str] = None
+    email: Optional[EmailStr] = None
+    full_name: Optional[str] = None
+    is_active: Optional[bool] = None
+    password: Optional[str] = None
+
+class UserResponse(UserBase):
+    """User response schema."""
+    id: int
+    is_superuser: bool = False
+    
+    class Config:
+        orm_mode = True
+
+# Token schemas
+class Token(BaseModel):
+    """Token schema."""
+    access_token: str
+    token_type: str = "bearer"
+
+class TokenPayload(BaseModel):
+    """Token payload schema."""
+    sub: Optional[int] = None
+
+# Dark Web Content schemas
+class DarkWebContentBase(BaseModel):
+    """Base schema for dark web content."""
+    url: str
+    title: Optional[str] = None
+    content: str
+    content_type: str
+    source_name: Optional[str] = None
+    source_type: Optional[str] = None
+    language: Optional[str] = None
+
+class DarkWebContentCreate(DarkWebContentBase):
+    """Schema for creating dark web content."""
+    relevance_score: float = 0.0
+    sentiment_score: float = 0.0
+    entity_data: Optional[str] = None
+
+class DarkWebContentUpdate(BaseModel):
+    """Schema for updating dark web content."""
+    title: Optional[str] = None
+    content_status: Optional[str] = None
+    relevance_score: Optional[float] = None
+    sentiment_score: Optional[float] = None
+    entity_data: Optional[str] = None
+
+class DarkWebContentResponse(DarkWebContentBase):
+    """Schema for dark web content response."""
+    id: int
+    domain: Optional[str] = None
+    content_status: str
+    scraped_at: datetime
+    relevance_score: float
+    sentiment_score: float
+    entity_data: Optional[str] = None
+    
+    class Config:
+        orm_mode = True
+
+# Dark Web Mention schemas
+class DarkWebMentionBase(BaseModel):
+    """Base schema for dark web mention."""
+    content_id: int
+    keyword: str
+    keyword_category: Optional[str] = None
+    context: Optional[str] = None
+    snippet: Optional[str] = None
+    mention_type: Optional[str] = None
+
+class DarkWebMentionCreate(DarkWebMentionBase):
+    """Schema for creating dark web mention."""
+    confidence: float = 0.0
+    is_verified: bool = False
+
+class DarkWebMentionUpdate(BaseModel):
+    """Schema for updating dark web mention."""
+    keyword_category: Optional[str] = None
+    mention_type: Optional[str] = None
+    confidence: Optional[float] = None
+    is_verified: Optional[bool] = None
+
+class DarkWebMentionResponse(DarkWebMentionBase):
+    """Schema for dark web mention response."""
+    id: int
+    confidence: float
+    is_verified: bool
+    created_at: datetime
+    
+    class Config:
+        orm_mode = True
+
+# Threat schemas
+class ThreatBase(BaseModel):
+    """Base schema for threat."""
+    title: str
+    description: str
+    severity: str
+    category: str
+
+class ThreatCreate(ThreatBase):
+    """Schema for creating threat."""
+    status: str = "New"
+    source_url: Optional[str] = None
+    source_name: Optional[str] = None
+    source_type: Optional[str] = None
+    affected_entity: Optional[str] = None
+    affected_entity_type: Optional[str] = None
+    confidence_score: float = 0.0
+    risk_score: float = 0.0
+
+class ThreatUpdate(BaseModel):
+    """Schema for updating threat."""
+    title: Optional[str] = None
+    description: Optional[str] = None
+    severity: Optional[str] = None
+    status: Optional[str] = None
+    category: Optional[str] = None
+    affected_entity: Optional[str] = None
+    affected_entity_type: Optional[str] = None
+    confidence_score: Optional[float] = None
+    risk_score: Optional[float] = None
+
+class ThreatResponse(ThreatBase):
+    """Schema for threat response."""
+    id: int
+    status: str
+    source_url: Optional[str] = None
+    source_name: Optional[str] = None
+    source_type: Optional[str] = None
+    discovered_at: datetime
+    affected_entity: Optional[str] = None
+    affected_entity_type: Optional[str] = None
+    confidence_score: float
+    risk_score: float
+    
+    class Config:
+        orm_mode = True
+
+# Indicator schemas
+class IndicatorBase(BaseModel):
+    """Base schema for indicator."""
+    threat_id: int
+    value: str
+    indicator_type: str
+    description: Optional[str] = None
+
+class IndicatorCreate(IndicatorBase):
+    """Schema for creating indicator."""
+    is_verified: bool = False
+    context: Optional[str] = None
+    source: Optional[str] = None
+    confidence_score: float = 0.0
+
+class IndicatorUpdate(BaseModel):
+    """Schema for updating indicator."""
+    description: Optional[str] = None
+    is_verified: Optional[bool] = None
+    context: Optional[str] = None
+    source: Optional[str] = None
+    confidence_score: Optional[float] = None
+
+class IndicatorResponse(IndicatorBase):
+    """Schema for indicator response."""
+    id: int
+    is_verified: bool
+    context: Optional[str] = None
+    source: Optional[str] = None
+    confidence_score: float
+    first_seen: datetime
+    last_seen: datetime
+    
+    class Config:
+        orm_mode = True
+
+# Alert schemas
+class AlertBase(BaseModel):
+    """Base schema for alert."""
+    title: str
+    description: str
+    severity: str
+    category: str
+
+class AlertCreate(AlertBase):
+    """Schema for creating alert."""
+    source_url: Optional[str] = None
+    threat_id: Optional[int] = None
+    mention_id: Optional[int] = None
+
+class AlertUpdate(BaseModel):
+    """Schema for updating alert."""
+    status: str
+    action_taken: Optional[str] = None
+    assigned_to_id: Optional[int] = None
+    is_read: Optional[bool] = None
+
+class AlertResponse(AlertBase):
+    """Schema for alert response."""
+    id: int
+    status: str
+    generated_at: datetime
+    source_url: Optional[str] = None
+    is_read: bool
+    threat_id: Optional[int] = None
+    mention_id: Optional[int] = None
+    assigned_to_id: Optional[int] = None
+    action_taken: Optional[str] = None
+    resolved_at: Optional[datetime] = None
+    
+    class Config:
+        orm_mode = True
+
+# Report schemas
+class ReportBase(BaseModel):
+    """Base schema for report."""
+    report_id: str
+    title: str
+    summary: str
+    content: str
+    report_type: str
+
+class ReportCreate(ReportBase):
+    """Schema for creating report."""
+    status: str = "Draft"
+    severity: Optional[str] = None
+    publish_date: Optional[datetime] = None
+    time_period_start: Optional[datetime] = None
+    time_period_end: Optional[datetime] = None
+    keywords: Optional[str] = None
+    author_id: int
+    threat_ids: List[int] = []
+
+class ReportUpdate(BaseModel):
+    """Schema for updating report."""
+    title: Optional[str] = None
+    summary: Optional[str] = None
+    content: Optional[str] = None
+    report_type: Optional[str] = None
+    status: Optional[str] = None
+    severity: Optional[str] = None
+    publish_date: Optional[datetime] = None
+    time_period_start: Optional[datetime] = None
+    time_period_end: Optional[datetime] = None
+    keywords: Optional[str] = None
+    threat_ids: Optional[List[int]] = None
+
+class ReportResponse(ReportBase):
+    """Schema for report response."""
+    id: int
+    status: str
+    severity: Optional[str] = None
+    publish_date: Optional[datetime] = None
+    time_period_start: Optional[datetime] = None
+    time_period_end: Optional[datetime] = None
+    keywords: Optional[str] = None
+    author_id: int
+    
+    class Config:
+        orm_mode = True
+
+# Statistics response schemas
+class ThreatStatisticsResponse(BaseModel):
+    """Schema for threat statistics response."""
+    total_count: int
+    severity_counts: Dict[str, int]
+    status_counts: Dict[str, int]
+    category_counts: Dict[str, int]
+    time_series: List[Dict[str, Any]]
+    from_date: str
+    to_date: str
+
+class ContentStatisticsResponse(BaseModel):
+    """Schema for content statistics response."""
+    total_count: int
+    content_type_counts: Dict[str, int]
+    content_status_counts: Dict[str, int]
+    source_counts: Dict[str, int]
+    time_series: List[Dict[str, Any]]
+    from_date: str
+    to_date: str

src/api/security.py

@@ -0,0 +1,382 @@
+"""
+API Security Module
+
+This module provides security features for the API, including:
+1. Authentication using JWT tokens
+2. Rate limiting to prevent abuse
+3. Role-based access control
+4. Request validation
+5. Audit logging
+"""
+import os
+import time
+import logging
+import secrets
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Union, Any, Callable
+
+from fastapi import Depends, HTTPException, Security, status, Request
+from fastapi.security import OAuth2PasswordBearer, APIKeyHeader
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+from pydantic import BaseModel, EmailStr
+
+from src.models.user import User
+from src.api.database import get_db
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+# Security configuration
+SECRET_KEY = os.getenv("JWT_SECRET_KEY", secrets.token_hex(32))
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+API_KEY_NAME = "X-API-Key"
+
+# Set up password hashing
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+# Set up security schemes
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
+
+# User models
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+    expires_at: datetime
+
+class TokenData(BaseModel):
+    username: Optional[str] = None
+    scopes: List[str] = []
+
+class UserInDB(BaseModel):
+    id: int
+    username: str
+    email: EmailStr
+    full_name: Optional[str] = None
+    is_active: bool = True
+    is_superuser: bool = False
+    scopes: List[str] = []
+    
+    class Config:
+        from_attributes = True
+
+# Rate limiting
+class RateLimiter:
+    """Simple in-memory rate limiter"""
+    
+    def __init__(self, rate_limit: int = 100, time_window: int = 60):
+        """
+        Initialize rate limiter.
+        
+        Args:
+            rate_limit: Maximum number of requests per time window
+            time_window: Time window in seconds
+        """
+        self.rate_limit = rate_limit
+        self.time_window = time_window
+        self.requests = {}
+    
+    def is_rate_limited(self, key: str) -> bool:
+        """
+        Check if a key is rate limited.
+        
+        Args:
+            key: Identifier for the client (IP address, API key, etc.)
+            
+        Returns:
+            True if rate limited, False otherwise
+        """
+        current_time = time.time()
+        
+        # Initialize or clean up old requests
+        if key not in self.requests:
+            self.requests[key] = []
+        else:
+            # Remove requests outside the time window
+            self.requests[key] = [t for t in self.requests[key] if t > current_time - self.time_window]
+        
+        # Check if rate limit is exceeded
+        if len(self.requests[key]) >= self.rate_limit:
+            return True
+        
+        # Add the current request
+        self.requests[key].append(current_time)
+        return False
+
+# Create global rate limiter instance
+rate_limiter = RateLimiter()
+
+# Role-based access control
+# Define roles and permissions
+ROLES = {
+    "admin": ["read:all", "write:all", "delete:all"],
+    "analyst": ["read:all", "write:threats", "write:indicators", "write:reports"],
+    "user": ["read:threats", "read:reports", "read:dashboard"],
+    "api": ["read:all", "write:threats", "write:indicators"]
+}
+
+# Security utility functions
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """Verify a password against a hash"""
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password: str) -> str:
+    """Hash a password for storage"""
+    return pwd_context.hash(password)
+
+async def get_user(db: AsyncSession, username: str) -> Optional[UserInDB]:
+    """Get a user from the database by username"""
+    result = await db.execute(select(User).filter(User.username == username))
+    user_db = result.scalars().first()
+    
+    if not user_db:
+        return None
+        
+    # Get user roles and scopes
+    scopes = []
+    if user_db.is_superuser:
+        scopes = ROLES["admin"]
+    else:
+        # In a real application, you would look up user roles in a database
+        # For simplicity, we'll assume non-superusers have the "user" role
+        scopes = ROLES["user"]
+    
+    return UserInDB(
+        id=user_db.id,
+        username=user_db.username,
+        email=user_db.email,
+        full_name=user_db.full_name,
+        is_active=user_db.is_active,
+        is_superuser=user_db.is_superuser,
+        scopes=scopes
+    )
+
+async def authenticate_user(db: AsyncSession, username: str, password: str) -> Optional[UserInDB]:
+    """Authenticate a user with username and password"""
+    user = await get_user(db, username)
+    if not user:
+        return None
+    
+    # Get the user from the database again to get the hashed password
+    result = await db.execute(select(User).filter(User.username == username))
+    user_db = result.scalars().first()
+    
+    if not verify_password(password, user_db.hashed_password):
+        return None
+        
+    return user
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
+    """Create a JWT access token"""
+    to_encode = data.copy()
+    
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+        
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+async def get_api_key_user(
+    api_key: str, 
+    db: AsyncSession
+) -> Optional[UserInDB]:
+    """Get user associated with an API key"""
+    # In a real application, you would look up API keys in a database
+    # For simplicity, we'll use a simple hardcoded mapping
+    # TODO: Replace with database-backed API key storage
+    API_KEYS = {
+        "test-api-key": "api_user",
+        # Add more API keys here
+    }
+    
+    if api_key not in API_KEYS:
+        return None
+        
+    username = API_KEYS[api_key]
+    user = await get_user(db, username)
+    
+    if not user:
+        return None
+        
+    # Override scopes with API role scopes
+    user.scopes = ROLES["api"]
+    
+    return user
+
+# Dependencies for FastAPI
+async def rate_limit(request: Request):
+    """Rate limiting dependency"""
+    # Use API key or IP address as the rate limit key
+    client_key = request.headers.get(API_KEY_NAME) or request.client.host
+    
+    if rate_limiter.is_rate_limited(client_key):
+        logger.warning(f"Rate limit exceeded for {client_key}")
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail="Rate limit exceeded. Please try again later."
+        )
+    
+    return True
+
+async def get_current_user(
+    token: str = Depends(oauth2_scheme),
+    api_key: str = Security(api_key_header),
+    db: AsyncSession = Depends(get_db)
+) -> UserInDB:
+    """
+    Get the current user from either JWT token or API key.
+    
+    This dependency can be used to require authentication for endpoints.
+    """
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    
+    # Check API key first
+    if api_key:
+        user = await get_api_key_user(api_key, db)
+        if user:
+            return user
+    
+    # Then check JWT token
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+            
+        token_data = TokenData(
+            username=username,
+            scopes=payload.get("scopes", [])
+        )
+    except JWTError:
+        raise credentials_exception
+        
+    user = await get_user(db, username=token_data.username)
+    if user is None:
+        raise credentials_exception
+        
+    return user
+
+async def get_current_active_user(
+    current_user: UserInDB = Depends(get_current_user)
+) -> UserInDB:
+    """
+    Get the current active user.
+    
+    This dependency can be used to require an active user for endpoints.
+    """
+    if not current_user.is_active:
+        raise HTTPException(status_code=400, detail="Inactive user")
+        
+    return current_user
+
+def has_scope(required_scopes: List[str]):
+    """
+    Create a dependency that checks if the user has the required scopes.
+    
+    Args:
+        required_scopes: List of required scopes
+        
+    Returns:
+        A dependency function that checks if the user has the required scopes
+    """
+    async def _has_scope(
+        current_user: UserInDB = Depends(get_current_active_user)
+    ) -> UserInDB:
+        for scope in required_scopes:
+            if scope not in current_user.scopes:
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    detail=f"Permission denied. Required scope: {scope}"
+                )
+                
+        return current_user
+        
+    return _has_scope
+
+def admin_only(
+    current_user: UserInDB = Depends(get_current_active_user)
+) -> UserInDB:
+    """
+    Dependency that requires an admin user.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Permission denied. Admin access required."
+        )
+        
+    return current_user
+
+# Audit logging middleware
+async def audit_log_middleware(request: Request, call_next):
+    """
+    Middleware for audit logging.
+    
+    Records details about API requests.
+    """
+    # Get request details
+    method = request.method
+    path = request.url.path
+    client_host = request.client.host
+    user_agent = request.headers.get("User-Agent", "Unknown")
+    
+    # Get user details if available
+    user = getattr(request.state, "user", None)
+    username = user.username if user else "Anonymous"
+    
+    # Log request
+    logger.info(
+        f"API Request: {method} {path} | User: {username} | "
+        f"Client: {client_host} | User-Agent: {user_agent}"
+    )
+    
+    # Process the request
+    start_time = time.time()
+    response = await call_next(request)
+    process_time = time.time() - start_time
+    
+    # Log response
+    logger.info(
+        f"API Response: {method} {path} | Status: {response.status_code} | "
+        f"Time: {process_time:.4f}s | User: {username}"
+    )
+    
+    return response
+
+# API key validation middleware
+def validate_api_key(request: Request):
+    """
+    Middleware function to validate API keys.
+    
+    This can be used as a dependency for FastAPI routes.
+    """
+    api_key = request.headers.get(API_KEY_NAME)
+    if not api_key:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="API key required",
+            headers={"WWW-Authenticate": f"{API_KEY_NAME}"},
+        )
+    
+    # In a real application, you would validate the API key against a database
+    # For simplicity, we'll use a hardcoded list
+    valid_keys = ["test-api-key"]  # Replace with database lookup
+    if api_key not in valid_keys:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid API key",
+            headers={"WWW-Authenticate": f"{API_KEY_NAME}"},
+        )
+    
+    return True

src/api/services/__init__.py

@@ -0,0 +1,3 @@
+"""
+Package initialization for API services.
+"""

src/api/services/alert_service.py

@@ -0,0 +1,316 @@
+"""
+Service for alert operations.
+"""
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+from sqlalchemy import func, or_, and_
+from datetime import datetime
+from typing import List, Optional, Dict, Any, Union
+
+from src.models.alert import Alert, AlertStatus, AlertCategory
+from src.models.threat import ThreatSeverity
+from src.api.schemas import PaginationParams
+
+async def create_alert(
+    db: AsyncSession,
+    title: str,
+    description: str,
+    severity: ThreatSeverity,
+    category: AlertCategory,
+    source_url: Optional[str] = None,
+    threat_id: Optional[int] = None,
+    mention_id: Optional[int] = None,
+) -> Alert:
+    """
+    Create a new alert.
+    
+    Args:
+        db: Database session
+        title: Alert title
+        description: Alert description
+        severity: Alert severity
+        category: Alert category
+        source_url: Source URL for the alert
+        threat_id: ID of related threat
+        mention_id: ID of related dark web mention
+        
+    Returns:
+        Alert: Created alert
+    """
+    db_alert = Alert(
+        title=title,
+        description=description,
+        severity=severity,
+        status=AlertStatus.NEW,
+        category=category,
+        generated_at=datetime.utcnow(),
+        source_url=source_url,
+        is_read=False,
+        threat_id=threat_id,
+        mention_id=mention_id,
+    )
+    
+    db.add(db_alert)
+    await db.commit()
+    await db.refresh(db_alert)
+    
+    return db_alert
+
+async def get_alert_by_id(db: AsyncSession, alert_id: int) -> Optional[Alert]:
+    """
+    Get alert by ID.
+    
+    Args:
+        db: Database session
+        alert_id: Alert ID
+        
+    Returns:
+        Optional[Alert]: Alert or None if not found
+    """
+    result = await db.execute(select(Alert).filter(Alert.id == alert_id))
+    return result.scalars().first()
+
+async def get_alerts(
+    db: AsyncSession,
+    pagination: PaginationParams,
+    severity: Optional[List[ThreatSeverity]] = None,
+    status: Optional[List[AlertStatus]] = None,
+    category: Optional[List[AlertCategory]] = None,
+    is_read: Optional[bool] = None,
+    search_query: Optional[str] = None,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> List[Alert]:
+    """
+    Get alerts with filtering and pagination.
+    
+    Args:
+        db: Database session
+        pagination: Pagination parameters
+        severity: Filter by severity
+        status: Filter by status
+        category: Filter by category
+        is_read: Filter by read status
+        search_query: Search in title and description
+        from_date: Filter by generated_at >= from_date
+        to_date: Filter by generated_at <= to_date
+        
+    Returns:
+        List[Alert]: List of alerts
+    """
+    query = select(Alert)
+    
+    # Apply filters
+    if severity:
+        query = query.filter(Alert.severity.in_(severity))
+    
+    if status:
+        query = query.filter(Alert.status.in_(status))
+    
+    if category:
+        query = query.filter(Alert.category.in_(category))
+    
+    if is_read is not None:
+        query = query.filter(Alert.is_read == is_read)
+    
+    if search_query:
+        search_filter = or_(
+            Alert.title.ilike(f"%{search_query}%"),
+            Alert.description.ilike(f"%{search_query}%")
+        )
+        query = query.filter(search_filter)
+    
+    if from_date:
+        query = query.filter(Alert.generated_at >= from_date)
+    
+    if to_date:
+        query = query.filter(Alert.generated_at <= to_date)
+    
+    # Apply pagination
+    query = query.order_by(Alert.generated_at.desc())
+    query = query.offset((pagination.page - 1) * pagination.size).limit(pagination.size)
+    
+    result = await db.execute(query)
+    return result.scalars().all()
+
+async def count_alerts(
+    db: AsyncSession,
+    severity: Optional[List[ThreatSeverity]] = None,
+    status: Optional[List[AlertStatus]] = None,
+    category: Optional[List[AlertCategory]] = None,
+    is_read: Optional[bool] = None,
+    search_query: Optional[str] = None,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> int:
+    """
+    Count alerts with filtering.
+    
+    Args:
+        db: Database session
+        severity: Filter by severity
+        status: Filter by status
+        category: Filter by category
+        is_read: Filter by read status
+        search_query: Search in title and description
+        from_date: Filter by generated_at >= from_date
+        to_date: Filter by generated_at <= to_date
+        
+    Returns:
+        int: Count of alerts
+    """
+    query = select(func.count(Alert.id))
+    
+    # Apply filters (same as in get_alerts)
+    if severity:
+        query = query.filter(Alert.severity.in_(severity))
+    
+    if status:
+        query = query.filter(Alert.status.in_(status))
+    
+    if category:
+        query = query.filter(Alert.category.in_(category))
+    
+    if is_read is not None:
+        query = query.filter(Alert.is_read == is_read)
+    
+    if search_query:
+        search_filter = or_(
+            Alert.title.ilike(f"%{search_query}%"),
+            Alert.description.ilike(f"%{search_query}%")
+        )
+        query = query.filter(search_filter)
+    
+    if from_date:
+        query = query.filter(Alert.generated_at >= from_date)
+    
+    if to_date:
+        query = query.filter(Alert.generated_at <= to_date)
+    
+    result = await db.execute(query)
+    return result.scalar()
+
+async def update_alert_status(
+    db: AsyncSession,
+    alert_id: int,
+    status: AlertStatus,
+    action_taken: Optional[str] = None,
+) -> Optional[Alert]:
+    """
+    Update alert status.
+    
+    Args:
+        db: Database session
+        alert_id: Alert ID
+        status: New status
+        action_taken: Description of action taken
+        
+    Returns:
+        Optional[Alert]: Updated alert or None if not found
+    """
+    alert = await get_alert_by_id(db, alert_id)
+    if not alert:
+        return None
+    
+    alert.status = status
+    
+    if action_taken:
+        alert.action_taken = action_taken
+    
+    if status == AlertStatus.RESOLVED:
+        alert.resolved_at = datetime.utcnow()
+    
+    alert.updated_at = datetime.utcnow()
+    
+    await db.commit()
+    await db.refresh(alert)
+    
+    return alert
+
+async def mark_alert_as_read(
+    db: AsyncSession,
+    alert_id: int,
+) -> Optional[Alert]:
+    """
+    Mark alert as read.
+    
+    Args:
+        db: Database session
+        alert_id: Alert ID
+        
+    Returns:
+        Optional[Alert]: Updated alert or None if not found
+    """
+    alert = await get_alert_by_id(db, alert_id)
+    if not alert:
+        return None
+    
+    alert.is_read = True
+    alert.updated_at = datetime.utcnow()
+    
+    await db.commit()
+    await db.refresh(alert)
+    
+    return alert
+
+async def assign_alert(
+    db: AsyncSession,
+    alert_id: int,
+    user_id: int,
+) -> Optional[Alert]:
+    """
+    Assign alert to a user.
+    
+    Args:
+        db: Database session
+        alert_id: Alert ID
+        user_id: User ID to assign to
+        
+    Returns:
+        Optional[Alert]: Updated alert or None if not found
+    """
+    alert = await get_alert_by_id(db, alert_id)
+    if not alert:
+        return None
+    
+    alert.assigned_to_id = user_id
+    alert.status = AlertStatus.ASSIGNED
+    alert.updated_at = datetime.utcnow()
+    
+    await db.commit()
+    await db.refresh(alert)
+    
+    return alert
+
+async def get_alert_counts_by_severity(
+    db: AsyncSession,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> Dict[str, int]:
+    """
+    Get count of alerts by severity.
+    
+    Args:
+        db: Database session
+        from_date: Filter by generated_at >= from_date
+        to_date: Filter by generated_at <= to_date
+        
+    Returns:
+        Dict[str, int]: Mapping of severity to count
+    """
+    result = {}
+    
+    for severity in ThreatSeverity:
+        query = select(func.count(Alert.id)).filter(Alert.severity == severity)
+        
+        if from_date:
+            query = query.filter(Alert.generated_at >= from_date)
+        
+        if to_date:
+            query = query.filter(Alert.generated_at <= to_date)
+        
+        count_result = await db.execute(query)
+        count = count_result.scalar() or 0
+        result[severity.value] = count
+    
+    return result

src/api/services/dark_web_content_service.py

@@ -0,0 +1,357 @@
+"""
+Service for dark web content operations.
+"""
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+from sqlalchemy import func, or_, text
+from datetime import datetime
+from typing import List, Optional, Dict, Any, Union
+
+from src.models.dark_web_content import DarkWebContent, DarkWebMention, ContentType, ContentStatus
+from src.models.threat import Threat, ThreatCategory, ThreatSeverity, ThreatStatus
+from src.api.schemas import PaginationParams
+
+async def create_content(
+    db: AsyncSession,
+    url: str,
+    content: str,
+    title: Optional[str] = None,
+    content_type: ContentType = ContentType.OTHER,
+    content_status: ContentStatus = ContentStatus.NEW,
+    source_name: Optional[str] = None,
+    source_type: Optional[str] = None,
+    language: Optional[str] = None,
+    relevance_score: float = 0.0,
+    sentiment_score: float = 0.0,
+    entity_data: Optional[str] = None,
+) -> DarkWebContent:
+    """
+    Create a new dark web content entry.
+    
+    Args:
+        db: Database session
+        url: URL of the content
+        content: Text content
+        title: Title of the content
+        content_type: Type of content
+        content_status: Status of content
+        source_name: Name of the source
+        source_type: Type of source
+        language: Language of the content
+        relevance_score: Relevance score (0-1)
+        sentiment_score: Sentiment score (-1 to 1)
+        entity_data: JSON string of extracted entities
+        
+    Returns:
+        DarkWebContent: Created content
+    """
+    # Extract domain from URL if possible
+    domain = None
+    if url:
+        try:
+            from urllib.parse import urlparse
+            parsed_url = urlparse(url)
+            domain = parsed_url.netloc
+        except:
+            pass
+    
+    db_content = DarkWebContent(
+        url=url,
+        domain=domain,
+        title=title,
+        content=content,
+        content_type=content_type,
+        content_status=content_status,
+        source_name=source_name,
+        source_type=source_type,
+        language=language,
+        scraped_at=datetime.utcnow(),
+        relevance_score=relevance_score,
+        sentiment_score=sentiment_score,
+        entity_data=entity_data,
+    )
+    
+    db.add(db_content)
+    await db.commit()
+    await db.refresh(db_content)
+    
+    return db_content
+
+async def get_content_by_id(db: AsyncSession, content_id: int) -> Optional[DarkWebContent]:
+    """
+    Get dark web content by ID.
+    
+    Args:
+        db: Database session
+        content_id: Content ID
+        
+    Returns:
+        Optional[DarkWebContent]: Content or None if not found
+    """
+    result = await db.execute(select(DarkWebContent).filter(DarkWebContent.id == content_id))
+    return result.scalars().first()
+
+async def get_contents(
+    db: AsyncSession,
+    pagination: PaginationParams,
+    content_type: Optional[List[ContentType]] = None,
+    content_status: Optional[List[ContentStatus]] = None,
+    source_name: Optional[str] = None,
+    search_query: Optional[str] = None,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> List[DarkWebContent]:
+    """
+    Get dark web contents with filtering and pagination.
+    
+    Args:
+        db: Database session
+        pagination: Pagination parameters
+        content_type: Filter by content type
+        content_status: Filter by content status
+        source_name: Filter by source name
+        search_query: Search in title and content
+        from_date: Filter by scraped_at >= from_date
+        to_date: Filter by scraped_at <= to_date
+        
+    Returns:
+        List[DarkWebContent]: List of dark web contents
+    """
+    query = select(DarkWebContent)
+    
+    # Apply filters
+    if content_type:
+        query = query.filter(DarkWebContent.content_type.in_(content_type))
+    
+    if content_status:
+        query = query.filter(DarkWebContent.content_status.in_(content_status))
+    
+    if source_name:
+        query = query.filter(DarkWebContent.source_name == source_name)
+    
+    if search_query:
+        search_filter = or_(
+            DarkWebContent.title.ilike(f"%{search_query}%"),
+            DarkWebContent.content.ilike(f"%{search_query}%")
+        )
+        query = query.filter(search_filter)
+    
+    if from_date:
+        query = query.filter(DarkWebContent.scraped_at >= from_date)
+    
+    if to_date:
+        query = query.filter(DarkWebContent.scraped_at <= to_date)
+    
+    # Apply pagination
+    query = query.order_by(DarkWebContent.scraped_at.desc())
+    query = query.offset((pagination.page - 1) * pagination.size).limit(pagination.size)
+    
+    result = await db.execute(query)
+    return result.scalars().all()
+
+async def count_contents(
+    db: AsyncSession,
+    content_type: Optional[List[ContentType]] = None,
+    content_status: Optional[List[ContentStatus]] = None,
+    source_name: Optional[str] = None,
+    search_query: Optional[str] = None,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> int:
+    """
+    Count dark web contents with filtering.
+    
+    Args:
+        db: Database session
+        content_type: Filter by content type
+        content_status: Filter by content status
+        source_name: Filter by source name
+        search_query: Search in title and content
+        from_date: Filter by scraped_at >= from_date
+        to_date: Filter by scraped_at <= to_date
+        
+    Returns:
+        int: Count of dark web contents
+    """
+    query = select(func.count(DarkWebContent.id))
+    
+    # Apply filters (same as in get_contents)
+    if content_type:
+        query = query.filter(DarkWebContent.content_type.in_(content_type))
+    
+    if content_status:
+        query = query.filter(DarkWebContent.content_status.in_(content_status))
+    
+    if source_name:
+        query = query.filter(DarkWebContent.source_name == source_name)
+    
+    if search_query:
+        search_filter = or_(
+            DarkWebContent.title.ilike(f"%{search_query}%"),
+            DarkWebContent.content.ilike(f"%{search_query}%")
+        )
+        query = query.filter(search_filter)
+    
+    if from_date:
+        query = query.filter(DarkWebContent.scraped_at >= from_date)
+    
+    if to_date:
+        query = query.filter(DarkWebContent.scraped_at <= to_date)
+    
+    result = await db.execute(query)
+    return result.scalar()
+
+async def create_mention(
+    db: AsyncSession,
+    content_id: int,
+    keyword: str,
+    keyword_category: Optional[str] = None,
+    context: Optional[str] = None,
+    snippet: Optional[str] = None,
+    mention_type: Optional[str] = None,
+    confidence: float = 0.0,
+    is_verified: bool = False,
+) -> DarkWebMention:
+    """
+    Create a new dark web mention.
+    
+    Args:
+        db: Database session
+        content_id: ID of the content where the mention was found
+        keyword: Keyword that was mentioned
+        keyword_category: Category of the keyword
+        context: Text surrounding the mention
+        snippet: Extract of text containing the mention
+        mention_type: Type of mention
+        confidence: Confidence score (0-1)
+        is_verified: Whether the mention is verified
+        
+    Returns:
+        DarkWebMention: Created mention
+    """
+    db_mention = DarkWebMention(
+        content_id=content_id,
+        keyword=keyword,
+        keyword_category=keyword_category,
+        context=context,
+        snippet=snippet,
+        mention_type=mention_type,
+        confidence=confidence,
+        is_verified=is_verified,
+    )
+    
+    db.add(db_mention)
+    await db.commit()
+    await db.refresh(db_mention)
+    
+    return db_mention
+
+async def get_mention_by_id(db: AsyncSession, mention_id: int) -> Optional[DarkWebMention]:
+    """
+    Get dark web mention by ID.
+    
+    Args:
+        db: Database session
+        mention_id: Mention ID
+        
+    Returns:
+        Optional[DarkWebMention]: Mention or None if not found
+    """
+    result = await db.execute(select(DarkWebMention).filter(DarkWebMention.id == mention_id))
+    return result.scalars().first()
+
+async def get_mentions(
+    db: AsyncSession,
+    pagination: PaginationParams,
+    keyword: Optional[str] = None,
+    content_id: Optional[int] = None,
+    is_verified: Optional[bool] = None,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> List[DarkWebMention]:
+    """
+    Get dark web mentions with filtering and pagination.
+    
+    Args:
+        db: Database session
+        pagination: Pagination parameters
+        keyword: Filter by keyword
+        content_id: Filter by content ID
+        is_verified: Filter by verification status
+        from_date: Filter by created_at >= from_date
+        to_date: Filter by created_at <= to_date
+        
+    Returns:
+        List[DarkWebMention]: List of dark web mentions
+    """
+    query = select(DarkWebMention)
+    
+    # Apply filters
+    if keyword:
+        query = query.filter(DarkWebMention.keyword.ilike(f"%{keyword}%"))
+    
+    if content_id:
+        query = query.filter(DarkWebMention.content_id == content_id)
+    
+    if is_verified is not None:
+        query = query.filter(DarkWebMention.is_verified == is_verified)
+    
+    if from_date:
+        query = query.filter(DarkWebMention.created_at >= from_date)
+    
+    if to_date:
+        query = query.filter(DarkWebMention.created_at <= to_date)
+    
+    # Apply pagination
+    query = query.order_by(DarkWebMention.created_at.desc())
+    query = query.offset((pagination.page - 1) * pagination.size).limit(pagination.size)
+    
+    result = await db.execute(query)
+    return result.scalars().all()
+
+async def create_threat_from_content(
+    db: AsyncSession,
+    content_id: int,
+    title: str,
+    description: str,
+    severity: ThreatSeverity,
+    category: ThreatCategory,
+    confidence_score: float = 0.0,
+) -> Threat:
+    """
+    Create a threat from dark web content.
+    
+    Args:
+        db: Database session
+        content_id: ID of the content
+        title: Threat title
+        description: Threat description
+        severity: Threat severity
+        category: Threat category
+        confidence_score: Confidence score (0-1)
+        
+    Returns:
+        Threat: Created threat
+    """
+    # Get the content
+    content = await get_content_by_id(db, content_id)
+    if not content:
+        raise ValueError(f"Content with ID {content_id} not found")
+    
+    # Create the threat
+    from src.api.services.threat_service import create_threat
+    
+    threat = await create_threat(
+        db=db,
+        title=title,
+        description=description,
+        severity=severity,
+        category=category,
+        status=ThreatStatus.NEW,
+        source_url=content.url,
+        source_name=content.source_name,
+        source_type=content.source_type,
+        confidence_score=confidence_score,
+    )
+    
+    return threat

src/api/services/report_service.py

@@ -0,0 +1,436 @@
+"""
+Service for working with intelligence reports.
+"""
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+from sqlalchemy import update, delete, func, desc, and_, or_
+from typing import List, Optional, Dict, Any, Union
+import logging
+from datetime import datetime
+
+from src.models.report import Report, ReportType, ReportStatus
+from src.models.threat import ThreatSeverity
+from src.api.schemas import PaginationParams
+
+# Configure logger
+logger = logging.getLogger(__name__)
+
+
+async def create_report(
+    db: AsyncSession,
+    title: str,
+    summary: str,
+    content: str,
+    report_type: ReportType,
+    report_id: str,
+    status: ReportStatus = ReportStatus.DRAFT,
+    severity: Optional[ThreatSeverity] = None,
+    publish_date: Optional[datetime] = None,
+    time_period_start: Optional[datetime] = None,
+    time_period_end: Optional[datetime] = None,
+    keywords: Optional[List[str]] = None,
+    source_data: Optional[Dict[str, Any]] = None,
+    author_id: Optional[int] = None,
+) -> Report:
+    """
+    Create a new intelligence report.
+    
+    Args:
+        db: Database session
+        title: Report title
+        summary: Report summary
+        content: Report content
+        report_type: Type of report
+        report_id: Custom ID for the report
+        status: Report status
+        severity: Report severity
+        publish_date: Publication date
+        time_period_start: Start of time period covered
+        time_period_end: End of time period covered
+        keywords: List of keywords related to the report
+        source_data: Sources and references
+        author_id: ID of the report author
+        
+    Returns:
+        Report: Created report
+    """
+    report = Report(
+        title=title,
+        summary=summary,
+        content=content,
+        report_type=report_type,
+        report_id=report_id,
+        status=status,
+        severity=severity,
+        publish_date=publish_date,
+        time_period_start=time_period_start,
+        time_period_end=time_period_end,
+        keywords=keywords or [],
+        source_data=source_data or {},
+        author_id=author_id,
+    )
+    
+    db.add(report)
+    await db.commit()
+    await db.refresh(report)
+    
+    return report
+
+
+async def get_report_by_id(db: AsyncSession, report_id: int) -> Optional[Report]:
+    """
+    Get report by ID.
+    
+    Args:
+        db: Database session
+        report_id: Report ID
+        
+    Returns:
+        Optional[Report]: Found report or None
+    """
+    result = await db.execute(
+        select(Report).where(Report.id == report_id)
+    )
+    return result.scalars().first()
+
+
+async def get_report_by_custom_id(db: AsyncSession, custom_id: str) -> Optional[Report]:
+    """
+    Get report by custom ID.
+    
+    Args:
+        db: Database session
+        custom_id: Custom report ID (e.g., RPT-2023-0001)
+        
+    Returns:
+        Optional[Report]: Found report or None
+    """
+    result = await db.execute(
+        select(Report).where(Report.report_id == custom_id)
+    )
+    return result.scalars().first()
+
+
+async def get_reports(
+    db: AsyncSession,
+    pagination: PaginationParams,
+    report_type: Optional[List[ReportType]] = None,
+    status: Optional[List[ReportStatus]] = None,
+    severity: Optional[List[ThreatSeverity]] = None,
+    search_query: Optional[str] = None,
+    keywords: Optional[List[str]] = None,
+    author_id: Optional[int] = None,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> List[Report]:
+    """
+    Get reports with filtering.
+    
+    Args:
+        db: Database session
+        pagination: Pagination parameters
+        report_type: Filter by report type
+        status: Filter by status
+        severity: Filter by severity
+        search_query: Search in title and summary
+        keywords: Filter by keywords
+        author_id: Filter by author ID
+        from_date: Filter by created_at >= from_date
+        to_date: Filter by created_at <= to_date
+        
+    Returns:
+        List[Report]: List of reports
+    """
+    query = select(Report)
+    
+    # Apply filters
+    if report_type:
+        query = query.where(Report.report_type.in_(report_type))
+    
+    if status:
+        query = query.where(Report.status.in_(status))
+    
+    if severity:
+        query = query.where(Report.severity.in_(severity))
+    
+    if search_query:
+        search_filter = or_(
+            Report.title.ilike(f"%{search_query}%"),
+            Report.summary.ilike(f"%{search_query}%"),
+            Report.content.ilike(f"%{search_query}%"),
+        )
+        query = query.where(search_filter)
+    
+    if keywords:
+        # For JSON arrays, need to use a more complex query
+        for keyword in keywords:
+            query = query.where(Report.keywords.contains([keyword]))
+    
+    if author_id:
+        query = query.where(Report.author_id == author_id)
+    
+    if from_date:
+        query = query.where(Report.created_at >= from_date)
+    
+    if to_date:
+        query = query.where(Report.created_at <= to_date)
+    
+    # Apply pagination
+    query = query.order_by(desc(Report.created_at))
+    query = query.offset((pagination.page - 1) * pagination.size).limit(pagination.size)
+    
+    result = await db.execute(query)
+    return result.scalars().all()
+
+
+async def count_reports(
+    db: AsyncSession,
+    report_type: Optional[List[ReportType]] = None,
+    status: Optional[List[ReportStatus]] = None,
+    severity: Optional[List[ThreatSeverity]] = None,
+    search_query: Optional[str] = None,
+    keywords: Optional[List[str]] = None,
+    author_id: Optional[int] = None,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> int:
+    """
+    Count reports with filtering.
+    
+    Args are the same as get_reports, except pagination.
+    
+    Returns:
+        int: Count of matching reports
+    """
+    query = select(func.count(Report.id))
+    
+    # Apply filters
+    if report_type:
+        query = query.where(Report.report_type.in_(report_type))
+    
+    if status:
+        query = query.where(Report.status.in_(status))
+    
+    if severity:
+        query = query.where(Report.severity.in_(severity))
+    
+    if search_query:
+        search_filter = or_(
+            Report.title.ilike(f"%{search_query}%"),
+            Report.summary.ilike(f"%{search_query}%"),
+            Report.content.ilike(f"%{search_query}%"),
+        )
+        query = query.where(search_filter)
+    
+    if keywords:
+        # For JSON arrays, need to use a more complex query
+        for keyword in keywords:
+            query = query.where(Report.keywords.contains([keyword]))
+    
+    if author_id:
+        query = query.where(Report.author_id == author_id)
+    
+    if from_date:
+        query = query.where(Report.created_at >= from_date)
+    
+    if to_date:
+        query = query.where(Report.created_at <= to_date)
+    
+    result = await db.execute(query)
+    return result.scalar()
+
+
+async def update_report(
+    db: AsyncSession,
+    report_id: int,
+    title: Optional[str] = None,
+    summary: Optional[str] = None,
+    content: Optional[str] = None,
+    report_type: Optional[ReportType] = None,
+    status: Optional[ReportStatus] = None,
+    severity: Optional[ThreatSeverity] = None,
+    publish_date: Optional[datetime] = None,
+    time_period_start: Optional[datetime] = None,
+    time_period_end: Optional[datetime] = None,
+    keywords: Optional[List[str]] = None,
+    source_data: Optional[Dict[str, Any]] = None,
+) -> Optional[Report]:
+    """
+    Update a report.
+    
+    Args:
+        db: Database session
+        report_id: Report ID
+        Other args: Fields to update
+        
+    Returns:
+        Optional[Report]: Updated report or None
+    """
+    report = await get_report_by_id(db, report_id)
+    
+    if not report:
+        return None
+    
+    # Update fields if provided
+    if title is not None:
+        report.title = title
+    
+    if summary is not None:
+        report.summary = summary
+    
+    if content is not None:
+        report.content = content
+    
+    if report_type is not None:
+        report.report_type = report_type
+    
+    if status is not None:
+        report.status = status
+    
+    if severity is not None:
+        report.severity = severity
+    
+    if publish_date is not None:
+        report.publish_date = publish_date
+    
+    if time_period_start is not None:
+        report.time_period_start = time_period_start
+    
+    if time_period_end is not None:
+        report.time_period_end = time_period_end
+    
+    if keywords is not None:
+        report.keywords = keywords
+    
+    if source_data is not None:
+        report.source_data = source_data
+    
+    await db.commit()
+    await db.refresh(report)
+    
+    return report
+
+
+async def add_threat_to_report(
+    db: AsyncSession,
+    report_id: int,
+    threat_id: int,
+) -> Optional[Report]:
+    """
+    Add a threat to a report.
+    
+    Args:
+        db: Database session
+        report_id: Report ID
+        threat_id: Threat ID
+        
+    Returns:
+        Optional[Report]: Updated report or None
+    """
+    from src.api.services.threat_service import get_threat_by_id
+    
+    # Get report and threat
+    report = await get_report_by_id(db, report_id)
+    threat = await get_threat_by_id(db, threat_id)
+    
+    if not report or not threat:
+        return None
+    
+    # Add threat to report
+    report.threats.append(threat)
+    await db.commit()
+    await db.refresh(report)
+    
+    return report
+
+
+async def remove_threat_from_report(
+    db: AsyncSession,
+    report_id: int,
+    threat_id: int,
+) -> Optional[Report]:
+    """
+    Remove a threat from a report.
+    
+    Args:
+        db: Database session
+        report_id: Report ID
+        threat_id: Threat ID
+        
+    Returns:
+        Optional[Report]: Updated report or None
+    """
+    from src.api.services.threat_service import get_threat_by_id
+    
+    # Get report and threat
+    report = await get_report_by_id(db, report_id)
+    threat = await get_threat_by_id(db, threat_id)
+    
+    if not report or not threat:
+        return None
+    
+    # Remove threat from report
+    if threat in report.threats:
+        report.threats.remove(threat)
+        await db.commit()
+        await db.refresh(report)
+    
+    return report
+
+
+async def publish_report(
+    db: AsyncSession,
+    report_id: int,
+) -> Optional[Report]:
+    """
+    Publish a report.
+    
+    Args:
+        db: Database session
+        report_id: Report ID
+        
+    Returns:
+        Optional[Report]: Updated report or None
+    """
+    report = await get_report_by_id(db, report_id)
+    
+    if not report:
+        return None
+    
+    # Update status and publish date
+    report.status = ReportStatus.PUBLISHED
+    
+    if not report.publish_date:
+        report.publish_date = datetime.utcnow()
+    
+    await db.commit()
+    await db.refresh(report)
+    
+    return report
+
+
+async def archive_report(
+    db: AsyncSession,
+    report_id: int,
+) -> Optional[Report]:
+    """
+    Archive a report.
+    
+    Args:
+        db: Database session
+        report_id: Report ID
+        
+    Returns:
+        Optional[Report]: Updated report or None
+    """
+    report = await get_report_by_id(db, report_id)
+    
+    if not report:
+        return None
+    
+    # Update status
+    report.status = ReportStatus.ARCHIVED
+    await db.commit()
+    await db.refresh(report)
+    
+    return report

src/api/services/search_history_service.py

@@ -0,0 +1,609 @@
+"""
+Search History and Trends Service
+
+This service manages search history, saved searches, and trend analysis.
+"""
+import logging
+import json
+from datetime import datetime, timedelta
+from typing import List, Dict, Any, Optional, Tuple, Union
+
+from sqlalchemy import func, desc, and_, or_, text
+from sqlalchemy.future import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from src.models.search_history import SearchHistory, SearchResult, SavedSearch, TrendTopic
+from src.models.dark_web_content import DarkWebContent
+from src.models.user import User
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+async def add_search_history(
+    db: AsyncSession,
+    query: str,
+    user_id: Optional[int] = None,
+    result_count: int = 0,
+    category: Optional[str] = None,
+    is_saved: bool = False,
+    notes: Optional[str] = None,
+    tags: Optional[str] = None
+) -> SearchHistory:
+    """
+    Add a new search history entry.
+    
+    Args:
+        db: Database session
+        query: Search query
+        user_id: ID of the user who performed the search (optional)
+        result_count: Number of results returned
+        category: Category of the search
+        is_saved: Whether this is a saved search
+        notes: Optional notes
+        tags: Optional tags (comma-separated)
+        
+    Returns:
+        The created SearchHistory object
+    """
+    search_history = SearchHistory(
+        query=query,
+        user_id=user_id,
+        result_count=result_count,
+        category=category,
+        is_saved=is_saved,
+        notes=notes,
+        tags=tags
+    )
+    
+    db.add(search_history)
+    await db.commit()
+    await db.refresh(search_history)
+    
+    # Update trend data
+    await update_trend_data(db, query, category)
+    
+    return search_history
+
+async def add_search_result(
+    db: AsyncSession,
+    search_id: int,
+    url: str,
+    title: Optional[str] = None,
+    snippet: Optional[str] = None,
+    source: Optional[str] = None,
+    relevance_score: float = 0.0,
+    content_id: Optional[int] = None
+) -> SearchResult:
+    """
+    Add a new search result.
+    
+    Args:
+        db: Database session
+        search_id: ID of the parent search
+        url: URL of the result
+        title: Title of the result
+        snippet: Text snippet from the result
+        source: Source of the result
+        relevance_score: Score indicating relevance to the search query
+        content_id: ID of the content in our database (if applicable)
+        
+    Returns:
+        The created SearchResult object
+    """
+    search_result = SearchResult(
+        search_id=search_id,
+        url=url,
+        title=title,
+        snippet=snippet,
+        source=source,
+        relevance_score=relevance_score,
+        content_id=content_id
+    )
+    
+    db.add(search_result)
+    await db.commit()
+    await db.refresh(search_result)
+    
+    return search_result
+
+async def get_search_history(
+    db: AsyncSession,
+    skip: int = 0,
+    limit: int = 100,
+    user_id: Optional[int] = None,
+    query_filter: Optional[str] = None,
+    date_from: Optional[datetime] = None,
+    date_to: Optional[datetime] = None,
+    category: Optional[str] = None,
+    is_saved: Optional[bool] = None,
+    include_results: bool = False
+) -> List[SearchHistory]:
+    """
+    Get search history with filtering options.
+    
+    Args:
+        db: Database session
+        skip: Number of items to skip
+        limit: Maximum number of items to return
+        user_id: Filter by user ID
+        query_filter: Filter by search query (partial match)
+        date_from: Filter by timestamp (from)
+        date_to: Filter by timestamp (to)
+        category: Filter by category
+        is_saved: Filter by saved status
+        include_results: Whether to include search results
+        
+    Returns:
+        List of SearchHistory objects
+    """
+    statement = select(SearchHistory)
+    
+    # Apply filters
+    if user_id is not None:
+        statement = statement.where(SearchHistory.user_id == user_id)
+    
+    if query_filter:
+        statement = statement.where(SearchHistory.query.ilike(f"%{query_filter}%"))
+    
+    if date_from:
+        statement = statement.where(SearchHistory.timestamp >= date_from)
+    
+    if date_to:
+        statement = statement.where(SearchHistory.timestamp <= date_to)
+    
+    if category:
+        statement = statement.where(SearchHistory.category == category)
+    
+    if is_saved is not None:
+        statement = statement.where(SearchHistory.is_saved == is_saved)
+    
+    # Load related data if requested
+    if include_results:
+        statement = statement.options(selectinload(SearchHistory.search_results))
+    
+    # Apply pagination
+    statement = statement.order_by(desc(SearchHistory.timestamp)).offset(skip).limit(limit)
+    
+    result = await db.execute(statement)
+    return result.scalars().all()
+
+async def get_search_by_id(
+    db: AsyncSession,
+    search_id: int,
+    include_results: bool = False
+) -> Optional[SearchHistory]:
+    """
+    Get a search history entry by ID.
+    
+    Args:
+        db: Database session
+        search_id: Search history ID
+        include_results: Whether to include search results
+        
+    Returns:
+        SearchHistory object or None if not found
+    """
+    statement = select(SearchHistory).where(SearchHistory.id == search_id)
+    
+    if include_results:
+        statement = statement.options(selectinload(SearchHistory.search_results))
+    
+    result = await db.execute(statement)
+    return result.scalars().first()
+
+async def delete_search_history(db: AsyncSession, search_id: int) -> bool:
+    """
+    Delete a search history entry.
+    
+    Args:
+        db: Database session
+        search_id: ID of the search to delete
+        
+    Returns:
+        True if successful, False otherwise
+    """
+    search = await get_search_by_id(db, search_id)
+    if not search:
+        return False
+    
+    await db.delete(search)
+    await db.commit()
+    return True
+
+async def save_search(
+    db: AsyncSession,
+    search_id: int,
+    is_saved: bool = True,
+    notes: Optional[str] = None,
+    tags: Optional[str] = None
+) -> Optional[SearchHistory]:
+    """
+    Save or unsave a search history entry.
+    
+    Args:
+        db: Database session
+        search_id: ID of the search
+        is_saved: Whether to save or unsave
+        notes: Optional notes to add
+        tags: Optional tags to add (comma-separated)
+        
+    Returns:
+        Updated SearchHistory object or None if not found
+    """
+    search = await get_search_by_id(db, search_id)
+    if not search:
+        return None
+    
+    search.is_saved = is_saved
+    
+    if notes:
+        search.notes = notes
+    
+    if tags:
+        search.tags = tags
+    
+    await db.commit()
+    await db.refresh(search)
+    return search
+
+async def create_saved_search(
+    db: AsyncSession,
+    name: str,
+    query: str,
+    user_id: int,
+    frequency: int = 24,
+    notification_enabled: bool = True,
+    threshold: int = 1,
+    category: Optional[str] = None
+) -> SavedSearch:
+    """
+    Create a new saved search with periodic monitoring.
+    
+    Args:
+        db: Database session
+        name: Name of the saved search
+        query: Search query
+        user_id: ID of the user
+        frequency: How often to run this search (in hours, 0 for manual only)
+        notification_enabled: Whether to send notifications for new results
+        threshold: Minimum number of new results for notification
+        category: Category of the search
+        
+    Returns:
+        The created SavedSearch object
+    """
+    saved_search = SavedSearch(
+        name=name,
+        query=query,
+        user_id=user_id,
+        frequency=frequency,
+        notification_enabled=notification_enabled,
+        threshold=threshold,
+        category=category
+    )
+    
+    db.add(saved_search)
+    await db.commit()
+    await db.refresh(saved_search)
+    
+    return saved_search
+
+async def get_saved_searches(
+    db: AsyncSession,
+    user_id: Optional[int] = None,
+    is_active: Optional[bool] = None,
+    skip: int = 0,
+    limit: int = 100
+) -> List[SavedSearch]:
+    """
+    Get saved searches with filtering options.
+    
+    Args:
+        db: Database session
+        user_id: Filter by user ID
+        is_active: Filter by active status
+        skip: Number of items to skip
+        limit: Maximum number of items to return
+        
+    Returns:
+        List of SavedSearch objects
+    """
+    statement = select(SavedSearch)
+    
+    # Apply filters
+    if user_id is not None:
+        statement = statement.where(SavedSearch.user_id == user_id)
+    
+    if is_active is not None:
+        statement = statement.where(SavedSearch.is_active == is_active)
+    
+    # Apply pagination
+    statement = statement.order_by(SavedSearch.name).offset(skip).limit(limit)
+    
+    result = await db.execute(statement)
+    return result.scalars().all()
+
+async def update_trend_data(
+    db: AsyncSession,
+    query: str,
+    category: Optional[str] = None
+) -> None:
+    """
+    Update trend data based on search queries.
+    
+    Args:
+        db: Database session
+        query: Search query
+        category: Category of the search
+    """
+    # Split query into individual terms/topics
+    topics = [t.strip() for t in query.split() if len(t.strip()) > 3]
+    
+    # Process each topic
+    for topic in topics:
+        # Check if topic already exists
+        statement = select(TrendTopic).where(TrendTopic.topic == topic)
+        result = await db.execute(statement)
+        trend_topic = result.scalars().first()
+        
+        if trend_topic:
+            # Update existing topic
+            trend_topic.last_seen = datetime.utcnow()
+            trend_topic.mention_count += 1
+            
+            # Calculate growth rate (percentage change over the last 24 hours)
+            time_diff = (trend_topic.last_seen - trend_topic.first_seen).total_seconds() / 3600  # hours
+            if time_diff > 0:
+                hourly_rate = trend_topic.mention_count / time_diff
+                trend_topic.growth_rate = hourly_rate * 24  # daily growth rate
+            
+            # Update category if provided and not already set
+            if category and not trend_topic.category:
+                trend_topic.category = category
+        else:
+            # Create a new trend topic
+            trend_topic = TrendTopic(
+                topic=topic,
+                category=category,
+                mention_count=1,
+                growth_rate=1.0  # Initial growth rate
+            )
+            db.add(trend_topic)
+    
+    await db.commit()
+
+async def get_trending_topics(
+    db: AsyncSession,
+    days: int = 7,
+    limit: int = 20,
+    category: Optional[str] = None,
+    min_mentions: int = 3
+) -> List[TrendTopic]:
+    """
+    Get trending topics over a specific time period.
+    
+    Args:
+        db: Database session
+        days: Number of days to consider
+        limit: Maximum number of topics to return
+        category: Filter by category
+        min_mentions: Minimum number of mentions
+        
+    Returns:
+        List of TrendTopic objects sorted by growth rate
+    """
+    cutoff_date = datetime.utcnow() - timedelta(days=days)
+    
+    statement = select(TrendTopic).where(
+        and_(
+            TrendTopic.last_seen >= cutoff_date,
+            TrendTopic.mention_count >= min_mentions,
+            TrendTopic.is_active == True
+        )
+    )
+    
+    if category:
+        statement = statement.where(TrendTopic.category == category)
+    
+    statement = statement.order_by(desc(TrendTopic.growth_rate)).limit(limit)
+    
+    result = await db.execute(statement)
+    return result.scalars().all()
+
+async def get_search_frequency(
+    db: AsyncSession,
+    days: int = 30,
+    interval: str = 'day'
+) -> List[Dict[str, Any]]:
+    """
+    Get search frequency over time for visualization.
+    
+    Args:
+        db: Database session
+        days: Number of days to analyze
+        interval: Time interval ('hour', 'day', 'week', 'month')
+        
+    Returns:
+        List of dictionaries with time intervals and search counts
+    """
+    cutoff_date = datetime.utcnow() - timedelta(days=days)
+    
+    # SQL query depends on the interval
+    if interval == 'hour':
+        date_format = "YYYY-MM-DD HH24:00"
+        trunc_expr = func.date_trunc('hour', SearchHistory.timestamp)
+    elif interval == 'day':
+        date_format = "YYYY-MM-DD"
+        trunc_expr = func.date_trunc('day', SearchHistory.timestamp)
+    elif interval == 'week':
+        date_format = "YYYY-WW"
+        trunc_expr = func.date_trunc('week', SearchHistory.timestamp)
+    else:  # month
+        date_format = "YYYY-MM"
+        trunc_expr = func.date_trunc('month', SearchHistory.timestamp)
+    
+    # Query for search count by interval
+    statement = select(
+        trunc_expr.label('interval'),
+        func.count(SearchHistory.id).label('count')
+    ).where(
+        SearchHistory.timestamp >= cutoff_date
+    ).group_by(
+        'interval'
+    ).order_by(
+        'interval'
+    )
+    
+    result = await db.execute(statement)
+    rows = result.all()
+    
+    # Convert to list of dictionaries
+    return [{"interval": row.interval, "count": row.count} for row in rows]
+
+async def get_popular_searches(
+    db: AsyncSession,
+    days: int = 30,
+    limit: int = 10
+) -> List[Dict[str, Any]]:
+    """
+    Get the most popular search terms.
+    
+    Args:
+        db: Database session
+        days: Number of days to analyze
+        limit: Maximum number of terms to return
+        
+    Returns:
+        List of dictionaries with search queries and counts
+    """
+    cutoff_date = datetime.utcnow() - timedelta(days=days)
+    
+    statement = select(
+        SearchHistory.query,
+        func.count(SearchHistory.id).label('count')
+    ).where(
+        SearchHistory.timestamp >= cutoff_date
+    ).group_by(
+        SearchHistory.query
+    ).order_by(
+        desc('count')
+    ).limit(limit)
+    
+    result = await db.execute(statement)
+    rows = result.all()
+    
+    return [{"query": row.query, "count": row.count} for row in rows]
+
+async def get_search_categories(
+    db: AsyncSession,
+    days: int = 30
+) -> List[Dict[str, Any]]:
+    """
+    Get distribution of search categories.
+    
+    Args:
+        db: Database session
+        days: Number of days to analyze
+        
+    Returns:
+        List of dictionaries with categories and counts
+    """
+    cutoff_date = datetime.utcnow() - timedelta(days=days)
+    
+    statement = select(
+        SearchHistory.category,
+        func.count(SearchHistory.id).label('count')
+    ).where(
+        and_(
+            SearchHistory.timestamp >= cutoff_date,
+            SearchHistory.category.is_not(None)
+        )
+    ).group_by(
+        SearchHistory.category
+    ).order_by(
+        desc('count')
+    )
+    
+    result = await db.execute(statement)
+    rows = result.all()
+    
+    return [{"category": row.category or "Uncategorized", "count": row.count} for row in rows]
+
+async def get_search_trend_analysis(
+    db: AsyncSession,
+    days: int = 90,
+    trend_days: int = 7,
+    limit: int = 10
+) -> Dict[str, Any]:
+    """
+    Get comprehensive analysis of search trends.
+    
+    Args:
+        db: Database session
+        days: Total days to analyze
+        trend_days: Days to calculate short-term trends
+        limit: Maximum number of items in each category
+        
+    Returns:
+        Dictionary with various trend analyses
+    """
+    # Get overall search frequency
+    frequency = await get_search_frequency(db, days, 'day')
+    
+    # Get popular searches
+    popular = await get_popular_searches(db, days, limit)
+    
+    # Get recent trending topics
+    trending = await get_trending_topics(db, trend_days, limit)
+    
+    # Get category distribution
+    categories = await get_search_categories(db, days)
+    
+    # Get recent (last 24 hours) vs. overall popular terms
+    recent_popular = await get_popular_searches(db, 1, limit)
+    
+    # Calculate velocity (rate of change)
+    # This compares the last 7 days to the previous 7 days
+    cutoff_recent = datetime.utcnow() - timedelta(days=trend_days)
+    cutoff_previous = cutoff_recent - timedelta(days=trend_days)
+    
+    # Query for velocity calculation
+    statement_recent = select(func.count(SearchHistory.id)).where(
+        SearchHistory.timestamp >= cutoff_recent
+    )
+    statement_previous = select(func.count(SearchHistory.id)).where(
+        and_(
+            SearchHistory.timestamp >= cutoff_previous,
+            SearchHistory.timestamp < cutoff_recent
+        )
+    )
+    
+    result_recent = await db.execute(statement_recent)
+    result_previous = await db.execute(statement_previous)
+    
+    count_recent = result_recent.scalar() or 0
+    count_previous = result_previous.scalar() or 0
+    
+    if count_previous > 0:
+        velocity = (count_recent - count_previous) / count_previous * 100  # percentage change
+    else:
+        velocity = 100.0 if count_recent > 0 else 0.0
+    
+    # Compile the results
+    return {
+        "frequency": frequency,
+        "popular_searches": popular,
+        "trending_topics": [
+            {"topic": t.topic, "mentions": t.mention_count, "growth_rate": t.growth_rate} 
+            for t in trending
+        ],
+        "categories": categories,
+        "recent_popular": recent_popular,
+        "velocity": velocity,
+        "total_searches": {
+            "total": count_recent + count_previous,
+            "recent": count_recent,
+            "previous": count_previous
+        }
+    }

src/api/services/subscription_service.py

@@ -0,0 +1,681 @@
+"""
+Subscription service.
+
+This module provides functions for managing subscriptions.
+"""
+import os
+import logging
+from datetime import datetime, timedelta
+from typing import List, Dict, Any, Optional, Tuple, Union
+
+import stripe
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, update, delete
+from sqlalchemy.orm import joinedload
+
+from src.models.subscription import (
+    SubscriptionPlan, UserSubscription, PaymentHistory,
+    SubscriptionTier, BillingPeriod, SubscriptionStatus, PaymentStatus
+)
+from src.models.user import User
+
+# Set up Stripe API key
+stripe.api_key = os.environ.get("STRIPE_SECRET_KEY")
+STRIPE_PUBLISHABLE_KEY = os.environ.get("STRIPE_PUBLISHABLE_KEY")
+
+# Set up logging
+logger = logging.getLogger(__name__)
+
+
+async def get_subscription_plans(
+    db: AsyncSession,
+    active_only: bool = True
+) -> List[SubscriptionPlan]:
+    """
+    Get all subscription plans.
+    
+    Args:
+        db: Database session
+        active_only: If True, only return active plans
+        
+    Returns:
+        List of subscription plans
+    """
+    query = select(SubscriptionPlan)
+    
+    if active_only:
+        query = query.where(SubscriptionPlan.is_active == True)
+    
+    result = await db.execute(query)
+    plans = result.scalars().all()
+    
+    return plans
+
+
+async def get_subscription_plan_by_id(
+    db: AsyncSession,
+    plan_id: int
+) -> Optional[SubscriptionPlan]:
+    """
+    Get a subscription plan by ID.
+    
+    Args:
+        db: Database session
+        plan_id: ID of the plan to get
+        
+    Returns:
+        Subscription plan or None if not found
+    """
+    query = select(SubscriptionPlan).where(SubscriptionPlan.id == plan_id)
+    result = await db.execute(query)
+    plan = result.scalars().first()
+    
+    return plan
+
+
+async def get_subscription_plan_by_tier(
+    db: AsyncSession,
+    tier: SubscriptionTier
+) -> Optional[SubscriptionPlan]:
+    """
+    Get a subscription plan by tier.
+    
+    Args:
+        db: Database session
+        tier: Tier of the plan to get
+        
+    Returns:
+        Subscription plan or None if not found
+    """
+    query = select(SubscriptionPlan).where(SubscriptionPlan.tier == tier)
+    result = await db.execute(query)
+    plan = result.scalars().first()
+    
+    return plan
+
+
+async def create_subscription_plan(
+    db: AsyncSession,
+    name: str,
+    tier: SubscriptionTier,
+    description: str,
+    price_monthly: float,
+    price_annually: float,
+    max_alerts: int = 10,
+    max_reports: int = 5,
+    max_searches_per_day: int = 20,
+    max_monitoring_keywords: int = 10,
+    max_data_retention_days: int = 30,
+    supports_api_access: bool = False,
+    supports_live_feed: bool = False,
+    supports_dark_web_monitoring: bool = False,
+    supports_export: bool = False,
+    supports_advanced_analytics: bool = False,
+    create_stripe_product: bool = True
+) -> Optional[SubscriptionPlan]:
+    """
+    Create a new subscription plan.
+    
+    Args:
+        db: Database session
+        name: Name of the plan
+        tier: Tier of the plan
+        description: Description of the plan
+        price_monthly: Monthly price of the plan
+        price_annually: Annual price of the plan
+        max_alerts: Maximum number of alerts allowed
+        max_reports: Maximum number of reports allowed
+        max_searches_per_day: Maximum number of searches per day
+        max_monitoring_keywords: Maximum number of monitoring keywords
+        max_data_retention_days: Maximum number of days to retain data
+        supports_api_access: Whether the plan supports API access
+        supports_live_feed: Whether the plan supports live feed
+        supports_dark_web_monitoring: Whether the plan supports dark web monitoring
+        supports_export: Whether the plan supports data export
+        supports_advanced_analytics: Whether the plan supports advanced analytics
+        create_stripe_product: Whether to create a Stripe product for this plan
+        
+    Returns:
+        Created subscription plan or None if creation failed
+    """
+    # Check if plan with the same tier already exists
+    existing_plan = await get_subscription_plan_by_tier(db, tier)
+    
+    if existing_plan:
+        logger.warning(f"Subscription plan with tier {tier} already exists")
+        return None
+    
+    # Create Stripe product if requested
+    stripe_product_id = None
+    stripe_monthly_price_id = None
+    stripe_annual_price_id = None
+    
+    if create_stripe_product and stripe.api_key:
+        try:
+            # Create Stripe product
+            product = stripe.Product.create(
+                name=name,
+                description=description,
+                metadata={
+                    "tier": tier.value,
+                    "max_alerts": max_alerts,
+                    "max_reports": max_reports,
+                    "max_searches_per_day": max_searches_per_day,
+                    "max_monitoring_keywords": max_monitoring_keywords,
+                    "max_data_retention_days": max_data_retention_days,
+                    "supports_api_access": "yes" if supports_api_access else "no",
+                    "supports_live_feed": "yes" if supports_live_feed else "no",
+                    "supports_dark_web_monitoring": "yes" if supports_dark_web_monitoring else "no",
+                    "supports_export": "yes" if supports_export else "no",
+                    "supports_advanced_analytics": "yes" if supports_advanced_analytics else "no"
+                }
+            )
+            
+            stripe_product_id = product.id
+            
+            # Create monthly price
+            monthly_price = stripe.Price.create(
+                product=product.id,
+                unit_amount=int(price_monthly * 100),  # Stripe uses cents
+                currency="usd",
+                recurring={"interval": "month"},
+                metadata={"billing_period": "monthly"}
+            )
+            
+            stripe_monthly_price_id = monthly_price.id
+            
+            # Create annual price
+            annual_price = stripe.Price.create(
+                product=product.id,
+                unit_amount=int(price_annually * 100),  # Stripe uses cents
+                currency="usd",
+                recurring={"interval": "year"},
+                metadata={"billing_period": "annually"}
+            )
+            
+            stripe_annual_price_id = annual_price.id
+            
+            logger.info(f"Created Stripe product {product.id} for plan {name}")
+        except Exception as e:
+            logger.error(f"Failed to create Stripe product for plan {name}: {e}")
+    
+    # Create plan in database
+    plan = SubscriptionPlan(
+        name=name,
+        tier=tier,
+        description=description,
+        price_monthly=price_monthly,
+        price_annually=price_annually,
+        max_alerts=max_alerts,
+        max_reports=max_reports,
+        max_searches_per_day=max_searches_per_day,
+        max_monitoring_keywords=max_monitoring_keywords,
+        max_data_retention_days=max_data_retention_days,
+        supports_api_access=supports_api_access,
+        supports_live_feed=supports_live_feed,
+        supports_dark_web_monitoring=supports_dark_web_monitoring,
+        supports_export=supports_export,
+        supports_advanced_analytics=supports_advanced_analytics,
+        stripe_product_id=stripe_product_id,
+        stripe_monthly_price_id=stripe_monthly_price_id,
+        stripe_annual_price_id=stripe_annual_price_id
+    )
+    
+    db.add(plan)
+    await db.commit()
+    await db.refresh(plan)
+    
+    return plan
+
+
+async def update_subscription_plan(
+    db: AsyncSession,
+    plan_id: int,
+    name: Optional[str] = None,
+    description: Optional[str] = None,
+    price_monthly: Optional[float] = None,
+    price_annually: Optional[float] = None,
+    is_active: Optional[bool] = None,
+    max_alerts: Optional[int] = None,
+    max_reports: Optional[int] = None,
+    max_searches_per_day: Optional[int] = None,
+    max_monitoring_keywords: Optional[int] = None,
+    max_data_retention_days: Optional[int] = None,
+    supports_api_access: Optional[bool] = None,
+    supports_live_feed: Optional[bool] = None,
+    supports_dark_web_monitoring: Optional[bool] = None,
+    supports_export: Optional[bool] = None,
+    supports_advanced_analytics: Optional[bool] = None,
+    update_stripe_product: bool = True
+) -> Optional[SubscriptionPlan]:
+    """
+    Update a subscription plan.
+    
+    Args:
+        db: Database session
+        plan_id: ID of the plan to update
+        name: New name of the plan
+        description: New description of the plan
+        price_monthly: New monthly price of the plan
+        price_annually: New annual price of the plan
+        is_active: New active status of the plan
+        max_alerts: New maximum number of alerts allowed
+        max_reports: New maximum number of reports allowed
+        max_searches_per_day: New maximum number of searches per day
+        max_monitoring_keywords: New maximum number of monitoring keywords
+        max_data_retention_days: New maximum number of days to retain data
+        supports_api_access: New API access support status
+        supports_live_feed: New live feed support status
+        supports_dark_web_monitoring: New dark web monitoring support status
+        supports_export: New data export support status
+        supports_advanced_analytics: New advanced analytics support status
+        update_stripe_product: Whether to update the Stripe product for this plan
+        
+    Returns:
+        Updated subscription plan or None if update failed
+    """
+    # Get existing plan
+    plan = await get_subscription_plan_by_id(db, plan_id)
+    
+    if not plan:
+        logger.warning(f"Subscription plan with ID {plan_id} not found")
+        return None
+    
+    # Prepare update data
+    update_data = {}
+    
+    if name is not None:
+        update_data["name"] = name
+    
+    if description is not None:
+        update_data["description"] = description
+    
+    if price_monthly is not None:
+        update_data["price_monthly"] = price_monthly
+    
+    if price_annually is not None:
+        update_data["price_annually"] = price_annually
+    
+    if is_active is not None:
+        update_data["is_active"] = is_active
+    
+    if max_alerts is not None:
+        update_data["max_alerts"] = max_alerts
+    
+    if max_reports is not None:
+        update_data["max_reports"] = max_reports
+    
+    if max_searches_per_day is not None:
+        update_data["max_searches_per_day"] = max_searches_per_day
+    
+    if max_monitoring_keywords is not None:
+        update_data["max_monitoring_keywords"] = max_monitoring_keywords
+    
+    if max_data_retention_days is not None:
+        update_data["max_data_retention_days"] = max_data_retention_days
+    
+    if supports_api_access is not None:
+        update_data["supports_api_access"] = supports_api_access
+    
+    if supports_live_feed is not None:
+        update_data["supports_live_feed"] = supports_live_feed
+    
+    if supports_dark_web_monitoring is not None:
+        update_data["supports_dark_web_monitoring"] = supports_dark_web_monitoring
+    
+    if supports_export is not None:
+        update_data["supports_export"] = supports_export
+    
+    if supports_advanced_analytics is not None:
+        update_data["supports_advanced_analytics"] = supports_advanced_analytics
+    
+    # Update Stripe product if requested
+    if update_stripe_product and plan.stripe_product_id and stripe.api_key:
+        try:
+            # Update Stripe product
+            product_update_data = {}
+            
+            if name is not None:
+                product_update_data["name"] = name
+            
+            if description is not None:
+                product_update_data["description"] = description
+            
+            metadata_update = {}
+            
+            if max_alerts is not None:
+                metadata_update["max_alerts"] = max_alerts
+            
+            if max_reports is not None:
+                metadata_update["max_reports"] = max_reports
+            
+            if max_searches_per_day is not None:
+                metadata_update["max_searches_per_day"] = max_searches_per_day
+            
+            if max_monitoring_keywords is not None:
+                metadata_update["max_monitoring_keywords"] = max_monitoring_keywords
+            
+            if max_data_retention_days is not None:
+                metadata_update["max_data_retention_days"] = max_data_retention_days
+            
+            if supports_api_access is not None:
+                metadata_update["supports_api_access"] = "yes" if supports_api_access else "no"
+            
+            if supports_live_feed is not None:
+                metadata_update["supports_live_feed"] = "yes" if supports_live_feed else "no"
+            
+            if supports_dark_web_monitoring is not None:
+                metadata_update["supports_dark_web_monitoring"] = "yes" if supports_dark_web_monitoring else "no"
+            
+            if supports_export is not None:
+                metadata_update["supports_export"] = "yes" if supports_export else "no"
+            
+            if supports_advanced_analytics is not None:
+                metadata_update["supports_advanced_analytics"] = "yes" if supports_advanced_analytics else "no"
+            
+            if metadata_update:
+                product_update_data["metadata"] = metadata_update
+            
+            if product_update_data:
+                stripe.Product.modify(plan.stripe_product_id, **product_update_data)
+            
+            # Update prices if needed
+            if price_monthly is not None and plan.stripe_monthly_price_id:
+                # Can't update existing price in Stripe, create a new one
+                new_monthly_price = stripe.Price.create(
+                    product=plan.stripe_product_id,
+                    unit_amount=int(price_monthly * 100),  # Stripe uses cents
+                    currency="usd",
+                    recurring={"interval": "month"},
+                    metadata={"billing_period": "monthly"}
+                )
+                
+                update_data["stripe_monthly_price_id"] = new_monthly_price.id
+            
+            if price_annually is not None and plan.stripe_annual_price_id:
+                # Can't update existing price in Stripe, create a new one
+                new_annual_price = stripe.Price.create(
+                    product=plan.stripe_product_id,
+                    unit_amount=int(price_annually * 100),  # Stripe uses cents
+                    currency="usd",
+                    recurring={"interval": "year"},
+                    metadata={"billing_period": "annually"}
+                )
+                
+                update_data["stripe_annual_price_id"] = new_annual_price.id
+            
+            logger.info(f"Updated Stripe product {plan.stripe_product_id} for plan {plan.name}")
+        except Exception as e:
+            logger.error(f"Failed to update Stripe product for plan {plan.name}: {e}")
+    
+    # Update plan in database
+    if update_data:
+        await db.execute(
+            update(SubscriptionPlan)
+            .where(SubscriptionPlan.id == plan_id)
+            .values(**update_data)
+        )
+        
+        await db.commit()
+        
+        # Refresh plan
+        plan = await get_subscription_plan_by_id(db, plan_id)
+    
+    return plan
+
+
+async def get_user_subscription(
+    db: AsyncSession,
+    user_id: int
+) -> Optional[UserSubscription]:
+    """
+    Get a user's active subscription.
+    
+    Args:
+        db: Database session
+        user_id: ID of the user
+        
+    Returns:
+        User subscription or None if not found
+    """
+    query = (
+        select(UserSubscription)
+        .where(UserSubscription.user_id == user_id)
+        .where(UserSubscription.status != SubscriptionStatus.CANCELED)
+        .options(joinedload(UserSubscription.plan))
+    )
+    
+    result = await db.execute(query)
+    subscription = result.scalars().first()
+    
+    return subscription
+
+
+async def get_user_subscription_by_id(
+    db: AsyncSession,
+    subscription_id: int
+) -> Optional[UserSubscription]:
+    """
+    Get a user subscription by ID.
+    
+    Args:
+        db: Database session
+        subscription_id: ID of the subscription
+        
+    Returns:
+        User subscription or None if not found
+    """
+    query = (
+        select(UserSubscription)
+        .where(UserSubscription.id == subscription_id)
+        .options(joinedload(UserSubscription.plan))
+    )
+    
+    result = await db.execute(query)
+    subscription = result.scalars().first()
+    
+    return subscription
+
+
+async def create_user_subscription(
+    db: AsyncSession,
+    user_id: int,
+    plan_id: int,
+    billing_period: BillingPeriod = BillingPeriod.MONTHLY,
+    create_stripe_subscription: bool = True,
+    payment_method_id: Optional[str] = None
+) -> Optional[UserSubscription]:
+    """
+    Create a new user subscription.
+    
+    Args:
+        db: Database session
+        user_id: ID of the user
+        plan_id: ID of the subscription plan
+        billing_period: Billing period (monthly or annually)
+        create_stripe_subscription: Whether to create a Stripe subscription
+        payment_method_id: ID of the payment method to use (required if create_stripe_subscription is True)
+        
+    Returns:
+        Created user subscription or None if creation failed
+    """
+    # Check if user exists
+    query = select(User).where(User.id == user_id)
+    result = await db.execute(query)
+    user = result.scalars().first()
+    
+    if not user:
+        logger.warning(f"User with ID {user_id} not found")
+        return None
+    
+    # Check if plan exists
+    plan = await get_subscription_plan_by_id(db, plan_id)
+    
+    if not plan:
+        logger.warning(f"Subscription plan with ID {plan_id} not found")
+        return None
+    
+    # Check if user already has an active subscription
+    existing_subscription = await get_user_subscription(db, user_id)
+    
+    if existing_subscription:
+        logger.warning(f"User with ID {user_id} already has an active subscription")
+        return None
+    
+    # Calculate subscription period
+    now = datetime.utcnow()
+    
+    if billing_period == BillingPeriod.MONTHLY:
+        current_period_end = now + timedelta(days=30)
+        price = plan.price_monthly
+        stripe_price_id = plan.stripe_monthly_price_id
+    elif billing_period == BillingPeriod.ANNUALLY:
+        current_period_end = now + timedelta(days=365)
+        price = plan.price_annually
+        stripe_price_id = plan.stripe_annual_price_id
+    else:
+        logger.warning(f"Invalid billing period: {billing_period}")
+        return None
+    
+    # Create Stripe subscription if requested
+    stripe_subscription_id = None
+    stripe_customer_id = None
+    
+    if create_stripe_subscription and stripe.api_key and plan.stripe_product_id:
+        if not payment_method_id:
+            logger.warning("Payment method ID is required to create a Stripe subscription")
+            return None
+        
+        try:
+            # Create or retrieve Stripe customer
+            customers = stripe.Customer.list(email=user.email)
+            
+            if customers.data:
+                customer = customers.data[0]
+                stripe_customer_id = customer.id
+            else:
+                customer = stripe.Customer.create(
+                    email=user.email,
+                    name=user.full_name,
+                    metadata={"user_id": user_id}
+                )
+                
+                stripe_customer_id = customer.id
+            
+            # Attach payment method to customer
+            stripe.PaymentMethod.attach(
+                payment_method_id,
+                customer=stripe_customer_id
+            )
+            
+            # Set as default payment method
+            stripe.Customer.modify(
+                stripe_customer_id,
+                invoice_settings={
+                    "default_payment_method": payment_method_id
+                }
+            )
+            
+            # Create subscription
+            subscription = stripe.Subscription.create(
+                customer=stripe_customer_id,
+                items=[
+                    {"price": stripe_price_id}
+                ],
+                expand=["latest_invoice.payment_intent"]
+            )
+            
+            stripe_subscription_id = subscription.id
+            
+            logger.info(f"Created Stripe subscription {subscription.id} for user {user_id}")
+        except Exception as e:
+            logger.error(f"Failed to create Stripe subscription for user {user_id}: {e}")
+            return None
+    
+    # Create subscription in database
+    subscription = UserSubscription(
+        user_id=user_id,
+        plan_id=plan_id,
+        status=SubscriptionStatus.ACTIVE,
+        billing_period=billing_period,
+        current_period_start=now,
+        current_period_end=current_period_end,
+        stripe_subscription_id=stripe_subscription_id,
+        stripe_customer_id=stripe_customer_id
+    )
+    
+    db.add(subscription)
+    await db.commit()
+    await db.refresh(subscription)
+    
+    # Record payment
+    if subscription.id:
+        payment_status = PaymentStatus.SUCCEEDED if stripe_subscription_id else PaymentStatus.PENDING
+        
+        payment = PaymentHistory(
+            user_id=user_id,
+            subscription_id=subscription.id,
+            amount=price,
+            currency="USD",
+            status=payment_status
+        )
+        
+        db.add(payment)
+        await db.commit()
+    
+    return subscription
+
+
+async def cancel_user_subscription(
+    db: AsyncSession,
+    subscription_id: int,
+    cancel_stripe_subscription: bool = True
+) -> Optional[UserSubscription]:
+    """
+    Cancel a user subscription.
+    
+    Args:
+        db: Database session
+        subscription_id: ID of the subscription to cancel
+        cancel_stripe_subscription: Whether to cancel the Stripe subscription
+        
+    Returns:
+        Canceled user subscription or None if cancellation failed
+    """
+    # Get subscription
+    subscription = await get_user_subscription_by_id(db, subscription_id)
+    
+    if not subscription:
+        logger.warning(f"Subscription with ID {subscription_id} not found")
+        return None
+    
+    # Cancel Stripe subscription if requested
+    if cancel_stripe_subscription and subscription.stripe_subscription_id and stripe.api_key:
+        try:
+            stripe.Subscription.modify(
+                subscription.stripe_subscription_id,
+                cancel_at_period_end=True
+            )
+            
+            logger.info(f"Canceled Stripe subscription {subscription.stripe_subscription_id} at period end")
+        except Exception as e:
+            logger.error(f"Failed to cancel Stripe subscription {subscription.stripe_subscription_id}: {e}")
+    
+    # Update subscription in database
+    now = datetime.utcnow()
+    
+    await db.execute(
+        update(UserSubscription)
+        .where(UserSubscription.id == subscription_id)
+        .values(
+            status=SubscriptionStatus.CANCELED,
+            canceled_at=now
+        )
+    )
+    
+    await db.commit()
+    
+    # Refresh subscription
+    subscription = await get_user_subscription_by_id(db, subscription_id)
+    
+    return subscription

src/api/services/threat_service.py

@@ -0,0 +1,411 @@
+"""
+Service for threat operations.
+"""
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+from sqlalchemy import func, or_, and_
+from datetime import datetime, timedelta
+from typing import List, Optional, Dict, Any, Union
+
+from src.models.threat import Threat, ThreatSeverity, ThreatStatus, ThreatCategory
+from src.models.indicator import Indicator, IndicatorType
+from src.api.schemas import PaginationParams
+
+async def create_threat(
+    db: AsyncSession,
+    title: str,
+    description: str,
+    severity: ThreatSeverity,
+    category: ThreatCategory,
+    status: ThreatStatus = ThreatStatus.NEW,
+    source_url: Optional[str] = None,
+    source_name: Optional[str] = None,
+    source_type: Optional[str] = None,
+    affected_entity: Optional[str] = None,
+    affected_entity_type: Optional[str] = None,
+    confidence_score: float = 0.0,
+    risk_score: float = 0.0,
+) -> Threat:
+    """
+    Create a new threat.
+    
+    Args:
+        db: Database session
+        title: Threat title
+        description: Threat description
+        severity: Threat severity
+        category: Threat category
+        status: Threat status
+        source_url: URL of the source
+        source_name: Name of the source
+        source_type: Type of source
+        affected_entity: Name of affected entity
+        affected_entity_type: Type of affected entity
+        confidence_score: Confidence score (0-1)
+        risk_score: Risk score (0-1)
+        
+    Returns:
+        Threat: Created threat
+    """
+    db_threat = Threat(
+        title=title,
+        description=description,
+        severity=severity,
+        category=category,
+        status=status,
+        source_url=source_url,
+        source_name=source_name,
+        source_type=source_type,
+        discovered_at=datetime.utcnow(),
+        affected_entity=affected_entity,
+        affected_entity_type=affected_entity_type,
+        confidence_score=confidence_score,
+        risk_score=risk_score,
+    )
+    
+    db.add(db_threat)
+    await db.commit()
+    await db.refresh(db_threat)
+    
+    return db_threat
+
+async def get_threat_by_id(db: AsyncSession, threat_id: int) -> Optional[Threat]:
+    """
+    Get threat by ID.
+    
+    Args:
+        db: Database session
+        threat_id: Threat ID
+        
+    Returns:
+        Optional[Threat]: Threat or None if not found
+    """
+    result = await db.execute(select(Threat).filter(Threat.id == threat_id))
+    return result.scalars().first()
+
+async def get_threats(
+    db: AsyncSession,
+    pagination: PaginationParams,
+    severity: Optional[List[ThreatSeverity]] = None,
+    status: Optional[List[ThreatStatus]] = None,
+    category: Optional[List[ThreatCategory]] = None,
+    search_query: Optional[str] = None,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> List[Threat]:
+    """
+    Get threats with filtering and pagination.
+    
+    Args:
+        db: Database session
+        pagination: Pagination parameters
+        severity: Filter by severity
+        status: Filter by status
+        category: Filter by category
+        search_query: Search in title and description
+        from_date: Filter by discovered_at >= from_date
+        to_date: Filter by discovered_at <= to_date
+        
+    Returns:
+        List[Threat]: List of threats
+    """
+    query = select(Threat)
+    
+    # Apply filters
+    if severity:
+        query = query.filter(Threat.severity.in_(severity))
+    
+    if status:
+        query = query.filter(Threat.status.in_(status))
+    
+    if category:
+        query = query.filter(Threat.category.in_(category))
+    
+    if search_query:
+        search_filter = or_(
+            Threat.title.ilike(f"%{search_query}%"),
+            Threat.description.ilike(f"%{search_query}%")
+        )
+        query = query.filter(search_filter)
+    
+    if from_date:
+        query = query.filter(Threat.discovered_at >= from_date)
+    
+    if to_date:
+        query = query.filter(Threat.discovered_at <= to_date)
+    
+    # Apply pagination
+    query = query.order_by(Threat.discovered_at.desc())
+    query = query.offset((pagination.page - 1) * pagination.size).limit(pagination.size)
+    
+    result = await db.execute(query)
+    return result.scalars().all()
+
+async def count_threats(
+    db: AsyncSession,
+    severity: Optional[List[ThreatSeverity]] = None,
+    status: Optional[List[ThreatStatus]] = None,
+    category: Optional[List[ThreatCategory]] = None,
+    search_query: Optional[str] = None,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> int:
+    """
+    Count threats with filtering.
+    
+    Args:
+        db: Database session
+        severity: Filter by severity
+        status: Filter by status
+        category: Filter by category
+        search_query: Search in title and description
+        from_date: Filter by discovered_at >= from_date
+        to_date: Filter by discovered_at <= to_date
+        
+    Returns:
+        int: Count of threats
+    """
+    query = select(func.count(Threat.id))
+    
+    # Apply filters (same as in get_threats)
+    if severity:
+        query = query.filter(Threat.severity.in_(severity))
+    
+    if status:
+        query = query.filter(Threat.status.in_(status))
+    
+    if category:
+        query = query.filter(Threat.category.in_(category))
+    
+    if search_query:
+        search_filter = or_(
+            Threat.title.ilike(f"%{search_query}%"),
+            Threat.description.ilike(f"%{search_query}%")
+        )
+        query = query.filter(search_filter)
+    
+    if from_date:
+        query = query.filter(Threat.discovered_at >= from_date)
+    
+    if to_date:
+        query = query.filter(Threat.discovered_at <= to_date)
+    
+    result = await db.execute(query)
+    return result.scalar()
+
+async def update_threat(
+    db: AsyncSession,
+    threat_id: int,
+    title: Optional[str] = None,
+    description: Optional[str] = None,
+    severity: Optional[ThreatSeverity] = None,
+    status: Optional[ThreatStatus] = None,
+    category: Optional[ThreatCategory] = None,
+    affected_entity: Optional[str] = None,
+    affected_entity_type: Optional[str] = None,
+    confidence_score: Optional[float] = None,
+    risk_score: Optional[float] = None,
+) -> Optional[Threat]:
+    """
+    Update threat.
+    
+    Args:
+        db: Database session
+        threat_id: Threat ID
+        title: New title
+        description: New description
+        severity: New severity
+        status: New status
+        category: New category
+        affected_entity: New affected entity
+        affected_entity_type: New affected entity type
+        confidence_score: New confidence score
+        risk_score: New risk score
+        
+    Returns:
+        Optional[Threat]: Updated threat or None if not found
+    """
+    threat = await get_threat_by_id(db, threat_id)
+    if not threat:
+        return None
+    
+    if title is not None:
+        threat.title = title
+    
+    if description is not None:
+        threat.description = description
+    
+    if severity is not None:
+        threat.severity = severity
+    
+    if status is not None:
+        threat.status = status
+    
+    if category is not None:
+        threat.category = category
+    
+    if affected_entity is not None:
+        threat.affected_entity = affected_entity
+    
+    if affected_entity_type is not None:
+        threat.affected_entity_type = affected_entity_type
+    
+    if confidence_score is not None:
+        threat.confidence_score = confidence_score
+    
+    if risk_score is not None:
+        threat.risk_score = risk_score
+    
+    threat.updated_at = datetime.utcnow()
+    
+    await db.commit()
+    await db.refresh(threat)
+    
+    return threat
+
+async def add_indicator_to_threat(
+    db: AsyncSession,
+    threat_id: int,
+    value: str,
+    indicator_type: IndicatorType,
+    description: Optional[str] = None,
+    is_verified: bool = False,
+    context: Optional[str] = None,
+    source: Optional[str] = None,
+    confidence_score: float = 0.0,
+) -> Indicator:
+    """
+    Add an indicator to a threat.
+    
+    Args:
+        db: Database session
+        threat_id: Threat ID
+        value: Indicator value
+        indicator_type: Indicator type
+        description: Description of the indicator
+        is_verified: Whether the indicator is verified
+        context: Context of the indicator
+        source: Source of the indicator
+        confidence_score: Confidence score (0-1)
+        
+    Returns:
+        Indicator: Created indicator
+    """
+    # Check if threat exists
+    threat = await get_threat_by_id(db, threat_id)
+    if not threat:
+        raise ValueError(f"Threat with ID {threat_id} not found")
+    
+    # Create indicator
+    db_indicator = Indicator(
+        threat_id=threat_id,
+        value=value,
+        indicator_type=indicator_type,
+        description=description,
+        is_verified=is_verified,
+        context=context,
+        source=source,
+        confidence_score=confidence_score,
+        first_seen=datetime.utcnow(),
+        last_seen=datetime.utcnow(),
+    )
+    
+    db.add(db_indicator)
+    await db.commit()
+    await db.refresh(db_indicator)
+    
+    return db_indicator
+
+async def get_threat_statistics(
+    db: AsyncSession,
+    from_date: Optional[datetime] = None,
+    to_date: Optional[datetime] = None,
+) -> Dict[str, Any]:
+    """
+    Get threat statistics.
+    
+    Args:
+        db: Database session
+        from_date: Filter by discovered_at >= from_date
+        to_date: Filter by discovered_at <= to_date
+        
+    Returns:
+        Dict[str, Any]: Threat statistics
+    """
+    # Set default time range if not provided
+    if not to_date:
+        to_date = datetime.utcnow()
+    
+    if not from_date:
+        from_date = to_date - timedelta(days=30)
+    
+    # Get count by severity
+    severity_counts = {}
+    for severity in ThreatSeverity:
+        query = select(func.count(Threat.id)).filter(and_(
+            Threat.severity == severity,
+            Threat.discovered_at >= from_date,
+            Threat.discovered_at <= to_date,
+        ))
+        result = await db.execute(query)
+        severity_counts[severity.value] = result.scalar() or 0
+    
+    # Get count by status
+    status_counts = {}
+    for status in ThreatStatus:
+        query = select(func.count(Threat.id)).filter(and_(
+            Threat.status == status,
+            Threat.discovered_at >= from_date,
+            Threat.discovered_at <= to_date,
+        ))
+        result = await db.execute(query)
+        status_counts[status.value] = result.scalar() or 0
+    
+    # Get count by category
+    category_counts = {}
+    for category in ThreatCategory:
+        query = select(func.count(Threat.id)).filter(and_(
+            Threat.category == category,
+            Threat.discovered_at >= from_date,
+            Threat.discovered_at <= to_date,
+        ))
+        result = await db.execute(query)
+        category_counts[category.value] = result.scalar() or 0
+    
+    # Get total count
+    query = select(func.count(Threat.id)).filter(and_(
+        Threat.discovered_at >= from_date,
+        Threat.discovered_at <= to_date,
+    ))
+    result = await db.execute(query)
+    total_count = result.scalar() or 0
+    
+    # Get count by day
+    time_series = []
+    current_date = from_date.date()
+    end_date = to_date.date()
+    
+    while current_date <= end_date:
+        next_date = current_date + timedelta(days=1)
+        query = select(func.count(Threat.id)).filter(and_(
+            Threat.discovered_at >= datetime.combine(current_date, datetime.min.time()),
+            Threat.discovered_at < datetime.combine(next_date, datetime.min.time()),
+        ))
+        result = await db.execute(query)
+        count = result.scalar() or 0
+        time_series.append({
+            "date": current_date.isoformat(),
+            "count": count
+        })
+        current_date = next_date
+    
+    # Return statistics
+    return {
+        "total_count": total_count,
+        "severity_counts": severity_counts,
+        "status_counts": status_counts,
+        "category_counts": category_counts,
+        "time_series": time_series,
+        "from_date": from_date.isoformat(),
+        "to_date": to_date.isoformat(),
+    }

src/api/services/user_service.py

@@ -0,0 +1,166 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+from sqlalchemy import update
+from passlib.context import CryptContext
+from typing import Optional, List, Dict, Any
+import logging
+
+from src.models.user import User
+from src.api.schemas import UserCreate, UserUpdate, UserInDB
+
+# Configure logger
+logger = logging.getLogger(__name__)
+
+# Password context for hashing and verification
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """
+    Verify password against hash.
+    
+    Args:
+        plain_password: Plain password
+        hashed_password: Hashed password
+        
+    Returns:
+        bool: True if password is correct
+    """
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password: str) -> str:
+    """
+    Hash password.
+    
+    Args:
+        password: Plain password
+        
+    Returns:
+        str: Hashed password
+    """
+    return pwd_context.hash(password)
+
+async def get_user_by_username(db: AsyncSession, username: str) -> Optional[UserInDB]:
+    """
+    Get user by username.
+    
+    Args:
+        db: Database session
+        username: Username
+        
+    Returns:
+        Optional[UserInDB]: User if found, None otherwise
+    """
+    try:
+        result = await db.execute(select(User).where(User.username == username))
+        user = result.scalars().first()
+        
+        if not user:
+            return None
+            
+        # Convert SQLAlchemy model to Pydantic model
+        user_dict = {c.name: getattr(user, c.name) for c in user.__table__.columns}
+        return UserInDB(**user_dict)
+    except Exception as e:
+        logger.error(f"Error getting user by username: {e}")
+        return None
+
+async def authenticate_user(db: AsyncSession, username: str, password: str) -> Optional[UserInDB]:
+    """
+    Authenticate user.
+    
+    Args:
+        db: Database session
+        username: Username
+        password: Plain password
+        
+    Returns:
+        Optional[UserInDB]: User if authenticated, None otherwise
+    """
+    user = await get_user_by_username(db, username)
+    
+    if not user:
+        return None
+        
+    if not verify_password(password, user.hashed_password):
+        return None
+        
+    return user
+
+async def create_user(db: AsyncSession, user_data: UserCreate) -> Optional[UserInDB]:
+    """
+    Create a new user.
+    
+    Args:
+        db: Database session
+        user_data: User data
+        
+    Returns:
+        Optional[UserInDB]: Created user
+    """
+    try:
+        # Check if user already exists
+        existing_user = await get_user_by_username(db, user_data.username)
+        if existing_user:
+            return None
+            
+        # Create new user
+        hashed_password = get_password_hash(user_data.password)
+        user = User(
+            username=user_data.username,
+            email=user_data.email,
+            full_name=user_data.full_name,
+            hashed_password=hashed_password,
+            is_active=user_data.is_active
+        )
+        
+        db.add(user)
+        await db.commit()
+        await db.refresh(user)
+        
+        # Convert SQLAlchemy model to Pydantic model
+        user_dict = {c.name: getattr(user, c.name) for c in user.__table__.columns}
+        return UserInDB(**user_dict)
+    except Exception as e:
+        logger.error(f"Error creating user: {e}")
+        await db.rollback()
+        return None
+
+async def update_user(db: AsyncSession, user_id: int, user_data: UserUpdate) -> Optional[UserInDB]:
+    """
+    Update user.
+    
+    Args:
+        db: Database session
+        user_id: User ID
+        user_data: User data
+        
+    Returns:
+        Optional[UserInDB]: Updated user
+    """
+    try:
+        # Create update dictionary
+        update_data = user_data.dict(exclude_unset=True)
+        
+        # Hash password if provided
+        if "password" in update_data:
+            update_data["hashed_password"] = get_password_hash(update_data.pop("password"))
+            
+        # Update user
+        stmt = update(User).where(User.id == user_id).values(**update_data)
+        await db.execute(stmt)
+        await db.commit()
+        
+        # Get updated user
+        result = await db.execute(select(User).where(User.id == user_id))
+        user = result.scalars().first()
+        
+        if not user:
+            return None
+            
+        # Convert SQLAlchemy model to Pydantic model
+        user_dict = {c.name: getattr(user, c.name) for c in user.__table__.columns}
+        return UserInDB(**user_dict)
+    except Exception as e:
+        logger.error(f"Error updating user: {e}")
+        await db.rollback()
+        return None