import sys
import os

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from amazon_params.params import starrocks_CONN
from amazon_params.params import REDIS_CONN
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool
import time
import traceback
from datetime import datetime
from kafka import KafkaProducer
import redis
from utils.secure_db_client import get_remote_engine

class BaseUtils(object):
    def __init__(self):
        self.site_name = 'us'
        self.engine = self.mysql_connect()

    def pg_connect(self):
        # db_type_alias_map = {
        #     "mysql": "mysql",  # 阿里云mysql
        #     "postgresql_14": "postgresql_14",  # pg14爬虫库-内网
        #     "postgresql_14_outer": "postgresql_14_outer",  # pg14爬虫库-外网
        #     "postgresql_15": "postgresql_15",  # pg15正式库-内网
        #     "postgresql_15_outer": "postgresql_15_outer",  # pg15正式库-外网
        #     "postgresql_cluster": "postgresql_cluster",  # pg集群-内网
        #     "postgresql_cluster_outer": "postgresql_cluster_outer",  # pg集群-外网
        #     "doris": "doris",  # doris集群-内网
        # }
        engine_pg = get_remote_engine(
            site_name=self.site_name,  # -> database "selection"
            db_type="postgresql_14_outer",  # -> 服务端 alias "mysql"
        )
        print('engine_pg::',engine_pg)
        return engine_pg
        # while True:
        #     try:
        #         if self.site_name == 'us':
        #             db = 'selection'
        #         else:
        #             db = f'selection_{self.site_name}'
        #         engine_pg = create_engine(
        #             f"postgresql+psycopg2://{PG_CONN_DICT['pg_user']}:{PG_CONN_DICT['pg_pwd']}@{PG_CONN_DICT['pg_host']}:{PG_CONN_DICT['pg_port']}/{db}",
        #             encoding='utf-8', connect_args={"connect_timeout": 10, "keepalives": 1,
        #                                             "keepalives_idle": 40,  # 40s 空闲后开始发 心跳链接
        #                                             "keepalives_interval": 20,  # 每 20s 发一次
        #                                             "keepalives_count": 10}, #在空闲 40 秒后，每 20 秒发一次探测，总共探测 10 次
        #             pool_recycle=900,  # 太老的连接（15mi）强制回收，避免中间网络设备回收后无法用，池中连接存活 15 min 后丢弃
        #             pool_size=60,  # 根据并发量适当设置
        #             max_overflow=40)
        #         return engine_pg
        #     except Exception as e:
        #         print("pg_connect 14 t11111111111111111111111:", e, f"\n{traceback.format_exc()}")
        #         time.sleep(3)
        #         continue

    def doris_connect(self):
        engine_doris = get_remote_engine(
            site_name=self.site_name,  # -> database "selection"
            db_type="doris",  # -> 服务端 alias "mysql"
        )
        print('engine_pg::', engine_doris)
        return engine_doris
        # nums = 0
        # while True:
        #     nums += 1
        #     try:
        #         db = 'us_spider'
        #         # 设置连接参数字典，包括连接超时参数
        #         connect_args = {
        #             "connect_timeout": 10
        #         }
        #         return create_engine(
        #             f'mysql+pymysql://{DORIS_CONN["mysql_user"]}:' + f'{DORIS_CONN["mysql_pwd"]}@{DORIS_CONN["mysql_host"]}:{DORIS_CONN["mysql_port"]}/{db}?charset=utf8mb4',
        #             connect_args=connect_args, poolclass=NullPool)
        #     except Exception as e:
        #         print("doris_connect22222222222222222222222222:", e, f"\n{traceback.format_exc()}")
        #         time.sleep(3)
        #         continue

    def pg_connect_6(self):
        engine_pg15 = get_remote_engine(
            site_name=self.site_name,  # -> database "selection"
            db_type="postgresql_15_outer",  # -> 服务端 alias "mysql"
        )
        print('engine_pg15::', engine_pg15)
        return engine_pg15

    def pg_reconnect(self, table_name=None, e=None):
        self.engine_pg = self.pg_connect()

    def mysql_connect(self):
        """
        Connection of mysql.
        """
        engine_mysql = get_remote_engine(
            site_name=self.site_name,  # -> database "selection"
            db_type="mysql",  # -> 服务端 alias "mysql"
        )
        return engine_mysql
        # nums = 0
        # while True:
        #     nums += 1
        #     try:
        #         if self.site_name == 'us':
        #             db = 'selection'
        #         else:
        #             db = f'selection_{self.site_name}'
        #         # 设置连接参数字典，包括连接超时参数
        #         url = URL.create(
        #             drivername="mysql+pymysql",
        #             username=DB_CONN_DICT["mysql_user"],
        #             password=DB_CONN_DICT["mysql_pwd"],  # 原始密码，含 @ 也没问题
        #             host=DB_CONN_DICT["mysql_host"],
        #             port=int(DB_CONN_DICT["mysql_port"]),
        #             database=db,
        #             query={"charset": "utf8mb4"}
        #         )
        #         return create_engine(url, connect_args={"connect_timeout": 10}, poolclass=NullPool )  # , pool_recycle=3600
        #     except Exception as e:
        #         print("error_mysql_connect22222222222222222222222222:", e, f"\n{traceback.format_exc()}")
        #         time.sleep(3)
        #         continue

    def starrocks_connect(self):
        """
        Connection of starrocks_connect.
        """
        nums = 0
        while True:
            nums += 1
            try:
                db = 'us_spider'
                # 设置连接参数字典，包括连接超时参数
                connect_args = {
                    "connect_timeout": 10
                }
                return create_engine(
                    f'mysql+pymysql://{starrocks_CONN["mysql_user"]}:' + f'{starrocks_CONN["mysql_pwd"]}@{starrocks_CONN["mysql_host"]}:{starrocks_CONN["mysql_port"]}/{db}?charset=utf8mb4',
                    connect_args=connect_args, poolclass=NullPool)
            except Exception as e:
                print("error_mysql_connect22222222222222222222222222:", e, f"\n{traceback.format_exc()}")
                time.sleep(3)
                continue

    def mysql_reconnect(self, table_name=None, e=None):
        print(f"{table_name}表报错，等待5s继续：", e, f"\n{traceback.format_exc()}")
        if "2013, 'Lost connection to MySQL server during query" in str(e):
            print("连接异常重新连接")
            self.engine = self.mysql_connect()
        time.sleep(5)

    def kafuka_connect(self, kafka_html_connect=None, bootstrap_servers=None, acks=None, connections_max_idle_ms=60000):
        request_timeout_ms = 30000
        if kafka_html_connect:
            bootstrap_servers = '61.145.136.61:20092'
        else:
            if bootstrap_servers:
                bootstrap_servers = '192.168.10.210:9092,192.168.10.218:9092,192.168.10.219:9092,192.168.10.220:9092'
            else:
                bootstrap_servers = '61.145.136.61:29092,61.145.136.61:49092,61.145.136.61:59092'
        print(bootstrap_servers)
        while True:
            try:
                producer = KafkaProducer(
                    bootstrap_servers=bootstrap_servers,
                    security_protocol='SASL_PLAINTEXT',
                    sasl_mechanism='PLAIN',
                    sasl_plain_username='producer',
                    sasl_plain_password='R8@xY3pL!qz',
                    value_serializer=str.encode,
                    max_request_size=10000120,
                    request_timeout_ms=request_timeout_ms,
                    max_block_ms=30000,  # 阻塞超时时间设置为60秒
                    compression_type='gzip',  # 启用消息压缩
                    acks=1 if acks else 0,  # 根据需要设置 acks,  # 等待所有副本确认接收
                    connections_max_idle_ms=connections_max_idle_ms,  # 一分钟释放链接
                    max_in_flight_requests_per_connection=1000,
                    linger_ms=150,  # 增加等待时间
                    batch_size=16384 if acks else 0,  # 增加批处理大小
                    api_version=(2, 4, 1),  # 我的kafka版本是2.4.1
                    retries=10,  # 自动重试
                    retry_backoff_ms=500
                )
                return producer
            except Exception as e:
                print("kafuka_connect 链接报错:", e, f"\n{traceback.format_exc()}")
                time.sleep(10)
                continue

    def redis_db(self):
        nums = 0
        while True:
            nums += 1
            try:
                redis_db14 = redis.Redis(host=REDIS_CONN['redis_host'], port=REDIS_CONN['redis_port'],
                                         password=REDIS_CONN['redis_pwd'], db=REDIS_CONN['redis_db'])
                return redis_db14
            except Exception as e:
                print("redis_db 链接报错:", e, f"\n{traceback.format_exc()}")
                time.sleep(3)
                continue

    def created_path(self, path_table, week):
        # 获取当前时间
        current_time = datetime.now()
        # 格式化为整点时间形式
        formatted_time = current_time.strftime('%Y%m%d%H')
        # 站点/周/每小时/表名/csv
        path_1 = fr"/run/{self.site_name}/{week}/{formatted_time}/{path_table}"
        if os.path.exists(path_1) == False:  # 判断路径是否存在
            os.makedirs(path_1)
        t = time.time()
        t_int = int(t * 10000000)
        created_at = current_time.strftime('%Y-%m-%d %H:%M:%S')
        # 获取当前登录用户的用户名
        current_user = os.getlogin()
        file_csv = f'{current_user}_{t_int}'
        return path_1, file_csv, formatted_time, created_at


if __name__ == '__main__':
    BaseUtils().kafuka_connect()
