235 lines
7.6 KiB
Python
235 lines
7.6 KiB
Python
from flask import Flask, render_template, request, jsonify, flash, redirect, url_for
|
||
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
|
||
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())
|
||
|
||
def verify_ad_content(content, character_name):
|
||
# 检查必须包含的关键词
|
||
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):
|
||
return False
|
||
|
||
# 确保包含角色名
|
||
if character_name not in content:
|
||
return False
|
||
|
||
return True
|
||
|
||
@app.route('/')
|
||
def index():
|
||
return render_template('index.html')
|
||
|
||
@app.route('/generate')
|
||
def generate():
|
||
return render_template('generate.html', server_info=SERVER_INFO)
|
||
|
||
@app.route('/create-ad', methods=['POST'])
|
||
def create_ad():
|
||
character_name = request.form.get('character_name')
|
||
name_type = request.form.get('name_type', 'character')
|
||
|
||
if not character_name:
|
||
return jsonify({'success': False, 'message': '请输入名称'})
|
||
|
||
ad_content = AD_TEMPLATE.format(
|
||
server_name=SERVER_INFO['server_name'],
|
||
name_type=NAME_TYPES[name_type],
|
||
name=character_name,
|
||
server_features=SERVER_INFO['server_features'],
|
||
website=SERVER_INFO['website']
|
||
)
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'ad_content': ad_content
|
||
})
|
||
|
||
@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()
|
||
|
||
# 验证广告内容
|
||
if not verify_ad_content(post_content, character_name):
|
||
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:
|
||
return jsonify({'success': False, 'message': f'验证失败：{str(e)}'})
|
||
|
||
@app.route('/redeem', methods=['GET', 'POST'])
|
||
def redeem():
|
||
if request.method == 'POST':
|
||
post_url = request.form.get('link')
|
||
if not post_url:
|
||
flash('请输入宣传链接', 'danger')
|
||
return redirect(url_for('redeem'))
|
||
|
||
try:
|
||
response = requests.get(post_url)
|
||
soup = BeautifulSoup(response.text, 'html.parser')
|
||
post_content = soup.get_text()
|
||
|
||
# 使用正则表达式提取角色名
|
||
character_match = re.search(r'推荐(角色|账号)：([^\n\r]+)', post_content)
|
||
if not character_match:
|
||
flash('未在帖子中找到有效的推荐信息', 'danger')
|
||
return redirect(url_for('redeem'))
|
||
|
||
character_name = character_match.group(2).strip()
|
||
|
||
# 验证广告内容
|
||
if not verify_ad_content(post_content, character_name):
|
||
flash('未找到有效的宣传内容，请确保完整复制宣传模板', 'danger')
|
||
return redirect(url_for('redeem'))
|
||
|
||
# 检查是否已经领取过
|
||
used = UsedCDK.query.filter_by(character_name=character_name).first()
|
||
if used:
|
||
flash('该角色已经领取过CDK', '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'恭喜！您的CDK是：{cdk.兑换码}', 'success')
|
||
return redirect(url_for('redeem'))
|
||
|
||
except Exception as e:
|
||
flash(f'验证失败：{str(e)}', 'danger')
|
||
return redirect(url_for('redeem'))
|
||
|
||
return render_template('redeem.html', server_info=SERVER_INFO)
|
||
|
||
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) |