Files
pdf-translate-web/models.py
coder 236b202e3e 添加后台管理功能
功能:
- 数据概览仪表盘
- 用户管理(列表/编辑/删除)
- 翻译记录管理
- 缓存管理
- 统计报表(图表)
- 操作日志

管理员账号: admin / admin123
后台地址: /admin/
2026-04-07 23:08:49 +08:00

271 lines
9.6 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 datetime import datetime
from flask_sqlalchemy import SQLAlchemy
from werkzeug.security import generate_password_hash, check_password_hash
import hashlib
db = SQLAlchemy()
# ==================== 用户模型 ====================
class User(db.Model):
"""用户表"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(256), nullable=False)
# 用户类型: guest, free, vip_basic, vip_pro, vip_enterprise, admin
user_type = db.Column(db.String(20), default='free')
# 会员信息
membership_expire = db.Column(db.DateTime, nullable=True) # 会员到期时间
# 使用统计
daily_count = db.Column(db.Integer, default=0) # 今日翻译次数
total_count = db.Column(db.Integer, default=0) # 总翻译次数
last_translate_date = db.Column(db.Date, nullable=True) # 最后翻译日期
# 时间戳
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 状态
is_active = db.Column(db.Boolean, default=True) # 是否启用
is_admin = db.Column(db.Boolean, default=False) # 是否管理员
# 关系
translations = db.relationship('Translation', backref='user', lazy=True)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def is_vip(self):
"""检查是否为付费会员"""
if self.user_type.startswith('vip'):
if self.membership_expire and self.membership_expire > datetime.utcnow():
return True
# 过期则降级为免费用户
self.user_type = 'free'
self.membership_expire = None
db.session.commit()
return False
def can_translate(self, pages, config):
"""检查是否可以翻译(次数、页数限制)"""
limits = config['USER_LIMITS'].get(self.user_type, config['USER_LIMITS']['free'])
# 检查页数限制
max_pages = limits['max_pages']
if max_pages > 0 and pages > max_pages:
return False, f"PDF页数超出限制最大{max_pages}页)"
# 检查每日次数限制
today = datetime.utcnow().date()
if self.last_translate_date != today:
self.daily_count = 0
self.last_translate_date = today
daily_limit = limits['daily_translations']
if daily_limit > 0 and self.daily_count >= daily_limit:
return False, f"今日翻译次数已达上限({daily_limit}次)"
return True, "OK"
def increment_count(self):
"""增加翻译计数"""
today = datetime.utcnow().date()
if self.last_translate_date != today:
self.daily_count = 0
self.last_translate_date = today
self.daily_count += 1
self.total_count += 1
db.session.commit()
def to_dict(self):
return {
'id': self.id,
'username': self.username,
'email': self.email,
'user_type': self.user_type,
'is_vip': self.is_vip(),
'is_admin': self.is_admin,
'is_active': self.is_active,
'daily_count': self.daily_count,
'total_count': self.total_count,
'created_at': self.created_at.isoformat() if self.created_at else None,
'membership_expire': self.membership_expire.isoformat() if self.membership_expire else None,
}
# ==================== 翻译记录模型 ====================
class Translation(db.Model):
"""翻译记录表"""
__tablename__ = 'translations'
id = db.Column(db.Integer, primary_key=True)
# 用户关联
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True) # guest可为null
# 文件信息
file_hash = db.Column(db.String(64), nullable=False) # 文件MD5哈希
original_filename = db.Column(db.String(255), nullable=False)
file_size = db.Column(db.Integer, nullable=False)
page_count = db.Column(db.Integer, nullable=False)
# 翻译信息
source_language = db.Column(db.String(10), default='en')
target_language = db.Column(db.String(10), default='zh')
translate_params = db.Column(db.Text, nullable=True) # JSON格式的翻译参数
# 状态
status = db.Column(db.String(20), default='pending') # pending, processing, completed, failed
progress = db.Column(db.Integer, default=0) # 翻译进度 0-100
error_message = db.Column(db.Text, nullable=True)
# 输出
output_path = db.Column(db.String(255), nullable=True) # 翻译结果文件路径
# 时间戳
created_at = db.Column(db.DateTime, default=datetime.utcnow)
completed_at = db.Column(db.DateTime, nullable=True)
# 是否来自缓存
from_cache = db.Column(db.Boolean, default=False)
# 重译信息
retranslate_request = db.Column(db.Text, nullable=True) # 重译要求
parent_id = db.Column(db.Integer, db.ForeignKey('translations.id'), nullable=True) # 原翻译ID
def to_dict(self):
return {
'id': self.id,
'filename': self.original_filename,
'pages': self.page_count,
'status': self.status,
'progress': self.progress,
'from_cache': self.from_cache,
'file_size': self.file_size,
'created_at': self.created_at.isoformat() if self.created_at else None,
'completed_at': self.completed_at.isoformat() if self.completed_at else None,
'user_id': self.user_id,
}
# ==================== 翻译缓存模型 ====================
class TranslationCache(db.Model):
"""翻译缓存表"""
__tablename__ = 'translation_cache'
id = db.Column(db.Integer, primary_key=True)
# 文件哈希
file_hash = db.Column(db.String(64), unique=True, nullable=False)
# 缓存信息
cache_path = db.Column(db.String(255), nullable=False) # 缓存文件路径
page_count = db.Column(db.Integer, nullable=False)
file_size = db.Column(db.Integer, default=0)
# 统计
hit_count = db.Column(db.Integer, default=0) # 缓存命中次数
# 时间戳
created_at = db.Column(db.DateTime, default=datetime.utcnow)
expires_at = db.Column(db.DateTime, nullable=True)
def increment_hit(self):
self.hit_count += 1
db.session.commit()
@staticmethod
def compute_hash(file_content):
"""计算文件哈希"""
return hashlib.md5(file_content).hexdigest()
# ==================== 访客翻译记录 ====================
class GuestTranslation(db.Model):
"""访客翻译记录基于IP或Session"""
__tablename__ = 'guest_translations'
id = db.Column(db.Integer, primary_key=True)
# 访客标识
session_id = db.Column(db.String(64), nullable=False) # Session ID
ip_address = db.Column(db.String(45), nullable=True)
# 统计
daily_count = db.Column(db.Integer, default=0)
total_count = db.Column(db.Integer, default=0)
last_translate_date = db.Column(db.Date, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
# ==================== 系统配置模型 ====================
class SystemConfig(db.Model):
"""系统配置表"""
__tablename__ = 'system_config'
id = db.Column(db.Integer, primary_key=True)
key = db.Column(db.String(100), unique=True, nullable=False)
value = db.Column(db.Text, nullable=True)
description = db.Column(db.String(255), nullable=True)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
@staticmethod
def get(key, default=None):
config = SystemConfig.query.filter_by(key=key).first()
return config.value if config else default
@staticmethod
def set(key, value, description=None):
config = SystemConfig.query.filter_by(key=key).first()
if config:
config.value = value
else:
config = SystemConfig(key=key, value=value, description=description)
db.session.add(config)
db.session.commit()
# ==================== 操作日志模型 ====================
class OperationLog(db.Model):
"""操作日志表"""
__tablename__ = 'operation_logs'
id = db.Column(db.Integer, primary_key=True)
# 操作者
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
username = db.Column(db.String(80), nullable=True)
# 操作信息
action = db.Column(db.String(50), nullable=False) # login, translate, register, etc.
target = db.Column(db.String(100), nullable=True) # 操作对象
detail = db.Column(db.Text, nullable=True) # 详细信息(JSON)
# IP地址
ip_address = db.Column(db.String(45), nullable=True)
# 时间
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def to_dict(self):
return {
'id': self.id,
'username': self.username,
'action': self.action,
'target': self.target,
'ip_address': self.ip_address,
'created_at': self.created_at.isoformat() if self.created_at else None,
}