scripts/bi/update_streamer.py

from lib.all import *
import datetime


# 主播当日收入明细

class S:
    def __init__(self):
        self.col_streamer = MongoDB(
            host="mongodb://root:Wishpal2024@dds-bp1da1ddd62bede41.mongodb.rds.aliyuncs.com:3717,dds-bp1da1ddd62bede42.mongodb.rds.aliyuncs.com:3717/admin?replicaSet=mgset-77304659",
            port=3717,
            db="streamer",
            collection="streamer"
        )
        self.col_account = MongoDB(
            host="mongodb://root:Wishpal2024@dds-bp1da1ddd62bede41.mongodb.rds.aliyuncs.com:3717,dds-bp1da1ddd62bede42.mongodb.rds.aliyuncs.com:3717/admin?replicaSet=mgset-77304659",
            port=3717,
            db="account",
            collection="account"
        )
        self.mysql_db_bi = Mysql(
            "172.31.37.71", 3306, "bi", "root", "Wishpal@2023"
        )

    def __del__(self):
        self.col_streamer.close()
        self.col_account.close()
        self.mysql_db_bi.close()

    # 更新主播信息
    def update_steamer_info(self):
        # 获取主播信息
        prj = {"mid": 1}
        docs = self.col_streamer.find({}, projection=prj)
        streamer_mids = list()
        for doc in docs:
            mid = safe_get_int(doc, "mid")
            streamer_mids.append(mid)
        print("len(docs): {}".format(len(docs)))

        # 获取账号信息并更新
        acnt_docs = self.col_account.find({"_id": {"$in": streamer_mids}})
        for doc in acnt_docs:
            mid = safe_get_int(doc, "_id")
            user_id = safe_get_int(doc, "user_id")
            name = safe_get_str(doc, "name")

            sql_q = "select * from streamer where mid='{}'".format(mid)
            rows = self.mysql_db_bi.query(sql_q)
            if len(rows) > 0:
                sql_u = "update streamer set user_id=%s, name=%s where mid=%s"
                self.mysql_db_bi.exec(sql_u, (user_id, name, mid))
            else:
                sql_i = "insert into streamer (mid, user_id, name) values (%s,%s,%s)"
                self.mysql_db_bi.exec(sql_i, (mid, user_id, name))

    def proc(self):
        self.update_steamer_info()


s = S()
s.proc()