renyupeng 3 vuotta sitten
vanhempi
commit
777a3e50fa

+ 103 - 0
Apietl/conn/TidbProConn.py

@@ -0,0 +1,103 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2021/9/15 11:06 上午
+# @Site    :
+# @File    : TidbProConn.py
+# @Software: PyCharm
+# @contact: renyupeng@c-top.com.cn
+import logging
+import time
+from sqlite3 import OperationalError
+
+import pymysql
+from Apietl.constant.ConfConstant import ConfConstant
+
+
+class TidbProConn:
+    """
+    Tidb连接管理
+    """
+
+    @staticmethod
+    def get_connection():
+        """
+        获取Tidb连接
+        :return: 连接
+        """
+        return pymysql.connect(host=ConfConstant.TIDB_PRO_HOST, user=ConfConstant.TIDB_PRO_USER,
+                               passwd=ConfConstant.TIDB_PRO_PASSWORD,
+                               port=ConfConstant.TIDB_PRO_PORT, charset="utf8")
+
+    @classmethod
+    def _reConn(cls, num=28800, stime=5):  # 重试连接总次数为1天,这里根据实际情况自己设置,如果服务器宕机1天都没发现就......
+        _number = 0
+        _status = True
+        while _status and _number <= num:
+            try:
+                TidbProConn.get_connection().ping()  # cping 校验连接是否异常
+                _status = False
+            except:
+                if TidbProConn.get_connection() == True:  # 重新连接,成功退出
+                    _status = False
+                    break
+                _number += 1
+                time.sleep(stime)  # 连接不成功,休眠3秒钟,继续循环,知道成功或重试次数结束
+
+    @classmethod
+    def quary(cls, sql):
+        mysql_conn = TidbProConn.get_connection()
+        TidbProConn._reCon()
+        cur = mysql_conn.cursor()
+        cur.execute(sql)
+        result = cur.fetchall()
+        return result
+
+    @staticmethod
+    def conn_close(conn: pymysql.connections.Connection):
+        """
+        关闭资源
+        """
+        if conn:
+            try:
+                conn.close()
+            except pymysql.err.Error:
+                pass
+
+    @staticmethod
+    def exec_sql(conn, sql_buff):
+        sql_list = sql_buff.split(";")
+        cur = conn.cursor()
+        for sql in sql_list:
+            if sql.strip():
+                cur.execute(sql)
+        return cur
+
+    @staticmethod
+    def upsert(conn, sql_buff):
+        TidbProConn._reCon()
+        try:
+            TidbProConn.exec_sql(conn, sql_buff)
+        except Exception as e:
+            logging.error('失败详细信息:', repr(e))
+        TidbProConn.get_connection().commit()
+
+    @classmethod
+    def _reCon(cls):
+        """ MySQLdb.OperationalError异常"""
+        # self.con.close()
+        while True:
+            try:
+                TidbProConn.get_connection().ping()
+                break
+            except OperationalError:
+                TidbProConn.get_connection().ping(True)
+
+
+if __name__ == '__main__':
+    db = TidbProConn.get_connection()
+    cursor = db.cursor()
+    # 执行sql语句
+    sql = "show databases"
+    cursor.execute(sql)
+    result = cursor.fetchall()
+    print("result====", result)

+ 6 - 0
Apietl/constant/ConfConstant.py

@@ -26,6 +26,12 @@ class ConfConstant:
     TIDB_USER = _cfp.get('tidb', 'tidb_user')
     TIDB_PASSWORD = _cfp.get('tidb', 'tidb_password')
 
+    # tidb 配置信息
+    TIDB_PRO_HOST = _cfp.get('tidb_pro', 'tidb_host')
+    TIDB_PRO_PORT = _cfp.getint('tidb_pro', 'tidb_port')
+    TIDB_PRO_USER = _cfp.get('tidb_pro', 'tidb_user')
+    TIDB_PRO_PASSWORD = _cfp.get('tidb_pro', 'tidb_password')
+
     # xxl-job 配置
     xxl_url = _cfp.get('xxl_job', 'xxl_url')
 

