api_server.py

#!/usr/bin/env python3
"""
Xianyu AutoAgent - REST API Server
提供 Web 管理后台的后端 API 服务
"""

import os
import sys
import json
import uuid
import time
import hashlib
import threading
from datetime import datetime, timedelta
from functools import wraps

from flask import Flask, request, jsonify
from flask_cors import CORS
from flask_socketio import SocketIO, emit, join_room, leave_room
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity
from loguru import logger
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 初始化 Flask 应用
app = Flask(__name__)
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'xianyu-secret-key-' + str(uuid.uuid4()))
app.config['JWT_SECRET_KEY'] = os.getenv('JWT_SECRET_KEY', 'jwt-secret-key-' + str(uuid.uuid4()))
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(days=int(os.getenv('JWT_EXPIRE_DAYS', 7)))

# 初始化扩展
CORS(app, resources={r"/api/*": {"origins": "*"}})
jwt = JWTManager(app)
socketio = SocketIO(app, cors_allowed_origins="*", async_mode='eventlet')

# ==================== 数据存储（内存 + 文件） ====================

class DataStore:
    """简单内存数据存储，生产环境应使用数据库"""
    
    def __init__(self):
        self.users = {
            'admin': {
                'id': 1,
                'username': 'admin',
                'password_hash': self._hash_password('123456'),
                'email': 'admin@example.com',
                'role': 'admin',
                'status': 1,
                'created_at': datetime.now().isoformat(),
                'last_login': None
            }
        }
        self.accounts = {}  # 闲鱼账号
        self.qr_sessions = {}  # 扫码登录会话
        self.logs = []  # 日志
        self.chat_sessions = {}  # 聊天会话
        self.config = {
            'model': {
                'apiKey': os.getenv('API_KEY', ''),
                'baseUrl': os.getenv('MODEL_BASE_URL', 'https://api.deepseek.com'),
                'modelName': os.getenv('MODEL_NAME', 'deepseek-chat'),
                'temperature': 0.7,
                'maxTokens': 1500
            },
            'behavior': {
                'simulateTyping': os.getenv('SIMULATE_HUMAN_TYPING', 'False').lower() == 'true',
                'typingDelay': 2,
                'toggleKeyword': os.getenv('TOGGLE_KEYWORDS', '。'),
                'filterSystemMsg': True,
                'ignoreIrrelevant': True
            },
            'notification': {
                'enableDingtalk': False,
                'dingtalkWebhook': '',
                'dingtalkSecret': '',
                'events': ['manual_takeover', 'system_error']
            }
        }
        self.prompts = self._load_prompts()
        self._next_id = 1
        
    def _hash_password(self, password):
        return hashlib.sha256(password.encode()).hexdigest()
    
    def _load_prompts(self):
        """从文件加载提示词模板"""
        prompts = {}
        prompt_dir = os.path.join(os.path.dirname(__file__), 'prompts')
        prompt_types = {
            'classify': 'classify_prompt.txt',
            'price': 'price_prompt.txt',
            'tech': 'tech_prompt.txt',
            'default': 'default_prompt.txt'
        }
        for key, filename in prompt_types.items():
            filepath = os.path.join(prompt_dir, filename)
            example_filepath = os.path.join(prompt_dir, filename.replace('.txt', '_example.txt'))
            
            # 优先读取用户自定义的，不存在则读取示例
            if os.path.exists(filepath):
                with open(filepath, 'r', encoding='utf-8') as f:
                    content = f.read()
            elif os.path.exists(example_filepath):
                with open(example_filepath, 'r', encoding='utf-8') as f:
                    content = f.read()
            else:
                content = ''
            
            prompts[key] = {
                'type': key,
                'content': content,
                'enabled': True,
                'updated_at': datetime.now().isoformat()
            }
        return prompts
    
    def save_prompt(self, prompt_type, content):
        """保存提示词到文件"""
        filename = f'{prompt_type}_prompt.txt'
        filepath = os.path.join(os.path.dirname(__file__), 'prompts', filename)
        
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(content)
        
        self.prompts[prompt_type]['content'] = content
        self.prompts[prompt_type]['updated_at'] = datetime.now().isoformat()
    
    def get_next_id(self):
        self._next_id += 1
        return self._next_id

# 全局数据存储
data_store = DataStore()

