api_server.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import json
import asyncio
from typing import Dict, List, Optional, Any
from datetime import datetime

from fastapi import FastAPI, HTTPException, BackgroundTasks, Query, Depends, UploadFile, File, Body, Header
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from pydantic import BaseModel, Field
import uvicorn

from unibo_jetbrains_activation import (
    DatabaseManager, JetbrainsSubmitter, LinkExtractor, ProcessController,
    STATUS_PENDING, STATUS_SUBMITTED, STATUS_FAILED, STATUS_LINK_EXTRACTED
)
from loguru import logger

# 配置日志
logger.remove()
logger.add("api_server.log", rotation="1 MB", level="INFO")
logger.add(lambda msg: print(msg, end=""), level="INFO")

# 创建FastAPI应用
app = FastAPI(
    title="JetBrains激活链接管理系统",
    description="管理JetBrains激活链接的获取流程，包括提交邮箱和提取链接",
    version="1.0.0"
)

# 允许跨域请求
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 在生产环境中应该替换为实际的前端域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 创建全局的ProcessController实例
controller = ProcessController()

# 定义环境变量
ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "admin123")
# API_TOKEN = os.environ.get("API_TOKEN", "unibo_token")

def verify_token(token: str = Header(None)):
    if token != ADMIN_PASSWORD:
        raise HTTPException(status_code=401, detail="无效的token")

# 定义请求和响应模型
class AccountItem(BaseModel):
    id: Optional[int] = None
    register_time: str
    username: str
    password: str
    security_email: Optional[str] = None
    status: int = 0
    activation_link: Optional[str] = None
    notes: Optional[str] = None
    updated_at: Optional[str] = None
    used: Optional[int] = 0

class AccountsResponse(BaseModel):
    accounts: List[AccountItem]
    total: int

class SubmitRequest(BaseModel):
    account_ids: List[int] = []
    max_workers: int = 3
    proxy: Optional[str] = None

class ExtractRequest(BaseModel):
    account_ids: List[int] = []
    max_workers: int = 1

class ProcessResponse(BaseModel):
    success_count: int
    error_count: int
    details: Optional[Dict[str, Any]] = None

class ImportRequest(BaseModel):
    filepath: str

class ImportResponse(BaseModel):
    imported_count: int
    total_accounts: int

class ExportRequest(BaseModel):
    status: Optional[int] = None
    filename: Optional[str] = "export_accounts.txt"

class ExportResponse(BaseModel):
    exported_count: int
    filepath: str

class StatusCounts(BaseModel):
    total: int = 0
    pending: int = 0
    submitted: int = 0
    failed: int = 0
    link_extracted: int = 0
    unused: int = 0  # 新增

class StatusUpdateRequest(BaseModel):
    status: int
    notes: Optional[str] = None
    activation_link: Optional[str] = None

class BatchToggleUsedRequest(BaseModel):
    ids: list[int]
    used: int

# 后台任务
running_tasks = {}

