zz_er-Qbot_manager
/
models.py

from typing import Iterable
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, BigInteger, DateTime, ForeignKey, Boolean
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine
from urllib.parse import quote
# from sqlalchemy.sql.sqltypes import DateTime
import datetime
from werkzeug.datastructures import CharsetAccept
from time import sleep
import config
from hashlib import md5
import os
from gen_settings import gen_setting
import json
Base = declarative_base()

# 学生信息表
class Student(Base):
    __tablename__="students"
    uid = Column(Integer, primary_key=True) # 学号
    name = Column(String(8), nullable=True) # 姓名
    sex = Column(String(3), default="男",nullable=False) # 性别
    # banji = Column(String(16), nullable=True) # 班级
    banji = Column(Integer, ForeignKey("classes.classesID"), nullable=True) # 班级
    phone = Column(String(16), nullable=True) # 电话
    QQ = Column(String(16), nullable=False) # QQ
    beizhu = Column(Text, nullable=True) # 备注

    stu_cls = relationship('Classes', backref='cls_student')
    # to_uploadFile 提交的文件

    def to_json(self):
            dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
            return dict

# 群聊信息表
class Group(Base):
    __tablename__="ugroups"

    gid = Column(Integer, primary_key=True) # 群ID
    groupName = Column(String(32), nullable=True) # 群名
    numbers = Column(Integer, nullable=True, default=0) # 群聊人数
    classesID = Column(Integer, nullable=True)
    status = Column(String(6), nullable=True, default="未加入") # 状态
    beizhu = Column(Text, nullable=True) # 备注

    # group_cls 与 classes关系

    def to_json(self):
        dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
        return dict


# 班级表
class Classes(Base):
    __tablename__="classes"

    classesID = Column(Integer, primary_key=True) # 班级ID
    memCount = Column(Integer, nullable=True) # 班级人数
    mainGroup = Column(Integer, ForeignKey("ugroups.gid"),nullable=True, default=0) # 主群ID
    beizhu = Column(Text, nullable=True) # 备注

    cls_group = relationship('Group', backref='group_cls')
    # cls_student 与 students 表关系
    # cls_group 与ugroups 表关系


    def to_json(self):
        dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
        return dict

# 文件库映射
class File(Base):
    __tablename__="files"

    fileID = Column(Integer, primary_key=True, autoincrement=True) # 文件ID
    userFileName = Column(Text, nullable=False) # 用户文件名
    serverFileName = Column(Text, nullable=False) # 服务端文件名
    fileSize = Column(BigInteger, nullable=False, default=0) # 文件大小 B
    datetime = Column(DateTime, nullable=False, default=datetime.datetime.now) # 创建日期
    beizhu = Column(Text, nullable=True) # 备注


    def to_json(self):
        dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
        return dict

# 自定义表单 form
class Form(Base):
    __tablename__="forms"

    formID = Column(Integer, primary_key=True, autoincrement=True) # 表单ID
    formTitle = Column(Text, nullable=False) # 表单主题
    serverFileName = Column(Text, nullable=False) # 服务端文件名
    datetime = Column(DateTime, nullable=False, default=datetime.datetime.now) # 创建日期
    beizhu = Column(Text, nullable=True) # 备注
    dbs_fields = Column(Text, nullable=False) # 数据库字段映射

    # to_dataTable 数据表

    def to_json(self):
        dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
        return dict

# 表单数据表 映射表
class FormDatabase(Base):
    __tablename__="formDatabase"

    tableID = Column(Integer, primary_key=True, autoincrement=True) # 数据表ID
    formID = Column(Integer, ForeignKey("forms.formID"), nullable=False) # 数据表对映数据表ID
    tableNickName = Column(Text, nullable=False) # 客户端数据表名
    datetime = Column(DateTime, nullable=False, default=datetime.datetime.now) # 创建日期
    beizhu = Column(Text, nullable=True) # 备注

    # 到Form 表的映射
    to_form = relationship('Form', backref='to_dataTable')
    # to_uploadFile 到提交文件

    def to_json(self):
        dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
        return dict