# ==================== 工具函数 ====================

def response_success(data=None, message='success'):
    return jsonify({'code': 200, 'message': message, 'data': data})

def response_error(message, code=400):
    return jsonify({'code': code, 'message': message, 'data': None}), code

# ==================== 认证 API ====================

@app.route('/api/auth/login', methods=['POST'])
def login():
    """用户登录"""
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    
    if not username or not password:
        return response_error('用户名和密码不能为空')
    
    user = data_store.users.get(username)
    if not user or user['password_hash'] != data_store._hash_password(password):
        return response_error('用户名或密码错误', 401)
    
    if user['status'] != 1:
        return response_error('账号已被禁用', 403)
    
    # 更新最后登录时间
    user['last_login'] = datetime.now().isoformat()
    
    # 生成 JWT token
    access_token = create_access_token(identity=username)
    
    return response_success({
        'token': access_token,
        'userInfo': {
            'id': user['id'],
            'username': user['username'],
            'email': user['email'],
            'role': user['role'],
            'avatar': user.get('avatar', '')
        }
    })

@app.route('/api/auth/logout', methods=['POST'])
@jwt_required()
def logout():
    """用户登出（前端清除 token 即可）"""
    return response_success(message='登出成功')

@app.route('/api/auth/profile', methods=['GET'])
@jwt_required()
def get_profile():
    """获取用户信息"""
    username = get_jwt_identity()
    user = data_store.users.get(username)
    if not user:
        return response_error('用户不存在', 404)
    
    return response_success({
        'id': user['id'],
        'username': user['username'],
        'email': user['email'],
        'role': user['role'],
        'avatar': user.get('avatar', '')
    })

# ==================== 账号管理 API ====================

@app.route('/api/accounts', methods=['GET'])
@jwt_required()
def get_accounts():
    """获取账号列表"""
    accounts = list(data_store.accounts.values())
    return response_success(accounts)

@app.route('/api/accounts', methods=['POST'])
@jwt_required()
def create_account():
    """创建账号"""
    data = request.get_json()
    
    account_id = data_store.get_next_id()
    account = {
        'id': account_id,
        'nickname': data.get('nickname', ''),
        'username': data.get('username', ''),
        'cookies': data.get('cookies', ''),
        'cookiesValid': True,
        'running': data.get('autoStart', False),
        'autoStart': data.get('autoStart', False),
        'remark': data.get('remark', ''),
        'todayMessages': 0,
        'todayReplies': 0,
        'todayDeals': 0,
        'lastActive': '-',
        'created_at': datetime.now().isoformat()
    }
    
    data_store.accounts[account_id] = account
    
    # 广播账号变更
    socketio.emit('account_updated', account, namespace='/admin')
    
    logger.info(f"创建账号: {account['nickname']}")
    return response_success(account)

@app.route('/api/accounts/<int:account_id>', methods=['PUT'])
@jwt_required()
def update_account(account_id):
    """更新账号"""
    if account_id not in data_store.accounts:
        return response_error('账号不存在', 404)
    
    data = request.get_json()
    account = data_store.accounts[account_id]
    
    account.update({
        'nickname': data.get('nickname', account['nickname']),
        'username': data.get('username', account['username']),
        'cookies': data.get('cookies', account['cookies']),
        'autoStart': data.get('autoStart', account['autoStart']),
        'remark': data.get('remark', account['remark'])
    })
    
    socketio.emit('account_updated', account, namespace='/admin')
    logger.info(f"更新账号: {account['nickname']}")
    return response_success(account)

@app.route('/api/accounts/<int:account_id>', methods=['DELETE'])
@jwt_required()
def delete_account(account_id):
    """删除账号"""
    if account_id not in data_store.accounts:
        return response_error('账号不存在', 404)
    
    account = data_store.accounts.pop(account_id)
    socketio.emit('account_deleted', {'id': account_id}, namespace='/admin')
    logger.info(f"删除账号: {account['nickname']}")
    return response_success(message='删除成功')

@app.route('/api/accounts/<int:account_id>/toggle', methods=['POST'])
@jwt_required()
def toggle_account(account_id):
    """切换账号运行状态"""
    if account_id not in data_store.accounts:
        return response_error('账号不存在', 404)
    
    account = data_store.accounts[account_id]
    account['running'] = not account['running']
    
    socketio.emit('account_updated', account, namespace='/admin')
    logger.info(f"切换账号状态: {account['nickname']} -> {account['running']}")
    return response_success(account)

