# !/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()