zfbox/core/DBManager.py

import pymysql
import sqlite3
import threading
import os
import json
from myutils.ConfigManager import myCongif
from myutils.MyLogger_logger import LogHandler

class DBManager():
    #实例化数据库管理对象，并连接数据库
    #itype=0  使用mysql数据库，1-使用sqlite数据库
    def __init__(self):
        self.logger = LogHandler().get_logger("DBManager")
        self.lock = threading.Lock()
        self.itype = myCongif.get_data("DBType")
        if self.itype ==0:
            self.host = myCongif.get_data('mysql.host')
            self.port = myCongif.get_data('mysql.port')
            self.user = myCongif.get_data('mysql.user')
            self.passwd = myCongif.get_data('mysql.passwd')
            self.database = myCongif.get_data('mysql.database')
            self.connection = None
            self.cursor = None
        elif self.itype ==1:
            self.dbfile = myCongif.get_data("sqlite")
            if not os.path.exists(self.dbfile):
                self.dbfile = "../" + self.dbfile   #直接运行DBManager时初始路径不是在根目录
                if not os.path.exists(self.dbfile):
                    raise FileNotFoundError(f"Database file {self.dbfile} does not exist.")
        else:
            self.logger.error("错误的数据库类型，请检查")


    def __del__(self):
        if self.ok:
            self.cursor.close()
            self.connection.close()
            self.cursor = None
            self.connection = None
        self.logger.debug("DBManager销毁")

    def connect(self):
        try:
            if self.itype ==0:
                self.connection = pymysql.connect(host=self.host, port=self.port, user=self.user,
                                                  passwd=self.passwd, db=self.database,charset='utf8')
                self.cursor = self.connection.cursor()
            elif self.itype ==1:
                self.connection = sqlite3.connect(self.dbfile)
                self.cursor = self.connection.cursor()
            self.ok = True
            return True
        except:
            self.logger.error("服务器端数据库连接失败")
            return False

    # 判断数据库连接是否正常，若不正常则重连接
    def Retest_conn(self):
        if self.itype == 0: #除了mysql，sqlite3不需要判断连接状态
            try:
                self.connection.ping()
            except:
                return self.connect()
        return True

    # 执行数据库查询操作 1-只查询一条记录，其他所有记录
    def do_select(self, strsql, itype=0):
        # self.conn.begin()
        self.lock.acquire()
        data = None
        if self.Retest_conn():
            try:
                self.cursor.execute(strsql)
                self.connection.commit()  # select要commit提交事务，是存在获取不到最新数据的问题（innoDB事务机制）
            except Exception as e:
                self.logger.error("do_select异常报错:%s" % str(e))
                self.lock.release()
                return None
            if itype == 1:
                data = self.cursor.fetchone()
            else:
                data = self.cursor.fetchall()
        self.lock.release()
        return data

    # 执行数据库语句
    def do_sql(self, strsql, data=None):
        self.lock.acquire()
        bok = False
        if self.Retest_conn():
            try:
                # self.conn.begin()
                if data:
                    iret = self.cursor.executemany(strsql, data)    #批量执行sql语句
                else:
                    iret = self.cursor.execute(strsql)
                self.connection.commit()
                if iret.rowcount > 0:  # 要有修改成功的记录才返回true
                    bok = True
            except Exception as e:
                self.logger.error("执行数据库语句%s出错:%s" % (strsql, str(e)))
                self.connection.rollback()
        self.lock.release()
        return bok

    #---------------------特定数据库操作函数---------------------
    #根据通道ID或者模型ID删除通道和模型间的关联数据  1-通道ID，2-模型ID
    def delC2M(self,ID,itype):
        #channel2model
        if itype ==1:
            strsql = f"select ID from channel2model where channel_id={ID};"
            datas = self.do_select(strsql)

            strsql = f"delete from channel2model where channel_id={ID};"
            ret = self.do_sql(strsql)
            if ret == False:
                return False
        elif itype ==2:
            strsql = f"select ID from channel2model where model_id={ID};"
            datas = self.do_select(strsql)

            strsql = f"delete from channel2model where model_id={ID};"
            ret = self.do_sql(strsql)
            if ret == False:
                return False
        else:
            return  False
        #schedule
        for data in datas:
            c2m_id = data[0]
            strsql = f"delete from schedule where channel2model_id={c2m_id};"
            ret = self.do_sql(strsql)
            if ret == False:
                return False
        return True

    #删除通道，需要关联删除布防时间，通道和算法的关联表
    def delchannel(self,ID):
        ret = self.delC2M(ID)
        if ret == False:
            return False
        #channel
        strsql = f"delete from channel where ID={ID};"
        ret = self.do_sql(strsql)
        if ret == False:
            return False
        return True

    #修改视频通道和算法间的关联关系
    #channel_id  通道ID
    #modell_list 最新配置的模型id list
    def updateC2M(self,channel_id,model_list):
        strsql = f"select model_id from channel2model where channel_id={channel_id};"
        datas = set(self.do_select(strsql))
        data_new = set(model_list)
        #计算要新增和修改的
        need_add = data_new - datas
        need_del = datas-data_new
        #新增
        for one in need_add:
            strsql = f"insert into channel2model (channel_id,model_id) values ({channel_id},{one});"
            if self.do_sql(strsql) == False:
                return False
            #初始化布防时间 -- 全1
            strsql = f"select ID from channel2model where channel_id={channel_id} and model_id={one};"
            data = mDBM.do_select(strsql,1)
            schedule_data_str = ("{'6': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], "
                                 "'0': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], "
                                 "'1': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], "
                                 "'2': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], "
                                 "'3': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],"
                                 "'4': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],"
                                 "'5': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]}")
            schedule_data = json.loads(schedule_data_str.replace("'", '"'))
            for day, hours in schedule_data.items():
                for hour, status in enumerate(hours):
                    strsql = (
                        f"insert into schedule (channel2model_id,day,hour,status) values ({ID},'{day}',{hour},{status})"
                        f" on conflict(channel2model_id,day,hour) do update set status=excluded.status;")
                    ret = mDBM.do_sql(strsql)
                    if not ret:
                        return ret

        #差删除
        for one in need_del:
            strsql = f"select ID from channel2model where channel_id={channel_id} and model_id={one};"
            c2m_id = mDBM.do_select(strsql,1)[0]
            #删除布防计划数据
            strsql = f"delete from schedule where channel2model_id={c2m_id};"
            if self.do_sql(strsql) == False:
                return False
            #删除关联记录
            strsql = f"delete from channel2model where ID = {c2m_id};"
            if self.do_sql(strsql) == False:
                return False
        return True

    #检查设备ID是否在数据库？
    def checkDevID(self,cID):
        pass

    def test(self):
        # 建立数据库连接
        conn = pymysql.connect(
            host='localhost',
            port=3306,
            user='username',
            password='password',
            database='database_name'
        )

        # 创建游标对象
        cursor = conn.cursor()

        # 执行 SQL 查询
        query = "SELECT * FROM table_name"
        cursor.execute(query)

        # 获取查询结果
        result = cursor.fetchall()

        # 输出结果
        for row in result:
            print(row)

        # 关闭游标和连接
        cursor.close()
        conn.close()

mDBM = DBManager()
mDBM.connect()

if __name__ == "__main__":
    lista = set([1,2,3,4,5,6,7])
    listb = set([4,6,8,9,10])
    nadd = lista -listb
    ndel = listb -lista
    for one in nadd:
        print(one)