async def process_submit_task(task_id: str, account_ids: List[int], max_workers: int, proxy: Optional[str] = None):
    """后台提交邮箱任务（并发版）"""
    try:
        # 获取要处理的账号
        accounts = []
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            if account_ids:
                # 处理指定的账号
                for account_id in account_ids:
                    cursor.execute(
                        "SELECT id, register_time, username, password, security_email, status, activation_link FROM accounts WHERE id = ?",
                        (account_id,)
                    )
                    row = cursor.fetchone()
                    if row:
                        accounts.append({
                            'id': row[0],
                            'register_time': row[1],
                            'username': row[2],
                            'password': row[3],
                            'security_email': row[4],
                            'status': row[5],
                            'activation_link': row[6]
                        })
            else:
                # 处理所有未提交的账号
                cursor.execute(
                    "SELECT id, register_time, username, password, security_email, status, activation_link FROM accounts WHERE status = ?",
                    (STATUS_PENDING,)
                )
                for row in cursor.fetchall():
                    accounts.append({
                        'id': row[0],
                        'register_time': row[1],
                        'username': row[2],
                        'password': row[3],
                        'security_email': row[4],
                        'status': row[5],
                        'activation_link': row[6]
                    })
        
        # 设置代理（如果提供）
        if proxy:
            submitter = JetbrainsSubmitter(controller.db_manager, proxy)
        else:
            submitter = controller.submitter

        success_count = 0
        error_count = 0
        results = []

        sem = asyncio.Semaphore(max_workers)

        async def handle_account(account):
            nonlocal success_count, error_count
            try:
                async with sem:
                    controller.db_manager.log_operation(
                        account['username'],
                        "submit_email_background",
                        "processing",
                        f"任务ID: {task_id}"
                    )
                    # 提交邮箱（同步方法用run_in_executor）
                    loop = asyncio.get_running_loop()
                    result = await loop.run_in_executor(None, submitter.submit_email, account)
                    if result:
                        success_count += 1
                    else:
                        error_count += 1
                    await asyncio.sleep(1)
            except Exception as e:
                error_count += 1
                error_msg = str(e)
                controller.db_manager.update_account_status(
                    account['id'],
                    STATUS_FAILED,
                    notes=f"后台任务异常: {error_msg[:100]}"
                )
                controller.db_manager.log_operation(
                    account['username'],
                    "submit_email_background",
                    "error",
                    error_msg[:200]
                )

        tasks_list = [handle_account(account) for account in accounts]
        await asyncio.gather(*tasks_list)

        running_tasks[task_id] = {
            "status": "completed",
            "success_count": success_count,
            "error_count": error_count,
            "total": len(accounts)
        }
    except Exception as e:
        logger.error(f"后台提交任务 {task_id} 执行失败: {str(e)}")
        running_tasks[task_id] = {
            "status": "failed",
            "error": str(e)
        }

async def process_extract_task(task_id: str, account_ids: List[int], max_workers: int):
    """后台提取链接任务（并发版）"""
    try:
        # 获取要处理的账号
        accounts = []
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            if account_ids:
                # 处理指定的账号
                for account_id in account_ids:
                    cursor.execute(
                        "SELECT id, register_time, username, password, security_email, status, activation_link FROM accounts WHERE id = ?",
                        (account_id,)
                    )
                    row = cursor.fetchone()
                    if row and row[5] == STATUS_SUBMITTED:  # 仅处理已提交的账号
                        accounts.append({
                            'id': row[0],
                            'register_time': row[1],
                            'username': row[2],
                            'password': row[3],
                            'security_email': row[4],
                            'status': row[5],
                            'activation_link': row[6]
                        })
            else:
                # 处理所有已提交但未提取链接的账号
                cursor.execute(
                    "SELECT id, register_time, username, password, security_email, status, activation_link FROM accounts WHERE status = ? AND (activation_link IS NULL OR activation_link = '')",
                    (STATUS_SUBMITTED,)
                )
                for row in cursor.fetchall():
                    accounts.append({
                        'id': row[0],
                        'register_time': row[1],
                        'username': row[2],
                        'password': row[3],
                        'security_email': row[4],
                        'status': row[5],
                        'activation_link': row[6]
                    })
        
        # 使用LinkExtractor提取链接
        extractor = controller.extractor
        success_count = 0
        error_count = 0
        sem = asyncio.Semaphore(max_workers)

        async def handle_account(account):
            nonlocal success_count, error_count
            try:
                async with sem:
                    controller.db_manager.log_operation(
                        account['username'],
                        "extract_link_background",
                        "processing",
                        f"任务ID: {task_id}"
                    )
                    loop = asyncio.get_running_loop()
                    link = await loop.run_in_executor(None, extractor.extract_link, account)
                    if link:
                        success_count += 1
                    else:
                        error_count += 1
                    await asyncio.sleep(1)
            except Exception as e:
                error_count += 1
                error_msg = str(e)
                controller.db_manager.update_account_status(
                    account['id'],
                    STATUS_SUBMITTED,
                    notes=f"后台任务异常: {error_msg[:100]}"
                )
                controller.db_manager.log_operation(
                    account['username'],
                    "extract_link_background",
                    "error",
                    error_msg[:200]
                )

        tasks_list = [handle_account(account) for account in accounts]
        await asyncio.gather(*tasks_list)

        running_tasks[task_id] = {
            "status": "completed",
            "success_count": success_count,
            "error_count": error_count,
            "total": len(accounts)
        }
    except Exception as e:
        logger.error(f"后台提取任务 {task_id} 执行失败: {str(e)}")
        running_tasks[task_id] = {
            "status": "failed",
            "error": str(e)
        }

