| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205 | 
							- #!/usr/bin/env python
 
- # -*- encoding: utf-8 -*-
 
- """
 
- Author renyupeng
 
- coding=utf-8
 
- @Time    : 2023/2/7 下午
 
- @Site    :
 
- @File    : mysql_utils.py
 
- @Software: PyCharm
 
- @contact: renyupeng@c-top.com.cn
 
- @Tel 1501435553
 
- encoding=utf-8
 
- """
 
- import pymysql
 
- # 导入所有Mysql配置常量,请自行指定文件
 
- from constant.ConfConstant import ConfConstant
 
- from utils.send_feishu_msg import SendFeiShuMsg
 
- class MysqlProUtils(object):
 
-     """
 
-     mysql操作类,对mysql数据库进行增删改查
 
-     """
 
-     def __init__(self):
 
-         config = dict(
 
-             host=ConfConstant.TIDB_PRO_HOST,
 
-             user=ConfConstant.TIDB_PRO_USER,
 
-             password=ConfConstant.TIDB_PRO_PASSWORD,
 
-             port=ConfConstant.TIDB_PRO_PORT,
 
-             charset='utf8mb4',  # 编码要加上,否则可能出现中文乱码问题
 
-             use_unicode=False,
 
-         )
 
-         # Connect to the database
 
-         self.connection = pymysql.connect(**config)
 
-         self.connection.autocommit(True)
 
-         self.cursor = self.connection.cursor()
 
-     def QueryAll(self, sql):
 
-         """
 
-         查询所有数据
 
-         :param sql:
 
-         :return:
 
-         """
 
-         # 数据库若断开即重连
 
-         self.reConnect()
 
-         self.cursor.execute(sql)
 
-         return self.cursor.fetchall()
 
-     def QueryMany(self, sql, n):
 
-         """
 
-         查询某几条数据数据
 
-         :param sql:
 
-         :return:
 
-         """
 
-         # 数据库若断开即重连
 
-         self.reConnect()
 
-         self.cursor.execute(sql)
 
-         return self.cursor.fetchmany(n)
 
-     def QueryOne(self, sql):
 
-         """
 
-         查询某几条数据数据
 
-         :param sql:
 
-         :return:
 
-         """
 
-         # 数据库若断开即重连
 
-         self.reConnect()
 
-         self.cursor.execute(sql)
 
-         return self.cursor.fetchone()
 
-     # return self.cursor.fetchone()
 
-     def reConnect(self):
 
-         """
 
-         重连机制
 
-         :return:
 
-         """
 
-         try:
 
-             self.connection.ping()
 
-         except:
 
-             self.connection()
 
-     def Operate(self, sql, params=None, DML=True):
 
-         """
 
-         数据库操作:增删改查
 
-         DML: insert / update / delete
 
-         DDL: CREATE TABLE/VIEW/INDEX/SYN/CLUSTER
 
-         """
 
-         try:
 
-             # 数据库若断开即重连
 
-             self.reConnect()
 
-             with self.connection.cursor() as cursor:
 
-                 cursor.execute(sql, params)
 
-                 self.connection.commit()
 
-         except Exception as e:
 
-             if DML:
 
-                 # 涉及DML操作时,若抛异常需要回滚
 
-                 self.connection.rollback()
 
-             print(e)
 
-     def __del__(self):
 
-         """
 
-         MysqlConnection实例对象被释放时调用此方法,用于关闭cursor和connection连接
 
-         """
 
-         self.cursor.close()
 
-         self.connection.close()
 
-     @staticmethod
 
-     def batch_insert(item_list):
 
-         con = pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
 
-                               passwd=ConfConstant.TIDB_PRO_PASSWORD, charset='utf8', port=3390)
 
-         # 对数据库进行插入操作,并不需要commit,twisted会自动commit
 
-         placeholders = ', '.join(['%s'] * len(tuple(item_list[0])))
 
-         print('====placeholders====', placeholders)
 
-         columns = ', '.join(item_list[0].keys())
 
-         print('====columns====', columns)
 
-         insert_sql = "REPLACE INTO ruixuan.promoter_info ( %s ) VALUES ( %s )" % (
 
-             columns, placeholders)
 
-         cur = con.cursor()
 
-         data = []
 
-         try:
 
-             for item in item_list:
 
-                 value = tuple(item.values())
 
-                 data.append(value)
 
-             cur.executemany(insert_sql, data)
 
-             con.commit()
 
-         except Exception as e:
 
-             SendFeiShuMsg.send_robot_msg('数据批量插入错误{e}'.format(e=e))
 
-             con.rollback()
 
-         con.close()
 
-     @staticmethod
 
-     def insert(item):
 
-         print(item, '---------')
 
-         con = pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
 
-                               passwd=ConfConstant.TIDB_PRO_PASSWORD, charset='utf8', port=3390)
 
-         async_item = tuple(item.values())
 
-         # 对数据库进行插入操作,并不需要commit,twisted会自动commit
 
-         placeholders = ', '.join(['%s'] * len(item))
 
-         columns = ', '.join(item.keys())
 
-         insert_sql = "REPLACE INTO ruixuan.promoter_info ( %s ) VALUES ( %s )" % (
 
-             columns, placeholders)
 
-         print(insert_sql)
 
-         cur = con.cursor()
 
-         try:
 
-             cur.execute(insert_sql, async_item)
 
-             con.commit()
 
-         except Exception as e:
 
-             SendFeiShuMsg.send_robot_msg('数据插入错误{e}'.format(e=e))
 
-             con.rollback()
 
-         con.close()
 
-     @staticmethod
 
-     def MgsInsert(item):
 
-         print(item, '---------')
 
-         con = pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
 
-                               passwd=ConfConstant.TIDB_PRO_PASSWORD, charset='utf8', port=3390)
 
-         async_item = tuple(item.values())
 
-         # 对数据库进行插入操作,并不需要commit,twisted会自动commit
 
-         placeholders = ', '.join(['%s'] * len(item))
 
-         columns = ', '.join(item.keys())
 
-         insert_sql = "REPLACE INTO miaogousi.promoter_info ( %s ) VALUES ( %s )" % (
 
-             columns, placeholders)
 
-         print(insert_sql)
 
-         cur = con.cursor()
 
-         try:
 
-             cur.execute(insert_sql, async_item)
 
-             con.commit()
 
-         except Exception as e:
 
-             SendFeiShuMsg.send_robot_msg('数据插入错误{e}'.format(e=e))
 
-             con.rollback()
 
-         con.close()
 
-     @staticmethod
 
-     def RocketInsert(item):
 
-         print(item, '---------')
 
-         con = pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
 
-                               passwd=ConfConstant.TIDB_PRO_PASSWORD, charset='utf8', port=3390)
 
-         async_item = tuple(item.values())
 
-         # 对数据库进行插入操作,并不需要commit,twisted会自动commit
 
-         placeholders = ', '.join(['%s'] * len(item))
 
-         columns = ', '.join(item.keys())
 
-         insert_sql = "REPLACE INTO rocket.promoter_info ( %s ) VALUES ( %s )" % (
 
-             columns, placeholders)
 
-         print(insert_sql)
 
-         cur = con.cursor()
 
-         try:
 
-             cur.execute(insert_sql, async_item)
 
-             con.commit()
 
-         except Exception as e:
 
-             SendFeiShuMsg.send_robot_msg('数据插入错误{e}'.format(e=e))
 
-             con.rollback()
 
-         con.close()
 
 
  |