添加2001 个性化数据(时长，累计用户数)

crontab/cron_2001.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+import datetime
+from ops.init_tga import GetFromTga
+from ops.plog import define_logger
+import logging
+
+define_logger("/data/logs/datacollect_2001.log")
+log = logging.getLogger(__name__)
+
+
+def write_tga(times, event_type, data):
+    kwargs = {}
+    kwargs['gameid'] = 2001
+    kwargs['channelid'] = 6001
+    kwargs['date'] = times
+
+
+class Mycollect2001(GetFromTga):
+    def workflow(self):
+        ad_channel = self.get_ad_challel()
+        ad_channel.append('')
+        self.collect_playtimes()
+        self.collect_grand_users()
+
+
+    def get_ad_challel(self):
+        sql = f"""SELECT distinct ad_channel  FROM v_event_22 where "$part_event"='event_11_1' and "$part_date"='{self.date}'"""
+        temp = []
+        data = self.tga.get_data(sql)
+        if data:
+            for line in data:
+                try:
+                    if line[0]:
+                        temp.append(line[0])
+                except Exception:
+                    log.error(f"get ad channel failed!", exc_info=True)
+        return temp
+
+
+    def collect_playtimes(self):
+        sql = f"""SELECT sum(cast(online_duration as int)),ad_channel  FROM v_event_22 where 
+"$part_event"='event_21_2' and "$part_date"='{self.date}'  group by ad_channel """
+        data = self.tga.get_data(sql)
+        event_type = "playtimes"
+        if data:
+            for line in data:
+                if line:
+                    temp = {}
+                    try:
+                        temp['playtimes'], temp['ad_channel'] = line
+                        self.write_tga_files(event_type, temp)
+                    except Exception:
+                        log.error(f"split {line} error", exc_info=True)
+
+
+    def collect_grand_users(self, ad_channel):
+        for channel in ad_channel:
+            sql = f"""SELECT count("#account_id") FROM v_user_22 where gameid='{self.gameid}' and first_ad_channel='{channel}'"""
+            data = self.tga.get_data(sql)
+            event_type = "grand_users"
+            if data:
+                for line in data:
+                    if line:
+                        temp = {}
+                        try:
+                            temp['num'] = line[0]
+                            temp['ad_channel'] = channel
+                            self.write_tga_files(event_type, temp)
+                        except Exception:
+                            log.error(f"split {line} error", exc_info=True)
+
+
+def main():
+    times = (datetime.date.today() - datetime.timedelta(days=1)).strftime('%Y-%m-%d')
+    kwargs = {}
+    kwargs['gameid'] = 2001
+    kwargs['channelid'] = 6001
+    kwargs['date'] = times
+
+    mc = Mycollect2001(**kwargs)
+    mc.workflow()
+
+
+if __name__ == "__main__":
+    main()

ops/init_tga.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+from ops.mtga import FromTga, GetTgaConfig
+from ops.plog import define_logger
+import logging
+import requests
+import json
+from config.config import is_debug, adv_id_state, ad_type
+
+
+define_logger("/data/logs/reports.log")
+log = logging.getLogger(__name__)
+
+
+class GetFromTga:
+    def __init__(self, **kwargs):
+        # log.info("begin collect gameid={},times={}!".format(kwargs.get('gameid'), kwargs.get('date')))
+        self.date = kwargs.get('date')
+        self.gameid = kwargs.get('gameid')
+        self.channelid = kwargs.get('channelid')
+        g = GetTgaConfig()
+        item = g.get_api_key(self.gameid)
+        self.url = item['url']
+        self.tgaid = item['appid']
+        self.suffix = item.get('suffix', 0)
+        self.api_key = item.get('api_key', None)
+        self.tga = FromTga(self.url, self.api_key)
+        self.tga.init_tga_write(self.tgaid)
+
+
+    def write_tga_files(self, event_type, data):
+        try:
+            tga_tap = {}
+            tga_tap['result'] = json.dumps(data, ensure_ascii=False)
+            tga_tap['account_id'] = f"{self.gameid}_{self.channelid}"
+            tga_tap['gameid'] = self.gameid
+            tga_tap['channelid'] = self.channelid
+            tga_tap['date'] = self.date
+            if data:
+                self.tga.put_event_data(tga_tap, f'rep_{event_type}')
+            else:
+                log.error(f"write {data} 2 tga!,event_type=f{event_type}", exc_info=True)
+        except Exception:
+            log.error(f"write 2 tga failed!", exc_info=True)