# 技術(shù)棧優(yōu)勢分析
tech_advantages = {
    "FastAPI": {
        "性能": "基于Starlette和Pydantic，性能接近NodeJS和Go",
        "開發(fā)效率": "自動API文檔、類型提示、異步支持",
        "現(xiàn)代特性": "OpenAPI、JSON Schema、依賴注入"
    },
    "SQLite": {
        "輕量級": "無服務(wù)器、零配置的數(shù)據(jù)庫引擎",
        "可靠性": "ACID事務(wù)，廣泛測試的代碼庫",
        "適用場景": "完美適合中小型應(yīng)用和原型開發(fā)"
    }
}

#!/usr/bin/env python3
"""
短鏈接服務(wù)數(shù)據(jù)庫模型設(shè)計
"""

from sqlalchemy import create_engine, Column, Integer, String, DateTime, Boolean, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import func
from datetime import datetime
import hashlib
import secrets
import string

Base = declarative_base()

class ShortURL(Base):
    """
    短鏈接數(shù)據(jù)模型
    存儲短鏈接與原始URL的映射關(guān)系
    """
    
    __tablename__ = 'short_urls'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    
    # 短鏈接代碼（唯一標(biāo)識）
    short_code = Column(String(10), unique=True, nullable=False, index=True)
    
    # 原始URL
    original_url = Column(Text, nullable=False)
    
    # 創(chuàng)建時間
    created_at = Column(DateTime, default=func.now(), nullable=False)
    
    # 過期時間（可選）
    expires_at = Column(DateTime, nullable=True)
    
    # 點擊次數(shù)統(tǒng)計
    click_count = Column(Integer, default=0, nullable=False)
    
    # 是否啟用
    is_active = Column(Boolean, default=True, nullable=False)
    
    # 創(chuàng)建者標(biāo)識（用于多用戶擴展）
    created_by = Column(String(50), nullable=True)
    
    # 自定義短代碼（如果用戶提供）
    custom_code = Column(String(10), unique=True, nullable=True)
    
    def __repr__(self):
        return f"<ShortURL(short_code='{self.short_code}', original_url='{self.original_url}')>"

class ClickAnalytics(Base):
    """
    點擊分析數(shù)據(jù)模型
    記錄每次點擊的詳細(xì)信息
    """
    
    __tablename__ = 'click_analytics'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    
    # 關(guān)聯(lián)的短鏈接ID
    short_url_id = Column(Integer, nullable=False, index=True)
    
    # 點擊時間
    clicked_at = Column(DateTime, default=func.now(), nullable=False)
    
    # 用戶代理
    user_agent = Column(Text, nullable=True)
    
    # IP地址
    ip_address = Column(String(45), nullable=True)  # 支持IPv6
    
    # 引用來源
    referrer = Column(Text, nullable=True)
    
    # 國家/地區(qū)（通過IP解析）
    country = Column(String(2), nullable=True)
    
    # 瀏覽器信息
    browser = Column(String(50), nullable=True)
    
    # 操作系統(tǒng)
    operating_system = Column(String(50), nullable=True)
    
    # 設(shè)備類型（桌面/移動/平板）
    device_type = Column(String(20), nullable=True)

class APIToken(Base):
    """
    API令牌管理
    用于API訪問認(rèn)證
    """
    
    __tablename__ = 'api_tokens'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    
    # 令牌標(biāo)識
    token_name = Column(String(50), nullable=False)
    
    # 令牌哈希（存儲哈希值而非原始令牌）
    token_hash = Column(String(64), unique=True, nullable=False)
    
    # 創(chuàng)建時間
    created_at = Column(DateTime, default=func.now(), nullable=False)
    
    # 過期時間
    expires_at = Column(DateTime, nullable=True)
    
    # 是否啟用
    is_active = Column(Boolean, default=True, nullable=False)
    
    # 權(quán)限級別
    permission_level = Column(String(20), default='user', nullable=False)

class DatabaseManager:
    """
    數(shù)據(jù)庫管理類
    負(fù)責(zé)數(shù)據(jù)庫連接、初始化和基本操作
    """
    
    def __init__(self, database_url: str = "sqlite:///./shortener.db"):
        """
        初始化數(shù)據(jù)庫管理器
        
        Args:
            database_url: 數(shù)據(jù)庫連接URL
        """
        self.database_url = database_url
        self.engine = None
        self.SessionLocal = None
        
    def init_database(self):
        """
        初始化數(shù)據(jù)庫連接和表結(jié)構(gòu)
        """
        # 創(chuàng)建引擎
        self.engine = create_engine(
            self.database_url, 
            connect_args={"check_same_thread": False}  # SQLite需要這個參數(shù)
        )
        
        # 創(chuàng)建會話工廠
        self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)
        
        # 創(chuàng)建所有表
        Base.metadata.create_all(bind=self.engine)
        
        print("? 數(shù)據(jù)庫初始化完成")
        print(f"   數(shù)據(jù)庫位置: {self.database_url}")
        print(f"   創(chuàng)建的表: {Base.metadata.tables.keys()}")
    
    def get_session(self):
        """
        獲取數(shù)據(jù)庫會話
        
        Returns:
            Session: 數(shù)據(jù)庫會話對象
        """
        if not self.SessionLocal:
            raise RuntimeError("數(shù)據(jù)庫未初始化，請先調(diào)用 init_database()")
        
        return self.SessionLocal()
    
    def close_session(self, session):
        """
        關(guān)閉數(shù)據(jù)庫會話
        
        Args:
            session: 數(shù)據(jù)庫會話對象
        """
        if session:
            session.close()

# 數(shù)據(jù)庫工具函數(shù)
class URLGenerator:
    """
    URL生成工具類
    負(fù)責(zé)生成短鏈接代碼和處理沖突
    """
    
    def __init__(self):
        self.attempts_limit = 5  # 最大嘗試次數(shù)
    
    def generate_short_code(self, original_url: str = None, length: int = 6) -> str:
        """
        生成短鏈接代碼
        
        Args:
            original_url: 原始URL（用于確定性生成）
            length: 代碼長度
            
        Returns:
            str: 短鏈接代碼
        """
        if original_url:
            # 基于URL內(nèi)容的確定性生成
            hash_object = hashlib.md5(original_url.encode())
            hex_digest = hash_object.hexdigest()
            return hex_digest[:length]
        else:
            # 隨機生成
            characters = string.ascii_letters + string.digits
            return ''.join(secrets.choice(characters) for _ in range(length))
    
    def generate_unique_code(self, db_session, original_url: str = None, custom_code: str = None) -> str:
        """
        生成唯一的短鏈接代碼
        
        Args:
            db_session: 數(shù)據(jù)庫會話
            original_url: 原始URL
            custom_code: 自定義代碼
            
        Returns:
            str: 唯一的短鏈接代碼
        """
        # 如果提供了自定義代碼，直接使用
        if custom_code:
            if self._is_code_available(db_session, custom_code):
                return custom_code
            else:
                raise ValueError(f"自定義代碼 '{custom_code}' 已被使用")
        
        # 生成并檢查唯一性
        for attempt in range(self.attempts_limit):
            if attempt == 0 and original_url:
                # 第一次嘗試使用確定性生成
                short_code = self.generate_short_code(original_url)
            else:
                # 后續(xù)嘗試使用隨機生成
                short_code = self.generate_short_code()
            
            if self._is_code_available(db_session, short_code):
                return short_code
        
        # 如果所有嘗試都失敗，增加長度再試一次
        return self.generate_short_code(length=8)
    
    def _is_code_available(self, db_session, short_code: str) -> bool:
        """
        檢查短鏈接代碼是否可用
        
        Args:
            db_session: 數(shù)據(jù)庫會話
            short_code: 要檢查的代碼
            
        Returns:
            bool: 是否可用
        """
        from sqlalchemy import exists
        
        # 檢查是否已存在
        exists_query = db_session.query(
            exists().where(ShortURL.short_code == short_code)
        )
        return not exists_query.scalar()

