sync_amz_data/sync_amz_data/tasks/_base.py

from functools import lru_cache
import logging
from typing import List
from urllib.parse import urljoin

from sync_amz_data.public import BaseClient, asj_api
from sync_amz_data.settings import AWS_LWA_CLIENT, DATA_PATH

logger = logging.getLogger(__name__)
ASJ_URL_BASE = "http://127.0.0.1:8000/api/ad_manage/"


@lru_cache(maxsize=100)
def query_shop_info(profile_id: str) -> dict:
    # todo 临时
    if profile_id == "3006125408623189":
        return {
            "profile_id": profile_id,
            "id": 1,
            "shop_name": "ZosiDirect",
            "region": "NA",
            "access_token": "Atza|IwEBION12cAeJLW1CrhirYbH8ianuTfh1JQtC2nbnZjWcCk_J0v9XVF00Pm7AYzZPHXAuFgdul0vuQt4XUAcwVdzQF9AzDfk5wKAUXe9fuGZhb0nq0mdJxg08u2BR_rUggxKWSd1sg6OW7szIEq8xQzok9hcr-Ai-WupyaA-CCznOt7STmyEZltNsK8VuJb7ySxTlxwf-DbuX2Tn9JdsEta7DwQIsHcYv2QwsfLYGnk2LcLUUXG-6TtwCWFGHMxHfoScOvN92hOHiPl3CsdTs5RmKO5eVFdf0XUu8OU5Z9icnjuP1tYBy7_e9s2oTL9fVLLfH_ATUJplPsfm1MhtW6ioX9IXxfJSmOJ0ntEv45ndb9t-wHE_vLukuy_4jwmy_50NU_TOU_9pbFJLQhVyAB0f4HVcb5fgPn--feAX89ANhRJnIn5zibVk_rY_rte7Xu7JMJNNW41PoCUvfVdxjnorhGTIGh2u2JHQPqdww2xLcZ93SQ",
            "status": 1,
            "refresh_token": "Atzr|IwEBIL4ur8kbcwRyxVu_srprAAoTYzujnBvA6jU-0SMxkRgOhGjYJSUNGKvw24EQwJa1jG5RM76mQD2P22AKSq8qSD94LddoXGdKDO74eQVYl0RhuqOMFqdrEZpp1p4bIR6_N8VeSJDHr7UCuo8FiabkSHrkq7tsNvRP-yI-bnpQv4EayPBh7YwHVX3hYdRbhxaBvgJENgCuiEPb35Q2-Z6w6ujjiKUAK2VSbCFpENlEfcHNsjDeY7RCvFlwlCoHj1IeiNIaFTE9yXFu3aEWlExe3LzHv6PZyunEi88QJSXKSh56Um0e0eEg05rMv-VBM83cAqc5POmZnTP1vUdZO8fQv3NFLZ-xU6e1WQVxVPi5Cyqk4jYhGf1Y9t98N654y0tVvw74qNIsTrB-8bGS0Uhfe24oBEWmzObvBY3zhtT1d42myGUJv4pMTU6yPoS83zhPKm3LbUDEpBA1hvvc_09jHk7vUEAuFB-UAZzlht2C1yklzQ",
            "update_time": 1688351991,
            "token_expires_time": 1688355491,
            "create_time": 1683702488
        }
    return {}


class BaseTask:
    AmzAdClientClass = BaseClient

    def __init__(self, profile_id: str):
        self.shop_info = query_shop_info(profile_id)
        self.ad_cil = self.AmzAdClientClass(
            profile_id=profile_id,
            refresh_token=self.shop_info["refresh_token"],
            data_path=DATA_PATH, **AWS_LWA_CLIENT)

    def do(self, task_info: dict):
        record = task_info["record"]
        iter_records = getattr(self.ad_cil, f"iter_{record}", None)
        change_func = getattr(self, f"change_{record}", None)
        ad_api_params = task_info.get("params")
        if ad_api_params is None:
            records_iterator = iter_records()
        else:
            records_iterator = iter_records(ad_api_params)
        for data in records_iterator:
            if change_func:
                data = change_func(data)
            self.to_mysql(record, data)
            logger.info(data)

    def batch_do(self, task_info: dict):
        """
        适用于使用pandas等进行批处理
        @param task_info:
        @return:
        """
        pass

    def to_mysql(self, record: str, data: [dict, List[dict]]):
        asj_api.create(urljoin(ASJ_URL_BASE, record), data)