_base.py 2.9 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  1. from functools import lru_cache
  2. import logging
  3. from typing import List
  4. from sync_amz_data.public import BaseClient, asj_api
  5. from sync_amz_data.settings import AWS_LWA_CLIENT, DATA_PATH
  6. logger = logging.getLogger(__name__)
  7. ASJ_URL_BASE = "http://127.0.0.1:8000/api/ad_manage/"
  8. @lru_cache(maxsize=100)
  9. def query_shop_info(profile_id: str) -> dict:
  10. # todo 临时
  11. if profile_id == "3006125408623189":
  12. return {
  13. "profile_id": profile_id,
  14. "id": 1,
  15. "shop_name": "ZosiDirect",
  16. "region": "NA",
  17. "access_token": "Atza|IwEBION12cAeJLW1CrhirYbH8ianuTfh1JQtC2nbnZjWcCk_J0v9XVF00Pm7AYzZPHXAuFgdul0vuQt4XUAcwVdzQF9AzDfk5wKAUXe9fuGZhb0nq0mdJxg08u2BR_rUggxKWSd1sg6OW7szIEq8xQzok9hcr-Ai-WupyaA-CCznOt7STmyEZltNsK8VuJb7ySxTlxwf-DbuX2Tn9JdsEta7DwQIsHcYv2QwsfLYGnk2LcLUUXG-6TtwCWFGHMxHfoScOvN92hOHiPl3CsdTs5RmKO5eVFdf0XUu8OU5Z9icnjuP1tYBy7_e9s2oTL9fVLLfH_ATUJplPsfm1MhtW6ioX9IXxfJSmOJ0ntEv45ndb9t-wHE_vLukuy_4jwmy_50NU_TOU_9pbFJLQhVyAB0f4HVcb5fgPn--feAX89ANhRJnIn5zibVk_rY_rte7Xu7JMJNNW41PoCUvfVdxjnorhGTIGh2u2JHQPqdww2xLcZ93SQ",
  18. "status": 1,
  19. "refresh_token": "Atzr|IwEBIL4ur8kbcwRyxVu_srprAAoTYzujnBvA6jU-0SMxkRgOhGjYJSUNGKvw24EQwJa1jG5RM76mQD2P22AKSq8qSD94LddoXGdKDO74eQVYl0RhuqOMFqdrEZpp1p4bIR6_N8VeSJDHr7UCuo8FiabkSHrkq7tsNvRP-yI-bnpQv4EayPBh7YwHVX3hYdRbhxaBvgJENgCuiEPb35Q2-Z6w6ujjiKUAK2VSbCFpENlEfcHNsjDeY7RCvFlwlCoHj1IeiNIaFTE9yXFu3aEWlExe3LzHv6PZyunEi88QJSXKSh56Um0e0eEg05rMv-VBM83cAqc5POmZnTP1vUdZO8fQv3NFLZ-xU6e1WQVxVPi5Cyqk4jYhGf1Y9t98N654y0tVvw74qNIsTrB-8bGS0Uhfe24oBEWmzObvBY3zhtT1d42myGUJv4pMTU6yPoS83zhPKm3LbUDEpBA1hvvc_09jHk7vUEAuFB-UAZzlht2C1yklzQ",
  20. "update_time": 1688351991,
  21. "token_expires_time": 1688355491,
  22. "create_time": 1683702488
  23. }
  24. return {}
  25. class BaseTask:
  26. AmzAdClientClass = BaseClient
  27. def __init__(self, profile_id: str):
  28. self.shop_info = query_shop_info(profile_id)
  29. self.ad_cil = self.AmzAdClientClass(
  30. profile_id=profile_id,
  31. refresh_token=self.shop_info["refresh_token"],
  32. data_path=DATA_PATH, **AWS_LWA_CLIENT)
  33. def do(self, task_info: dict):
  34. record = task_info["record"]
  35. iter_records = getattr(self.ad_cil, f"iter_{record}", None)
  36. change_func = getattr(self, f"change_{record}", None)
  37. ad_api_params = task_info.get("params")
  38. if ad_api_params is None:
  39. records_iterator = iter_records()
  40. else:
  41. records_iterator = iter_records(ad_api_params)
  42. for data in records_iterator:
  43. if change_func:
  44. data = change_func(data)
  45. self.to_mysql(record, data)
  46. logger.info(data)
  47. def batch_do(self, task_info: dict):
  48. """
  49. 适用于使用pandas等进行批处理
  50. @param task_info:
  51. @return:
  52. """
  53. pass
  54. def to_mysql(self, record: str, data: [dict, List[dict]]):
  55. asj_api.create(f"{ASJ_URL_BASE}{record}/", data)