military-tech-voice3.0/backend/app/routes/admin.py

from flask import Blueprint, request, jsonify
from app.db import get_db
from app.routes.auth import admin_required
from werkzeug.security import generate_password_hash
import secrets
import string

bp = Blueprint('admin', __name__, url_prefix='/api/admin')


def generate_invite_code():
    """生成8位邀请码（大写字母+数字，去掉易混淆字符 O/I/0/1）"""
    alphabet = (
        string.ascii_uppercase.replace('O', '').replace('I', '')
        + string.digits.replace('0', '').replace('1', '')
    )
    return ''.join(secrets.choice(alphabet) for _ in range(8))


# ========== 邀请码管理 ==========

@bp.route('/invites', methods=['GET'])
@admin_required
def list_invites():
    conn = get_db()
    invites = conn.execute('''
        SELECT ic.*, u.username as used_by_username
        FROM invite_codes ic
        LEFT JOIN users u ON ic.used_by_user_id = u.id
        ORDER BY ic.created_at DESC
    ''').fetchall()
    conn.close()
    return jsonify({'invites': [dict(row) for row in invites]})


@bp.route('/invites', methods=['POST'])
@admin_required
def create_invite():
    data = request.get_json()
    editor_name = data.get('editor_name', '').strip()
    if not editor_name:
        return jsonify({'error': '请填写编导姓名'}), 400

    code = generate_invite_code()
    conn = get_db()
    try:
        conn.execute(
            'INSERT INTO invite_codes (code, editor_name) VALUES (?, ?)',
            (code, editor_name)
        )
        conn.commit()
        return jsonify({'success': True, 'code': code, 'editor_name': editor_name})
    finally:
        conn.close()


@bp.route('/invites/<int:invite_id>', methods=['DELETE'])
@admin_required
def delete_invite(invite_id):
    conn = get_db()
    try:
        invite = conn.execute(
            'SELECT * FROM invite_codes WHERE id = ? AND is_used = 0', (invite_id,)
        ).fetchone()
        if not invite:
            return jsonify({'error': '邀请码不存在或已被使用，无法删除'}), 400
        conn.execute('DELETE FROM invite_codes WHERE id = ?', (invite_id,))
        conn.commit()
        return jsonify({'success': True})
    finally:
        conn.close()


# ========== 用户管理 ==========

@bp.route('/users', methods=['GET'])
@admin_required
def list_users():
    conn = get_db()
    users = conn.execute('''
        SELECT u.id, u.username, u.display_name, u.role, u.is_active, u.created_at,
               COUNT(ul.id) as usage_count,
               COALESCE(SUM(ul.text_length), 0) as total_chars
        FROM users u
        LEFT JOIN usage_logs ul ON u.id = ul.user_id
        GROUP BY u.id
        ORDER BY u.created_at DESC
    ''').fetchall()
    conn.close()
    return jsonify({'users': [dict(row) for row in users]})


@bp.route('/users/<int:user_id>', methods=['PUT'])
@admin_required
def update_user(user_id):
    data = request.get_json()
    conn = get_db()
    try:
        user = conn.execute('SELECT * FROM users WHERE id = ?', (user_id,)).fetchone()
        if not user:
            return jsonify({'error': '用户不存在'}), 404

        if 'is_active' in data:
            conn.execute(
                'UPDATE users SET is_active = ? WHERE id = ?',
                (1 if data['is_active'] else 0, user_id)
            )

        if 'password' in data and data['password']:
            conn.execute(
                'UPDATE users SET password_hash = ? WHERE id = ?',
                (generate_password_hash(data['password']), user_id)
            )

        if 'display_name' in data and data['display_name'].strip():
            conn.execute(
                'UPDATE users SET display_name = ? WHERE id = ?',
                (data['display_name'].strip(), user_id)
            )

        conn.commit()
        return jsonify({'success': True})
    finally:
        conn.close()


# ========== 用量日志 ==========

@bp.route('/logs', methods=['GET'])
@admin_required
def list_logs():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 50, type=int)
    user_id = request.args.get('user_id', type=int)
    date_from = request.args.get('date_from', '')
    date_to = request.args.get('date_to', '')

    conn = get_db()

    where_clauses = []
    params = []

    if user_id:
        where_clauses.append('user_id = ?')
        params.append(user_id)
    if date_from:
        where_clauses.append('created_at >= ?')
        params.append(date_from)
    if date_to:
        where_clauses.append('created_at <= ?')
        params.append(date_to + ' 23:59:59')

    where_sql = (' WHERE ' + ' AND '.join(where_clauses)) if where_clauses else ''

    total = conn.execute(
        f'SELECT COUNT(*) FROM usage_logs{where_sql}', params
    ).fetchone()[0]

    offset = (page - 1) * per_page
    logs = conn.execute(
        f'SELECT * FROM usage_logs{where_sql} ORDER BY created_at DESC LIMIT ? OFFSET ?',
        params + [per_page, offset]
    ).fetchall()
    conn.close()

    return jsonify({
        'logs': [dict(row) for row in logs],
        'total': total,
        'page': page,
        'per_page': per_page,
        'total_pages': (total + per_page - 1) // per_page
    })


@bp.route('/stats', methods=['GET'])
@admin_required
def stats():
    conn = get_db()
    user_stats = conn.execute('''
        SELECT username, display_name,
               COUNT(*) as call_count,
               SUM(text_length) as total_chars,
               MAX(created_at) as last_used
        FROM usage_logs
        GROUP BY user_id
        ORDER BY call_count DESC
    ''').fetchall()

    totals = conn.execute('''
        SELECT COUNT(*) as total_calls,
               COALESCE(SUM(text_length), 0) as total_chars,
               COUNT(DISTINCT user_id) as active_users
        FROM usage_logs
    ''').fetchone()

    conn.close()
    return jsonify({
        'user_stats': [dict(row) for row in user_stats],
        'totals': dict(totals)
    })