# ==================== 扫码登录 API ====================

@app.route('/api/accounts/qrcode/generate', methods=['POST'])
@jwt_required()
def generate_qrcode():
    """生成扫码登录二维码"""
    try:
        import qrcode
        import qrcode.image.pil
        from io import BytesIO
        import base64
        
        # 生成唯一 token
        token = f"qr_{uuid.uuid4().hex[:16]}"
        
        # 创建二维码（使用淘宝登录 URL 作为示例）
        # 实际应调用闲鱼/淘宝的扫码登录接口
        qr_url = f"https://login.taobao.com/qrcode.htm?token={token}"
        
        # 生成二维码图片
        qr = qrcode.QRCode(version=1, box_size=10, border=2)
        qr.add_data(qr_url)
        qr.make(fit=True)
        
        img = qr.make_image(fill_color="black", back_color="white")
        buffer = BytesIO()
        img.save(buffer, format='PNG')
        img_str = base64.b64encode(buffer.getvalue()).decode()
        
        # 存储会话
        data_store.qr_sessions[token] = {
            'token': token,
            'status': 'waiting',  # waiting, scanned, confirmed, expired
            'created_at': time.time(),
            'expires_at': time.time() + 120,  # 2分钟过期
            'cookies': None,
            'account_info': None
        }
        
        return response_success({
            'token': token,
            'qr_url': f"data:image/png;base64,{img_str}",
            'expires_in': 120
        })
        
    except Exception as e:
        logger.error(f"生成二维码失败: {e}")
        return response_error('生成二维码失败')

@app.route('/api/accounts/qrcode/status', methods=['GET'])
@jwt_required()
def get_qrcode_status():
    """查询扫码状态"""
    token = request.args.get('token')
    
    if not token or token not in data_store.qr_sessions:
        return response_error('无效的 token', 404)
    
    session = data_store.qr_sessions[token]

    # 检查是否过期
    if time.time() > session['expires_at'] and session['status'] != 'confirmed':
        session['status'] = 'expired'

    # TODO: 接入真实的闲鱼/淘宝扫码登录 API
    # 当前仅返回状态，不模拟虚假数据

    return response_success({
        'token': token,
        'status': session['status'],
        'cookies': session['cookies'] if session['status'] == 'confirmed' else None,
        'account_info': session['account_info'] if session['status'] == 'confirmed' else None
    })

# ==================== 配置管理 API ====================

@app.route('/api/config', methods=['GET'])
@jwt_required()
def get_config():
    """获取系统配置"""
    # 隐藏 API Key 部分信息
    config = data_store.config.copy()
    if config['model']['apiKey']:
        key = config['model']['apiKey']
        config['model']['apiKey'] = key[:4] + '*' * (len(key) - 8) + key[-4:]
    return response_success(config)

@app.route('/api/config', methods=['PUT'])
@jwt_required()
def update_config():
    """更新系统配置"""
    data = request.get_json()
    
    if 'model' in data:
        data_store.config['model'].update(data['model'])
        # 更新环境变量
        if data['model'].get('apiKey'):
            os.environ['API_KEY'] = data['model']['apiKey']
        if data['model'].get('baseUrl'):
            os.environ['MODEL_BASE_URL'] = data['model']['baseUrl']
        if data['model'].get('modelName'):
            os.environ['MODEL_NAME'] = data['model']['modelName']
    
    if 'behavior' in data:
        data_store.config['behavior'].update(data['behavior'])
        if data['behavior'].get('toggleKeyword'):
            os.environ['TOGGLE_KEYWORDS'] = data['behavior']['toggleKeyword']
        os.environ['SIMULATE_HUMAN_TYPING'] = str(data['behavior'].get('simulateTyping', False))
    
    if 'notification' in data:
        data_store.config['notification'].update(data['notification'])
    
    socketio.emit('config_updated', data_store.config, namespace='/admin')
    logger.info("更新系统配置")
    return response_success(data_store.config)

