sonar/sonar_process/hard_disk_storage.py

import datetime
import math
import re

# import openpyxl
import pymysql
import traceback
import time
# from AES_crypt import decrypt
import configparser


class HardDiskStorage:
    def __init__(self, port=3306, charset='utf8'):
        self.config = configparser.ConfigParser()
        self.config.read("database.conf", encoding="utf-8")
        
        # for section in self.config.sections():
        #     print(f"[{section}]")
        #     for key, value in self.config.items(section):
        #         print(f"{key} = {value}")
        
        self.host = self.config.get("db", "db_host")
        self.user = self.config.get("db", "db_user")
        self.passwd = self.config.get("db", "db_pass")
        self.db = self.config.get("db", "db_name")
        self.port = port
        self.charset = charset
        self.conn = None
        if not self._conn():
            self._reConn()

    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 Exception as e:
            print(f'failed to connect to {self.host}:{self.port}:{self.db} by [{self.user}:{self.passwd}]:{e}')
            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:
                if self._conn():  # 重新连接,成功退出
                    _status = False
                    break
                _number += 1
                time.sleep(stime)  # 连接不成功,休眠3秒钟,继续循环，知道成功或重试次数结束

    # 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:
    #         print(traceback.format_exc())
    #         return False

    # def get_table_data(self, senect_info):
    #     table_name = "table_" + senect_info['deviceName']
    #     time_begin = senect_info['timeBegin']
    #     time_end = senect_info['timeEnd']
    #     column = senect_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:
    #         print(traceback.format_exc())
    #         return None

    # # 历史查询接口（new）--------------------------------------------\\
    # def get_total_count_and_first_id(self, search_info):
    #     table_name = "table_" + search_info['deviceName']
    #     time_begin = search_info['timeBegin']
    #     time_end = search_info['timeEnd']
    #     sql = "select count(*) from %s where times >= '%s' and times <= '%s';" % (table_name, time_begin, time_end)
    #     sql_1 = "select id from %s where times >= '%s' limit 1;" % (table_name, time_begin)
    #     try:
    #         self._reConn()
    #         self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
    #         self.cursor.execute(sql)
    #         count = self.cursor.fetchall()
    #         self.cursor.execute(sql_1)
    #         first_id = self.cursor.fetchall()
    #         if isinstance(first_id, tuple):
    #             first_id = list(first_id)
    #         result = count + first_id
    #         return result
    #     except:
    #         print(traceback.format_exc())
    #         return None

    # def get_item_by_id_offset(self, search_info):
    #     table_name = "table_" + search_info['deviceName']
    #     point_list = search_info['pointList']
    #     id_offset = search_info['idOffset']
    #     quantity = search_info['quantity']
    #     sql = "select times, %s from %s where id  >= %s limit %s" % (','.join(point_list), table_name, id_offset, quantity)
    #     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 quary_table_data(self, search_info):
    #     table_name = "table_" + search_info['deviceName']
    #     time_begin = search_info['timeBegin']
    #     time_end = search_info['timeEnd']
    #     point_list = search_info['pointList']
    #     point_list_1 = str([i[1:] for i in point_list])[1:-1]
    #
    #     sql = "select times, %s from %s where times >= '%s' and times <= '%s';" % (','.join(point_list), table_name, time_begin, time_end)
    #     sql1 = "select io_point_name from data_point_tbl where serial_number in ( %s );" % point_list_1
    #
    #     try:
    #         self._reConn()
    #         self.cursor = self.conn.cursor()
    #         self.cursor.execute(sql)
    #         res = self.cursor.fetchall()
    #         self.cursor.execute(sql1)
    #         res1 = self.cursor.fetchall()
    #         title = [item[0] for item in res1]
    #         title.insert(0, '日期')
    #         self.cursor.close()
    #     except:
    #         print(traceback.format_exc())
    #         return None
    #     book = openpyxl.Workbook()
    #     sheet = book.create_sheet(index=0)
    #     # 循环将表头写入到sheet页
    #     for i in range(len(title)):
    #         if search_info['deviceName'] == 'adcp':  # adcp需要重新计算垂直方向表头
    #             name = re.findall(r"\d+\.?\d*", title[i])  # 提取数字字符
    #             if name:
    #                 title[i] = re.sub(r"\d+\.?\d*", str(round((int(name[0]) * math.sin(14 * math.pi / 180)), 2)), title[i])  # 14为adcp安装倾斜角度，2保留小数的位数
    #         sheet.cell(1, i + 1).value = title[i]
    #     # 写数据
    #     for row in range(0, len(res)):
    #         for col in range(0, len(res[row])):
    #             cell_val = res[row][col]
    #             if isinstance(cell_val, datetime.datetime):
    #                 times = cell_val.strftime("%Y-%m-%d %H:%M:%S")
    #                 sheet.cell(row + 2, col + 1).value = times
    #             else:
    #                 sheet.cell(row + 2, col + 1).value = cell_val
    #     file_path = (table_name + '.xlsx')
    #     savepath = file_path
    #     book.save(savepath)
    #     return savepath
    #
    # # ------------------------------------------------//

    # # 获取insitu指令接口
    # def get_in_situ_command(self):
    #     sql = "select * from shuizhi_insitu_instruct;"
    #     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_connectors(self):
    #     sql = "SELECT * FROM station_info_tbl WHERE status = 1"
    #     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 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:
    #             print(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()
    #             print(e)
    #     except Exception as e:
    #         self._reConn()
    #         print(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:
    #         print(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'" % 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, var):
        try:
            self._reConn()
            self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
            self.cursor.execute(sql, var)
            results = self.cursor.fetchall()
            self.cursor.close()
            return results
        except:
            print(traceback.format_exc())
            return None

    def get_all(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(traceback.format_exc())
            return False