yixue_gzh_dj/wx/dbTools.py

from typing import Any, Dict, Callable

from sqlalchemy import create_engine, Integer, Column, String, Boolean, Float, DateTime, Date
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine('mysql+pymysql://yixue_gzh:Lai123@test.db.cxhy.cn/yixue_gzh', pool_pre_ping=True)

Base = declarative_base()


def not_none_of_all(*args: Any):
    result = True
    for arg in args:
        if arg is None:
            result = False
            break
    return result


class DBBase(Base):
    __abstract__ = True

    def __repr__(self):
        result = self.__class__.__name__ + "("
        for attr in self.__dict__:
            if attr.startswith("_") or attr.startswith("__") or attr == "__tablename__":
                continue
            result += attr + "=" + str(getattr(self, attr)) + " "
        result += ")"
        return result

    def to_json_object(self, **kwargs: Any):
        obj = {}
        for attr in self.__dict__:
            if attr.startswith("_") or attr.startswith("__") or attr == "__tablename__":
                continue
            obj[attr] = getattr(self, attr)
        for key, value in kwargs.items():
            obj[key] = value
        return obj

    @classmethod
    def new_session(cls):
        Session = sessionmaker(bind=engine)
        session = Session()
        return session

    @classmethod
    def insert(cls, **kwargs: Any):
        ss = cls.new_session()
        try:
            obj = cls()
            for key in kwargs.keys():
                if hasattr(obj, key):
                    setattr(obj, key, kwargs[key])
            ss.add(obj)
            ss.commit()
        finally:
            ss.close()

    @classmethod
    def insert_all(cls, items: []):
        ss = cls.new_session()
        try:
            lst = []
            for item in items:
                obj = cls()
                for key in item.keys():
                    if hasattr(obj, key):
                        setattr(obj, key, item[key])
                lst.append(obj)
            ss.add_all(lst)
            ss.commit()
        finally:
            ss.close()

    @classmethod
    def delete_by(cls, **kwargs: Any):
        ss = cls.new_session()
        try:
            data = ss.query(cls).filter_by(**kwargs).all()
            for dt in data:
                ss.delete(dt)
            ss.commit()
        finally:
            ss.close()

    @classmethod
    def query_by(cls, **kwargs: Any):
        ss = cls.new_session()
        try:
            data = ss.query(cls).filter_by(**kwargs).all()
            return data
        finally:
            ss.close()

    @classmethod
    def query_first_by(cls, **kwargs: Any):
        ss = cls.new_session()
        try:
            data = ss.query(cls).filter_by(**kwargs).first()
            return data
        finally:
            ss.close()

    @classmethod
    def check_exist(cls, **kwargs: Any):
        return cls.query_first_by(**kwargs) is not None

    @classmethod
    def update(cls, conditions: Callable, values: Dict[str, Any]):
        # DBSysUser.update(conditions=(DBSysUser.id == user.id), values={"finger": finger})
        session = cls.new_session()
        try:
            session.query(cls).filter(conditions).update(values)
            session.commit()
        finally:
            session.close()


class DB_WX_GZH_User(DBBase):
    __tablename__ = 'wx_gzh_user'
    id = Column(Integer, primary_key=True)
    open_id = Column(String)
    user_name = Column(String)
    is_man = Column(Boolean)
    date_mode = Column(Integer)
    born_year = Column(Integer)
    born_month = Column(Integer)
    born_day = Column(Integer)
    born_hour = Column(Integer)
    born_minute = Column(Integer)
    know_time = Column(Integer)
    born_sheng = Column(String)
    born_shi = Column(String)
    born_qu = Column(String)
    nian_gan = Column(String)
    nian_zhi = Column(String)
    yue_gan = Column(String)
    yue_zhi = Column(String)
    ri_gan = Column(String)
    ri_zhi = Column(String)
    shi_gan = Column(String)
    shi_zhi = Column(String)
    nian_cang1 = Column(String)
    nian_cang2 = Column(String)
    nian_cang3 = Column(String)
    yue_cang1 = Column(String)
    yue_cang2 = Column(String)
    yue_cang3 = Column(String)
    ri_cang1 = Column(String)
    ri_cang2 = Column(String)
    ri_cang3 = Column(String)
    shi_cang1 = Column(String)
    shi_cang2 = Column(String)
    shi_cang3 = Column(String)

    def is_know_time(self):
        return self.know_time == 0

    def has_sizhu(self):
        return not_none_of_all(
            self.nian_gan, self.nian_zhi,
            self.yue_gan, self.yue_zhi,
            self.ri_gan, self.ri_zhi,
            self.shi_gan, self.shi_zhi)

    def get_sizhu_text(self):
        return self.nian_gan + self.nian_zhi + " " + self.yue_gan + self.yue_zhi + " " + self.ri_gan + self.ri_zhi + " " + self.shi_gan + self.shi_zhi


