zlwl-algos-new/COMMON/utils/ProUtils.py

import configparser
import logging
import logging.handlers
import os
import re
from urllib import parse
import json

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import oss2
import redis
from kafka import KafkaConsumer,KafkaProducer
from redis import RedisCluster
import pandas as pd
import requests
from alibabacloud_dingtalk.robot_1_0.client import Client as dingtalkrobot_1_0Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_dingtalk.robot_1_0 import models as dingtalkrobot__1__0_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class SysUtils:
    """通过配置文件目录，获取配置文件对应内容
    """
    def __init__(self, cur_env='dev', app_path=os.path.dirname(os.path.dirname(os.path.dirname((os.path.abspath(__file__))))), type=1):
        if type == 1:
            # 项目内部调用
            self._init__(cur_env, app_path)
        else:
            # 外部调用，只使用内部的部分功能
            pass
    def _init__(self, cur_env='dev', app_path=os.path.dirname(os.path.dirname(os.path.dirname((os.path.abspath(__file__)))))):
        self.cur_env = os.environ.get("PRO_ENV", cur_env)  # 环境覆盖
        self.app_path = os.environ.get("APP_PATH", app_path)  # 路径覆盖
        if not os.path.exists(os.path.join(self.app_path, "RESOURCES")):
            raise Exception("$APP_PATH exit, {} is error. {}/RESOURCES is not exist".format(self.app_path, self.app_path))
        if not os.path.exists(os.path.join(self.app_path, "RESOURCES", f"config-{self.cur_env}.ini")):
            raise Exception("config-{} is not exist".format(self.cur_env))
        # if not os.path.exists("{}/Log".format(self.app_path)):
        #     os.makedirs("{}/Log".format(self.app_path))
        print("start env: {} ".format(self.cur_env))
        self._get_cf()
        self.env_params = os.environ # 获取环境变量参数
    
    def _get_cf(self):
        self.cf = configparser.ConfigParser()
        self.cf.read(os.path.join(self.app_path, "RESOURCES", f"config-{self.cur_env}.ini"))
    
    def get_cf_param(self, section):
        """根据不同的type 返回不同的参数
        """
        options = self.cf.options(section)
        params = {}
        for o in options:
            params.update({o:self.cf.get(section, o)})
        return params
        # if type == 'mysql':
        #     host = self.cf.get(tag, 'host')
        #     port = int(self.cf.get(tag, 'port'))
        #     db = self.cf.get(tag, 'db')
        #     user = self.cf.get(tag, 'user')
        #     password = parse.quote_plus(self.cf.get(tag, 'password'))
        #     print("mysql: {} - {}".format(host, port))
        #     return  {'host':host, 'port':port, 'db':db, 'user':user, 'password':password}
        # elif type == 'redis':
        #     host = self.cf.get(tag, 'host')
        #     port = int(self.cf.get(tag, 'port'))
        #     password = self.cf.get(tag, 'password')
        #     cluster_enable = self.cf.get(tag, 'cluster_enable', fallback=None)
        #     print("redis: {} - {}".format(host, port))
        #     return  {'host':host, 'port':port, 'password':password, 'cluster_enable':cluster_enable}
        # elif type == 'hbase':
        #     host = self.cf.get(tag, 'host')
        #     port = int(self.cf.get(tag, 'port'))
        #     db = self.cf.get(tag, 'db', fallback=None)
        #     print("hbase: {} - {}".format(host, port))
        #     return  {'host':host, 'port':port, 'db':db}
        # elif type == 'kafka':
        #     host = self.cf.get(tag, 'host')
        #     port = int(self.cf.get(tag, 'port'))
        #     topic = self.cf.get(tag, 'topic', fallback=None)
        #     group_id = self.cf.get(tag, 'group_id', fallback=None)
        #     print("kafka: {} - {}".format(host, port))
        #     return  {'host':host, 'port':port, 'topic':topic, 'group_id':group_id}
        # elif type == 'oss':
        #     bucket = self.cf.get(tag, 'bucket')
        #     accesskeyId = self.cf.get(tag, 'accesskeyId')
        #     accesskeySecret = self.cf.get(tag, 'accesskeySecret')
        #     endpoint = self.cf.get(tag, 'endpointout')
        #     print("hbase: {} - {}".format(host, port))
        #     return  {'bucket':bucket, 'accesskeyId':accesskeyId, 'accesskeySecret':accesskeySecret, 'endpoint':endpoint}
        
    def get_log_handler(self, log_file, level=logging.INFO):
         # 根据日期滚动
        if not os.path.exists(os.path.dirname(log_file)):
            os.makedirs(os.path.dirname(log_file))
        fh = logging.handlers.TimedRotatingFileHandler(filename=log_file, when="D", interval=1, backupCount=7,
                                                        encoding="utf-8")
        formatter = logging.Formatter("%(asctime)s - %(name)s-%(levelname)s %(message)s")
        fh.suffix = "%Y-%m-%d_%H-%M-%S"
        fh.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}")
        fh.setFormatter(formatter)
        fh.setLevel(level)
        return fh
        
    def get_logger(self, log_name, log_path):
        # 日志配置
        logger = logging.getLogger(log_name)
        logger.setLevel(logging.INFO)
        logger.addHandler(self.get_log_handler("{}/{}.info.log".format(log_path, log_name), logging.INFO))
        logger.addHandler(self.get_log_handler("{}/{}.error.log".format(log_path, log_name), logging.ERROR))
        return logger
    
    def get_loggers(self, algo_list, log_path, process_num):
        '''根据算啦列表，返回多个logger
        '''
        loggers = {}
        for algo in algo_list:
            algo_log_path = f"{log_path}/{algo}/process_{process_num}" # 设置日志路径
            logger = self.get_logger(algo, algo_log_path)
            logger.info("pid is " + str(os.getpid()))
            loggers.update({algo:logger})
        return loggers
    

    
    