+ 3 - 1
Apietl/kwai_etl/report/AccountDailyReport.py

@@ -9,9 +9,10 @@
 import logging
 from math import ceil
 
+from Apietl.kwai_etl.report.ReplaceAccountInfo import ReplaceAccountInfo
 from Apietl.utlis.Logger import Logger
 from Apietl.utlis.Utils import Utils
-from TidbConn import TidbConn
+from Apietl.conn.TidbConn import TidbConn
 
 
 class AccountDailyReport:
@@ -22,6 +23,7 @@ class AccountDailyReport:
         self.end_time = self.parm[1]
 
     def daily_report(self):
+        ReplaceAccountInfo.handler()
         try:
             sql = """select count(1) from media_api.kuaishou_account_report_daily where 
             stat_date between {start_time} and {end_time}""".format(

+ 2 - 0
Apietl/kwai_etl/report/AccountHourlyReport.py

@@ -9,6 +9,7 @@
 import logging
 from math import ceil
 
+from Apietl.kwai_etl.report.ReplaceAccountInfo import ReplaceAccountInfo
 from Apietl.utlis.Logger import Logger
 from Apietl.utlis.Utils import Utils
 from TidbConn import TidbConn
@@ -22,6 +23,7 @@ class AccountHourlyReport:
         self.end_time = self.parm[1]
 
     def daily_report(self):
+        ReplaceAccountInfo.handler()
         try:
             sql = """select count(1) from media_api.kuaishou_account_report_hourly where 
             stat_date between {start_time} and {end_time}""".format(

+ 2 - 0
Apietl/kwai_etl/report/AccountReport.py

@@ -8,6 +8,7 @@
 
 import logging
 from Apietl.conn.TidbConn import TidbConn
+from Apietl.kwai_etl.report.ReplaceAccountInfo import ReplaceAccountInfo
 from Apietl.utlis.Utils import Utils
 
 
@@ -16,6 +17,7 @@ class AccountReport:
         self.conn = TidbConn.get_connection()
 
     def daily_report(self, account_id, start_date, end_date):
+        ReplaceAccountInfo.handler()
         try:
             sql = """
             replace into application.kuaishou_account_report_daily_dw

+ 2 - 0
Apietl/kwai_etl/report/CampaignDailyReport.py

@@ -9,6 +9,7 @@
 import logging
 from math import ceil
 
+from Apietl.kwai_etl.report.ReplaceAccountInfo import ReplaceAccountInfo
 from Apietl.utlis.Logger import Logger
 from Apietl.utlis.Utils import Utils
 from TidbConn import TidbConn
@@ -22,6 +23,7 @@ class CampaignDailyReport:
         self.end_time = self.parm[1]
 
     def daily_report(self):
+        ReplaceAccountInfo.handler()
         try:
             sql = """select count(1) from media_api.kuaishou_campaign_report_daily where 
             stat_date between {start_time} and {end_time}""".format(

+ 2 - 0
Apietl/kwai_etl/report/CampaignHourlyReport.py

@@ -9,6 +9,7 @@
 import logging
 from math import ceil
 
+from Apietl.kwai_etl.report.ReplaceAccountInfo import ReplaceAccountInfo
 from Apietl.utlis.Logger import Logger
 from Apietl.utlis.Utils import Utils
 from TidbConn import TidbConn
@@ -22,6 +23,7 @@ class CampaignHourlyReport:
         self.end_time = self.parm[1]
 
     def hourly_report(self):
+        ReplaceAccountInfo.handler()
         try:
             sql = """select advertiser_id from media_api.kuaishou_campaign_report_hourly where 
             stat_date between {start_time} and {end_time} group by advertiser_id""".format(

+ 1 - 0
Apietl/kwai_etl/report/MateirialVideoDailyReport.py

@@ -9,6 +9,7 @@
 import logging
 from math import ceil
 
+from Apietl.kwai_etl.report.ReplaceAccountInfo import ReplaceAccountInfo
 from Apietl.utlis.Logger import Logger
 from Apietl.utlis.Utils import Utils
 from TidbConn import TidbConn

+ 1 - 0
Apietl/kwai_etl/report/MateirialVideoHourlyReport.py

@@ -9,6 +9,7 @@
 import logging
 from math import ceil
 
+from Apietl.kwai_etl.report.ReplaceAccountInfo import ReplaceAccountInfo
 from Apietl.utlis.Logger import Logger
 from Apietl.utlis.Utils import Utils
 from Apietl.conn.TidbConn import TidbConn

+ 50 - 0
Apietl/kwai_etl/report/ReplaceAccountInfo.py

@@ -0,0 +1,50 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2021/10/26 10:15 上午
+# @Site    :
+# @File    : ReplaceAccountInfo.py
+# @Software: PyCharm
+# @contact: renyupeng@c-top.com.cn
+# @Tel 1501435553
+import logging
+
+from Apietl.conn.TidbConn import TidbConn
+
+
+class ReplaceAccountInfo:
+
+    @staticmethod
+    def handler():
+        try:
+            sql = """
+replace into `jeecg-boot`.ctop_account_attribution_info(id, account_id, project_id,product_id, advertiser_id, company_id,account_name, project_name, advertiser_name, company_name, sale_id, user_id,org_code)
+select cua.account_id,
+       cua.account_id,
+       cp.id,
+       product.id,
+       ca.id,
+       uc.company_id,
+       cua.auth_name,
+       cp.project_name,
+       ca.name,
+       uc.company_name,
+       cp.sale_id,
+       su.id,
+       su.org_code
+from `jeecg-boot`.ctop_user_allocation cua
+         left join `jeecg-boot`.ctop_project cp on cua.project_id = cp.id
+         left join `jeecg-boot`.ctop_advertiser ca on ca.id = cua.advertiser_id
+         left join `jeecg-boot`.sys_user su on su.id = cua.user_id
+         left join `jeecg-boot`.user_company uc on uc.user_id = cua.user_id
+         left join `jeecg-boot`.ctop_product product on product.id = cp.product_id
+where cua.account_id is not null
+            """
+            TidbConn.upsert(conn=TidbConn.get_connection(), sql_buff=sql)
+            TidbConn.get_connection().commit()
+            TidbConn.conn_close(conn=TidbConn.get_connection())
+        except Exception as e:
+            logging.error(e)
+
+
+if __name__ == '__main__':
+    ReplaceAccountInfo.handler()

+ 2 - 0
Apietl/kwai_etl/report/UnitDailyReport.py

@@ -9,6 +9,7 @@
 import logging
 from math import ceil
 
+from Apietl.kwai_etl.report.ReplaceAccountInfo import ReplaceAccountInfo
 from Apietl.utlis.Logger import Logger
 from Apietl.utlis.Utils import Utils
 from TidbConn import TidbConn
@@ -22,6 +23,7 @@ class UnitDailyReport:
         self.end_time = self.parm[1]
 
     def daily_report(self):
+        ReplaceAccountInfo.handler()
         try:
             sql = """
             select count(1) from media_api.kuaishou_unit_report_daily where 

+ 2 - 0
Apietl/kwai_etl/report/UnitHourlyReport.py

@@ -9,6 +9,7 @@
 import logging
 from math import ceil
 
+from Apietl.kwai_etl.report.ReplaceAccountInfo import ReplaceAccountInfo
 from Apietl.utlis.Utils import Utils
 from TidbConn import TidbConn
 
@@ -21,6 +22,7 @@ class UnitHourlyReport:
         self.end_time = self.parm[1]
 
     def hourly_report(self):
+        ReplaceAccountInfo.handler()
         try:
             sql = """select advertiser_id from media_api.kuaishou_unit_report_hourly where 
             stat_date between {start_time} and {end_time} group by advertiser_id""".format(