@app.route('/api/config/test', methods=['POST'])
@jwt_required()
def test_model_connection():
    """测试模型连接"""
    data = request.get_json()
    
    try:
        from openai import OpenAI
        
        client = OpenAI(
            api_key=data.get('apiKey', os.getenv('API_KEY')),
            base_url=data.get('baseUrl', os.getenv('MODEL_BASE_URL'))
        )
        
        # 简单测试请求
        response = client.chat.completions.create(
            model=data.get('modelName', 'deepseek-chat'),
            messages=[{"role": "user", "content": "Hello"}],
            max_tokens=5
        )
        
        return response_success({'connected': True, 'message': '连接成功'})
        
    except Exception as e:
        logger.error(f"测试模型连接失败: {e}")
        return response_success({'connected': False, 'message': str(e)})

# ==================== 提示词管理 API ====================

@app.route('/api/prompts', methods=['GET'])
@jwt_required()
def get_prompts():
    """获取所有提示词"""
    return response_success(list(data_store.prompts.values()))

@app.route('/api/prompts/<prompt_type>', methods=['GET'])
@jwt_required()
def get_prompt(prompt_type):
    """获取单个提示词"""
    if prompt_type not in data_store.prompts:
        return response_error('提示词类型不存在', 404)
    return response_success(data_store.prompts[prompt_type])

@app.route('/api/prompts/<prompt_type>', methods=['PUT'])
@jwt_required()
def update_prompt(prompt_type):
    """更新提示词"""
    if prompt_type not in data_store.prompts:
        return response_error('提示词类型不存在', 404)
    
    data = request.get_json()
    content = data.get('content', '')
    
    data_store.save_prompt(prompt_type, content)
    
    socketio.emit('prompt_updated', data_store.prompts[prompt_type], namespace='/admin')
    logger.info(f"更新提示词: {prompt_type}")
    return response_success(data_store.prompts[prompt_type])

@app.route('/api/prompts/<prompt_type>/reset', methods=['POST'])
@jwt_required()
def reset_prompt(prompt_type):
    """重置提示词为默认"""
    if prompt_type not in data_store.prompts:
        return response_error('提示词类型不存在', 404)
    
    # 从示例文件恢复
    example_filepath = os.path.join(os.path.dirname(__file__), 'prompts', f'{prompt_type}_prompt_example.txt')
    if os.path.exists(example_filepath):
        with open(example_filepath, 'r', encoding='utf-8') as f:
            content = f.read()
        data_store.save_prompt(prompt_type, content)
    
    socketio.emit('prompt_updated', data_store.prompts[prompt_type], namespace='/admin')
    logger.info(f"重置提示词: {prompt_type}")
    return response_success(data_store.prompts[prompt_type])

# ==================== 日志管理 API ====================

@app.route('/api/logs', methods=['GET'])
@jwt_required()
def get_logs():
    """获取日志列表"""
    level = request.args.get('level', '')
    page = int(request.args.get('page', 1))
    page_size = int(request.args.get('pageSize', 50))
    
    logs = data_store.logs
    
    # 筛选
    if level:
        logs = [log for log in logs if log['level'] == level]
    
    # 分页
    total = len(logs)
    start = (page - 1) * page_size
    end = start + page_size
    logs_page = logs[start:end]
    
    return response_success({
        'list': logs_page,
        'total': total,
        'page': page,
        'pageSize': page_size
    })

@app.route('/api/logs', methods=['DELETE'])
@jwt_required()
def clear_logs():
    """清空日志"""
    data_store.logs.clear()
    socketio.emit('logs_cleared', {}, namespace='/admin')
    logger.info("清空日志")
    return response_success(message='日志已清空')

# ==================== 统计数据 API ====================

@app.route('/api/stats/dashboard', methods=['GET'])
@jwt_required()
def get_dashboard_stats():
    """获取仪表盘统计数据"""
    # 计算统计数据
    total_messages = sum(a.get('todayMessages', 0) for a in data_store.accounts.values())
    total_replies = sum(a.get('todayReplies', 0) for a in data_store.accounts.values())
    total_deals = sum(a.get('todayDeals', 0) for a in data_store.accounts.values())
    active_accounts = sum(1 for a in data_store.accounts.values() if a.get('running', False))
    
    return response_success({
        'todayMessages': total_messages,
        'todayReplies': total_replies,
        'todayDeals': total_deals,
        'activeAccounts': active_accounts,
        'totalAccounts': len(data_store.accounts),
        'aiReplyRate': 92 if total_messages > 0 else 0,
        'avgResponseTime': 1.2
    })