# 演示數(shù)據(jù)庫初始化
def demo_database_setup():
    """演示數(shù)據(jù)庫設(shè)置"""
    print("短鏈接服務(wù)數(shù)據(jù)庫演示")
    print("=" * 50)
    
    # 創(chuàng)建數(shù)據(jù)庫管理器
    db_manager = DatabaseManager()
    db_manager.init_database()
    
    # 獲取會話并演示一些操作
    session = db_manager.get_session()
    
    try:
        # 創(chuàng)建URL生成器
        url_generator = URLGenerator()
        
        # 生成一些示例短鏈接
        test_urls = [
            "https://www.example.com/very/long/url/path/that/needs/shortening",
            "https://docs.python.org/3/library/sqlalchemy.html",
            "https://fastapi.tiangolo.com/tutorial/sql-databases/"
        ]
        
        print("\n生成示例短鏈接:")
        for url in test_urls:
            short_code = url_generator.generate_unique_code(session, url)
            print(f"  {url[:50]}... -> {short_code}")
        
        # 顯示數(shù)據(jù)庫統(tǒng)計
        table_count = len(Base.metadata.tables)
        print(f"\n數(shù)據(jù)庫狀態(tài):")
        print(f"  表數(shù)量: {table_count}")
        print(f"  表名稱: {list(Base.metadata.tables.keys())}")
        
    finally:
        db_manager.close_session(session)

if __name__ == "__main__":
    demo_database_setup()

#!/usr/bin/env python3
"""
FastAPI應(yīng)用核心配置和依賴注入
"""

from fastapi import FastAPI, Depends, HTTPException, status, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse, JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
from pydantic import BaseModel, validator, HttpUrl
from typing import Optional, List
from datetime import datetime, timedelta
import secrets
import hashlib
import re

# 導(dǎo)入數(shù)據(jù)庫相關(guān)
from database import DatabaseManager, ShortURL, ClickAnalytics, URLGenerator

# Pydantic模型定義
class ShortURLCreate(BaseModel):
    """創(chuàng)建短鏈接請求模型"""
    original_url: HttpUrl
    custom_code: Optional[str] = None
    expires_in_days: Optional[int] = None
    
    @validator('custom_code')
    def validate_custom_code(cls, v):
        if v is not None:
            # 只允許字母、數(shù)字、連字符和下劃線
            if not re.match(r'^[a-zA-Z0-9_-]{3,10}$', v):
                raise ValueError('自定義代碼只能包含字母、數(shù)字、連字符和下劃線，長度3-10個字符')
        return v
    
    @validator('expires_in_days')
    def validate_expires_in_days(cls, v):
        if v is not None and v <= 0:
            raise ValueError('過期天數(shù)必須大于0')
        return v

class ShortURLResponse(BaseModel):
    """短鏈接響應(yīng)模型"""
    short_code: str
    short_url: str
    original_url: str
    created_at: datetime
    expires_at: Optional[datetime]
    click_count: int
    
    class Config:
        from_attributes = True

class AnalyticsResponse(BaseModel):
    """分析數(shù)據(jù)響應(yīng)模型"""
    short_code: str
    total_clicks: int
    clicks_last_24h: int
    clicks_last_7d: int
    top_referrers: List[str]
    country_stats: dict
    browser_stats: dict

class ErrorResponse(BaseModel):
    """錯誤響應(yīng)模型"""
    error: str
    detail: Optional[str] = None
    code: int

# 安全相關(guān)
security = HTTPBearer()

class ShortenerService:
    """
    短鏈接服務(wù)核心類
    包含所有業(yè)務(wù)邏輯
    """
    
    def __init__(self, db_manager: DatabaseManager):
        self.db_manager = db_manager
        self.url_generator = URLGenerator()
    
    def create_short_url(self, 
                        original_url: str, 
                        custom_code: Optional[str] = None,
                        expires_in_days: Optional[int] = None,
                        created_by: Optional[str] = None) -> ShortURL:
        """
        創(chuàng)建短鏈接
        
        Args:
            original_url: 原始URL
            custom_code: 自定義代碼
            expires_in_days: 過期天數(shù)
            created_by: 創(chuàng)建者標(biāo)識
            
        Returns:
            ShortURL: 創(chuàng)建的短鏈接對象
        """
        session = self.db_manager.get_session()
        
        try:
            # 生成唯一短代碼
            short_code = self.url_generator.generate_unique_code(
                session, original_url, custom_code
            )
            
            # 計算過期時間
            expires_at = None
            if expires_in_days:
                expires_at = datetime.now() + timedelta(days=expires_in_days)
            
            # 創(chuàng)建短鏈接記錄
            short_url = ShortURL(
                short_code=short_code,
                original_url=str(original_url),
                expires_at=expires_at,
                created_by=created_by,
                custom_code=custom_code
            )
            
            session.add(short_url)
            session.commit()
            session.refresh(short_url)
            
            return short_url
            
        except Exception as e:
            session.rollback()
            raise e
        finally:
            self.db_manager.close_session(session)
    
    def get_short_url(self, short_code: str) -> Optional[ShortURL]:
        """
        獲取短鏈接信息
        
        Args:
            short_code: 短鏈接代碼
            
        Returns:
            Optional[ShortURL]: 短鏈接對象，如果不存在則返回None
        """
        session = self.db_manager.get_session()
        
        try:
            short_url = session.query(ShortURL).filter(
                ShortURL.short_code == short_code,
                ShortURL.is_active == True
            ).first()
            
            return short_url
            
        finally:
            self.db_manager.close_session(session)
    
    def record_click(self, 
                    short_url_id: int, 
                    request: Request,
                    user_agent: Optional[str] = None,
                    referrer: Optional[str] = None):
        """
        記錄點擊數(shù)據(jù)
        
        Args:
            short_url_id: 短鏈接ID
            request: 請求對象
            user_agent: 用戶代理
            referrer: 引用來源
        """
        session = self.db_manager.get_session()
        
        try:
            # 更新點擊計數(shù)
            short_url = session.query(ShortURL).filter(ShortURL.id == short_url_id).first()
            if short_url:
                short_url.click_count += 1
            
            # 記錄詳細(xì)點擊數(shù)據(jù)
            click_analytics = ClickAnalytics(
                short_url_id=short_url_id,
                user_agent=user_agent,
                ip_address=request.client.host if request.client else None,
                referrer=referrer
            )
            
            session.add(click_analytics)
            session.commit()
            
        except Exception as e:
            session.rollback()
            # 點擊記錄失敗不應(yīng)影響重定向
            print(f"記錄點擊數(shù)據(jù)失敗: {e}")
        finally:
            self.db_manager.close_session(session)
    
    def get_analytics(self, short_code: str) -> AnalyticsResponse:
        """
        獲取分析數(shù)據(jù)
        
        Args:
            short_code: 短鏈接代碼
            
        Returns:
            AnalyticsResponse: 分析數(shù)據(jù)
        """
        session = self.db_manager.get_session()
        
        try:
            short_url = session.query(ShortURL).filter(
                ShortURL.short_code == short_code
            ).first()
            
            if not short_url:
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail="短鏈接不存在"
                )
            
            # 獲取基本統(tǒng)計
            total_clicks = short_url.click_count
            
            # 獲取時間范圍統(tǒng)計
            now = datetime.now()
            day_ago = now - timedelta(days=1)
            week_ago = now - timedelta(days=7)
            
            clicks_last_24h = session.query(ClickAnalytics).filter(
                ClickAnalytics.short_url_id == short_url.id,
                ClickAnalytics.clicked_at >= day_ago
            ).count()
            
            clicks_last_7d = session.query(ClickAnalytics).filter(
                ClickAnalytics.short_url_id == short_url.id,
                ClickAnalytics.clicked_at >= week_ago
            ).count()
            
            # 獲取引用來源統(tǒng)計
            referrer_stats = session.query(
                ClickAnalytics.referrer
            ).filter(
                ClickAnalytics.short_url_id == short_url.id,
                ClickAnalytics.referrer.isnot(None)
            ).group_by(ClickAnalytics.referrer).all()
            
            top_referrers = [ref[0] for ref in referrer_stats[:5]]  # 前5個引用來源
            
            return AnalyticsResponse(
                short_code=short_code,
                total_clicks=total_clicks,
                clicks_last_24h=clicks_last_24h,
                clicks_last_7d=clicks_last_7d,
                top_referrers=top_referrers,
                country_stats={},  # 簡化實現(xiàn)，實際中可以通過IP地址解析
                browser_stats={}   # 簡化實現(xiàn)，實際中可以解析user_agent
            )
            
        finally:
            self.db_manager.close_session(session)