class DBWanNianLi(DBBase):
    __tablename__ = 'wan_nian_li'
    id = Column(Integer, primary_key=True)
    nian = Column(Integer)
    yue = Column(Integer)
    ri = Column(Integer)
    nian_gan = Column(String)
    nian_zhi = Column(String)
    yue_gan = Column(String)
    yue_zhi = Column(String)
    ri_gan = Column(String)
    ri_zhi = Column(String)


class DBJieQi(DBBase):
    __tablename__ = 'jie_qi'
    id = Column(Integer, primary_key=True)
    nian = Column(Integer)
    chuxi = Column(String)
    lichun = Column(String)
    jingzhe = Column(String)
    qingming = Column(String)
    lixia = Column(String)
    mangzhong = Column(String)
    xiaoshu = Column(String)
    liqiu = Column(String)
    bailu = Column(String)
    hanlu = Column(String)
    lidong = Column(String)
    daxue = Column(String)
    xiaohan = Column(String)
    yiyue = Column(String)
    yiri = Column(String)


class DBTaiYangShi(DBBase):
    __tablename__ = 'tai_yang_shi'
    id = Column(Integer, primary_key=True)
    month = Column(Integer)
    day = Column(Integer)
    diff = Column(Integer)


class DBAreaInfo(DBBase):
    __tablename__ = 'area_info'
    id = Column(Integer, primary_key=True)
    sheng = Column(String)
    shi = Column(String)
    qu = Column(String)
    weidu = Column(Float)
    jingdu = Column(Float)
    hour = Column(Integer)
    minute = Column(Integer)
    second = Column(Integer)
    shi_pinyin = Column(String)
    qu_pinyin = Column(String)


class DBDaYunNan(DBBase):
    __tablename__ = 'dayun_nan'
    id = Column(Integer, primary_key=True)
    time = Column(DateTime)
    year = Column(Integer)
    month = Column(Integer)
    day = Column(Integer)
    hour = Column(Integer)
    sign = Column(String)


class DBDaYunNv(DBBase):
    __tablename__ = 'dayun_nv'
    id = Column(Integer, primary_key=True)
    time = Column(DateTime)
    year = Column(Integer)
    month = Column(Integer)
    day = Column(Integer)
    hour = Column(Integer)
    sign = Column(String)


class DBNongLi(DBBase):
    __tablename__ = 'nong_li'
    id = Column(Integer, primary_key=True)
    new_date = Column(Date)
    month = Column(String)
    day = Column(String)
    year = Column(Integer)


class DBYiXueUser(DBBase):
    __tablename__ = 'yixue_user'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    isMan = Column(Boolean)
    dateMode = Column(Integer)
    year = Column(Integer)
    month = Column(Integer)
    day = Column(Integer)
    hourMode = Column(Integer)
    hour = Column(Integer)
    minute = Column(Integer)
    areaMode = Column(Integer)
    sheng = Column(String)
    shi = Column(String)
    qu = Column(String)
    extra = Column(String)
    nongli_year = Column(Integer)
    nongli_month = Column(String)
    nongli_day = Column(String)
    req_nian_gan = Column(String)
    req_nian_zhi = Column(String)
    req_yue_gan = Column(String)
    req_yue_zhi = Column(String)
    req_ri_gan = Column(String)
    req_ri_zhi = Column(String)
    req_shi_gan = Column(String)
    req_shi_zhi = Column(String)
    rsp_nian_gan = Column(String)
    rsp_nian_zhi = Column(String)
    rsp_yue_gan = Column(String)
    rsp_yue_zhi = Column(String)
    rsp_ri_gan = Column(String)
    rsp_ri_zhi = Column(String)
    rsp_shi_gan = Column(String)
    rsp_shi_zhi = Column(String)
    owner_id = Column(Integer)


class DBSysUser(DBBase):
    __tablename__ = 'sys_user'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    account = Column(String)
    password = Column(String)
    finger = Column(String)
    dev_change = Column(Integer)
    allow_change_dev = Column(Boolean)