jikong_backend_code/004_H515/jikong/hard_disk_storage.py

import pymysql
import traceback
import time
import datetime
from log import OutPutLog

class HardDiskStorage():
    def __init__(self, config, port=3306, charset='utf8'):
        self._log = OutPutLog()
        self.host = config['ip']
        self.user = config['username']
        self.passwd = config['password']
        self.db = config['dataBaseName']
        self.port = port
        self.charset = charset
        self.conn = None
        self._conn()

    def _conn(self):
        try:
            self.conn = pymysql.connect(host=self.host, user=self.user, password=self.passwd, database=self.db, port=self.port, autocommit=True)
            return True
        except:
            self._log.error(traceback.format_exc())
            return False

    def _reConn(self, num=28800, stime=3):  # 重试连接总次数为1天,这里根据实际情况自己设置,如果服务器宕机1天都没发现就......
        _number = 0
        _status = True
        while _status and _number <= num:
            try:
                self.conn.ping()  # cping 校验连接是否异常
                _status = False
            except Exception as e:
                self._log.error(e)
                if self._conn() == True:  # 重新连接,成功退出
                    print('haha')
                    _status = False
                    break
                _number += 1
                time.sleep(stime)  # 连接不成功,休眠3秒钟,继续循环，知道成功或重试次数结束

    def get_station_info(self, station_name):
        sql = "SELECT * FROM data_point_tbl where station_name = '%s'" % (station_name)
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            self._log.error(traceback.format_exc())
            return False

    def get_point_info(self, point_tuple):
        if point_tuple:
            sql = "SELECT * FROM data_point_tbl where serial_number in %s" % (str(point_tuple))
        else:
            sql = "SELECT * FROM data_point_tbl"
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            self._log.error(traceback.format_exc())
            return False

    def get_table_data(self, select_info):
        table_name = "table_" + select_info['deviceName']
        time_begin = select_info['timeBegin']
        time_end = select_info['timeEnd']
        column = select_info['pointList']
        if len(column) > 0:
            sql = "SELECT times"
            for column_name in column:
                sql = sql + "," + column_name
            sql = sql + " FROM %s WHERE times > '%s' AND times < '%s'" % (table_name, time_begin, time_end)
        else:
            sql = "SELECT * FROM %s WHERE times > '%s' AND times < '%s';" % (table_name, time_begin, time_end)
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            self._log.error(traceback.format_exc())
            return None



    def get_connectors(self, read_write):
        sql = f"SELECT * FROM station_info_tbl WHERE status = 1 AND `read_write`={read_write};"
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            self._log.error(traceback.format_exc())
            return None

    # def create_delete_stale_data_event(self, eventName, table_name, day):
    #     self.cursor = self.conn.cursor()
    #     sql = "create event %s on SCHEDULE every 1 day do delete from %s where times<(CURRENT_TIMESTAMP() + INTERVAL -%s DAY);" % (
    #         eventName, table_name, day)
    #     self.cursor.execute(sql)
    #     self.cursor.close()
    #
    # def create_many_column_table(self, table_name, list):
    #     self.cursor = self.conn.cursor()
    #     for index in range(len(list)):
    #         dataType = list[index]['storageType']
    #         columnName = "c" + str(list[index]['serialNumber'])
    #         sql_c = "CREATE TABLE IF NOT EXISTS %s (times datetime NOT NULL,INDEX (times)) \
    #             ENGINE=InnoDB DEFAULT CHARSET=utf8;" % (table_name)
    #         sql_add = "ALTER TABLE %s ADD %s %s " % (table_name, columnName, dataType)
    #         try:
    #             self.cursor.execute(sql_c)
    #             self.cursor.execute(sql_add)
    #         except:
    #             self._log.error(traceback.format_exc())
    #     sql_send = "ALTER TABLE %s ADD 'is_send'tinyint NOT NULL DEFAULT '0'" % (table_name)
    #     self.cursor.execute(sql_send)
    #     self.cursor.close()

    def insert_column_many(self, table_name, timeNow, dict):
        try:
            self.cursor = self.conn.cursor()
            sql = "INSERT INTO %s (times" % (table_name)
            for key_name in dict.keys():
                sql = sql + "," + key_name
            sql = sql + ") VALUE ('" + str(timeNow) + "'"
            for key_name in dict.keys():
                data = "," + str(dict[key_name])
                sql = sql + data
            sql = sql + ")"
            try:
                self.cursor.execute(sql)
                # 提交到数据库执行
                self.conn.commit()
            except Exception as e:
                # 如果发生错误则回滚
                self.conn.rollback()
                self._log.error(e)
        except Exception as e:
            self._reConn()
            self._log.error(e)
        else:
            self.cursor.close()

    def close(self):
        self.conn.close()

    def get_command_info(self, station_name):
        sql = "SELECT command FROM command where station_name = %s" % ("'" + station_name + "'")
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            self._log.error(traceback.format_exc())
            return None

    # lee
    def get_device_name_by_station_name(self, station_name):
        sql = "SELECT DISTINCT device_name FROM data_point_tbl WHERE station_name = '%s' " % station_name
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            print(traceback.format_exc())
            return None

    def get_data_point_by_device_name(self, device_name):
        sql = "SELECT * FROM data_point_tbl WHERE device_name = '%s' AND hard_storage=1;" % device_name
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            print(traceback.format_exc())
            return None

    # ----------------------------------------------------------------------------------------------
    def execute_sql(self, sql):
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            print("execute_sql"+str(traceback.format_exc()))
            self._log.error("execute_sql"+str(traceback.format_exc()))
            return None

    def get_table_name(self):
        '''[{'device_name': 'zfdj_electric'}, {'device_name': 'jbd1_electric'}, {'device_name': 'yysbdlfdx_electric'}, ...]'''
        sql = "SELECT DISTINCT device_name FROM data_point_tbl;"
        res = self.execute_sql(sql)

        return res

    # def table_xxx_insert_data(self, format_data_dict, tables_name, save_time):
    #     try:
    #         for table in tables_name:
    #             table_name = "table_" + table['device_name']  # zfdj_electric
    #             sql_columm = "SHOW FIELDS FROM %s WHERE Field like \'c%%\';" % (table_name)
    #             columns = []  # ['c1', 'c2',...,'c10', 'times', 'is_send']
    #             value = []  # [1.401298464324817e-45, 0.0,...,1.401298464324817e-45, 'NOW()', 0]
    #             for each in self.execute_sql(sql_columm):
    #                 columns.append(each['Field'])
    #                 value.append(format_data_dict[each['Field']])
    #             columns.append('times')
    #             value.append(save_time)
    #             columns.append('is_send')
    #             value.append(0)
    #             sql_insert = "INSERT INTO %s %s VALUES %s;" % (table_name, format(tuple(columns)).replace('\'', ''), format(tuple(value)))
    #             print(sql_insert)
    #             # self.execute_sql(sql_insert)
    #     except:
    #         self._log.error(traceback.format_exc())



    def get_device_info(self, device_name):
        sql = "SELECT * FROM data_point_tbl where device_name = '%s'" % (device_name)
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            self._log.error(traceback.format_exc())
            return False


    def execute_update_sql(self, sql):
        """执行UPDATE、INSERT、DELETE语句"""
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql)
            self.conn.commit()
            self.cursor.close()
            return True
        except pymysql.Error as mysql_error:
            self._log.error(f"MYSQL执行失败！{sql}")
            self._log.error(mysql_error)
            return False


    def gather_fish_light_info(self, location):
        """水下聚鱼灯远程控制功能：查找指定位置的聚鱼灯的相关控制信息"""
        sql = f"SELECT control_mode,status,definite_time,conn_status  FROM `gather_fish_light_control` WHERE location=\'{location}\';"

        data = self.execute_sql(sql)
        return data

    def get_gather_fish_light_command(self):
        """水下聚鱼灯远程控制功能：查看控制聚鱼灯的指令"""
        sql = "SELECT control_mode,status,command FROM gather_fish_light_command;"
        data = self.execute_sql(sql)

        return data