from flask import Flask, render_template, request, jsonify, flash, redirect, url_for, session from flask_sqlalchemy import SQLAlchemy import requests from bs4 import BeautifulSoup import os from dotenv import load_dotenv import random import string import pymysql import re import hashlib from functools import wraps from config import AD_TEMPLATE, SERVER_INFO, VERIFICATION_CONFIG, NAME_TYPES from sqlalchemy import create_engine # 设置PyMySQL作为MySQL驱动 pymysql.install_as_MySQLdb() load_dotenv() app = Flask(__name__) # 设置 Flask secret key app.secret_key = os.getenv('SECRET_KEY', 'dev_key_123') # 在生产环境中使用环境变量设置 # 数据库配置 database_url = os.getenv('DATABASE_URL') app.config['SQLALCHEMY_DATABASE_URI'] = database_url app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # 创建数据库连接 def create_database(): try: # 解析数据库URL url_parts = database_url.split('/') db_name = url_parts[-1] base_url = '/'.join(url_parts[:-1]) # 创建数据库引擎（不指定数据库名） engine = create_engine(base_url) # 连接并创建数据库 with engine.connect() as conn: conn.execute(f"CREATE DATABASE IF NOT EXISTS {db_name}") print(f"数据库 {db_name} 创建成功") except Exception as e: print(f"创建数据库失败：{str(e)}") db = SQLAlchemy(app) # 定义CDK模型 class CDKCode(db.Model): __tablename__ = '兑换码' ID = db.Column(db.Integer, primary_key=True) 兑换码 = db.Column(db.String(20), unique=True, nullable=False) 奖励模板ID = db.Column(db.Integer, nullable=False) # 定义已使用的CDK记录 class UsedCDK(db.Model): __tablename__ = 'used_cdks' id = db.Column(db.Integer, primary_key=True) character_name = db.Column(db.String(50), nullable=False) post_url = db.Column(db.String(255), nullable=False) cdk = db.Column(db.String(20), nullable=False) used_at = db.Column(db.DateTime, server_default=db.func.now()) # 定义管理员模型 class Admin(db.Model): __tablename__ = 'admins' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(50), unique=True, nullable=False) password_hash = db.Column(db.String(128), nullable=False) role = db.Column(db.String(20), default='user') # 'admin' 或 'user' created_at = db.Column(db.DateTime, server_default=db.func.now()) def set_password(self, password): self.password_hash = hashlib.sha256(password.encode()).hexdigest() def check_password(self, password): return self.password_hash == hashlib.sha256(password.encode()).hexdigest() def is_admin(self): return self.role == 'admin' # 定义数据库配置模型 class DatabaseConfig(db.Model): __tablename__ = 'database_configs' id = db.Column(db.Integer, primary_key=True) config_name = db.Column(db.String(50), nullable=False) config_value = db.Column(db.String(255), nullable=False) updated_at = db.Column(db.DateTime, server_default=db.func.now(), onupdate=db.func.now()) # 管理员登录装饰器 def admin_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'admin_id' not in session: flash('请先登录', 'danger') return redirect(url_for('admin_login')) # 检查是否为管理员 admin = Admin.query.get(session['admin_id']) if not admin or not admin.is_admin(): flash('您没有权限访问此页面', 'danger') return redirect(url_for('user_dashboard')) return f(*args, **kwargs) return decorated_function # 用户登录装饰器（包括管理员和普通用户） def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'admin_id' not in session: flash('请先登录', 'danger') return redirect(url_for('admin_login')) return f(*args, **kwargs) return decorated_function def verify_ad_content(content, character_name): # 清理内容中的空白字符 content = ' '.join(content.split()) # 构建预期的广告内容 expected_ad = AD_TEMPLATE.format( server_name=SERVER_INFO["server_name"], name_type=NAME_TYPES["character"], name=character_name, server_features=SERVER_INFO["server_features"], website=SERVER_INFO["website"] ) # 清理预期内容中的空白字符 expected_ad = ' '.join(expected_ad.split()) # 直接比较核心内容 core_content = re.sub(r'\s+', '', content) core_expected = re.sub(r'\s+', '', expected_ad) # 检查核心内容是否包含在发帖内容中 if core_expected in core_content: return True # 如果完全匹配失败，进行更宽松的验证 for keyword in VERIFICATION_CONFIG['required_keywords']: if keyword not in content: return False for pattern in VERIFICATION_CONFIG['required_patterns']: if not re.search(pattern, content, re.DOTALL): return False if character_name not in content: return False return True @app.route('/') def index(): return render_template('index.html') @app.route('/generate') @login_required def generate(): # 获取URL参数中的角色名 character_name = request.args.get('character_name') # 获取当前用户的游戏账号 user_id = session.get('admin_id') user = Admin.query.get(user_id) # 获取用户的角色列表 characters = [] conn = get_auth_db_connection() if conn: try: account_table = get_db_config('auth_account_table', 'account') # 获取与用户名匹配的游戏账号 query = f"SELECT id FROM {account_table} WHERE username = %s" result = conn.execute(query, (user.username,)) account_ids = [row[0] for row in result] if account_ids: char_conn = get_char_db_connection() if char_conn: try: char_table = get_db_config('char_table', 'characters') placeholders = ', '.join(['%s'] * len(account_ids)) query = f""" SELECT guid, account, name, level, race, class, gender FROM {char_table} WHERE account IN ({placeholders}) ORDER BY level DESC """ result = char_conn.execute(query, account_ids) characters = [dict(row) for row in result] # 添加种族和职业的文本描述 for character in characters: character['race_text'] = get_race_name(character['race']) character['class_text'] = get_class_name(character['class']) except Exception as e: flash(f'获取角色列表失败: {str(e)}', 'danger') finally: char_conn.close() except Exception as e: flash(f'获取游戏账号失败: {str(e)}', 'danger') finally: conn.close() # 如果没有角色，显示提示信息 if not characters: flash('您需要先在游戏中创建角色才能生成宣传内容', 'warning') return render_template('generate.html', server_info=SERVER_INFO, characters=characters, selected_character=character_name) @app.route('/create-ad', methods=['POST']) @login_required def create_ad(): character_name = request.form.get('character_name') if not character_name: return jsonify({'success': False, 'message': '请选择角色'}) # 验证角色是否属于当前用户 user_id = session.get('admin_id') user = Admin.query.get(user_id) # 获取用户的游戏账号ID account_ids = [] conn = get_auth_db_connection() if conn: try: account_table = get_db_config('auth_account_table', 'account') query = f"SELECT id FROM {account_table} WHERE username = %s" result = conn.execute(query, (user.username,)) account_ids = [row[0] for row in result] finally: conn.close() # 验证角色是否属于用户 if account_ids: char_conn = get_char_db_connection() if char_conn: try: char_table = get_db_config('char_table', 'characters') placeholders = ', '.join(['%s'] * len(account_ids)) query = f""" SELECT guid, account, name, level, race, class FROM {char_table} WHERE name = %s AND account IN ({placeholders}) """ params = [character_name] + account_ids result = char_conn.execute(query, params) character = result.fetchone() if not character: return jsonify({'success': False, 'message': '您选择的角色不存在或不属于您'}) # 获取角色的种族和职业文本 race_text = get_race_name(character['race']) class_text = get_class_name(character['class']) # 生成宣传内容 ad_content = AD_TEMPLATE.format( server_name=SERVER_INFO['server_name'], name_type=f"推荐{race_text}{class_text}", name=character_name, server_features=SERVER_INFO['server_features'], website=SERVER_INFO['website'] ) return jsonify({ 'success': True, 'ad_content': ad_content }) finally: char_conn.close() return jsonify({'success': False, 'message': '无法验证角色信息'}) @app.route('/verify', methods=['POST']) def verify_post(): character_name = request.form.get('character_name') post_url = request.form.get('post_url') if not character_name or not post_url: return jsonify({'success': False, 'message': '请填写所有必要信息'}) # 检查该角色是否已经领取过CDK used = UsedCDK.query.filter_by(character_name=character_name).first() if used: return jsonify({'success': False, 'message': '该角色已经领取过CDK'}) try: # 获取帖子内容 response = requests.get(post_url) soup = BeautifulSoup(response.text, 'html.parser') # 尝试多种方式提取文本 post_content = soup.get_text() # 打印调试信息 print(f"URL: {post_url}") print(f"获取到的内容长度: {len(post_content)}") print(f"内容预览: {post_content[:200]}...") # 验证广告内容 if not verify_ad_content(post_content, character_name): # 尝试其他方法获取内容 # 例如尝试获取特定的div或段落 post_divs = soup.find_all(['div', 'p', 'article', 'section']) # 遍历所有可能的内容区块 for div in post_divs: div_text = div.get_text() if len(div_text) > 100: # 只检查长度合理的文本块 print(f"尝试验证区块: {div_text[:100]}...") if verify_ad_content(div_text, character_name): # 找到了有效的广告内容 post_content = div_text break else: # 如果所有区块都验证失败 return jsonify({'success': False, 'message': '未找到有效的宣传内容，请确保完整复制宣传模板或检查发帖网站兼容性'}) # 获取未使用的CDK cdk = CDKCode.query.filter( ~CDKCode.兑换码.in_( db.session.query(UsedCDK.cdk) ) ).first() if not cdk: return jsonify({'success': False, 'message': 'CDK已经用完，请联系管理员'}) # 记录使用情况 used_cdk = UsedCDK( character_name=character_name, post_url=post_url, cdk=cdk.兑换码 ) db.session.add(used_cdk) db.session.commit() return jsonify({ 'success': True, 'message': '验证成功！', 'cdk': cdk.兑换码 }) except Exception as e: print(f"验证过程中出错: {str(e)}") return jsonify({'success': False, 'message': f'验证失败：{str(e)}'}) @app.route('/redeem', methods=['GET', 'POST']) @login_required def redeem(): if request.method == 'POST': character_name = request.form.get('character_name') post_url = request.form.get('link') if not character_name or not post_url: flash('请填写所有必要信息', 'danger') return redirect(url_for('redeem')) # 验证角色是否属于当前用户 user_id = session.get('admin_id') user = Admin.query.get(user_id) # 获取用户的游戏账号ID account_ids = [] conn = get_auth_db_connection() if conn: try: account_table = get_db_config('auth_account_table', 'account') query = f"SELECT id FROM {account_table} WHERE username = %s" result = conn.execute(query, (user.username,)) account_ids = [row[0] for row in result] finally: conn.close() # 验证角色是否属于用户 character_exists = False if account_ids: char_conn = get_char_db_connection() if char_conn: try: char_table = get_db_config('char_table', 'characters') placeholders = ', '.join(['%s'] * len(account_ids)) query = f""" SELECT guid, account, name FROM {char_table} WHERE name = %s AND account IN ({placeholders}) """ params = [character_name] + account_ids result = char_conn.execute(query, params) character = result.fetchone() if character: character_exists = True finally: char_conn.close() if not character_exists: flash('您选择的角色不存在或不属于您', 'danger') return redirect(url_for('redeem')) # 检查该角色是否已经领取过CDK used = UsedCDK.query.filter_by(character_name=character_name).first() if used: flash(f'该角色已经领取过CDK: {used.cdk}，领取时间: {used.used_at}', 'danger') return redirect(url_for('redeem')) # 获取帖子内容并验证 try: response = requests.get(post_url) soup = BeautifulSoup(response.text, 'html.parser') post_content = soup.get_text() # 验证广告内容 if not verify_ad_content(post_content, character_name): flash('未找到有效的宣传内容，请确保完整复制宣传模板2', 'danger') flash(post_content, 'danger') flash(character_name, 'danger') return redirect(url_for('redeem')) # 获取未使用的CDK cdk = CDKCode.query.filter( ~CDKCode.兑换码.in_( db.session.query(UsedCDK.cdk) ) ).first() if not cdk: flash('CDK已经用完，请联系管理员', 'danger') return redirect(url_for('redeem')) # 记录使用情况 used_cdk = UsedCDK( character_name=character_name, post_url=post_url, cdk=cdk.兑换码 ) db.session.add(used_cdk) db.session.commit() flash(f'恭喜！您的角色 {character_name} 已成功兑换CDK: {cdk.兑换码}', 'success') except Exception as e: flash(f'验证失败: {str(e)}', 'danger') return redirect(url_for('redeem')) # 获取当前用户的角色列表 characters = [] user_id = session.get('admin_id') user = Admin.query.get(user_id) conn = get_auth_db_connection() if conn: try: account_table = get_db_config('auth_account_table', 'account') query = f"SELECT id FROM {account_table} WHERE username = %s" result = conn.execute(query, (user.username,)) account_ids = [row[0] for row in result] if account_ids: char_conn = get_char_db_connection() if char_conn: try: char_table = get_db_config('char_table', 'characters') placeholders = ', '.join(['%s'] * len(account_ids)) query = f""" SELECT guid, account, name, level, race, class FROM {char_table} WHERE account IN ({placeholders}) ORDER BY level DESC """ result = char_conn.execute(query, account_ids) characters = [dict(row) for row in result] # 添加种族和职业的文本描述 for character in characters: character['race_text'] = get_race_name(character['race']) character['class_text'] = get_class_name(character['class']) # 检查角色是否已经兑换过CDK used = UsedCDK.query.filter_by(character_name=character['name']).first() character['has_cdk'] = used is not None if used: character['cdk'] = used.cdk character['redeem_time'] = used.used_at finally: char_conn.close() finally: conn.close() return render_template('redeem.html', server_info=SERVER_INFO, characters=characters) @app.route('/admin/login', methods=['GET', 'POST']) def admin_login(): if request.method == 'POST': username = request.form.get('username') password = request.form.get('password') admin = Admin.query.filter_by(username=username).first() if admin and admin.check_password(password): session['admin_id'] = admin.id session['admin_username'] = admin.username session['admin_role'] = admin.role flash('登录成功', 'success') # 根据角色重定向到不同的仪表盘 if admin.is_admin(): return redirect(url_for('admin_dashboard')) else: return redirect(url_for('user_dashboard')) else: flash('用户名或密码错误', 'danger') return render_template('admin_login.html') @app.route('/admin/logout') def admin_logout(): session.pop('admin_id', None) session.pop('admin_username', None) session.pop('admin_role', None) flash('已退出登录', 'success') return redirect(url_for('admin_login')) @app.route('/admin/dashboard') @admin_required def admin_dashboard(): return render_template('admin_dashboard.html') @app.route('/user/dashboard') @login_required def user_dashboard(): # 获取当前用户信息 user_id = session.get('admin_id') user = Admin.query.get(user_id) # 获取用户的游戏账号信息 game_accounts = [] conn = get_auth_db_connection() if conn: try: account_table = get_db_config('auth_account_table', 'account') # 这里假设用户名与游戏账号用户名相同，实际情况可能需要调整 query = f""" SELECT id, username, email, last_login, expansion, CASE WHEN online > 0 THEN '在线' ELSE '离线' END as status FROM {account_table} WHERE username = %s """ result = conn.execute(query, (user.username,)) game_accounts = [dict(row) for row in result] except Exception as e: flash(f'获取游戏账号失败: {str(e)}', 'danger') finally: conn.close() # 获取用户的角色信息 characters = [] if game_accounts: conn = get_char_db_connection() if conn: try: char_table = get_db_config('char_table', 'characters') account_ids = [acc['id'] for acc in game_accounts] placeholders = ', '.join(['%s'] * len(account_ids)) query = f""" SELECT guid, account, name, level, race, class, gender, CASE WHEN online > 0 THEN '在线' ELSE '离线' END as online_status, money/10000 as gold, totaltime/3600 as hours_played, map FROM {char_table} WHERE account IN ({placeholders}) ORDER BY level DESC """ result = conn.execute(query, account_ids) characters = [dict(row) for row in result] # 添加种族、职业和性别的文本描述 for character in characters: character['race_text'] = get_race_name(character['race']) character['class_text'] = get_class_name(character['class']) character['gender_text'] = '女' if character['gender'] == 1 else '男' except Exception as e: flash(f'获取角色列表失败: {str(e)}', 'danger') finally: conn.close() return render_template('user_dashboard.html', user=user, game_accounts=game_accounts, characters=characters) def generate_srp6_verifier(username, password): import hashlib from random import getrandbits N = int('894B645E89E1535BBDAD5B8B290650530801B18EBFBF5E8FAB3C82872A3E9BB7', 16) g = 7 # 生成随机salt salt = getrandbits(256).to_bytes(32, 'little') # 计算验证器 x = hashlib.sha1(salt + hashlib.sha1(f"{username}:{password}".encode()).digest()).digest() x = int.from_bytes(x, 'little') verifier = pow(g, x, N) verifier = verifier.to_bytes(32, 'little') return salt, verifier def create_game_account(username, password): username = username.upper() password = password.upper() salt, verifier = generate_srp6_verifier(username, password) conn = get_auth_db_connection() if conn: try: sql = """ INSERT INTO account (username, salt, verifier, expansion, joindate) VALUES (%s, %s, %s, %s, NOW()) """ conn.execute(sql, (username, salt, verifier, 2)) return True, "账号创建成功" except Exception as e: return False, f"数据库错误: {str(e)}" finally: conn.close() # 然后定义路由处理函数 @app.route('/admin/register', methods=['GET', 'POST']) def admin_register(): has_admin = Admin.query.filter_by(role='admin').first() is not None if request.method == 'POST': username = request.form.get('username') password = request.form.get('password') confirm_password = request.form.get('confirm_password') if not username or not password: flash('请填写所有必填字段', 'danger') elif password != confirm_password: flash('两次输入的密码不一致', 'danger') elif Admin.query.filter_by(username=username).first(): flash('用户名已存在', 'danger') else: admin = Admin(username=username) admin.set_password(password) if not has_admin: admin.role = 'admin' flash('管理员账号创建成功，请登录', 'success') else: admin.role = 'user' # 正确的调用方式应该是: success, message = create_game_account(username, password) if not success: flash(f'创建游戏账号失败: {message}', 'danger') return render_template('admin_register.html', is_admin_exists=has_admin) flash('用户账号创建成功，同时已创建游戏账号', 'success') db.session.add(admin) db.session.commit() return redirect(url_for('admin_login')) return render_template('admin_register.html', is_admin_exists=has_admin) @app.route('/admin/create_admin', methods=['GET', 'POST']) @admin_required def create_admin(): if request.method == 'POST': username = request.form.get('username') password = request.form.get('password') role = request.form.get('role', 'user') if not username or not password: flash('请填写所有必填字段', 'danger') elif Admin.query.filter_by(username=username).first(): flash('用户名已存在', 'danger') else: admin = Admin(username=username, role=role) admin.set_password(password) db.session.add(admin) db.session.commit() flash(f'{"管理员" if role == "admin" else "用户"}账号创建成功', 'success') return redirect(url_for('admin_dashboard')) return render_template('create_admin.html') @app.route('/admin/db_config', methods=['GET', 'POST']) @admin_required def db_config(): if request.method == 'POST': # 保存数据库配置 auth_host = request.form.get('auth_host') auth_port = request.form.get('auth_port') auth_user = request.form.get('auth_user') auth_pass = request.form.get('auth_pass') auth_db = request.form.get('auth_db') auth_account_table = request.form.get('auth_account_table') char_host = request.form.get('char_host') char_port = request.form.get('char_port') char_user = request.form.get('char_user') char_pass = request.form.get('char_pass') char_db = request.form.get('char_db') char_table = request.form.get('char_table') # 更新或创建配置 configs = { 'auth_host': auth_host, 'auth_port': auth_port, 'auth_user': auth_user, 'auth_pass': auth_pass, 'auth_db': auth_db, 'auth_account_table': auth_account_table, 'char_host': char_host, 'char_port': char_port, 'char_user': char_user, 'char_pass': char_pass, 'char_db': char_db, 'char_table': char_table } for key, value in configs.items(): config = DatabaseConfig.query.filter_by(config_name=key).first() if config: config.config_value = value else: config = DatabaseConfig(config_name=key, config_value=value) db.session.add(config) db.session.commit() flash('数据库配置已保存', 'success') return redirect(url_for('db_config')) # 获取现有配置 configs = {} db_configs = DatabaseConfig.query.all() for config in db_configs: configs[config.config_name] = config.config_value return render_template('db_config.html', configs=configs) @app.route('/admin/test_connection', methods=['POST']) @admin_required def test_connection(): connection_type = request.form.get('type') if connection_type == 'auth': host = request.form.get('auth_host') port = request.form.get('auth_port') user = request.form.get('auth_user') password = request.form.get('auth_pass') db_name = request.form.get('auth_db') else: # char host = request.form.get('char_host') port = request.form.get('char_port') user = request.form.get('char_user') password = request.form.get('char_pass') db_name = request.form.get('char_db') try: connection_string = f"mysql+pymysql://{user}:{password}@{host}:{port}/{db_name}" engine = create_engine(connection_string) connection = engine.connect() connection.close() return jsonify({'success': True, 'message': '连接成功'}) except Exception as e: return jsonify({'success': False, 'message': f'连接失败: {str(e)}'}) # 在应用启动时创建管理员账号（如果不存在） @app.before_first_request def create_admin(): db.create_all() if Admin.query.filter_by(role='admin').count() == 0: # 如果没有管理员账号，创建一个默认账号 admin = Admin(username='admin', role='admin') admin.set_password('admin') db.session.add(admin) db.session.commit() print("已创建默认管理员账号: admin/admin") # 获取数据库配置 def get_db_config(config_name, default=None): config = DatabaseConfig.query.filter_by(config_name=config_name).first() if config: return config.config_value return default # 连接到AzerothCore的Auth数据库 def get_auth_db_connection(): host = get_db_config('auth_host', '127.0.0.1') port = get_db_config('auth_port', '3306') user = get_db_config('auth_user', 'acore') password = get_db_config('auth_pass', 'acore') db_name = get_db_config('auth_db', 'acore_auth') try: connection_string = f"mysql+pymysql://{user}:{password}@{host}:{port}/{db_name}" engine = create_engine(connection_string) return engine.connect() except Exception as e: print(f"连接Auth数据库失败: {str(e)}") return None # 连接到AzerothCore的Characters数据库 def get_char_db_connection(): host = get_db_config('char_host', '127.0.0.1') port = get_db_config('char_port', '3306') user = get_db_config('char_user', 'acore') password = get_db_config('char_pass', 'acore') db_name = get_db_config('char_db', 'acore_characters') try: connection_string = f"mysql+pymysql://{user}:{password}@{host}:{port}/{db_name}" engine = create_engine(connection_string) return engine.connect() except Exception as e: print(f"连接Characters数据库失败: {str(e)}") return None @app.route('/admin/account_management', methods=['GET', 'POST']) @admin_required def account_management(): if request.method == 'POST': action = request.form.get('action') if action == 'create_account': username = request.form.get('username') password = request.form.get('password') email = request.form.get('email', '') if not username or not password: flash('用户名和密码不能为空', 'danger') return redirect(url_for('account_management')) # 连接到Auth数据库 conn = get_auth_db_connection() if not conn: flash('无法连接到Auth数据库', 'danger') return redirect(url_for('account_management')) try: # 检查账号是否已存在 account_table = get_db_config('auth_account_table', 'account') result = conn.execute(f"SELECT id FROM {account_table} WHERE username = %s", (username,)) if result.fetchone(): flash('账号已存在', 'danger') return redirect(url_for('account_management')) # 根据提供的表结构，我们需要生成salt和verifier # 这里使用简化的方法，实际应该使用SRP6算法 import os import hashlib import binascii # 生成随机salt (32字节) salt = os.urandom(32) # 计算verifier (简化版本，实际应使用SRP6) # 在实际的SRP6中，verifier的计算更复杂 # 这里我们使用一个简化的方法，仅用于演示 h1 = hashlib.sha1((username.upper() + ":" + password.upper()).encode()).digest() h2 = hashlib.sha1(salt + h1).digest() verifier = h2 # 构建SQL sql = f""" INSERT INTO {account_table} (username, salt, verifier, email, reg_mail, joindate) VALUES (%s, %s, %s, %s, %s, NOW()) """ # 执行SQL conn.execute(sql, (username, salt, verifier, email, email)) flash('账号创建成功', 'success') except Exception as e: flash(f'创建账号失败: {str(e)}', 'danger') finally: conn.close() return redirect(url_for('account_management')) # 获取账号列表 accounts = [] conn = get_auth_db_connection() if conn: try: account_table = get_db_config('auth_account_table', 'account') # 根据提供的表结构构建查询 query = f""" SELECT id, username, email, last_login, expansion, CASE WHEN online > 0 THEN '在线' ELSE '离线' END as status FROM {account_table} ORDER BY id DESC LIMIT 100 """ result = conn.execute(query) accounts = [dict(row) for row in result] except Exception as e: flash(f'获取账号列表失败: {str(e)}', 'danger') finally: conn.close() return render_template('account_management.html', accounts=accounts) @app.route('/admin/character_management') @admin_required def character_management(): # 获取角色列表 characters = [] conn = get_char_db_connection() if conn: try: char_table = get_db_config('char_table', 'characters') # 根据提供的表结构构建查询 query = f""" SELECT guid, account, name, level, race, class, gender, CASE WHEN online > 0 THEN '在线' ELSE '离线' END as online_status, money/10000 as gold, totaltime/3600 as hours_played, map FROM {char_table} ORDER BY guid DESC LIMIT 100 """ result = conn.execute(query) characters = [dict(row) for row in result] # 添加种族、职业和性别的文本描述 for character in characters: character['race_text'] = get_race_name(character['race']) character['class_text'] = get_class_name(character['class']) character['gender_text'] = '女' if character['gender'] == 1 else '男' except Exception as e: flash(f'获取角色列表失败: {str(e)}', 'danger') finally: conn.close() return render_template('character_management.html', characters=characters) @app.route('/admin/link_account', methods=['POST']) @admin_required def link_account(): character_name = request.form.get('character_name') post_url = request.form.get('post_url') if not character_name or not post_url: flash('请填写所有必要信息', 'danger') return redirect(url_for('character_management')) # 验证角色是否存在 char_conn = get_char_db_connection() if not char_conn: flash('无法连接到Characters数据库', 'danger') return redirect(url_for('character_management')) try: char_table = get_db_config('char_table', 'characters') # 查询角色 result = char_conn.execute(f"SELECT guid, account, name, level FROM {char_table} WHERE name = %s", (character_name,)) character = result.fetchone() if not character: flash('角色不存在', 'danger') return redirect(url_for('character_management')) # 检查该角色是否已经领取过CDK used = UsedCDK.query.filter_by(character_name=character_name).first() if used: flash(f'该角色已经领取过CDK: {used.cdk}，领取时间: {used.used_at}', 'warning') return redirect(url_for('character_management')) # 获取未使用的CDK cdk = CDKCode.query.filter( ~CDKCode.兑换码.in_( db.session.query(UsedCDK.cdk) ) ).first() if not cdk: flash('CDK已经用完，请添加新的CDK', 'danger') return redirect(url_for('character_management')) # 记录使用情况 used_cdk = UsedCDK( character_name=character_name, post_url=post_url, cdk=cdk.兑换码 ) db.session.add(used_cdk) db.session.commit() flash(f'已成功为角色 {character_name} (等级 {character["level"]}) 分配CDK: {cdk.兑换码}', 'success') except Exception as e: flash(f'操作失败: {str(e)}', 'danger') finally: char_conn.close() return redirect(url_for('character_management')) # 获取种族名称 def get_race_name(race_id): races = { 1: '人类', 2: '兽人', 3: '矮人', 4: '暗夜精灵', 5: '亡灵', 6: '牛头人', 7: '侏儒', 8: '巨魔', 9: '地精', 10: '血精灵', 11: '德莱尼', 22: '狼人', 24: '熊猫人', 25: '熊猫人', 26: '熊猫人' } return races.get(race_id, f'未知({race_id})') # 获取职业名称 def get_class_name(class_id): classes = { 1: '战士', 2: '圣骑士', 3: '猎人', 4: '盗贼', 5: '牧师', 6: '死亡骑士', 7: '萨满', 8: '法师', 9: '术士', 10: '武僧', 11: '德鲁伊' } return classes.get(class_id, f'未知({class_id})') if __name__ == '__main__': # 创建数据库 create_database() with app.app_context(): try: # 创建所有表 db.create_all() print("数据库表创建成功") except Exception as e: print(f"数据库表创建失败：{str(e)}") app.run(debug=True)