# 依賴注入
def get_db_manager():
    """獲取數(shù)據(jù)庫管理器依賴"""
    db_manager = DatabaseManager()
    db_manager.init_database()
    return db_manager

def get_shortener_service(db_manager: DatabaseManager = Depends(get_db_manager)):
    """獲取短鏈接服務(wù)依賴"""
    return ShortenerService(db_manager)

def verify_api_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """驗證API令牌（簡化實現(xiàn)）"""
    # 在實際應(yīng)用中，這里應(yīng)該查詢數(shù)據(jù)庫驗證令牌
    token = credentials.credentials
    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="無效的API令牌"
        )
    return token

# 創(chuàng)建FastAPI應(yīng)用
app = FastAPI(
    title="短鏈接生成器服務(wù)",
    description="基于FastAPI和SQLite的高性能短鏈接服務(wù)",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

# 添加CORS中間件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生產(chǎn)環(huán)境中應(yīng)該限制來源
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 全局異常處理
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            error=exc.detail,
            code=exc.status_code
        ).dict()
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content=ErrorResponse(
            error="內(nèi)部服務(wù)器錯誤",
            detail=str(exc),
            code=status.HTTP_500_INTERNAL_SERVER_ERROR
        ).dict()
    )

# 演示應(yīng)用啟動
def demo_app_setup():
    """演示應(yīng)用設(shè)置"""
    print("FastAPI短鏈接服務(wù)演示")
    print("=" * 50)
    print("應(yīng)用特性:")
    print("  ? 自動API文檔 (Swagger UI)")
    print("  ? 類型安全和數(shù)據(jù)驗證")
    print("  ? 異步支持")
    print("  ? CORS中間件")
    print("  ? 全局異常處理")
    print("  ? 依賴注入系統(tǒng)")
    print("  ? 安全認(rèn)證框架")

if __name__ == "__main__":
    demo_app_setup()

#!/usr/bin/env python3
"""
FastAPI路由實現(xiàn)
包含所有API端點的實現(xiàn)
"""

from fastapi import APIRouter, Depends, HTTPException, status, Request, Query
from fastapi.responses import RedirectResponse
from typing import Optional, List
from datetime import datetime

# 導(dǎo)入之前定義的模型和服務(wù)
from fastapi_app import (
    app, ShortURLCreate, ShortURLResponse, AnalyticsResponse, ErrorResponse,
    get_shortener_service, verify_api_token, ShortenerService
)

# 創(chuàng)建路由器
router = APIRouter()

@router.post(
    "/shorten",
    response_model=ShortURLResponse,
    status_code=status.HTTP_201_CREATED,
    summary="創(chuàng)建短鏈接",
    description="將長URL轉(zhuǎn)換為短鏈接"
)
async def create_short_url(
    request: ShortURLCreate,
    service: ShortenerService = Depends(get_shortener_service),
    token: str = Depends(verify_api_token)
):
    """
    創(chuàng)建短鏈接端點
    
    Args:
        request: 創(chuàng)建短鏈接請求
        service: 短鏈接服務(wù)
        token: API令牌
        
    Returns:
        ShortURLResponse: 創(chuàng)建的短鏈接信息
    """
    try:
        short_url = service.create_short_url(
            original_url=str(request.original_url),
            custom_code=request.custom_code,
            expires_in_days=request.expires_in_days,
            created_by=f"api_user_{hash(token) % 1000}"  # 簡化用戶標(biāo)識
        )
        
        # 構(gòu)建完整的短鏈接URL
        base_url = "http://localhost:8000"  # 實際部署時應(yīng)從配置讀取
        short_url_str = f"{base_url}/{short_url.short_code}"
        
        return ShortURLResponse(
            short_code=short_url.short_code,
            short_url=short_url_str,
            original_url=short_url.original_url,
            created_at=short_url.created_at,
            expires_at=short_url.expires_at,
            click_count=short_url.click_count
        )
        
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e)
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"創(chuàng)建短鏈接失敗: {str(e)}"
        )

@router.get(
    "/{short_code}",
    summary="重定向到原始URL",
    description="通過短鏈接代碼重定向到原始URL"
)
async def redirect_to_original(
    short_code: str,
    request: Request,
    service: ShortenerService = Depends(get_shortener_service)
):
    """
    重定向端點
    
    Args:
        short_code: 短鏈接代碼
        request: 請求對象
        service: 短鏈接服務(wù)
        
    Returns:
        RedirectResponse: 重定向響應(yīng)
    """
    # 獲取短鏈接信息
    short_url = service.get_short_url(short_code)
    
    if not short_url:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="短鏈接不存在或已失效"
        )
    
    # 檢查是否過期
    if short_url.expires_at and short_url.expires_at < datetime.now():
        raise HTTPException(
            status_code=status.HTTP_410_GONE,
            detail="短鏈接已過期"
        )
    
    # 檢查是否啟用
    if not short_url.is_active:
        raise HTTPException(
            status_code=status.HTTP_410_GONE,
            detail="短鏈接已被禁用"
        )
    
    # 記錄點擊
    service.record_click(
        short_url_id=short_url.id,
        request=request,
        user_agent=request.headers.get("user-agent"),
        referrer=request.headers.get("referer")
    )
    
    # 重定向到原始URL
    return RedirectResponse(url=short_url.original_url, status_code=status.HTTP_302_FOUND)

