data_analysis/Database/database_.py

import sys
sys.path.append("..")
import pandas as pd
import pymysql
from utils.tools import fmt_date
import config

class DatabaseParam:
    def __init__(self, db_user: str, db_password: str, db_host: str, db_name: str, db_port: int, db_charset: str='utf8mb4'):
        self.db_user = db_user
        self.db_password = db_password
        self.db_host = db_host
        self.db_name = db_name
        self.db_port = db_port
        self.db_charset = db_charset

    @property
    def params(self) -> dict:
        # 执行一些转换或者参数检查, 待补充
        pass
        return {'db_user': self.db_user,
                'db_password': self.db_password,
                'db_host': self.db_host,
                'db_name': self.db_name,
                'db_port': self.db_port,
                'db_charset': self.db_charset
        }


class Database:
    def __init__(self, params: DatabaseParam):
        self.params = params.params  # 参数
        self.db_conn = None  # 连接
        self.cursor = None  # 游标

    def __enter__(self):
        try:
            # 连接失败仍为None
            self.db_conn = pymysql.connect(host=self.params.get('db_host'),
                                         user=self.params.get('db_user'),
                                         password=self.params.get('db_password'),
                                         database=self.params.get('db_name'),
                                         port=self.params.get('db_port'),
                                         charset='utf8mb4')
            self.db_cursor = self.db_conn.cursor()
        except pymysql.MySQLError as e:
            print('数据库连接失败：', e)
            print(f'请检查 host: {self.params.get('db_host')}, user: {self.params.get('db_user')}, password: , database: {self.params.get('db_name')}, port: {self.params.get('db_port')}')
            return None
        if self.db_cursor and self.db_conn: print(f'数据库{self.params.get('db_name')}已连接！')
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):

        if self.db_cursor:
            self.db_cursor.close()
            self.db_cursor = None
        if self.db_conn:
            self.db_conn.close()
            self.db_conn = None
        if self.db_cursor is None and self.db_conn is None: print(f'数据库{self.params.get('db_name')}已断开！')
    def sheet_exists(self, sheet_name: str) -> bool:
        sql = f"""SHOW TABLES FROM {self.params.get('db_name')} LIKE '{sheet_name}'"""
        self.db_cursor.execute(sql)
        result = self.db_cursor.fetchall()
        if len(result) == 0:
            return False
        else:
            return True
    def query_sql_time_series2data_frame(self,
                              project_id:int ,
                              sheet_name:str ,
                              data_code:str,
                              start_year:int , end_year:int ,
                              start_month:int=1, end_month:int=12,
                              start_day:int=1, end_day:int=31,
                              start_hour:int=0, end_hour:int=23,
                              start_minute:int=0, end_minute:int=59,
                              start_second:int=0, end_second:int=59):
        # 时间格式化
        start_datetime, end_datetime = fmt_date(start_year=start_year,start_month=start_month,start_day=start_day,
                 end_year=end_year,end_month=end_month,end_day=end_day,
                 start_hour=start_hour,start_minute=start_minute,start_second=start_second,
                 end_hour=end_hour,end_minute=end_minute,end_second=end_second)
        # 查询语句
        sql = f"""SELECT * FROM {sheet_name} WHERE item_name = '{data_code}' AND project_id = '{project_id}' AND h_time >= '{start_datetime}' AND h_time <= '{end_datetime}'"""
        #print(sql)
        if self.db_cursor is None: raise TypeError('数据库可能未连接，值不能为None.', self.db_cursor)
        # 查询数据
        self.db_cursor.execute(sql)
        result = self.db_cursor.fetchall()
        #result = self.db_cursor.fetchmany(3)
        df = pd.DataFrame(result, columns=[desc[0] for desc in self.db_cursor.description])

        if not len(df):
            print(f'查询到0条数据,序列标签：{data_code}, 数量：{len(df)}')
            return None
        if df.iloc[0]['item_name'].strip() != data_code:
            raise RuntimeError(f'数据库中序列名称与输入不一致，输入：{data_code}, 数据库：{df.iloc[0]['item_name']}')
        # # 消除Nan
        # df.dropna(subset=['val'], inplace=True)  # 不要在数据库这里消除Nan
        # 修改标签
        val_label = df.iloc[0]['item_name'].strip()# + '_val'
        df.rename(columns={'val': f'{val_label}', 'h_time':'time'}, inplace=True)
        # 删除无关列
        df.drop(columns=['project_id', 'item_name'], inplace=True, axis=1)
        # 转换值数据类型
        df[val_label] = df[val_label].astype("float32")
        return df[['time', val_label]]

    def query_sql_time_series_group2data_frame(self,
                              code_name_dict: dict,
                              project_id:int ,
                              sheet_name:str ,
                              data_codes:list,
                              start_year:int , end_year:int ,
                              start_month:int=1, end_month:int=1,
                              start_day:int=1, end_day:int=1,
                              start_hour:int=0, end_hour:int=0,
                              start_minute:int=0, end_minute:int=0,
                              start_second:int=0, end_second:int=0):
        """从数据库中查询多个字段，返回统一结果"""
        frame_list = []
        data_codes = set(data_codes)
        for data_code in data_codes:
            frame = self.query_sql_time_series2data_frame(project_id=project_id,
                                                  sheet_name=sheet_name,
                                                  data_code=str(data_code),
                                                  start_year=start_year,end_year=end_year,
                                                  start_month=start_month,end_month=end_month,
                                                  start_day=start_day, end_day=end_day,
                                                  start_hour=start_hour, end_hour=end_hour,
                                                  start_minute=start_minute, end_minute=end_minute,
                                                  start_second=start_second, end_second=end_second,
                                                )
            if frame is None: continue
            # 过滤常数序列
            if frame[frame.columns[1]].nunique() <= 2:
                print(f'跳过常数列{frame.columns[1]}')
                continue
            frame_list.append(frame)
        # 融合所有字段
        if len(frame_list) == 0: return None

        df_merge = frame_list[0]
        for i in range(1, len(frame_list)):
            df_merge = pd.merge(df_merge, frame_list[i], how='outer', on='time') # 外连接融合所有结果
        # 按照日期排序
        df_merge.sort_values('time', kind='mergesort', inplace=True)

        return df_merge

if  __name__ == '__main__':
    # 创建参数
    db_param = DatabaseParam(
        db_host= '192.168.50.4',
        db_user='root',
        db_password='*B-@p2b+97D5xAF1e6',
        db_name='ws_data',
        db_port=4000)

    # 数据库操作应在内部
    with Database(db_param) as db:
        df_ = db.query_sql_time_series2data_frame(92,
                                 'dc_item_history_data_day',
                                 'QSWGB3_n',
                                 2025, 2025,
                                 3, 9,
                                 25,10)
        print(df_)