You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
117 lines
3.5 KiB
117 lines
3.5 KiB
import json
|
|
import os
|
|
from datetime import datetime
|
|
import pymysql
|
|
from sqlalchemy import create_engine, Column, String, Float, Integer, DateTime, Enum, ForeignKey
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker, relationship
|
|
from dotenv import load_dotenv
|
|
|
|
# 加载配置
|
|
load_dotenv()
|
|
Base = declarative_base()
|
|
|
|
|
|
# 修复数据库连接 - 处理端口号转换
|
|
def get_database_url():
|
|
host = os.getenv('MYSQL_HOST', 'localhost')
|
|
port = os.getenv('MYSQL_PORT', '3306')
|
|
# 确保端口是整数
|
|
try:
|
|
port = int(port)
|
|
except (ValueError, TypeError):
|
|
port = 3306 # 默认端口
|
|
|
|
user = os.getenv('MYSQL_USER', 'root')
|
|
password = os.getenv('MYSQL_PWD', '')
|
|
database = os.getenv('MYSQL_DB', 'roll_call_system')
|
|
|
|
return f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}"
|
|
|
|
|
|
# 数据库连接
|
|
try:
|
|
engine = create_engine(
|
|
get_database_url(),
|
|
echo=False,
|
|
pool_recycle=3600,
|
|
connect_args={'charset': 'utf8mb4'}
|
|
)
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
except Exception as e:
|
|
print(f"数据库连接失败: {e}")
|
|
# 使用SQLite作为备用数据库
|
|
engine = create_engine('sqlite:///./roll_call.db', connect_args={"check_same_thread": False})
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
|
class Student(Base):
|
|
__tablename__ = "students"
|
|
student_id = Column(String(20), primary_key=True)
|
|
name = Column(String(50), nullable=False)
|
|
major = Column(String(50), nullable=False)
|
|
class_name = Column(String(50), nullable=False)
|
|
total_score = Column(Float, default=0)
|
|
call_count = Column(Integer, default=0)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'student_id': self.student_id,
|
|
'name': self.name,
|
|
'major': self.major,
|
|
'class_name': self.class_name,
|
|
'total_score': self.total_score,
|
|
'call_count': self.call_count
|
|
}
|
|
|
|
|
|
class RollCallRecord(Base):
|
|
__tablename__ = "roll_call_records"
|
|
record_id = Column(Integer, autoincrement=True, primary_key=True)
|
|
student_id = Column(String(20), ForeignKey("students.student_id"), nullable=False)
|
|
call_time = Column(DateTime, default=datetime.now)
|
|
call_mode = Column(Enum('random', 'order'), nullable=False)
|
|
status = Column(Enum('present', 'absent'), default='present')
|
|
answer_type = Column(String(20), nullable=True)
|
|
performance = Column(String(20), nullable=True)
|
|
score_delta = Column(Float, default=0)
|
|
|
|
# 添加关系
|
|
student = relationship("Student", backref="records")
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'record_id': self.record_id,
|
|
'student_id': self.student_id,
|
|
'call_time': self.call_time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
'call_mode': self.call_mode,
|
|
'status': self.status,
|
|
'answer_type': self.answer_type,
|
|
'performance': self.performance,
|
|
'score_delta': self.score_delta
|
|
}
|
|
|
|
|
|
class OrderCallStatus(Base):
|
|
__tablename__ = "order_call_status"
|
|
class_name = Column(String(50), primary_key=True)
|
|
current_index = Column(Integer, default=0)
|
|
|
|
|
|
# 初始化数据库
|
|
def init_db():
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
# 数据库会话依赖
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
init_db()
|
|
print("数据库初始化完成！") |