@router.get(
    "/{short_code}/info",
    response_model=ShortURLResponse,
    summary="獲取短鏈接信息",
    description="獲取短鏈接的詳細(xì)信息"
)
async def get_short_url_info(
    short_code: str,
    service: ShortenerService = Depends(get_shortener_service),
    token: str = Depends(verify_api_token)
):
    """
    獲取短鏈接信息端點
    
    Args:
        short_code: 短鏈接代碼
        service: 短鏈接服務(wù)
        token: API令牌
        
    Returns:
        ShortURLResponse: 短鏈接信息
    """
    short_url = service.get_short_url(short_code)
    
    if not short_url:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="短鏈接不存在"
        )
    
    base_url = "http://localhost:8000"
    short_url_str = f"{base_url}/{short_url.short_code}"
    
    return ShortURLResponse(
        short_code=short_url.short_code,
        short_url=short_url_str,
        original_url=short_url.original_url,
        created_at=short_url.created_at,
        expires_at=short_url.expires_at,
        click_count=short_url.click_count
    )

@router.get(
    "/{short_code}/analytics",
    response_model=AnalyticsResponse,
    summary="獲取分析數(shù)據(jù)",
    description="獲取短鏈接的點擊分析數(shù)據(jù)"
)
async def get_analytics(
    short_code: str,
    service: ShortenerService = Depends(get_shortener_service),
    token: str = Depends(verify_api_token)
):
    """
    獲取分析數(shù)據(jù)端點
    
    Args:
        short_code: 短鏈接代碼
        service: 短鏈接服務(wù)
        token: API令牌
        
    Returns:
        AnalyticsResponse: 分析數(shù)據(jù)
    """
    try:
        analytics = service.get_analytics(short_code)
        return analytics
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"獲取分析數(shù)據(jù)失敗: {str(e)}"
        )

@router.delete(
    "/{short_code}",
    status_code=status.HTTP_200_OK,
    summary="刪除短鏈接",
    description="禁用或刪除短鏈接"
)
async def delete_short_url(
    short_code: str,
    service: ShortenerService = Depends(get_shortener_service),
    token: str = Depends(verify_api_token)
):
    """
    刪除短鏈接端點
    
    Args:
        short_code: 短鏈接代碼
        service: 短鏈接服務(wù)
        token: API令牌
        
    Returns:
        dict: 操作結(jié)果
    """
    session = service.db_manager.get_session()
    
    try:
        short_url = session.query(ShortURL).filter(ShortURL.short_code == short_code).first()
        
        if not short_url:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="短鏈接不存在"
            )
        
        # 軟刪除：禁用短鏈接
        short_url.is_active = False
        session.commit()
        
        return {"message": f"短鏈接 {short_code} 已成功禁用"}
        
    except HTTPException:
        raise
    except Exception as e:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"刪除短鏈接失敗: {str(e)}"
        )
    finally:
        service.db_manager.close_session(session)

@router.get(
    "/",
    summary="服務(wù)狀態(tài)",
    description="獲取服務(wù)狀態(tài)和基本信息"
)
async def get_service_status(service: ShortenerService = Depends(get_shortener_service)):
    """
    服務(wù)狀態(tài)端點
    
    Args:
        service: 短鏈接服務(wù)
        
    Returns:
        dict: 服務(wù)狀態(tài)信息
    """
    session = service.db_manager.get_session()
    
    try:
        # 獲取基本統(tǒng)計
        total_urls = session.query(ShortURL).count()
        active_urls = session.query(ShortURL).filter(ShortURL.is_active == True).count()
        total_clicks = session.query(ClickAnalytics).count()
        
        # 獲取今日點擊
        from datetime import datetime, timedelta
        today_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
        today_clicks = session.query(ClickAnalytics).filter(
            ClickAnalytics.clicked_at >= today_start
        ).count()
        
        return {
            "status": "running",
            "version": "1.0.0",
            "total_urls": total_urls,
            "active_urls": active_urls,
            "total_clicks": total_clicks,
            "today_clicks": today_clicks,
            "timestamp": datetime.now().isoformat()
        }
        
    finally:
        service.db_manager.close_session(session)

# 注冊路由
app.include_router(router)

# 根路徑重定向到文檔
@app.get("/")
async def root():
    """根路徑重定向到API文檔"""
    return RedirectResponse(url="/docs")

# 演示API使用
def demo_api_usage():
    """演示API使用方法"""
    print("\nAPI使用示例:")
    print("=" * 50)
    
    examples = {
        "創(chuàng)建短鏈接": {
            "method": "POST",
            "url": "http://localhost:8000/shorten",
            "headers": {"Authorization": "Bearer your_token"},
            "body": {
                "original_url": "https://www.example.com/very/long/url",
                "custom_code": "example",
                "expires_in_days": 30
            }
        },
        "重定向": {
            "method": "GET", 
            "url": "http://localhost:8000/abc123"
        },
        "獲取信息": {
            "method": "GET",
            "url": "http://localhost:8000/abc123/info",
            "headers": {"Authorization": "Bearer your_token"}
        },
        "獲取分析": {
            "method": "GET", 
            "url": "http://localhost:8000/abc123/analytics",
            "headers": {"Authorization": "Bearer your_token"}
        }
    }
    
    for endpoint, info in examples.items():
        print(f"\n{endpoint}:")
        print(f"  {info['method']} {info['url']}")
        if 'headers' in info:
            for key, value in info['headers'].items():
                print(f"  Header: {key}: {value}")
        if 'body' in info:
            print(f"  Body: {info['body']}")

if __name__ == "__main__":
    import uvicorn
    
    print("啟動短鏈接服務(wù)...")
    demo_api_usage()
    
    # 啟動服務(wù)器
    uvicorn.run(
        "fastapi_routes:app",
        host="0.0.0.0",
        port=8000,
        reload=True,  # 開發(fā)時啟用熱重載
        log_level="info"
    )

#!/usr/bin/env python3
"""
短鏈接生成器服務(wù) - 完整集成版本
主應(yīng)用入口點
"""

import os
import uvicorn
from fastapi import FastAPI
from contextlib import asynccontextmanager

# 導(dǎo)入之前定義的模塊
from database import DatabaseManager, Base
from fastapi_app import app, get_db_manager
from fastapi_routes import router

