scripts/bi/vas_yoy_flow.py

from lib.all import *
import datetime


# 流水同比

class S:
    def __init__(self):
        self.mysql_db_vas = Mysql(
            "rm-bp11t1616a1kjvmx5.mysql.rds.aliyuncs.com", 3306, "vas", "root", "Wishpal2024"
        )
        self.mysql_db_bi = Mysql(
            "127.0.0.1", 3306, "metabase_bi", "metabase_bi", "Wishpal_2024"
        )

    def __del__(self):
        self.mysql_db_vas.close()
        self.mysql_db_bi.close()

    def get_last_ptime(self):
        sql = '''select * from vas_yoy_flow order by ptime desc limit 1'''
        rows = self.mysql_db_bi.query(sql)
        if len(rows) > 0:
            dt = rows[0].get("ptime")
            return int(dt.timestamp())
        return 0

    def get_flow(self, st, et):
        sql = '''select sum(pay_amount) flow from vas_order where ct>={} and ct<{} and order_status in (1,2);'''.format(st, et)
        rows = self.mysql_db_vas.query(sql)
        if len(rows) > 0:
            return safe_get_int(rows[0], "flow")
        return 0

    def insert_flow(self, ptime, flow):
        sql = '''insert into vas_yoy_flow (ptime, flow) values ({}, {})'''.format(ptime, flow)
        return self.mysql_db_bi.exec(sql, ())

    def proc(self):
        time_now = int(time.time())
        zero_time = get_today_zero_time() - 86400
        last_time = self.get_last_ptime()

        while last_time <= get_today_zero_time():
            next_time = last_time + 300
            flow = self.get_flow(zero_time, next_time)

            dt = datetime.datetime.fromtimestamp(next_time)
            formatted_dt = dt.strftime('%Y-%m-%d %H:%M:%S')

            self.insert_flow(formatted_dt, flow)
            print(get_time_str_by_ts(next_time), flow)

            last_time += 300
            time.sleep(0.5)


s = S()
s.proc()