@app.route('/api/stats/realtime', methods=['GET'])
@jwt_required()
def get_realtime_stats():
    """获取实时统计数据"""
    return response_success({
        'running': True,
        'accounts': len(data_store.accounts),
        'messages': sum(a.get('todayMessages', 0) for a in data_store.accounts.values()),
        'online': True
    })

@app.route('/api/stats/trend', methods=['GET'])
@jwt_required()
def get_trend_stats():
    """获取趋势数据"""
    # 模拟趋势数据
    hours = ['00:00', '04:00', '08:00', '12:00', '16:00', '20:00', '24:00']
    return response_success({
        'hours': hours,
        'total': [120, 80, 250, 380, 420, 350, 200],
        'ai': [100, 70, 220, 340, 380, 310, 180],
        'manual': [20, 10, 30, 40, 40, 40, 20]
    })

# ==================== 聊天会话 API ====================

@app.route('/api/chat/sessions', methods=['GET'])
@jwt_required()
def get_chat_sessions():
    """获取聊天会话列表"""
    sessions = list(data_store.chat_sessions.values())
    return response_success(sessions)

@app.route('/api/chat/sessions/<session_id>/messages', methods=['GET'])
@jwt_required()
def get_chat_messages(session_id):
    """获取会话消息"""
    if session_id not in data_store.chat_sessions:
        return response_error('会话不存在', 404)
    
    return response_success(data_store.chat_sessions[session_id].get('messages', []))

@app.route('/api/chat/sessions/<session_id>/takeover', methods=['POST'])
@jwt_required()
def takeover_session(session_id):
    """接管会话"""
    if session_id not in data_store.chat_sessions:
        data_store.chat_sessions[session_id] = {'id': session_id, 'messages': [], 'status': 'manual'}
    
    data_store.chat_sessions[session_id]['status'] = 'manual'
    socketio.emit('session_takeover', {'sessionId': session_id}, namespace='/admin')
    logger.info(f"接管会话: {session_id}")
    return response_success({'sessionId': session_id, 'status': 'manual'})

@app.route('/api/chat/sessions/<session_id>/release', methods=['POST'])
@jwt_required()
def release_session(session_id):
    """释放会话"""
    if session_id not in data_store.chat_sessions:
        return response_error('会话不存在', 404)
    
    data_store.chat_sessions[session_id]['status'] = 'ai'
    socketio.emit('session_release', {'sessionId': session_id}, namespace='/admin')
    logger.info(f"释放会话: {session_id}")
    return response_success({'sessionId': session_id, 'status': 'ai'})

# ==================== WebSocket 事件 ====================

@socketio.on('connect', namespace='/admin')
def handle_connect():
    """管理员连接"""
    logger.info('管理员连接 WebSocket')
    emit('connected', {'message': 'Connected to admin namespace'})

@socketio.on('disconnect', namespace='/admin')
def handle_disconnect():
    """管理员断开"""
    logger.info('管理员断开 WebSocket')

@socketio.on('join_room', namespace='/admin')
def handle_join_room(data):
    """加入房间"""
    room = data.get('room', 'default')
    join_room(room)
    emit('joined', {'room': room})

@socketio.on('leave_room', namespace='/admin')
def handle_leave_room(data):
    """离开房间"""
    room = data.get('room', 'default')
    leave_room(room)
    emit('left', {'room': room})

# ==================== 日志 WebSocket 推送 ====================

def broadcast_log(level, module, message):
    """广播日志到所有连接的客户端"""
    log_entry = {
        'id': str(uuid.uuid4()),
        'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        'level': level,
        'module': module,
        'message': message
    }
    data_store.logs.append(log_entry)
    
    # 限制日志数量
    if len(data_store.logs) > 10000:
        data_store.logs = data_store.logs[-5000:]
    
    socketio.emit('new_log', log_entry, namespace='/admin')

# ==================== 主程序入口 ====================

if __name__ == '__main__':
    import os
    port = int(os.getenv('API_PORT', '8150'))
    
    logger.info("启动 API 服务器...")
    logger.info(f"API 地址: http://0.0.0.0:{port}")
    logger.info(f"WebSocket 地址: ws://0.0.0.0:{port}")
    
    socketio.run(app, host='0.0.0.0', port=port, debug=False)