# 應(yīng)用生命周期管理
@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    應(yīng)用生命周期管理
    - 啟動時初始化數(shù)據(jù)庫
    - 關(guān)閉時清理資源
    """
    # 啟動時
    print("?? 啟動短鏈接服務(wù)...")
    
    # 初始化數(shù)據(jù)庫
    db_manager = DatabaseManager()
    db_manager.init_database()
    
    # 創(chuàng)建示例數(shù)據(jù)（僅開發(fā)環(huán)境）
    if os.getenv("ENVIRONMENT") == "development":
        await create_sample_data(db_manager)
    
    yield  # 應(yīng)用運行期間
    
    # 關(guān)閉時
    print("?? 關(guān)閉短鏈接服務(wù)...")
    # 這里可以添加資源清理代碼

# 更新應(yīng)用的生命周期
app.router.lifespan_context = lifespan

# 確保路由已注冊
app.include_router(router)

async def create_sample_data(db_manager: DatabaseManager):
    """
    創(chuàng)建示例數(shù)據(jù)（僅用于演示）
    
    Args:
        db_manager: 數(shù)據(jù)庫管理器
    """
    from shortener_service import ShortenerService
    from sqlalchemy import text
    
    service = ShortenerService(db_manager)
    session = db_manager.get_session()
    
    try:
        # 檢查是否已有數(shù)據(jù)
        result = session.execute(text("SELECT COUNT(*) FROM short_urls"))
        count = result.scalar()
        
        if count == 0:
            print("?? 創(chuàng)建示例數(shù)據(jù)...")
            
            # 創(chuàng)建一些示例短鏈接
            sample_urls = [
                "https://www.github.com",
                "https://www.python.org",
                "https://fastapi.tiangolo.com",
                "https://www.sqlite.org/index.html",
                "https://www.docker.com"
            ]
            
            for url in sample_urls:
                try:
                    service.create_short_url(
                        original_url=url,
                        created_by="system"
                    )
                except Exception as e:
                    print(f"創(chuàng)建示例數(shù)據(jù)失敗 {url}: {e}")
            
            print(f"? 創(chuàng)建了 {len(sample_urls)} 個示例短鏈接")
        else:
            print(f"?? 數(shù)據(jù)庫中已有 {count} 個短鏈接")
            
    except Exception as e:
        print(f"創(chuàng)建示例數(shù)據(jù)時出錯: {e}")
    finally:
        db_manager.close_session(session)

def get_application() -> FastAPI:
    """
    獲取FastAPI應(yīng)用實例
    
    Returns:
        FastAPI: 應(yīng)用實例
    """
    return app

# 配置類
class Config:
    """應(yīng)用配置"""
    DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./shortener.db")
    HOST = os.getenv("HOST", "0.0.0.0")
    PORT = int(os.getenv("PORT", 8000))
    RELOAD = os.getenv("RELOAD", "true").lower() == "true"
    LOG_LEVEL = os.getenv("LOG_LEVEL", "info")

def print_startup_banner():
    """打印啟動橫幅"""
    banner = """
    ╔═══════════════════════════════════════════════╗
    ║          短鏈接生成器服務(wù)                      ║
    ║          FastAPI + SQLite                    ║
    ║                                               ║
    ║          ?? 服務(wù)已啟動！                      ║
    ║          ?? API文檔: http://{host}:{port}/docs ║
    ║          ?? Redoc文檔: http://{host}:{port}/redoc ║
    ║          ?? 服務(wù)狀態(tài): http://{host}:{port}/    ║
    ╚═══════════════════════════════════════════════╝
    """
    
    config = Config()
    print(banner.format(host=config.HOST, port=config.PORT))

def main():
    """主函數(shù)"""
    config = Config()
    
    # 打印啟動信息
    print_startup_banner()
    print(f"?? 數(shù)據(jù)庫: {config.DATABASE_URL}")
    print(f"?? 服務(wù)地址: http://{config.HOST}:{config.PORT}")
    print(f"?? 開發(fā)模式: {config.RELOAD}")
    print(f"?? 日志級別: {config.LOG_LEVEL}")
    
    # 啟動服務(wù)器
    uvicorn.run(
        "main:app",
        host=config.HOST,
        port=config.PORT,
        reload=config.RELOAD,
        log_level=config.LOG_LEVEL,
        access_log=True
    )

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
配置管理和環(huán)境設(shè)置
"""

import os
from typing import Optional
from pydantic import BaseSettings

class Settings(BaseSettings):
    """
    應(yīng)用配置設(shè)置
    使用pydantic的BaseSettings管理環(huán)境變量
    """
    
    # 應(yīng)用設(shè)置
    app_name: str = "短鏈接生成器服務(wù)"
    app_version: str = "1.0.0"
    app_description: str = "基于FastAPI和SQLite的高性能短鏈接服務(wù)"
    
    # 服務(wù)器設(shè)置
    host: str = "0.0.0.0"
    port: int = 8000
    reload: bool = True
    log_level: str = "info"
    
    # 數(shù)據(jù)庫設(shè)置
    database_url: str = "sqlite:///./shortener.db"
    
    # 安全設(shè)置
    secret_key: str = "your-secret-key-here"  # 生產(chǎn)環(huán)境應(yīng)該使用環(huán)境變量
    token_expire_minutes: int = 60 * 24 * 7  # 7天
    
    # 業(yè)務(wù)邏輯設(shè)置
    default_short_code_length: int = 6
    max_custom_code_length: int = 10
    max_retry_attempts: int = 5
    default_expiry_days: int = 365
    
    # 速率限制設(shè)置
    rate_limit_requests: int = 100
    rate_limit_minutes: int = 1
    
    # CORS設(shè)置
    cors_origins: list = ["*"]
    
    class Config:
        env_file = ".env"
        case_sensitive = False

def create_env_file():
    """
    創(chuàng)建環(huán)境變量示例文件
    """
    env_content = """# 短鏈接服務(wù)環(huán)境配置

# 應(yīng)用設(shè)置
APP_NAME=短鏈接生成器服務(wù)
APP_VERSION=1.0.0

# 服務(wù)器設(shè)置
HOST=0.0.0.0
PORT=8000
RELOAD=true
LOG_LEVEL=info

# 數(shù)據(jù)庫設(shè)置
DATABASE_URL=sqlite:///./shortener.db

# 安全設(shè)置
SECRET_KEY=your-secret-key-change-in-production
TOKEN_EXPIRE_MINUTES=10080

# 業(yè)務(wù)設(shè)置
DEFAULT_SHORT_CODE_LENGTH=6
MAX_CUSTOM_CODE_LENGTH=10
MAX_RETRY_ATTEMPTS=5
DEFAULT_EXPIRY_DAYS=365

# 速率限制
RATE_LIMIT_REQUESTS=100
RATE_LIMIT_MINUTES=1

# CORS設(shè)置
CORS_ORIGINS=["*"]
"""
    
    with open(".env.example", "w", encoding="utf-8") as f:
        f.write(env_content)
    
    print("? 已創(chuàng)建環(huán)境變量示例文件: .env.example")
    print("?? 請復(fù)制為 .env 并根據(jù)需要修改配置")

def get_settings() -> Settings:
    """
    獲取配置實例
    
    Returns:
        Settings: 配置實例
    """
    return Settings()

# 配置驗證和演示
def validate_config():
    """驗證配置"""
    settings = get_settings()
    
    print("?? 配置驗證")
    print("=" * 50)
    
    config_items = [
        ("應(yīng)用名稱", settings.app_name),
        ("服務(wù)器地址", f"{settings.host}:{settings.port}"),
        ("數(shù)據(jù)庫", settings.database_url),
        ("短代碼長度", settings.default_short_code_length),
        ("默認(rèn)過期天數(shù)", settings.default_expiry_days),
        ("速率限制", f"{settings.rate_limit_requests} 請求/{settings.rate_limit_minutes} 分鐘"),
    ]
    
    for name, value in config_items:
        print(f"  {name}: {value}")
    
    # 檢查關(guān)鍵安全配置
    if settings.secret_key == "your-secret-key-here":
        print("??  警告: 請在生產(chǎn)環(huán)境中修改 SECRET_KEY")
    
    print("? 配置驗證完成")

if __name__ == "__main__":
    create_env_file()
    validate_config()

#!/usr/bin/env python3
"""
緩存和性能優(yōu)化模塊
使用Redis或內(nèi)存緩存提高性能
"""

