342 lines
16 KiB
Python
342 lines
16 KiB
Python
# !/usr/bin/python
|
||
# -*- coding: utf-8 -*-
|
||
# @Author:: Arthur Wu
|
||
# @Date:: 2024/11/14-14:35
|
||
# @Description::
|
||
import requests,json, logging
|
||
from commons.SignatureYM import SignatureYM2
|
||
|
||
|
||
class YMClientApi(object):
|
||
def __init__(self, ShoppingCartMac):
|
||
self.Domain = "https://api.test.yimaogo.com/cart"
|
||
self.headerss = SignatureYM2(ShoppingCartMac)._headers()
|
||
|
||
def get_ads_list(self):
|
||
logging.info("========== [获取广告列表] get_ads_list ==========")
|
||
''' method 1 '''
|
||
payload = {}
|
||
url = self.Domain + "/v1/ads/list?areaCode&userId&barcode&adsAreaIds=1,2,3,4,5,6"
|
||
logging.info(f"---headers: {self.headerss}---")
|
||
response = requests.request("GET", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def query_ad_detail(self, ADsId: str):
|
||
logging.info("========== [查询广告详情] query_ad_detail ==========")
|
||
payload = {}
|
||
url = self.Domain + "/v1/ads/{adsId}".replace("{adsId}", str(ADsId))
|
||
response = requests.request("GET", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def session_start(self):
|
||
logging.info("========== [前置] session_start ==========")
|
||
url = self.Domain+"/v1/session/start"
|
||
payload = ""
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
|
||
def session_end(self):
|
||
logging.info("========== [后置] session_end ==========")
|
||
url = self.Domain+"/v1/session/end?reason"
|
||
payload = {}
|
||
response = requests.request("PUT", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
|
||
def login_app_v2(self, Payload):
|
||
'''
|
||
1-匿名登录
|
||
{"action": 0, "isAnon": True}
|
||
2-会员登录
|
||
{"action":1,"code":"18052753212","isAnon":False}
|
||
:param Payload:
|
||
:return:
|
||
'''
|
||
logging.info("========== [登录] login_app_v2 ==========")
|
||
payload = json.dumps(Payload)
|
||
url = self.Domain+"/v2/login"
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def get_login_type(self):
|
||
logging.info("========== [获取登录方式] get_login_type ==========")
|
||
payload = {}
|
||
url = self.Domain+"/v2/login/type?action=1"
|
||
response = requests.request("GET", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
|
||
def get_goods_info(self, InputCode):
|
||
logging.info("========== [获取商品信息] get_goods_info ==========")
|
||
payload = {}
|
||
url = self.Domain+"/v2/shopping/{inputCode}".replace("{inputCode}", str(InputCode))
|
||
logging.info(f"\n-----------获取商品信息url：{url}\n")
|
||
response = requests.request("GET", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def add_cart_goods(self, GoodsInfoData, AddPurchaseQuantity, LoginData):
|
||
logging.info("========== [加购 商品] add_retire_purchase ==========")
|
||
payload = json.dumps({
|
||
"addGoods":[{
|
||
"inputCode": GoodsInfoData["data"]["inputCode"],
|
||
"isNormalAddPurchase": True,
|
||
"qty": int(AddPurchaseQuantity), # AddPurchaseQuantity，GoodsInfoData["data"]["qty"]
|
||
"weight": int(GoodsInfoData["data"]["weight"]),
|
||
}],
|
||
"autoSelectCoupon": True,
|
||
"coupons": [],
|
||
"deleteGoods": [],
|
||
"existGoods": [],
|
||
"orderNo": LoginData["data"][0]["orderNo"]
|
||
}, indent=4, ensure_ascii=False)
|
||
url = self.Domain+"/v2/shopping/add/retire/purchase"
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def delete_cart_goods(self, GetGoodsInfoList, LoginData, AddGoodsRespList):
|
||
''' 退购 商品
|
||
:param GetGoodsInfoList:
|
||
:param LoginData:
|
||
:param AddGoodsResp:
|
||
:return:
|
||
'''
|
||
import random
|
||
logging.info("========== [退购 商品] delete_cart_goods ==========")
|
||
delGoodsList, existGoodsList = [], []
|
||
for add_goods_resp in AddGoodsRespList:
|
||
# for add_goods_info in AddGoodsResp["data"]["orderItemList"]:
|
||
add_goods_info = add_goods_resp["data"]["orderItemList"][0]
|
||
exist_goods_dict = {}
|
||
exist_goods_dict["inputCode"] = add_goods_info["inputCode"]
|
||
exist_goods_dict["isNormalAddPurchase"] = True
|
||
exist_goods_dict["qty"] = add_goods_info["qty"]
|
||
exist_goods_dict["uuid"] = add_goods_info["uuid"]
|
||
if add_goods_info["weight"] > 0:
|
||
exist_goods_dict["weight"] = add_goods_info["weight"]
|
||
else:
|
||
for get_goods_data in GetGoodsInfoList:
|
||
if get_goods_data["data"]["inputCode"] == add_goods_info["inputCode"]:
|
||
if get_goods_data["data"]["weight"] > 0:
|
||
exist_goods_dict["weight"] = get_goods_data["data"]["weight"]
|
||
existGoodsList.append(exist_goods_dict)
|
||
|
||
del_goods_dict = {}
|
||
delGoodsInfo = {
|
||
"inputCode": "",
|
||
"isNormalAddPurchase": True,
|
||
"qty": 1,
|
||
"weight": 0,
|
||
}
|
||
delGoodsInfo = random.choice(GetGoodsInfoList)
|
||
del_goods_dict["inputCode"] = delGoodsInfo["data"]["inputCode"]
|
||
del_goods_dict["isNormalAddPurchase"] = True
|
||
del_goods_dict["qty"] = 1
|
||
del_goods_dict["weight"] = 0
|
||
for goodData in delGoodsList:
|
||
if goodData["inputCode"] == del_goods_dict["inputCode"]:
|
||
del_goods_dict["uuid"] = goodData["uuid"]
|
||
delGoodsList.append(del_goods_dict)
|
||
|
||
payload = json.dumps({
|
||
"addGoods": [],
|
||
"autoSelectCoupon": True,
|
||
"coupons": [],
|
||
"deleteGoods": delGoodsList,
|
||
"existGoods": existGoodsList,
|
||
"orderNo": LoginData["data"][0]["orderNo"]
|
||
})
|
||
logging.info(f"-----------退购商品请求体为：{payload}\n")
|
||
url = self.Domain + "/v2/shopping/add/retire/purchase"
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
if response.json()["msg"] == '成功':
|
||
return json.loads(payload)
|
||
|
||
def delete_shopping_bag(self, GetGoodsInfoList, LoginData, AddGoodsResp):
|
||
''' 退购 购物嗲
|
||
:param GetGoodsInfoList:
|
||
:param LoginData:
|
||
:param AddGoodsResp:
|
||
:return:
|
||
'''
|
||
import random
|
||
logging.info("========== [退购 购物袋] delete_shopping_bag ==========")
|
||
|
||
delGoodsList, existGoodsList = [], []
|
||
for add_goods_info in AddGoodsResp["data"]["bagInfoList"]:
|
||
exist_goods_dict = {}
|
||
exist_goods_dict["inputCode"] = add_goods_info["inputCode"]
|
||
exist_goods_dict["isNormalAddPurchase"] = True
|
||
exist_goods_dict["qty"] = add_goods_info["qty"]
|
||
exist_goods_dict["uuid"] = add_goods_info["uuid"]
|
||
if add_goods_info["weight"] > 0:
|
||
exist_goods_dict["weight"] = add_goods_info["weight"]
|
||
else:
|
||
for get_goods_data in GetGoodsInfoList:
|
||
if get_goods_data["data"]["inputCode"] == add_goods_info["inputCode"]:
|
||
if get_goods_data["data"]["weight"] > 0:
|
||
exist_goods_dict["weight"] = get_goods_data["data"]["weight"]
|
||
existGoodsList.append(exist_goods_dict)
|
||
|
||
del_goods_dict = {}
|
||
delGoodsInfo = random.choice(GetGoodsInfoList)
|
||
del_goods_dict["inputCode"] = delGoodsInfo["data"]["inputCode"]
|
||
del_goods_dict["isNormalAddPurchase"] = True
|
||
del_goods_dict["qty"] = 1
|
||
del_goods_dict["weight"] = 0
|
||
for goodData in delGoodsList:
|
||
if goodData["inputCode"] == del_goods_dict["inputCode"]:
|
||
del_goods_dict["uuid"] = goodData["uuid"]
|
||
delGoodsList.append(del_goods_dict)
|
||
|
||
payload = json.dumps({
|
||
"addGoods": [],
|
||
"autoSelectCoupon": True,
|
||
"coupons": [],
|
||
"deleteGoods": delGoodsList,
|
||
"existGoods": existGoodsList,
|
||
"orderNo": LoginData["data"][0]["orderNo"]
|
||
})
|
||
url = self.Domain + "/v2/shopping/add/retire/purchase"
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def get_coupon_list(self):
|
||
logging.info("========== [获取优惠券列表] get_coupon_list ==========")
|
||
payload = json.dumps({
|
||
"code":"4",
|
||
"value":""
|
||
})
|
||
url = self.Domain+"/v1/coupon/list"
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def receive_coupon(self, CouponListData):
|
||
logging.info("========== [领取优惠券] receive_coupon ==========")
|
||
CouponList = []
|
||
for i in CouponListData["data"]:
|
||
CouponList.append(i["name"])
|
||
payload = json.dumps({"activityIds": CouponList})
|
||
url = self.Domain+"/v1/coupon/receive"
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def query_coupon_list_user(self):
|
||
logging.info("========== [查询优惠券列表] query_coupon_list_user ==========")
|
||
payload = {}
|
||
url = self.Domain+"/v1/coupon/list/user?state=2&page=0&pageSize=10"
|
||
response = requests.request("GET", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def get_cart_goods_info(self, GoodsDataList=[]):
|
||
logging.info("========== [获取购物车商品信息] get_cart_goods_info ==========")
|
||
# goods_param_list = []
|
||
# for goods_data in GoodsDataList:
|
||
# goods_dict = {}
|
||
# goods_dict["inputCode"] = goods_data["data"]["inputCode"]
|
||
# goods_dict["qty"] = int(float((goods_data["data"]["qty"])))
|
||
# goods_dict["weight"] = int(goods_data["data"]["weight"])
|
||
# goods_param_list.append(goods_dict)
|
||
# payload = json.dumps({"goodsParamList": goods_param_list}, indent=4, ensure_ascii=False)
|
||
# logging.info(f"-----------请求体为：{payload}")
|
||
|
||
payload = json.dumps({"goodsParamList": []})
|
||
url = self.Domain+"/v2/shopping/cart/goods/info"
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def request_order_settlement(self, LoginData):
|
||
logging.info("========== [请求订单结算] request_order_settlement ==========")
|
||
payload = json.dumps({
|
||
"orderNo": LoginData["data"][0]["orderNo"],
|
||
"setCouponFlag": False,
|
||
"setPayMethodFlag": False
|
||
})
|
||
url = self.Domain+"/v2/shopping/order/settle"
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def verify_the_goods_through_verification(self):
|
||
logging.info("========== [验证商品通过验货] ==========")
|
||
payload = json.dumps({
|
||
"manualId": 65,
|
||
"type": 1,
|
||
"barcodes": []
|
||
})
|
||
url = self.Domain+"/v1/check/goods/update"
|
||
response = requests.request("POST", url, headers=self.headerss, data=payload)
|
||
logging.info(f"-----------接口返回状态码：{response.status_code}")
|
||
logging.info(f"-----------接口返回数据：{response.json()}\n\n")
|
||
return response.json()
|
||
|
||
def traverse_the_barcode_of_the_prod(self, FP):
|
||
'''
|
||
根据商品Barcode遍历商品信息
|
||
:return:
|
||
'''
|
||
logging.info("========== [根据商品Barcode遍历商品信息] ==========")
|
||
'''
|
||
1- 遍历读取Barcode
|
||
2、根据Barcode遍历请求接口
|
||
3、根据Barcode查询情况，回写商品信息
|
||
'''
|
||
def get_barcode_data():
|
||
import xlrd
|
||
workbook = xlrd.open_workbook(FP)
|
||
sheet = workbook.sheet_by_index(0)
|
||
first_column = []
|
||
for row in range(sheet.nrows):
|
||
first_column.append(sheet.cell_value(row, 0))
|
||
return first_column
|
||
for barcode in get_barcode_data():
|
||
logging.info(f"-----------当前Barcode为：{barcode}")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
ymc = YMClientApi(ShoppingCartMac='')
|
||
''' 01- '''
|
||
ymc.session_start() # step1：session开始
|
||
Payload01 = {"action": 0, "isAnon": True}
|
||
LoginData = ymc.login_app_v2(Payload01) # step2：匿名登录
|
||
ymc.get_login_type() # step3：获取登录方式
|
||
|
||
Payload02 = {"action": 1, "code": "17786408673", "isAnon": False}
|
||
ymc.login_app_v2(Payload02) # step4：切换会员登录
|
||
|
||
GoodsInfoData = ymc.get_goods_info(InputCode=6924882486100) # step5：获取商品信息 ------ 6924882486100
|
||
|
||
CouponListData = ymc.get_coupon_list() # step6：获取优惠券列表
|
||
ymc.receive_coupon(CouponListData) # step7：领取优惠券
|
||
ymc.query_coupon_list_user() # step8：查询用户持有的优惠券列表
|
||
|
||
ymc.add_cart_goods(GoodsInfoData, 1, LoginData) # step9：加购商品 ------ 6924882486100
|
||
ymc.get_cart_goods_info(GoodsInfoData) # step10：获取购物车商品信息
|
||
ymc.request_order_settlement(LoginData) # step11：请求订单结算
|
||
|
||
ymc.session_end()
|
||
|