Kamixitong/app/api/license.py
2025-11-12 09:19:32 +08:00

722 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from flask import request, jsonify, current_app, send_file
from datetime import datetime, timedelta
from app import db
from app.models import Product, License
from . import api_bp
import io
import csv
import xlsxwriter
from functools import wraps
from sqlalchemy import or_, func
def require_admin(f):
"""管理员权限验证装饰器"""
@wraps(f)
def decorated_function(*args, **kwargs):
# 从Flask-Login获取当前用户
from flask_login import current_user
from flask import current_app, jsonify
try:
current_app.logger.info(f"装饰器检查开始,函数: {f.__name__}")
# 检查用户是否已认证
current_app.logger.info(f"检查用户认证状态: {current_user.is_authenticated if hasattr(current_user, 'is_authenticated') else 'Unknown'}")
if not hasattr(current_user, 'is_authenticated'):
current_app.logger.error("current_user对象缺少is_authenticated属性")
return jsonify({
'success': False,
'message': '用户认证状态异常'
}), 500
if not current_user.is_authenticated:
current_app.logger.warning("用户未认证")
return jsonify({
'success': False,
'message': '需要登录'
}), 401
# 检查是否为管理员
current_app.logger.info(f"检查管理员权限: {current_user.is_super_admin() if hasattr(current_user, 'is_super_admin') else 'Unknown'}")
if not hasattr(current_user, 'is_super_admin'):
current_app.logger.error("current_user对象缺少is_super_admin属性")
return jsonify({
'success': False,
'message': '用户权限状态异常'
}), 500
if not current_user.is_super_admin():
current_app.logger.warning("用户不是管理员")
return jsonify({
'success': False,
'message': '需要管理员权限'
}), 403
current_app.logger.info("权限检查通过")
result = f(*args, **kwargs)
current_app.logger.info("函数执行完成")
return result
except Exception as e:
current_app.logger.error(f"装饰器执行失败: {str(e)}")
current_app.logger.error(f"错误类型: {type(e)}")
import traceback
current_app.logger.error(f"错误堆栈: {traceback.format_exc()}")
return jsonify({
'success': False,
'message': f'服务器内部错误: {str(e)}'
}), 500
return decorated_function
@api_bp.route('/licenses', methods=['GET'])
@require_admin
def get_licenses():
"""获取卡密列表"""
try:
# 获取查询参数
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 100)
product_id = request.args.get('product_id')
status = request.args.get('status', type=int)
license_type = request.args.get('type', type=int)
keyword = request.args.get('keyword', '').strip()
# 构建查询
query = License.query
# 产品筛选
if product_id:
query = query.filter_by(product_id=product_id)
# 状态筛选
if status is not None:
query = query.filter_by(status=status)
# 类型筛选
if license_type is not None:
query = query.filter_by(type=license_type)
# 关键词搜索(卡密)
if keyword:
query = query.filter(func.lower(License.license_key).like(f'%{keyword.lower()}%'))
# 按创建时间倒序
query = query.order_by(License.create_time.desc())
# 分页
pagination = query.paginate(
page=page,
per_page=per_page,
error_out=False
)
# 格式化结果
licenses = [license.to_dict() for license in pagination.items]
return jsonify({
'success': True,
'data': {
'licenses': licenses,
'pagination': {
'page': page,
'per_page': per_page,
'total': pagination.total,
'pages': pagination.pages,
'has_prev': pagination.has_prev,
'has_next': pagination.has_next
}
}
})
except Exception as e:
current_app.logger.error(f"获取卡密列表失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
@api_bp.route('/licenses', methods=['POST'])
@require_admin
def generate_licenses():
"""批量生成卡密"""
try:
data = request.get_json()
if not data:
return jsonify({
'success': False,
'message': '请求数据为空'
}), 400
# 获取参数
product_id = data.get('product_id')
count = data.get('count', 1)
license_type = data.get('type', 1) # 0=试用, 1=正式
valid_days = data.get('valid_days') # 不设置默认值保持None
prefix = data.get('prefix', '')
length = data.get('length', 32)
# 验证参数
if not product_id:
return jsonify({
'success': False,
'message': '产品ID不能为空'
}), 400
if count < 1 or count > 10000:
return jsonify({
'success': False,
'message': '生成数量必须在1-10000之间'
}), 400
if length < 16 or length > 35:
return jsonify({
'success': False,
'message': '卡密长度必须在16-35之间'
}), 400
# 验证产品存在
product = Product.query.filter_by(product_id=product_id).first()
if not product:
return jsonify({
'success': False,
'message': '产品不存在'
}), 404
# 设置默认有效期
if valid_days is None:
valid_days = 365
# 试用卡密最大有效期限制
if license_type == 0 and valid_days > 90:
return jsonify({
'success': False,
'message': '试用卡密有效期不能超过90天'
}), 400
# 生成卡密
try:
licenses = License.generate_batch(
product_id=product_id,
count=count,
license_type=license_type,
valid_days=valid_days,
prefix=prefix,
length=length
)
# 批量保存
db.session.add_all(licenses)
db.session.commit()
# 格式化结果
license_data = [license.to_dict() for license in licenses]
return jsonify({
'success': True,
'message': f'成功生成 {count} 个卡密',
'data': {
'licenses': license_data,
'count': len(licenses)
}
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f"生成卡密失败: {str(e)}")
return jsonify({
'success': False,
'message': f'生成卡密失败: {str(e)}'
}), 500
except Exception as e:
current_app.logger.error(f"生成卡密失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
@api_bp.route('/licenses/<int:license_id>', methods=['GET'])
@require_admin
def get_license(license_id):
"""获取单个卡密详情"""
try:
license_obj = License.query.get(license_id)
if not license_obj:
return jsonify({
'success': False,
'message': '卡密不存在'
}), 404
return jsonify({
'success': True,
'data': license_obj.to_dict()
})
except Exception as e:
current_app.logger.error(f"获取卡密详情失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
@api_bp.route('/licenses/<int:license_id>', methods=['PUT'])
@require_admin
def update_license(license_id):
"""更新卡密信息"""
try:
license_obj = License.query.get(license_id)
if not license_obj:
return jsonify({
'success': False,
'message': '卡密不存在'
}), 404
data = request.get_json()
if not data:
return jsonify({
'success': False,
'message': '请求数据为空'
}), 400
# 更新状态
if 'status' in data:
new_status = data['status']
# 已激活的卡密不能改为未激活
if license_obj.status == 1 and new_status == 0:
return jsonify({
'success': False,
'message': '已激活的卡密不能改为未激活状态'
}), 400
license_obj.status = new_status
# 更新备注
if 'remark' in data:
license_obj.remark = data['remark']
db.session.commit()
return jsonify({
'success': True,
'message': '更新成功',
'data': license_obj.to_dict()
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f"更新卡密失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
@api_bp.route('/licenses/<int:license_id>/extend', methods=['POST'])
@require_admin
def extend_license(license_id):
"""延长卡密有效期"""
try:
license_obj = License.query.get(license_id)
if not license_obj:
return jsonify({
'success': False,
'message': '卡密不存在'
}), 404
data = request.get_json()
if not data or 'days' not in data:
return jsonify({
'success': False,
'message': '缺少延期天数参数'
}), 400
days = data['days']
if days <= 0:
return jsonify({
'success': False,
'message': '延期天数必须大于0'
}), 400
success, message = license_obj.extend_validity(days)
if not success:
return jsonify({
'success': False,
'message': message
}), 400
return jsonify({
'success': True,
'message': message,
'data': license_obj.to_dict()
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f"延长卡密有效期失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
@api_bp.route('/licenses/<int:license_id>/convert', methods=['POST'])
@require_admin
def convert_license(license_id):
"""试用卡密转正式"""
try:
license_obj = License.query.get(license_id)
if not license_obj:
return jsonify({
'success': False,
'message': '卡密不存在'
}), 404
data = request.get_json()
if not data or 'valid_days' not in data:
return jsonify({
'success': False,
'message': '缺少有效期参数'
}), 400
valid_days = data['valid_days']
if valid_days <= 0:
return jsonify({
'success': False,
'message': '有效期必须大于0天'
}), 400
success, message = license_obj.convert_to_formal(valid_days)
if not success:
return jsonify({
'success': False,
'message': message
}), 400
return jsonify({
'success': True,
'message': message,
'data': license_obj.to_dict()
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f"转换卡密失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
@api_bp.route('/licenses/<int:license_id>/unbind', methods=['POST'])
@require_admin
def unbind_license(license_id):
"""解绑设备"""
try:
license_obj = License.query.get(license_id)
if not license_obj:
return jsonify({
'success': False,
'message': '卡密不存在'
}), 404
# 获取最大解绑次数配置
max_unbind_times = current_app.config.get('MAX_UNBIND_TIMES', 3)
success, message = license_obj.unbind()
if not success:
return jsonify({
'success': False,
'message': message
}), 400
return jsonify({
'success': True,
'message': message,
'data': license_obj.to_dict()
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f"解绑设备失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
@api_bp.route('/licenses/export', methods=['POST'])
@require_admin
def export_licenses():
"""导出卡密"""
try:
data = request.get_json()
if not data:
return jsonify({
'success': False,
'message': '请求数据为空'
}), 400
product_id = data.get('product_id')
status = data.get('status')
license_type = data.get('type')
export_format = data.get('format', 'excel') # excel 或 csv
# 构建查询
query = License.query
if product_id:
query = query.filter_by(product_id=product_id)
if status is not None:
query = query.filter_by(status=status)
if license_type is not None:
query = query.filter_by(type=license_type)
# 获取数据
licenses = query.order_by(License.create_time.desc()).all()
if not licenses:
return jsonify({
'success': False,
'message': '没有找到符合条件的卡密'
}), 404
if export_format == 'csv':
# 导出CSV
output = io.StringIO()
writer = csv.writer(output)
# 写入表头
writer.writerow(['卡密', '产品ID', '产品名称', '类型', '状态', '有效期',
'绑定机器码', '激活时间', '过期时间', '创建时间'])
# 写入数据
for license_obj in licenses:
writer.writerow([
license_obj.license_key,
license_obj.product_id,
license_obj.product.product_name if license_obj.product else '',
'试用' if license_obj.is_trial() else '正式',
license_obj.get_status_name(),
'永久' if license_obj.valid_days == -1 else f'{license_obj.valid_days}',
license_obj.bind_machine_code or '',
license_obj.activate_time.strftime('%Y-%m-%d %H:%M:%S') if license_obj.activate_time else '',
license_obj.expire_time.strftime('%Y-%m-%d %H:%M:%S') if license_obj.expire_time else '',
license_obj.create_time.strftime('%Y-%m-%d %H:%M:%S')
])
output.seek(0)
return send_file(
io.BytesIO(output.getvalue().encode('utf-8-sig')),
mimetype='text/csv',
as_attachment=True,
download_name=f'licenses_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv'
)
else:
# 导出Excel
output = io.BytesIO()
workbook = xlsxwriter.Workbook(output)
worksheet = workbook.add_worksheet('卡密列表')
# 设置表头格式
header_format = workbook.add_format({
'bold': True,
'text_wrap': True,
'valign': 'top',
'fg_color': '#D7E4BC',
'border': 1
})
# 写入表头
headers = ['卡密', '产品ID', '产品名称', '类型', '状态', '有效期',
'绑定机器码', '激活时间', '过期时间', '创建时间']
for col, header in enumerate(headers):
worksheet.write(0, col, header, header_format)
# 写入数据
for row, license_obj in enumerate(licenses, 1):
worksheet.write(row, 0, license_obj.license_key)
worksheet.write(row, 1, license_obj.product_id)
worksheet.write(row, 2, license_obj.product.product_name if license_obj.product else '')
worksheet.write(row, 3, '试用' if license_obj.is_trial() else '正式')
worksheet.write(row, 4, license_obj.get_status_name())
worksheet.write(row, 5, '永久' if license_obj.valid_days == -1 else f'{license_obj.valid_days}')
worksheet.write(row, 6, license_obj.bind_machine_code or '')
worksheet.write(row, 7, license_obj.activate_time.strftime('%Y-%m-%d %H:%M:%S') if license_obj.activate_time else '')
worksheet.write(row, 8, license_obj.expire_time.strftime('%Y-%m-%d %H:%M:%S') if license_obj.expire_time else '')
worksheet.write(row, 9, license_obj.create_time.strftime('%Y-%m-%d %H:%M:%S'))
# 设置列宽
worksheet.set_column(0, 0, 35) # 卡密
worksheet.set_column(1, 1, 20) # 产品ID
worksheet.set_column(2, 2, 25) # 产品名称
worksheet.set_column(3, 5, 12) # 类型、状态、有效期
worksheet.set_column(6, 6, 35) # 机器码
worksheet.set_column(7, 9, 20) # 时间字段
workbook.close()
output.seek(0)
return send_file(
output,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=f'licenses_{datetime.now().strftime("%Y%m%d_%H%M%S")}.xlsx'
)
except Exception as e:
current_app.logger.error(f"导出卡密失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
@api_bp.route('/licenses/import', methods=['POST'])
@require_admin
def import_licenses():
"""导入卡密"""
try:
if 'file' not in request.files:
return jsonify({
'success': False,
'message': '没有上传文件'
}), 400
file = request.files['file']
if file.filename == '':
return jsonify({
'success': False,
'message': '没有选择文件'
}), 400
# 验证文件格式
if not file.filename or not (file.filename.endswith('.xlsx') or file.filename.endswith('.csv')):
return jsonify({
'success': False,
'message': '只支持Excel或CSV文件格式'
}), 400
# 读取文件内容
import pandas as pd
if file.filename.endswith('.xlsx'):
# 读取Excel文件
file.stream.seek(0) # 重置文件指针
df = pd.read_excel(file.stream)
else:
# 读取CSV文件
file.stream.seek(0) # 重置文件指针
df = pd.read_csv(file.stream)
# 验证必要列
required_columns = ['卡密', '产品ID', '有效期(天)', '类型']
missing_columns = [col for col in required_columns if col not in df.columns]
if missing_columns:
return jsonify({
'success': False,
'message': f'缺少必要列: {", ".join(missing_columns)}'
}), 400
success_count = 0
error_count = 0
error_messages = []
# 逐行处理
for index, row in df.iterrows():
try:
license_key = str(row['卡密']).strip()
product_id = str(row['产品ID']).strip()
valid_days = int(row['有效期(天)'])
license_type = 1 if str(row['类型']).strip() == '正式' else 0
# 验证产品存在
product = Product.query.filter_by(product_id=product_id).first()
if not product:
error_messages.append(f'{int(str(index))+1}行: 产品ID {product_id} 不存在')
error_count += 1
continue
# 检查卡密是否已存在
existing = License.query.filter_by(license_key=license_key).first()
if existing:
error_messages.append(f'{int(str(index))+1}行: 卡密 {license_key} 已存在')
error_count += 1
continue
# 创建新卡密
license_obj = License(
license_key=license_key,
product_id=product_id,
type=license_type,
valid_days=valid_days,
status=0 # 未激活
)
db.session.add(license_obj)
success_count += 1
except Exception as e:
error_messages.append(f'{int(str(index))+1}行: 处理失败 - {str(e)}')
error_count += 1
# 提交成功的记录
try:
db.session.commit()
except Exception as e:
db.session.rollback()
return jsonify({
'success': False,
'message': f'保存失败: {str(e)}'
}), 500
return jsonify({
'success': True,
'message': f'导入完成,成功 {success_count} 条,失败 {error_count}',
'data': {
'success_count': success_count,
'error_count': error_count,
'error_messages': error_messages[:10] # 只返回前10条错误信息
}
})
except Exception as e:
current_app.logger.error(f"导入卡密失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
@api_bp.route('/licenses/<string:license_key>', methods=['DELETE'])
@require_admin
def delete_license(license_key):
"""删除卡密"""
try:
# 获取查询参数,检查是否允许强制删除
force = request.args.get('force', 'false').lower() == 'true'
# 根据卡密key查找卡密
license_obj = License.query.filter_by(license_key=license_key).first()
if not license_obj:
return jsonify({
'success': False,
'message': '卡密不存在'
}), 404
# 检查卡密状态,已激活的卡密不能直接删除,除非强制删除
if license_obj.status == 1 and not force:
return jsonify({
'success': False,
'message': '已激活的卡密不能删除,请先禁用或解绑设备'
}), 400
# 执行删除操作
db.session.delete(license_obj)
db.session.commit()
return jsonify({
'success': True,
'message': '卡密删除成功'
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f"删除卡密失败: {str(e)}")
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500