class MysqlUtils:
    def __init__(self):
        pass

    
    def get_mysql_engine(self, params):
        self.db_engine = create_engine(
            "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8".format(
                params['user'],
                parse.quote_plus(params['password']),
                params['host'],
                int(params['port']),
                params['db']
            ),pool_recycle=3600,pool_size=5)
        self.Session = sessionmaker(bind=self.db_engine)
        return self.db_engine, self.Session
    
class RedisUtils:
    def __init__(self):
        pass
    
    def get_redis_conncect(self, params):
        rc = None
        if (r:=params.get('cluster_enable', None)) and r == 'false':
            rc = redis.Redis(host=params['host'], port=int(params['port']), password=params['password'],decode_responses=True) 
        elif (r:=params.get('cluster_enable', None)) and r == 'true':
            rc = RedisCluster(startup_nodes=[{"host":params['host'], "port":int(params['port'])}], password=params['password'],decode_responses=True)
        return rc
    
class KafkaUtils:
    def __init__(self):
        pass
    
    def get_kafka_consumer(self, params, topic_key, group_id_key, client_id="none"):
        consumer = KafkaConsumer(params[topic_key], bootstrap_servers=['{}:{}'.format(params['host'], params['port'])],
                                    client_id=client_id,
                                    auto_offset_reset='latest',# 消费kafka中最近的数据，如果设置为earliest则消费最早的数据，不管这些数据是否消费
                                    enable_auto_commit=True, # 自动提交消费者的offset
                                    auto_commit_interval_ms=3000, ## 自动提交消费者offset的时间间隔,
                                    max_poll_records=1, # 每次拉取消息的最大上限
                                    max_poll_interval_ms=6000000, # 两次拉取消息的最大间隔时长
                                    group_id=params[group_id_key]
            )
        return consumer
    def get_kafka_producer(self, params, client_id="none"):
        producer = KafkaProducer(bootstrap_servers=['{}:{}'.format(params['host'], params['port'])],
                    client_id=client_id, retries=10)
        return producer
    
class OssUtils:
    def __init__(self):
        pass 
    
    def get_bucket(self):
        oss_bucket, oss_accesskeyID, oss_accesskeySecret, oss_endpoint = self.sys_util.get_oss()
        auth = oss2.Auth(oss_accesskeyID, oss_accesskeySecret)
        bucket = oss2.Bucket(auth, "https://{}".format(oss_endpoint), oss_bucket)
        return bucket

    def download_file_oss(self, url, filepath='', filename=''):
        bucket = self.get_bucket()
        url = url.replace('://', '')
        path = url[url.index('/') + 1:]
        bucket.get_object_to_file(path, filepath + filename)
        
class CleanUtils:
    
    def clean(self, mysql_algo_conn=None, mysql_algo_engine=None, mysql_iotp_conn=None, 
              mysql_iotp_engine=None, kafka_consumer=None, rc=None):
        if not mysql_algo_conn is None:
            mysql_algo_conn.close()
            mysql_algo_engine.dispose()
        if not mysql_iotp_conn is None:
            mysql_iotp_conn.close()
            mysql_iotp_engine.dispose()
        if not kafka_consumer is None:
            kafka_consumer.close()
        if not rc is None:
            rc.close()
            