import time
from typing import Optional, Any
from functools import wraps
import redis
import pickle

class CacheManager:
    """
    緩存管理器
    提供多級緩存支持
    """
    
    def __init__(self, redis_url: Optional[str] = None):
        """
        初始化緩存管理器
        
        Args:
            redis_url: Redis連接URL，如果為None則使用內(nèi)存緩存
        """
        self.redis_client = None
        self.memory_cache = {}
        
        if redis_url:
            try:
                self.redis_client = redis.from_url(redis_url)
                print("? Redis緩存已啟用")
            except Exception as e:
                print(f"? Redis連接失敗: {e}, 使用內(nèi)存緩存")
    
    def get(self, key: str) -> Optional[Any]:
        """
        獲取緩存值
        
        Args:
            key: 緩存鍵
            
        Returns:
            Any: 緩存值，如果不存在返回None
        """
        # 首先嘗試Redis
        if self.redis_client:
            try:
                cached = self.redis_client.get(key)
                if cached:
                    return pickle.loads(cached)
            except Exception as e:
                print(f"Redis獲取失敗: {e}")
        
        # 回退到內(nèi)存緩存
        if key in self.memory_cache:
            cached_data, expiry = self.memory_cache[key]
            if expiry is None or time.time() < expiry:
                return cached_data
            else:
                del self.memory_cache[key]
        
        return None
    
    def set(self, key: str, value: Any, expire_seconds: Optional[int] = None):
        """
        設(shè)置緩存值
        
        Args:
            key: 緩存鍵
            value: 緩存值
            expire_seconds: 過期時間（秒）
        """
        # 設(shè)置Redis緩存
        if self.redis_client:
            try:
                serialized = pickle.dumps(value)
                if expire_seconds:
                    self.redis_client.setex(key, expire_seconds, serialized)
                else:
                    self.redis_client.set(key, serialized)
            except Exception as e:
                print(f"Redis設(shè)置失敗: {e}")
        
        # 設(shè)置內(nèi)存緩存
        expiry = time.time() + expire_seconds if expire_seconds else None
        self.memory_cache[key] = (value, expiry)
        
        # 清理過期的內(nèi)存緩存項
        self._cleanup_memory_cache()
    
    def delete(self, key: str):
        """
        刪除緩存值
        
        Args:
            key: 緩存鍵
        """
        # 刪除Redis緩存
        if self.redis_client:
            try:
                self.redis_client.delete(key)
            except Exception as e:
                print(f"Redis刪除失敗: {e}")
        
        # 刪除內(nèi)存緩存
        if key in self.memory_cache:
            del self.memory_cache[key]
    
    def _cleanup_memory_cache(self):
        """清理過期的內(nèi)存緩存"""
        current_time = time.time()
        expired_keys = [
            key for key, (_, expiry) in self.memory_cache.items()
            if expiry and expiry < current_time
        ]
        
        for key in expired_keys:
            del self.memory_cache[key]
    
    def clear(self):
        """清空所有緩存"""
        # 清空Redis緩存
        if self.redis_client:
            try:
                self.redis_client.flushdb()
            except Exception as e:
                print(f"Redis清空失敗: {e}")
        
        # 清空內(nèi)存緩存
        self.memory_cache.clear()

def cache_response(expire_seconds: int = 300):
    """
    緩存響應(yīng)裝飾器
    
    Args:
        expire_seconds: 緩存過期時間（秒）
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 從依賴注入中獲取緩存管理器
            cache_manager = None
            for arg in args:
                if isinstance(arg, CacheManager):
                    cache_manager = arg
                    break
            
            if not cache_manager:
                # 如果沒有緩存管理器，直接執(zhí)行函數(shù)
                return await func(*args, **kwargs)
            
            # 生成緩存鍵
            cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            
            # 嘗試從緩存獲取
            cached_result = cache_manager.get(cache_key)
            if cached_result is not None:
                return cached_result
            
            # 執(zhí)行函數(shù)并緩存結(jié)果
            result = await func(*args, **kwargs)
            cache_manager.set(cache_key, result, expire_seconds)
            
            return result
        return wrapper
    return decorator

class RateLimiter:
    """
    速率限制器
    防止API濫用
    """
    
    def __init__(self, cache_manager: CacheManager, requests: int = 100, minutes: int = 1):
        """
        初始化速率限制器
        
        Args:
            cache_manager: 緩存管理器
            requests: 允許的請求數(shù)
            minutes: 時間窗口（分鐘）
        """
        self.cache_manager = cache_manager
        self.requests = requests
        self.window_seconds = minutes * 60
    
    def is_rate_limited(self, identifier: str) -> bool:
        """
        檢查是否被限速
        
        Args:
            identifier: 用戶標(biāo)識（IP地址或用戶ID）
            
        Returns:
            bool: 是否被限速
        """
        current_window = int(time.time() / self.window_seconds)
        key = f"rate_limit:{identifier}:{current_window}"
        
        # 獲取當(dāng)前計數(shù)
        current_count = self.cache_manager.get(key) or 0
        
        if current_count >= self.requests:
            return True
        
        # 增加計數(shù)
        self.cache_manager.set(key, current_count + 1, self.window_seconds)
        return False
    
    def get_remaining_requests(self, identifier: str) -> int:
        """
        獲取剩余請求數(shù)
        
        Args:
            identifier: 用戶標(biāo)識
            
        Returns:
            int: 剩余請求數(shù)
        """
        current_window = int(time.time() / self.window_seconds)
        key = f"rate_limit:{identifier}:{current_window}"
        
        current_count = self.cache_manager.get(key) or 0
        return max(0, self.requests - current_count)

# 性能監(jiān)控裝飾器
def timing_decorator(func):
    """執(zhí)行時間監(jiān)控裝飾器"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        result = await func(*args, **kwargs)
        end_time = time.time()
        
        execution_time = end_time - start_time
        print(f"??  {func.__name__} 執(zhí)行時間: {execution_time:.4f}秒")
        
        return result
    return wrapper

# 演示緩存功能
def demo_cache_functionality():
    """演示緩存功能"""
    print("?? 緩存和性能優(yōu)化演示")
    print("=" * 50)
    
    # 創(chuàng)建緩存管理器
    cache = CacheManager()  # 使用內(nèi)存緩存
    
    # 演示基本緩存操作
    cache.set("test_key", "test_value", 60)
    value = cache.get("test_key")
    print(f"緩存設(shè)置和獲取: {value}")
    
    # 演示速率限制
    rate_limiter = RateLimiter(cache, requests=5, minutes=1)
    
    print("\n速率限制演示:")
    for i in range(7):
        limited = rate_limiter.is_rate_limited("test_user")
        remaining = rate_limiter.get_remaining_requests("test_user")
        status = "限速" if limited else "允許"
        print(f"  請求 {i+1}: {status} (剩余: {remaining})")
    
    cache.clear()
    print("? 緩存演示完成")

if __name__ == "__main__":
    demo_cache_functionality()

#!/usr/bin/env python3
"""
測試模塊
包含單元測試和集成測試
"""

import pytest
import pytest_asyncio
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from unittest.mock import Mock, patch

# 導(dǎo)入應(yīng)用和模型
from main import app, get_db_manager
from database import Base, DatabaseManager, ShortURL
from shortener_service import ShortenerService

