datacollect/data_collect_interface.py

# -*- coding: utf-8 -*-
# pip install apscheduler
# python data_collect_interface.py --port=5021
# 广告数据接口
# 请求方式为 http://154.8.214.202:5022/webapp/index.php?c=Ops&a=gamereport1004&body={"times":"2019-07-22","is_new":0}

import tornado.ioloop
import tornado.httpserver
import tornado.web
from ops.plog import define_logger
from ops.mtga import FromTga, GetTgaConfig
import logging
from tornado import gen
import tornado.options
import json
from config.config import *
from ops.mmysql import MysqlBase
#from apscheduler.schedulers.tornado import TornadoScheduler
from tasks import run_tasks
from urllib.parse import unquote
from config import *

#sched = TornadoScheduler()
define_logger("/data/logs/data_collect_interface.log")
log = logging.getLogger(__name__)
tornado.options.define("port", default=5022, type=int, help="run server on the given port.")


class DispatchHandler(tornado.web.RequestHandler):
    @gen.coroutine
    def get(self):
        if self.get_query_argument('c') == 'Ops' and self.get_query_argument('a') == 'selfChecking':
            self._selfCheckingHandler()
        elif self.get_query_argument('c') == 'Ops' and self.get_query_argument('a') == 'getreport':
            yield self._selfGetReport()
        elif self.get_query_argument('c') == 'Ops' and self.get_query_argument('a') == 'getmethods':
            yield self._selfgetmethods()
        else:
            self.write("pls check args!")

    def _selfgetmethods(self):
        gameid = self.get_query_argument('gameid')
        try:
            channelid = self.get_query_argument('channelid')
        except:
            channelid = 6001

        sql = f"""SELECT
            a.method_name,
            a.method_display,
            a.need_sort,
            b.fields,
            b.comment,
            b.sort_num
        FROM
            (SELECT
                method_name,
                method_display,
                need_sort
            FROM
                methods
            WHERE
                in_used=1 
                AND  gameid={gameid}
                AND  channelid={channelid})a ,
            (SELECT
                method_name,
                fields,
                sort_num,
                comment
            FROM
                methods_detail )b
                WHERE a.method_name=b.method_name"""
        my_db = MysqlBase(**DB)
        data = my_db.query(sql)
        output = []
        if data:
            for line in data:
                temp = {}
                try:
                    temp['methods'], temp['method_display'],temp['need_sort'],temp['fields'], temp['comment'],\
                        temp['sort_num'] = line
                    output.append(temp)
                except Exception:
                    log.error(f"split {line} error, get data from methods!", exc_info=True)
                    self.write(
                        {'errcode': 1, "errmsg": f"split {line} error, get data from methods!", "result": output})
        self.write({'errcode': 0, "errmsg": "", "result": output})


    def _selfCheckingHandler(self):
        self.write(json.dumps({'errcode': 0, 'errmsg': '', 'healthy': 1, 'max_rundelay': 10}, separators=(',', ':')))

    def _selfGetReport(self):
        self.suffix = 23
        self.api_key = 'OgOWG9CMq12HcE6d8nIisKELDQMWe0XJobj9evyA5NbcmHClQYV4oIorR2NdaiMk'
        # self.suffix = 7
        # self.api_key = 'UZUfYAwJoODl03IIEDGDV7FjKGpgepY6WGWQ4E1FcN1cP4Y5QX3xUBz4nnqhPsnU'
        gc = GetTgaConfig()
        self.url = gc.url
        self.tga = FromTga(url=self.url, token=self.api_key)

        gameid = self.get_query_argument('gameid')
        print(f"gameid was {gameid}")
        times = self.get_query_argument('times')
        try:
            channel = self.get_query_argument('channel')
        except:
            channel = 6001

        methods = self.get_query_argument('methods')

        if not (gameid and times and channel and methods):
            result = {'errcode': 2, "errmsg": f"get args failed`"}
            log.error(result, exc_info=True)
            self.write(result)

        try:
            all = self._get_dailyreport_data(gameid, times, channel, methods)
            return self.write({'errcode': 0, "errmsg": "", "result": all})
        except Exception:
            log.error(f"run self._get_tap_data(**kwargs) failed", exc_info=True)
            return self.write({'errcode': 2, "errmsg": "get data failed`"})


    def _get_dailyreport_data(self, gameid, times, channel, methods):
        sql = f"""SELECT result FROM v_event_{self.suffix} where "$part_event"='rep_{methods}'and date='{times}' 
        and "gameid"={gameid} and channelid={channel} """
        data = self.tga.get_data(sql)
        print(f"sql={sql}")
        all = []
        if data:
            try:
                for line in data:
                    if line:
                        all.append(line[0])
            except Exception:
                log.error("split data from tga failed !", exc_info=True)
        return all


def make_app():
    return tornado.web.Application([(r"/webapp/index[\.]php", DispatchHandler)])


if __name__ == "__main__":
    print('start!')
    tornado.options.parse_command_line()
    app = make_app()
    app.listen(tornado.options.options.port)
    #sched.add_job(run_tasks, 'cron', hour=Cron_hour, minute=Cron_minute)

    tornado.ioloop.IOLoop.current().start()