class DingdingUtils():
    
    def __init__(self):
        pass
    # 获取 Access Token
    def get_access_token(self, APP_KEY,APP_SECRET):
        url = f"https://oapi.dingtalk.com/gettoken?appkey={APP_KEY}&appsecret={APP_SECRET}"
        response = requests.get(url)
        data = response.json()
        return data.get('access_token')
    # 上传文件并获取 MediaId
    def upload_file(self, access_token, file_path):
        url = f"https://oapi.dingtalk.com/media/upload?access_token={access_token}&type=file"
        files = {'media': open(file_path, 'rb')}
        response = requests.post(url, files=files)
        data = response.json()
        return data.get('media_id')
    def _create_client(self) -> dingtalkrobot_1_0Client:
        """
        使用 Token 初始化账号Client
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config()
        config.protocol = 'https'
        config.region_id = 'central'
        return dingtalkrobot_1_0Client(config)

    def send_file_to_dingGroup(self, media_id, fileName, date_string, Group_Chat_access_token, access_token, suffix='docx', filetype='docx'):
        client = self._create_client()
        org_group_send_headers = dingtalkrobot__1__0_models.OrgGroupSendHeaders()
        org_group_send_headers.x_acs_dingtalk_access_token = access_token
        File_send_request = dingtalkrobot__1__0_models.OrgGroupSendRequest(
            msg_param=str({    'mediaId':media_id,    'fileName':'{}_{}.{}'.format(fileName,date_string,suffix),   'fileType':filetype}),
            msg_key='sampleFile',
            token=Group_Chat_access_token
        ) 
        try:
            client.org_group_send_with_options(File_send_request, org_group_send_headers, util_models.RuntimeOptions())
        except Exception as err:
            if not UtilClient.empty(err.code) and not UtilClient.empty(err.message):
                # err 中含有 code 和 message 属性，可帮助开发定位问题
                    pass
    def send_markdown_to_dingtalk(self, message_title, markdown_content,Group_Chat_access_token,access_token):
        client = self._create_client()
        org_group_send_headers = dingtalkrobot__1__0_models.OrgGroupSendHeaders()
        org_group_send_headers.x_acs_dingtalk_access_token = access_token
        File_send_request = dingtalkrobot__1__0_models.OrgGroupSendRequest(
            msg_param=str({"title": message_title,"text": markdown_content}),
            msg_key='sampleMarkdown',
            token=Group_Chat_access_token
        ) 
        try:
            client.org_group_send_with_options(File_send_request, org_group_send_headers, util_models.RuntimeOptions())
        except Exception as err:
            if not UtilClient.empty(err.code) and not UtilClient.empty(err.message):
                # err 中含有 code 和 message 属性，可帮助开发定位问题
                    pass  
                
                
            
class DocxUtils():
    def __init__(self):
        pass
    
    def add_table_to_word(self,doc, df):
        # 将DataFrame数据写入表格
        table = doc.add_table(rows=1, cols=len(df.columns))
        table.style = 'Table Grid'
        # 添加表头
        header_cells = table.rows[0].cells
        for col, header in enumerate(df.columns):
            header_cells[col].text = header
            header_cells[col].paragraphs[0].runs[0].bold = True
            header_cells[col].paragraphs[0].alignment = True
        # 添加数据行
        for row_index, row_data in df.iterrows():
            row_cells = table.add_row().cells
            for col, value in enumerate(row_data):
                row_cells[col].text = str(value) 
                row_cells[col].paragraphs[0].alignment = 1
                row_cells[col].vertical_alignment  = 1
                
    def add_table_to_word_with_conbine(self, doc, df, combine_by, combine_columns):
        # 带合并单元格的数据写入word
        # combine_by: 分组列，根据该列决定那些行需要合并
        # combine_columns： 决定哪些列需要合并
        table = doc.add_table(rows=1, cols=len(df.columns))
        table.style = 'Table Grid'
        # 添加表头
        header_cells = table.rows[0].cells
        for col, header in enumerate(df.columns):
            header_cells[col].text = header
            header_cells[col].paragraphs[0].runs[0].bold = True
            header_cells[col].paragraphs[0].alignment = True

        last_row_count = 0
        cur_row_count = 0
        if len(df) > 0:
            for g, df_t in df.groupby(by=combine_by, sort=False):
                # 根据分组列，将数据分组写入表格，然后合并单元格；重复该过程
                cur_row_count = cur_row_count + len(df_t)
                for row_index, row_data in df_t.iterrows():
                    row_cells = table.add_row().cells
                    for col, value in enumerate(row_data):
                        # print(col, value)
                        row_cells[col].text = str(value) 
                        row_cells[col].paragraphs[0].alignment = 1
                        row_cells[col].vertical_alignment  = 1
                # 合并单元格
                
                start_row_num = 1 + last_row_count
                for c in combine_columns:
                    c_idx = df.columns.get_loc(c)
                    text = table.cell(start_row_num, c_idx).text
                    for i in range(start_row_num, start_row_num + len(df_t)):
                        table.cell(start_row_num, c_idx).merge(table.cell(i, c_idx))
                    table.cell(start_row_num, c_idx).text = text
                last_row_count = len(df_t) + last_row_count
    def _set_font_and_style(self, run, font_name, font_size, bold=False, italic=False, underline=False):
        font = run.font
        font.name = font_name
        #font.size = Pt(font_size)
        run.bold = bold
        run.italic = italic
        run.underline = underline
    def set_font_and_style(self, doc):

        # 遍历文档中的段落和文本并设置字体
        for paragraph in doc.paragraphs:
            for run in paragraph.runs:
                self._set_font_and_style(run, 'Arial', 12, bold=True, italic=False, underline=False)

        # 遍历文档中的表格并设置字体
        for table in doc.tables:
            for row in table.rows:
                for cell in row.cells:
                    for paragraph in cell.paragraphs:
                        for run in paragraph.runs:
                            self._set_font_and_style(run, 'Arial', 12, bold=True, italic=False, underline=False)