# 測試數(shù)據(jù)庫URL
TEST_DATABASE_URL = "sqlite:///./test_shortener.db"

@pytest.fixture(scope="function")
def test_db():
    """
    測試數(shù)據(jù)庫fixture
    為每個測試函數(shù)創(chuàng)建獨立的數(shù)據(jù)庫
    """
    # 創(chuàng)建測試引擎
    engine = create_engine(TEST_DATABASE_URL)
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    
    # 創(chuàng)建所有表
    Base.metadata.create_all(bind=engine)
    
    # 創(chuàng)建數(shù)據(jù)庫管理器
    db_manager = DatabaseManager(TEST_DATABASE_URL)
    db_manager.engine = engine
    db_manager.SessionLocal = TestingSessionLocal
    
    yield db_manager
    
    # 清理
    Base.metadata.drop_all(bind=engine)

@pytest.fixture
def test_client(test_db):
    """
    測試客戶端fixture
    """
    # 覆蓋依賴
    def override_get_db():
        return test_db
    
    app.dependency_overrides[get_db_manager] = lambda: test_db
    
    with TestClient(app) as client:
        yield client
    
    # 清理覆蓋
    app.dependency_overrides.clear()

@pytest.fixture
def shortener_service(test_db):
    """
    短鏈接服務(wù)fixture
    """
    return ShortenerService(test_db)

class TestURLGenerator:
    """URL生成器測試"""
    
    def test_generate_short_code(self):
        """測試短代碼生成"""
        from database import URLGenerator
        
        generator = URLGenerator()
        
        # 測試確定性生成
        code1 = generator.generate_short_code("https://example.com")
        code2 = generator.generate_short_code("https://example.com")
        assert code1 == code2
        
        # 測試隨機生成
        code3 = generator.generate_short_code()
        code4 = generator.generate_short_code()
        assert code3 != code4
        
        # 測試長度
        assert len(code1) == 6
        assert len(generator.generate_short_code(length=8)) == 8
    
    def test_generate_unique_code(self, test_db):
        """測試唯一代碼生成"""
        from database import URLGenerator
        
        generator = URLGenerator()
        session = test_db.get_session()
        
        try:
            # 生成唯一代碼
            code1 = generator.generate_unique_code(session, "https://example1.com")
            assert code1 is not None
            
            # 再次生成相同URL應(yīng)該得到相同代碼
            code2 = generator.generate_unique_code(session, "https://example1.com")
            assert code1 == code2
            
            # 生成不同URL應(yīng)該得到不同代碼
            code3 = generator.generate_unique_code(session, "https://example2.com")
            assert code1 != code3
            
        finally:
            test_db.close_session(session)

class TestShortenerService:
    """短鏈接服務(wù)測試"""
    
    def test_create_short_url(self, shortener_service):
        """測試創(chuàng)建短鏈接"""
        original_url = "https://www.example.com/test"
        
        # 創(chuàng)建短鏈接
        short_url = shortener_service.create_short_url(original_url)
        
        assert short_url is not None
        assert short_url.original_url == original_url
        assert len(short_url.short_code) == 6
        assert short_url.click_count == 0
        assert short_url.is_active == True
    
    def test_create_short_url_with_custom_code(self, shortener_service):
        """測試使用自定義代碼創(chuàng)建短鏈接"""
        original_url = "https://www.example.com/custom"
        custom_code = "custom123"
        
        short_url = shortener_service.create_short_url(
            original_url, custom_code=custom_code
        )
        
        assert short_url.short_code == custom_code
    
    def test_get_short_url(self, shortener_service):
        """測試獲取短鏈接"""
        # 先創(chuàng)建
        original_url = "https://www.example.com/get"
        short_url = shortener_service.create_short_url(original_url)
        
        # 再獲取
        retrieved = shortener_service.get_short_url(short_url.short_code)
        
        assert retrieved is not None
        assert retrieved.original_url == original_url
        assert retrieved.short_code == short_url.short_code
    
    def test_get_nonexistent_short_url(self, shortener_service):
        """測試獲取不存在的短鏈接"""
        retrieved = shortener_service.get_short_url("nonexistent")
        assert retrieved is None

class TestAPIRoutes:
    """API路由測試"""
    
    def test_create_short_url_endpoint(self, test_client):
        """測試創(chuàng)建短鏈接端點"""
        # 模擬認(rèn)證
        with patch('fastapi_routes.verify_api_token') as mock_auth:
            mock_auth.return_value = "test_token"
            
            response = test_client.post(
                "/shorten",
                json={
                    "original_url": "https://www.example.com/api-test",
                    "custom_code": "apitest"
                },
                headers={"Authorization": "Bearer test_token"}
            )
            
            assert response.status_code == 201
            data = response.json()
            assert data["short_code"] == "apitest"
            assert data["original_url"] == "https://www.example.com/api-test"
    
    def test_redirect_endpoint(self, test_client, shortener_service):
        """測試重定向端點"""
        # 先創(chuàng)建短鏈接
        short_url = shortener_service.create_short_url("https://www.example.com/redirect")
        
        # 測試重定向
        response = test_client.get(f"/{short_url.short_code}", follow_redirects=False)
        
        assert response.status_code == 302
        assert response.headers["location"] == "https://www.example.com/redirect"
    
    def test_redirect_nonexistent_endpoint(self, test_client):
        """測試重定向到不存在的短鏈接"""
        response = test_client.get("/nonexistent")
        assert response.status_code == 404
    
    def test_get_analytics_endpoint(self, test_client, shortener_service):
        """測試獲取分析數(shù)據(jù)端點"""
        # 先創(chuàng)建短鏈接并模擬一些點擊
        short_url = shortener_service.create_short_url("https://www.example.com/analytics")
        
        # 模擬認(rèn)證
        with patch('fastapi_routes.verify_api_token') as mock_auth:
            mock_auth.return_value = "test_token"
            
            response = test_client.get(
                f"/{short_url.short_code}/analytics",
                headers={"Authorization": "Bearer test_token"}
            )
            
            assert response.status_code == 200
            data = response.json()
            assert data["short_code"] == short_url.short_code
            assert data["total_clicks"] == 0  # 還沒有點擊
    
    def test_service_status_endpoint(self, test_client):
        """測試服務(wù)狀態(tài)端點"""
        response = test_client.get("/")
        assert response.status_code == 200

class TestErrorHandling:
    """錯誤處理測試"""
    
    def test_invalid_url_creation(self, test_client):
        """測試創(chuàng)建無效URL"""
        with patch('fastapi_routes.verify_api_token') as mock_auth:
            mock_auth.return_value = "test_token"
            
            response = test_client.post(
                "/shorten",
                json={"original_url": "not-a-valid-url"},
                headers={"Authorization": "Bearer test_token"}
            )
            
            assert response.status_code == 422  # 驗證錯誤
    
    def test_duplicate_custom_code(self, test_client):
        """測試重復(fù)的自定義代碼"""
        with patch('fastapi_routes.verify_api_token') as mock_auth:
            mock_auth.return_value = "test_token"
            
            # 第一次創(chuàng)建
            response1 = test_client.post(
                "/shorten",
                json={
                    "original_url": "https://www.example.com/first",
                    "custom_code": "duplicate"
                },
                headers={"Authorization": "Bearer test_token"}
            )
            assert response1.status_code == 201
            
            # 第二次使用相同自定義代碼
            response2 = test_client.post(
                "/shorten",
                json={
                    "original_url": "https://www.example.com/second", 
                    "custom_code": "duplicate"
                },
                headers={"Authorization": "Bearer test_token"}
            )
            assert response2.status_code == 400

