tap order 数据写入mysql

taptap/taptap_report.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8 -*-
+from ops.mtga import FromTga
+from ops.plog import define_logger
+from ops.mmysql import MysqlBase
+import logging
+import datetime
+import sys
+from bson.objectid import ObjectId
+import json
+
+define_logger("/data/logs/ops/taptap_report.log")
+log = logging.getLogger(__name__)
+
+
+class TapTapReport:
+    def __init__(self, day):
+        self.db_conf = {'user': 'mytga', 'pswd': 'gzVwh4HGR68G', 'host': '10.10.3.5', 'db': 'external_data'}
+        self.url = "http://10.10.3.17:8992/querySql"
+        self.api_secret = "n9H4R32ZcjtSeN89ljCY6ESzTmOlnwwnOB3r4YsggnP5M1AXLtKtiS4sS1KKLOEQ"
+        self.tga = FromTga(url=self.url, token=self.api_secret)
+        self.day = day
+
+
+    def build_report(self):
+        pass
+
+    def get_order(self):
+        sql = f"""
+                SELECT
+                    gameid,
+                    catename,
+                    "order" 
+                FROM
+                    v_event_25 
+                WHERE
+                    "$part_date"='{self.day}'
+                
+                """
+        data = self.tga.get_data(sql)
+        if data:
+            self.order2mysql(data)
+            return True
+        else:
+            return False
+
+    def order2mysql(self, data):
+        mydb = MysqlBase(**self.db_conf)
+        table_name = "external_data"
+        for line in data:
+            temp = {}
+            try:
+                temp["gameid"], temp["catename"], temp["order"] = line
+                mydb.insert(table_name, temp)
+            except Exception:
+                log.error(f"instert {line} 2 db failed", exc_info=True)
+
+
+def main():
+    day = (datetime.date.today() - datetime.timedelta(days=1)).strftime('%Y-%m-%d')
+    tap = TapTapReport(day)
+    tap.get_order()
+
+
+if __name__ == "__main__":
+    main()