# 数据表提交文件
class FormTableUploadFile(Base):
    __tablename__="formTableUploadFile"

    fileID = Column(Text, primary_key=True) # 文件ID
    formTableID = Column(Integer, ForeignKey("formDatabase.tableID"),nullable=False) # 数据表ID
    formTableFileField = Column(Text, nullable=False) # 对应数据表中提交文件字段
    stuID = Column(Integer, ForeignKey("students.uid"), nullable=False) # 提交者ID
    userFileName = Column(Text, nullable=False) # 用户文件名
    serverFileName = Column(Text, nullable=False) # 服务端文件名
    fileSize = Column(BigInteger, nullable=False, default=0) # 文件大小 B
    datetime = Column(DateTime, nullable=False, default=datetime.datetime.now) # 创建日期
    beizhu = Column(Text, nullable=True) # 备注

    to_formTable = relationship("FormDatabase", backref="to_uploadFile")
    to_stu =relationship("Student", backref="to_uploadFile")

    def to_json(self):
        dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
        return dict

class Timer(Base):
    __tablename__="timer"

    timerID = Column(Integer, primary_key=True) # timerID
    title = Column(Text, nullable=False)
    func_type = Column(String(20), nullable=False)
    week = Column(String(8), nullable=True, default="0000000") #  周1 周2 周3 周4 周5 周6 周日
    date = Column(String(11), nullable=True) # 年-月-日
    time = Column(String(9), nullable=False) # 时:分:秒
    cmp_time = Column(Integer, nullable=True, default=0)
    exc_count = Column(Integer, nullable=False, default=0) # 今日执行次数
    use_flag = Column(Boolean, nullable=False, default=True) # 是否可用
    task_list = Column(Text, nullable=False)
    log = Column(Text, nullable=True)
    beizhu = Column(Text)  # 备注

    def to_json(self):
        dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
        return dict


class User(Base):
    __tablename__="users"

    userID = Column(String(15), primary_key=True) # UserID
    password = Column(String(15), nullable=False)

    def to_json(self):
        dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
        return dict


# 设置
class Setting(Base):
    __tablename__="settings"

    keyname = Column(String(30), primary_key=True)
    value = Column(Text, nullable=True, default="")
    type = Column(String(10), nullable=True, default="str")

    def to_json(self):
        dict = { k:self.__dict__[k] for k in self.__dict__ if k != "_sa_instance_state" }
        return dict
    @staticmethod
    def get_type(value):
        if isinstance(value, int):return "int"
        elif isinstance(value, float): return "float"
        elif isinstance(value, str): return "str"
        elif isinstance(value, Iterable): "Iterable"
        else: return "to_str"


def init_tables():
    def init_admin(session):
        admin = User(userID="admin", password=md5("admin".encode()).hexdigest())
        session.add(admin)

    def init_settings(session):
        settings = gen_setting()
        for k in settings:
            type_ = Setting.get_type(settings[k])
            if type_ == "Iterable":
                s = Setting(keyname=k, value=json.dumps(settings[k], ensure_ascii=False), type=type_)
            else:
                s = Setting(keyname=k, value=str(settings[k]), type=type_)
            session.add(s)

    print("=====创建表=====")
    engine = create_engine(config.CONNECTION, echo=True)
    Base.metadata.create_all(engine)
    session = config.makeNewSession()
    # 初始化管理员账号
    init_admin(session)
    # 初始化设置
    init_settings(session)

    session.commit()
    session.close()

def drop_tables():
    print("=====删除表=====")
    engine = create_engine(config.CONNECTION, echo=True)
    Base.metadata.drop_all(engine)

if __name__=="__main__":
    drop_tables()
    sleep(3)
    init_tables()