def run_tests():
    """運行測試套件"""
    print("?? 運行測試套件")
    print("=" * 50)
    
    # 這里可以添加特定的測試運行邏輯
    # 實際中應(yīng)該使用 pytest main() 
    pytest.main([__file__, "-v"])

if __name__ == "__main__":
    run_tests()

#!/usr/bin/env python3
"""
部署配置和Docker支持
"""

import os
import subprocess
from pathlib import Path

# Dockerfile內(nèi)容
DOCKERFILE_CONTENT = """
FROM python:3.9-slim

WORKDIR /app

# 安裝系統(tǒng)依賴
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# 復(fù)制依賴文件
COPY requirements.txt .

# 安裝Python依賴
RUN pip install --no-cache-dir -r requirements.txt

# 復(fù)制應(yīng)用代碼
COPY . .

# 創(chuàng)建非root用戶
RUN useradd --create-home --shell /bin/bash app
USER app

# 暴露端口
EXPOSE 8000

# 啟動命令
CMD ["python", "main.py"]
"""

# Docker Compose配置
DOCKER_COMPOSE_CONTENT = """
version: '3.8'

services:
  shortener-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=sqlite:///./shortener.db
      - HOST=0.0.0.0
      - PORT=8000
      - RELOAD=false
      - LOG_LEVEL=info
    volumes:
      - ./data:/app/data
    restart: unless-stopped

  # 可選：添加Redis用于緩存
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    restart: unless-stopped

  # 可選：添加Nginx用于反向代理和靜態(tài)文件
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - shortener-api
    restart: unless-stopped
"""

# Nginx配置
NGINX_CONFIG = """
events {
    worker_connections 1024;
}

http {
    upstream shortener_api {
        server shortener-api:8000;
    }

    server {
        listen 80;
        server_name localhost;

        # API請求轉(zhuǎn)發(fā)到FastAPI應(yīng)用
        location / {
            proxy_pass http://shortener_api;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
        }

        # 靜態(tài)文件服務(wù)（如果有）
        location /static/ {
            alias /app/static/;
            expires 1y;
            add_header Cache-Control "public, immutable";
        }
    }
}
"""

# 依賴文件
REQUIREMENTS_CONTENT = """
fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
pydantic==2.5.0
python-multipart==0.0.6
python-dotenv==1.0.0
redis==5.0.1
pytest==7.4.3
pytest-asyncio==0.21.1
requests==2.31.0
"""

def create_deployment_files():
    """創(chuàng)建部署文件"""
    print("?? 創(chuàng)建部署文件")
    print("=" * 50)
    
    files = {
        "Dockerfile": DOCKERFILE_CONTENT,
        "docker-compose.yml": DOCKER_COMPOSE_CONTENT,
        "nginx.conf": NGINX_CONFIG,
        "requirements.txt": REQUIREMENTS_CONTENT,
    }
    
    for filename, content in files.items():
        with open(filename, "w", encoding="utf-8") as f:
            f.write(content.strip())
        print(f"? 已創(chuàng)建: {filename}")
    
    # 創(chuàng)建數(shù)據(jù)目錄
    Path("data").mkdir(exist_ok=True)
    print("? 已創(chuàng)建: data/ 目錄")

def generate_environment_file():
    """生成生產(chǎn)環(huán)境配置文件"""
    env_content = """
# 生產(chǎn)環(huán)境配置

# 應(yīng)用設(shè)置
APP_NAME=短鏈接生成器服務(wù)
APP_VERSION=1.0.0

# 服務(wù)器設(shè)置
HOST=0.0.0.0
PORT=8000
RELOAD=false
LOG_LEVEL=info

# 數(shù)據(jù)庫設(shè)置
DATABASE_URL=sqlite:///./data/shortener.db

# 安全設(shè)置
SECRET_KEY=change-this-in-production-with-secure-random-key
TOKEN_EXPIRE_MINUTES=10080

# 業(yè)務(wù)設(shè)置
DEFAULT_SHORT_CODE_LENGTH=6
MAX_CUSTOM_CODE_LENGTH=10
MAX_RETRY_ATTEMPTS=5
DEFAULT_EXPIRY_DAYS=365

# 速率限制
RATE_LIMIT_REQUESTS=1000
RATE_LIMIT_MINUTES=1

# Redis緩存（可選）
REDIS_URL=redis://redis:6379/0
"""
    
    with open(".env.production", "w", encoding="utf-8") as f:
        f.write(env_content.strip())
    
    print("? 已創(chuàng)建: .env.production")

def deployment_commands():
    """顯示部署命令"""
    commands = {
        "開發(fā)模式運行": "python main.py",
        "生產(chǎn)模式運行": "uvicorn main:app --host 0.0.0.0 --port 8000",
        "Docker構(gòu)建": "docker build -t shortener-service .",
        "Docker運行": "docker run -p 8000:8000 shortener-service",
        "Docker Compose啟動": "docker-compose up -d",
        "Docker Compose停止": "docker-compose down",
        "查看日志": "docker-compose logs -f",
    }
    
    print("\n?? 部署命令參考:")
    print("=" * 50)
    for description, command in commands.items():
        print(f"{description}:")
        print(f"  $ {command}")

def health_check_script():
    """健康檢查腳本"""
    script_content = """#!/bin/bash

# 健康檢查腳本
# 用于Docker健康檢查或監(jiān)控

URL="http://localhost:8000/"

response=$(curl -s -o /dev/null -w "%{http_code}" $URL)

if [ $response -eq 200 ]; then
    echo "? 服務(wù)健康狀態(tài): 正常"
    exit 0
else
    echo "? 服務(wù)健康狀態(tài): 異常 (HTTP $response)"
    exit 1
fi
"""
    
    with open("healthcheck.sh", "w", encoding="utf-8") as f:
        f.write(script_content)
    
    # 設(shè)置執(zhí)行權(quán)限
    os.chmod("healthcheck.sh", 0o755)
    print("? 已創(chuàng)建: healthcheck.sh")

def main():
    """主函數(shù)"""
    print("短鏈接服務(wù)部署配置")
    print("=" * 50)
    
    create_deployment_files()
    generate_environment_file()
    health_check_script()
    deployment_commands()
    
    print("\n?? 部署配置完成!")
    print("接下來可以:")
    print("  1. 使用 'docker-compose up -d' 啟動服務(wù)")
    print("  2. 訪問 http://localhost:8000/docs 查看API文檔")
    print("  3. 修改 .env.production 配置生產(chǎn)環(huán)境參數(shù)")

if __name__ == "__main__":
    main()

# 未來擴展方向
extension_ideas = {
    "多租戶支持": "為不同組織提供獨立的短鏈接空間",
    "高級分析": "實時儀表板、地理分布分析",
    "批量操作": "批量創(chuàng)建和管理短鏈接",
    "自定義域名": "支持用戶綁定自己的域名",
    "QR碼生成": "自動生成短鏈接的QR碼",
    "鏈接預(yù)覽": "生成鏈接的預(yù)覽信息",
    "A/B測試": "為同一目標(biāo)URL創(chuàng)建多個短鏈接進行測試"
}