# API路由
@app.get("/")
async def root():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    index_path = os.path.join(current_dir, "index.html")
    if os.path.exists(index_path):
        return FileResponse(index_path)
    return {"message": "index.html not found in the current directory"}

@app.get("/accounts", response_model=AccountsResponse, dependencies=[Depends(verify_token)])
async def get_accounts(
    status: Optional[int] = None,
    page: int = Query(1, ge=1),
    per_page: int = Query(10, ge=1, le=100),
    search: Optional[str] = None,
    register_time_start: Optional[str] = None,
    register_time_end: Optional[str] = None,
    id_min: Optional[int] = None,
    id_max: Optional[int] = None,
    has_activation_link: Optional[str] = None,
    used: Optional[int] = Query(None)
):
    """获取账号列表，支持多条件筛选"""
    offset = (page - 1) * per_page
    try:
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            query_conditions = []
            query_params = []
            if status is not None:
                query_conditions.append("status = ?")
                query_params.append(status)
            if search:
                query_conditions.append("(username LIKE ? OR security_email LIKE ? OR notes LIKE ? OR activation_link LIKE ?)")
                search_param = f"%{search}%"
                query_params.extend([search_param, search_param, search_param, search_param])
            if register_time_start:
                query_conditions.append("register_time >= ?")
                query_params.append(register_time_start)
            if register_time_end:
                query_conditions.append("register_time <= ?")
                query_params.append(register_time_end)
            if id_min:
                query_conditions.append("id >= ?")
                query_params.append(id_min)
            if id_max:
                query_conditions.append("id <= ?")
                query_params.append(id_max)
            if has_activation_link == 'yes':
                query_conditions.append("activation_link IS NOT NULL AND activation_link != ''")
            elif has_activation_link == 'no':
                query_conditions.append("(activation_link IS NULL OR activation_link = '')")
            if used is not None and used != '':
                query_conditions.append("used = ?")
                query_params.append(int(used))
            if query_conditions:
                where_clause = "WHERE " + " AND ".join(query_conditions)
            else:
                where_clause = ""
            count_query = f"SELECT COUNT(*) FROM accounts {where_clause}"
            cursor.execute(count_query, query_params)
            total = cursor.fetchone()[0]
            data_query = f"""
            SELECT id, register_time, username, password, security_email, status, activation_link, updated_at, notes, used 
            FROM accounts 
            {where_clause}
            ORDER BY id DESC
            LIMIT ? OFFSET ?
            """
            query_params_data = query_params + [per_page, offset]
            cursor.execute(data_query, query_params_data)
            accounts = []
            for row in cursor.fetchall():
                accounts.append(AccountItem(
                    id=row[0],
                    register_time=row[1],
                    username=row[2],
                    password=row[3],
                    security_email=row[4],
                    status=row[5],
                    activation_link=row[6],
                    updated_at=row[7],
                    notes=row[8],
                    used=row[9] if len(row) > 9 else 0
                ))
            return AccountsResponse(accounts=accounts, total=total)
    except Exception as e:
        logger.error(f"获取账号列表出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"获取账号列表出错: {str(e)}")

@app.get("/accounts/{account_id}", response_model=AccountItem, dependencies=[Depends(verify_token)])
async def get_account(account_id: int):
    """获取单个账号详情"""
    try:
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            cursor.execute(
                "SELECT id, register_time, username, password, security_email, status, activation_link, updated_at, notes FROM accounts WHERE id = ?",
                (account_id,)
            )
            row = cursor.fetchone()
            
            if not row:
                raise HTTPException(status_code=404, detail=f"未找到ID为{account_id}的账号")
                
            return AccountItem(
                id=row[0],
                register_time=row[1],
                username=row[2],
                password=row[3],
                security_email=row[4],
                status=row[5],
                activation_link=row[6],
                updated_at=row[7],
                notes=row[8]
            )
            
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"获取账号详情出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"获取账号详情出错: {str(e)}")

@app.post("/accounts/{account_id}/status", response_model=AccountItem, dependencies=[Depends(verify_token)])
async def update_account_status(account_id: int, status_update: StatusUpdateRequest):
    """更新账号状态"""
    try:
        username = None
        updated_account = None
        
        # 使用锁进行数据库操作
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            
            # 先检查账号是否存在
            cursor.execute("SELECT username FROM accounts WHERE id = ?", (account_id,))
            result = cursor.fetchone()
            if not result:
                raise HTTPException(status_code=404, detail=f"未找到ID为{account_id}的账号")
            
            username = result[0]
            
            # 直接在锁内执行更新操作，避免嵌套锁
            if status_update.activation_link and status_update.notes:
                cursor.execute('''
                UPDATE accounts 
                SET status = ?, activation_link = ?, notes = ?, updated_at = datetime('now', 'localtime')
                WHERE id = ?
                ''', (status_update.status, status_update.activation_link, status_update.notes, account_id))
            elif status_update.activation_link:
                cursor.execute('''
                UPDATE accounts 
                SET status = ?, activation_link = ?, updated_at = datetime('now', 'localtime')
                WHERE id = ?
                ''', (status_update.status, status_update.activation_link, account_id))
            elif status_update.notes:
                cursor.execute('''
                UPDATE accounts 
                SET status = ?, notes = ?, updated_at = datetime('now', 'localtime')
                WHERE id = ?
                ''', (status_update.status, status_update.notes, account_id))
            else:
                cursor.execute('''
                UPDATE accounts 
                SET status = ?, updated_at = datetime('now', 'localtime')
                WHERE id = ?
                ''', (status_update.status, account_id))
            
            # 提交事务
            controller.db_manager.conn.commit()
            
            # 返回更新后的账号
            cursor.execute(
                "SELECT id, register_time, username, password, security_email, status, activation_link, updated_at, notes FROM accounts WHERE id = ?",
                (account_id,)
            )
            row = cursor.fetchone()
            
            updated_account = AccountItem(
                id=row[0],
                register_time=row[1],
                username=row[2],
                password=row[3],
                security_email=row[4],
                status=row[5],
                activation_link=row[6],
                updated_at=row[7],
                notes=row[8]
            )
        
        # 锁外记录操作日志，避免死锁
        if username:
            controller.db_manager.log_operation(
                username, 
                "update_status", 
                "success", 
                f"手动更新状态为: {status_update.status}"
            )
            
        return updated_account
            
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"更新账号状态出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"更新账号状态出错: {str(e)}")

@app.post("/accounts/{account_id}/toggle-used", dependencies=[Depends(verify_token)])
async def toggle_account_used(account_id: int):
    """切换账号的used状态"""
    try:
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            cursor.execute("SELECT used FROM accounts WHERE id = ?", (account_id,))
            row = cursor.fetchone()
            if not row:
                raise HTTPException(status_code=404, detail=f"未找到ID为{account_id}的账号")
            current = row[0] or 0
            new_value = 0 if current else 1
            cursor.execute("UPDATE accounts SET used = ?, updated_at = datetime('now', 'localtime') WHERE id = ?", (new_value, account_id))
            controller.db_manager.conn.commit()
        return {"id": account_id, "used": new_value}
    except Exception as e:
        logger.error(f"切换账号used状态出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"切换账号used状态出错: {str(e)}")

@app.post("/accounts/batch-toggle-used", dependencies=[Depends(verify_token)])
async def batch_toggle_used(req: BatchToggleUsedRequest = Body(...)):
    """批量修改账号的used状态"""
    try:
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            if not req.ids:
                raise HTTPException(status_code=400, detail="未指定账号ID")
            cursor.execute(f"UPDATE accounts SET used = ?, updated_at = datetime('now', 'localtime') WHERE id IN ({','.join(['?']*len(req.ids))})", [req.used, *req.ids])
            controller.db_manager.conn.commit()
        return {"updated": cursor.rowcount}
    except Exception as e:
        logger.error(f"批量切换账号used状态出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"批量切换账号used状态出错: {str(e)}")

@app.delete("/accounts/{account_id}", dependencies=[Depends(verify_token)])
async def delete_account(account_id: int):
    logger.info(f"收到删除账号请求: {account_id}")
    try:
        with controller.db_manager.lock:
            logger.info(f"已获取数据库锁，准备查询账号: {account_id}")
            cursor = controller.db_manager.conn.cursor()
            cursor.execute("SELECT username FROM accounts WHERE id = ?", (account_id,))
            row = cursor.fetchone()
            if not row:
                logger.warning(f"未找到账号: {account_id}")
                raise HTTPException(status_code=404, detail=f"未找到ID为{account_id}的账号")
            username = row[0]
            logger.info(f"准备删除账号: {account_id}, 用户名: {username}")
            cursor.execute("DELETE FROM accounts WHERE id = ?", (account_id,))
            controller.db_manager.conn.commit()
            logger.info(f"账号已删除: {account_id}")
        # 注意：操作日志写入要放到锁外，避免死锁
        controller.db_manager.log_operation(username, "delete_account", "success", f"删除账号ID: {account_id}")
        logger.info(f"删除账号流程结束: {account_id}")
        return {"message": f"账号 {account_id} 已删除"}
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"删除账号出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"删除账号出错: {str(e)}")

@app.post("/submit", response_model=Dict, dependencies=[Depends(verify_token)])
async def submit_emails(request: SubmitRequest, background_tasks: BackgroundTasks):
    """提交邮箱到JetBrains（异步任务）"""
    try:
        task_id = f"submit_{datetime.now().strftime('%Y%m%d%H%M%S')}_{len(running_tasks) + 1}"
        
        # 设置初始任务状态
        running_tasks[task_id] = {
            "status": "running",
            "type": "submit",
            "start_time": datetime.now().isoformat(),
            "account_ids": request.account_ids,
            "max_workers": request.max_workers
        }
        
        # 启动后台任务
        background_tasks.add_task(
            process_submit_task,
            task_id=task_id,
            account_ids=request.account_ids,
            max_workers=request.max_workers,
            proxy=request.proxy
        )
        
        return {
            "task_id": task_id,
            "message": "邮箱提交任务已启动",
            "status": "running"
        }
        
    except Exception as e:
        logger.error(f"启动提交任务出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"启动提交任务出错: {str(e)}")

@app.post("/extract", response_model=Dict, dependencies=[Depends(verify_token)])
async def extract_links(request: ExtractRequest, background_tasks: BackgroundTasks):
    """从邮箱提取激活链接（异步任务）"""
    try:
        task_id = f"extract_{datetime.now().strftime('%Y%m%d%H%M%S')}_{len(running_tasks) + 1}"
        
        # 设置初始任务状态
        running_tasks[task_id] = {
            "status": "running",
            "type": "extract",
            "start_time": datetime.now().isoformat(),
            "account_ids": request.account_ids,
            "max_workers": request.max_workers
        }
        
        # 启动后台任务
        background_tasks.add_task(
            process_extract_task,
            task_id=task_id,
            account_ids=request.account_ids,
            max_workers=request.max_workers
        )
        
        return {
            "task_id": task_id,
            "message": "链接提取任务已启动",
            "status": "running"
        }
        
    except Exception as e:
        logger.error(f"启动提取任务出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"启动提取任务出错: {str(e)}")

@app.get("/tasks/{task_id}", dependencies=[Depends(verify_token)])
async def get_task_status(task_id: str):
    """获取任务状态"""
    if task_id not in running_tasks:
        raise HTTPException(status_code=404, detail=f"未找到任务ID: {task_id}")
    
    return running_tasks[task_id]

@app.get("/tasks", dependencies=[Depends(verify_token)])
async def get_all_tasks():
    """获取所有任务状态"""
    return running_tasks

@app.post("/import", response_model=ImportResponse, dependencies=[Depends(verify_token)])
async def import_accounts(file: UploadFile = File(...)):
    """从上传的文件导入账号"""
    try:
        content = await file.read()
        # 假设文件是utf-8编码文本
        text = content.decode("utf-8")
        # 将内容写入临时文件，再用原有的import_data逻辑导入
        import tempfile
        with tempfile.NamedTemporaryFile(delete=False, mode="w", encoding="utf-8", suffix=".txt") as tmp:
            tmp.write(text)
            tmp_path = tmp.name
        imported_count = controller.import_data(tmp_path)
        # 获取总账号数
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            cursor.execute("SELECT COUNT(*) FROM accounts")
            total = cursor.fetchone()[0]
        # 删除临时文件
        import os
        os.remove(tmp_path)
        return ImportResponse(
            imported_count=imported_count,
            total_accounts=total
        )
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"导入账号出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"导入账号出错: {str(e)}")

@app.post("/export", dependencies=[Depends(verify_token)])
async def export_accounts(request: dict = Body(...)):
    """导出账号到文件并下载到本地，支持所有筛选条件"""
    try:
        import io
        buffer = io.StringIO()
        # 动态拼接筛选条件
        query_conditions = []
        query_params = []
        status = request.get('status', None)
        used = request.get('used', None)
        search = request.get('search', None)
        has_activation_link = request.get('has_activation_link', None)
        id_min = request.get('id_min', None)
        id_max = request.get('id_max', None)
        register_time_start = request.get('register_time_start', None)
        register_time_end = request.get('register_time_end', None)
        if status is not None and status != '':
            query_conditions.append("status = ?")
            query_params.append(status)
        if search:
            query_conditions.append("(username LIKE ? OR security_email LIKE ? OR notes LIKE ? OR activation_link LIKE ?)")
            search_param = f"%{search}%"
            query_params.extend([search_param, search_param, search_param, search_param])
        if register_time_start:
            query_conditions.append("register_time >= ?")
            query_params.append(register_time_start)
        if register_time_end:
            query_conditions.append("register_time <= ?")
            query_params.append(register_time_end)
        if id_min:
            query_conditions.append("id >= ?")
            query_params.append(id_min)
        if id_max:
            query_conditions.append("id <= ?")
            query_params.append(id_max)
        if has_activation_link == 'yes':
            query_conditions.append("activation_link IS NOT NULL AND activation_link != ''")
        elif has_activation_link == 'no':
            query_conditions.append("(activation_link IS NULL OR activation_link = '')")
        if used is not None and used != '':
            query_conditions.append("used = ?")
            query_params.append(int(used))
        if query_conditions:
            where_clause = "WHERE " + " AND ".join(query_conditions)
        else:
            where_clause = ""
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            cursor.execute(f"""
                SELECT activation_link 
                FROM accounts 
                {where_clause}
                ORDER BY id DESC
            """, query_params)
            for row in cursor.fetchall():
                if row[0]:
                    buffer.write(f"{row[0]}\n")
        buffer.seek(0)
        content = buffer.getvalue().encode('utf-8')
        byte_io = io.BytesIO(content)
        filename = request.get('filename', 'export_accounts.txt')
        headers = {
            'Content-Disposition': f'attachment; filename="{filename}"'
        }
        return StreamingResponse(
            byte_io, 
            media_type="text/plain",
            headers=headers
        )
    except Exception as e:
        logger.error(f"导出账号出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"导出账号出错: {str(e)}")

@app.get("/statistics", response_model=StatusCounts, dependencies=[Depends(verify_token)])
async def get_statistics():
    """获取账号统计信息"""
    try:
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            
            # 获取总账号数
            cursor.execute("SELECT COUNT(*) FROM accounts")
            total = cursor.fetchone()[0]
            
            # 获取各状态账号数
            cursor.execute("SELECT COUNT(*) FROM accounts WHERE status = ?", (STATUS_PENDING,))
            pending = cursor.fetchone()[0]
            
            cursor.execute("SELECT COUNT(*) FROM accounts WHERE status = ?", (STATUS_SUBMITTED,))
            submitted = cursor.fetchone()[0]
            
            cursor.execute("SELECT COUNT(*) FROM accounts WHERE status = ?", (STATUS_FAILED,))
            failed = cursor.fetchone()[0]
            
            cursor.execute("SELECT COUNT(*) FROM accounts WHERE status = ?", (STATUS_LINK_EXTRACTED,))
            link_extracted = cursor.fetchone()[0]
            
            cursor.execute("SELECT COUNT(*) FROM accounts WHERE used = 0")
            unused = cursor.fetchone()[0]
            
            return StatusCounts(
                total=total,
                pending=pending,
                submitted=submitted,
                failed=failed,
                link_extracted=link_extracted,
                unused=unused
            )
            
    except Exception as e:
        logger.error(f"获取统计信息出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"获取统计信息出错: {str(e)}")

@app.post("/reset-failed", dependencies=[Depends(verify_token)])
async def reset_failed_accounts():
    """将失败的账号状态重置为未提交状态"""
    try:
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            cursor.execute(
                "UPDATE accounts SET status = ?, notes = 'Reset from failed status', updated_at = datetime('now', 'localtime') WHERE status = ?",
                (STATUS_PENDING, STATUS_FAILED)
            )
            controller.db_manager.conn.commit()
            
            # 获取重置的账号数
            reset_count = cursor.rowcount
            
            return {"reset_count": reset_count, "message": f"已将{reset_count}个失败账号重置为未提交状态"}
            
    except Exception as e:
        logger.error(f"重置失败账号出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"重置失败账号出错: {str(e)}")

@app.get("/logs", dependencies=[Depends(verify_token)])
async def get_logs(
    username: Optional[str] = None,
    operation: Optional[str] = None,
    status: Optional[str] = None,
    page: int = Query(1, ge=1),
    per_page: int = Query(20, ge=1, le=100)
):
    """获取操作日志"""
    offset = (page - 1) * per_page
    
    try:
        with controller.db_manager.lock:
            cursor = controller.db_manager.conn.cursor()
            
            # 构建查询条件
            query_conditions = []
            query_params = []
            
            if username:
                query_conditions.append("username LIKE ?")
                query_params.append(f"%{username}%")
            
            if operation:
                query_conditions.append("operation = ?")
                query_params.append(operation)
            
            if status:
                query_conditions.append("status = ?")
                query_params.append(status)
            
            # 构建完整查询
            if query_conditions:
                where_clause = "WHERE " + " AND ".join(query_conditions)
            else:
                where_clause = ""
            
            # 计算总记录数
            count_query = f"SELECT COUNT(*) FROM operation_logs {where_clause}"
            cursor.execute(count_query, query_params)
            total = cursor.fetchone()[0]
            
            # 获取分页数据
            data_query = f"""
            SELECT id, username, operation, status, message, created_at 
            FROM operation_logs 
            {where_clause}
            ORDER BY id DESC
            LIMIT ? OFFSET ?
            """
            
            # 添加分页参数
            query_params.extend([per_page, offset])
            cursor.execute(data_query, query_params)
            
            logs = []
            for row in cursor.fetchall():
                logs.append({
                    "id": row[0],
                    "username": row[1],
                    "operation": row[2],
                    "status": row[3],
                    "message": row[4],
                    "created_at": row[5]
                })
            
            return {"logs": logs, "total": total}
            
    except Exception as e:
        logger.error(f"获取操作日志出错: {str(e)}")
        raise HTTPException(status_code=500, detail=f"获取操作日志出错: {str(e)}")

@app.post("/login")
async def login(data: dict = Body(...)):
    password = data.get("password", "")
    if password == ADMIN_PASSWORD:
        return {"token": ADMIN_PASSWORD}
    else:
        raise HTTPException(status_code=401, detail="密码错误")

if __name__ == "__main__":
    # 运行FastAPI应用
    uvicorn.run("api_server:app", host="0.0.0.0", port=8000, reload=True)