import base64
import requests
import settings
import json
import numpy as np
class JsonEncoder(json.JSONEncoder):
"""Convert numpy classes to JSON serializable objects."""
def default(self, obj):
if isinstance(obj, (np.integer, np.floating, np.bool_)):
return obj.item()
elif isinstance(obj, np.ndarray):
return obj.tolist()
else:
return super(JsonEncoder, self).default(obj)
class OnlineDataRequest(object):
def __init__(self, token):
self.s = requests.session()
self.token = token
self.post_headers = {
"Authorization": token,
# "Origin": settings.Headers["Origin"],
# "Host": settings.Headers["Host"],
"Content-Length": "0",
"Content-Type": "application/json",
"Accept": "application/json",
}
print("28 Authorization：", self.post_headers["Authorization"])
def refresh_headers(self, token):
self.post_headers = {
"Authorization": token,
# "Origin": settings.Headers["Origin"],
# "Host": settings.Headers["Host"],
"Content-Length": "0",
"Content-Type": "application/json",
"Accept": "application/json",
}
def auth_user(self):
# 用户登录
url = "{domain}/api/auth/user".format(domain=settings.DOMAIN)
s = requests.session()
_s = s.get(url=url, headers=settings.Headers)
response_data = _s.json()
return response_data
def logout(self):
url = "{domain}/api/auth/logout".format(domain=settings.DOMAIN)
s = requests.session()
_s = s.post(url=url, headers=settings.Headers)
def get_change_bar_code(self, code):
url = "{domain}/api/hct/open/sting_search_goods?string={code}".format(
domain=settings.DOMAIN, code=code
)
try:
s = requests.get(url)
goods_art_no = s.json()["data"]["goods_art_no"]
return goods_art_no
except BaseException as e:
print(e)
return
def get_goods_art_no_info(self, numbers_list=None, goods_art_list=None, token=None):
# 获取商品基础信息，入参为商品的编号
url = "{domain}/api/backend/produce/goods/info".format(domain=settings.DOMAIN)
print("执行 get_goods_art_no_info ", url)
if numbers_list:
data = {"numbers": numbers_list}
print("请求编码：", numbers_list)
else:
data = {"goods_art_nos": goods_art_list}
print("请求货号：", goods_art_list)
print("请求货号=====>", self.token)
print("执行 get_goods_art_no_info----------------", data)
post_headers = {
"Authorization": token,
# "Origin": settings.Headers["Origin"],
# "Host": settings.Headers["Host"],
"Content-Length": "",
"Content-Type": "application/json",
"Accept": "application/json",
}
data = json.dumps(data)
post_headers["Content-Length"] = str(len(data))
_s = self.s.post(url=url, data=data, headers=post_headers)
# _s = self.s.get(url=url, params=params, headers=settings.Headers)
response_data = _s.json()
# print(response_data)
# print("\n")
goods_number_data = {}
# ["", "", "", "", "", "", "", "", "", "", "", ]
if "data" not in response_data:
return {}
for data in response_data["data"]:
if numbers_list:
number = data["number"]
else:
number = data["goods_art_no"].upper()
goods_number_data[number] = {}
goods_number_data[number]["商品面料"] = data["fabric"]
goods_number_data[number]["商品内里"] = data["lining"]
goods_number_data[number]["商品鞋底"] = data["sole"]
goods_number_data[number]["后帮高"] = data["back_height"]
goods_number_data[number]["前掌宽"] = data["forefoot_width"]
goods_number_data[number]["鞋跟高"] = data["heel_height"]
goods_number_data[number]["FAB介绍"] = data["fab_info"]
goods_number_data[number]["编号"] = data["number"]
goods_number_data[number]["商品货号"] = data["goods_art_no"].upper()
goods_number_data[number]["款号"] = data["goods_number"].upper()
goods_number_data[number]["颜色名称"] = data["color"]
goods_number_data[number]["所属企划"] = data["projects"][0]
goods_number_data[number]["设计方名称"] = data["purchasing_unit"]
goods_number_data[number]["供应商"] = data["supplier_name"]
goods_number_data[number]["供应商编码"] = data["supplier_code"].lstrip("0")
goods_number_data[number]["供应商货号"] = data["supplier_goods_artno"]
goods_number_data[number]["销售工厂"] = data["sales_factory_name"]
goods_number_data[number]["销售组织"] = data["man_org_name"]
goods_number_data[number]["是否SAP"] = data["source"]
goods_number_data[number]["OEM报价"] = data["oem_price"]
goods_number_data[number]["出厂价"] = data["ex_factory_price"]
goods_number_data[number]["首单货期"] = data["earliest_delivery_date"]
goods_number_data[number]["包装"] = data["package_specification"]
goods_number_data[number]["创建日期"] = data["created_at"]
goods_number_data[number]["货号图"] = data["image"]
return goods_number_data
def get_on_goods_all_art(self, number):
# 获取商品基础信息，入参为商品的编号
url = "{domain}/api/backend/produce/goods/query/numbers?number={number}".format(
domain=settings.DOMAIN, number=number
)
_s = self.s.get(url=url, headers=self.post_headers)
response_data = _s.json()
print(number, response_data)
"""
14250230 {'data': {'goods_number': 'AC5200117', 'brother_goods_arts': [{'number': '14250232', 'goods_art_no': 'AC52001173', 'color': '杏色'}, {'number': '14250231', 'goods_art_no': 'AC52001172', 'color': '灰色'}, {'number': '14250230', 'goods_art_no': 'AC52001171', 'color': '黑色'}]}, 'code': 0, 'message': 'success'}
"""
return (
response_data["data"]["goods_number"],
response_data["data"]["brother_goods_arts"],
response_data["data"]["goods_art_no"],
)
def get_views(self, image_url):
url = "http://{}/shoes_category".format(settings.VIEW_DEAL_DOMAIN)
data = {
"train_path": "./datasets/Shoes_Dataset/Train/angle",
"model_filename": "./models/0320/output0320.pth",
"validate_path": image_url,
}
_s = requests.post(
url=url,
data=json.dumps(data),
)
response_data = _s.json()
return response_data["classify_result"]
def uploadImage(self, local_path: str) -> str:
post_headers = {"Authorization": self.token}
url = settings.DOMAIN + "/api/upload"
resultData = self.s.post(
url, files={"file": open(local_path, "rb")}, headers=post_headers
).json()
return resultData["data"]["url"]
def get_current_menu(self):
def get_menu(_menu_dict, _data):
for menu in _data:
_menu_dict[menu["key"]] = {}
for mods in menu["mods_arr"]:
_menu_dict[menu["key"]][mods["key"]] = mods["name"]
if "_child" in menu:
get_menu(_menu_dict, menu["_child"])
return _menu_dict
url = "{domain}/api/backend/basic/get_current_menu".format(
domain=settings.DOMAIN,
)
_s = self.s.get(url=url, headers=settings.Headers)
response_data = _s.json()
try:
menu_data = response_data["data"]["pc_menu"]
menu_dict = {}
menu_dict = get_menu(menu_dict, menu_data)
except:
menu_dict = {}
# print(json.dumps(menu_dict,ensure_ascii=False))
# raise 1
return menu_dict
# 获取所有资源的配置
def get_resource_config(self):
url = "{domain}/api/openai/query_client_addons".format(domain=settings.DOMAIN)
_s = self.s.get(
url=url, headers=self.post_headers, params={"type": "client_camera"}
)
response_data = _s.json()
return response_data
# 拍照日志上报
def add_auto_photo_logs(self, data):
url = "{domain}/api/openai/add_auto_photo_logs".format(domain=settings.DOMAIN)
post_data = {
"goods_no": data["goods_art_no"],
"take_photo_created_at": data["take_photo_created_at"],
"photo_created_at": data["photo_create_time"],
"image_dispose_mode": data["image_deal_mode"],
"photo_serial_number": data["image_index"],
}
post_data = json.dumps(post_data)
_s = self.s.post(url=url, headers=self.post_headers, data=post_data)
response_data = _s.json()
if settings.IS_TEST:
print("209-----拍照日志上报 add_auto_photo_logs")
print(response_data)
return response_data
def upload_pic_list_data(self, data):
url = "{domain}/api/backend/goods/save/images".format(domain=settings.DOMAIN)
data = json.dumps(data)
self.post_headers["Content-Length"] = str(len(data))
_s = self.s.post(url=url, data=data, headers=self.post_headers)
response_data = _s.json()
try:
if response_data["code"] == 0 and response_data["message"] == "success":
return True
else:
print(data)
print(response_data)
return False
except BaseException as e:
print(data)
print(e)
print(response_data)
return False
def upload_pic(self, goods_data):
# 检查货号图是否存在
url = "{domain}/api/backend/upload".format(domain=settings.DOMAIN)
# print(url)
headers = {
"Authorization": settings.Headers["Authorization"],
"User-Agent": settings.Headers["User-Agent"],
"Origin": settings.Headers["Origin"],
"Host": settings.Headers["Host"],
}
files = [
(
"file",
(
goods_data["file_path"],
goods_data["image_io"],
"image/{}".format(goods_data["e"]),
),
)
]
_s = requests.post(url=url, headers=headers, files=files)
response_data = _s.json()
return response_data["data"]["url"]
# 查询是否已有详情图
def check_detail_image(self, goods_art_no, token):
url = "{domain}/api/backend/goods/check_detail_image?number={number}".format(
domain=settings.DOMAIN, number=goods_art_no
)
_s = self.s.get(url=url, headers=self.post_headers)
response_data = _s.json()
# print(response_data)
return response_data["data"]["hasDetailImage"]
# 调用API识别是否是拖鞋
def yolo_shoes_category(self, image_url):
url = "{domain}/api/ai_image/main/yolo_shoes_category".format(
domain=settings.DOMAIN
)
post_data = {
"image_url": image_url,
}
post_data = json.dumps(post_data)
_s = self.s.post(url=url, headers=self.post_headers, data=post_data)
response_data = _s.json()
if settings.IS_TEST:
print("278-----yolo_shoes_category")
print(response_data)
r_data = None
try:
r_data = response_data["data"]["category"]
except BaseException as e:
print("285", e)
return r_data
# 图片上传by IO
def upload_image_by_io(self, image_io) -> str:
post_headers = {"Authorization": settings.Authorization}
url = settings.DOMAIN + "/api/upload"
resultData = self.s.post(
url, files={"file": image_io}, headers=post_headers
).json()
return resultData["data"]["url"]
def upload_goods_api(self, params):
'''上传商品api'''
post_headers = {
"Authorization": self.token,
"Content-Type": "application/json",
}
url = settings.DOMAIN+"/api/ai_image/camera_machine/publish_goods"
postData = json.dumps(params)
print("上传商品api==>url", url)
print("上传第三方数据打印", params)
resultData = self.s.post(url, data=postData, headers=post_headers).json()
print("上传商品api==>resultData", resultData)
return resultData
def uploadGoods2ThirdParty(self, goods_no_dict=None,online_stores=[]):
params = []
if goods_no_dict == None:
return
for goods_no in goods_no_dict.keys():
goods_data = goods_no_dict[goods_no]
detail_path = goods_data.get("detail_path", "")
if detail_path == "":
continue
goods_title = goods_data.get("商品标题", "")
if goods_title == "":
continue
goods_price = goods_data.get("商品价格", 0)
if goods_title == 0:
continue
skuList = []
itemImageInfoList = []
itemSkuImageList = []
sku_list_basic = goods_data.get("货号资料", [])
quantity = 9999
skuPropValueList = []
for skuIdx, sku_data in enumerate(sku_list_basic):
sku_goods_art_no = sku_data.get("货号", "")
color_name = sku_data.get("颜色名称", "")
mainImages = sku_data.get("800x800", [])
if not mainImages:
continue
mainImagePath = mainImages[0]
imageUrl = self.uploadImage(local_path=mainImagePath)
skuItemData = {
"skuNo": sku_goods_art_no,
"originalPrice": float(goods_price),
"newSkuWeight": int(1),
"skuMainImageUrl": str(imageUrl),
"skuName": f"颜色:{color_name}",
"sellingPrice": float(goods_price),
"quantity": int(quantity),
"showOrder": int(skuIdx + 1),
}
skuList.append(skuItemData)
itemImage = {
"imageUrl": str(imageUrl),
"imageType": 0,
"imageItem": int(skuIdx),
"imageIndex": 10,
}
itemImageInfoList.append(itemImage)
imageJson = {
"imageUrl": str(imageUrl),
"imageType": 1,
"showOrder": 1,
}
skuPropValueList.append(
{
"imageJson": imageJson,
"propValue": str(color_name),
"showOrder": 1,
}
)
itemSkuImageList.append(
{
"propName": "颜色",
"isImageProp": 1,
"propShowOrder": 1,
"skuPropValueList": skuPropValueList,
}
)
detailImageUrl = self.uploadImage(local_path=detail_path)
category_info = "流行男鞋>>休闲鞋>>时尚休闲鞋"
itemData = {
"catePathName": category_info, # 分类
"itemName": str(goods_title), # 商品标题
"itemNo": str(goods_no),
"brandName": "vali", # 品牌名称
"sellingPrice": float(goods_price), # 售价(未划线价)
"originalPrice": float(goods_price), # 划线价
"quantity": int(quantity), # 库存数量
"propInfoList": [
{"propName": "品牌", "propIndex": 2, "propValue": "Vali"},
{
"propName": "平台类目",
"propIndex": 3,
"propValue": category_info,
},
],
"skuList": skuList,
"itemImageInfoList": itemImageInfoList,
"itemSkuImageList": itemSkuImageList,
"wapDescription": f'
',
"pcDescription": f'',
}
params.append(itemData)
json_params = str(params) # 直接转换为字符串表示
# 使用base64编码
encoded = base64.b64encode(json_params.encode("utf-8")).decode("utf-8")
self.upload_goods_api({"bizcontent": encoded, "online_stores": online_stores})
print("商品上传第三方成功")
class GetOnlineDataHLM(OnlineDataRequest):
def __init__(self, token):
super().__init__(token)
self.token = token
def upload_pic(self, goods_data, token):
# 检查货号图是否存在
url = "{domain}/api/backend/upload".format(domain=settings.DOMAIN)
# print(url)
headers = {
"Authorization": self.token,
# 'User-Agent': settings.Headers["User-Agent"],
# 'Origin': settings.Headers["Origin"],
# 'Host': settings.Headers["Host"],
}
files = [
(
"file",
(
goods_data["file_path"],
goods_data["image_io"],
"image/{}".format(goods_data["e"]),
),
)
]
_s = requests.post(url=url, headers=headers, files=files)
response_data = _s.json()
return response_data["data"]["url"]
def upload_pic_list_data(self, data, token):
url = "{domain}/api/backend/goods/save/images".format(domain=settings.DOMAIN)
data = json.dumps(data)
self.post_headers["Content-Length"] = str(len(data))
_s = self.s.post(url=url, data=data, headers=self.post_headers)
response_data = _s.json()
try:
if response_data["code"] == 0 and response_data["message"] == "success":
return True
else:
print(response_data)
return False
except BaseException as e:
print(e)
print(response_data)
return False
def get_goods_art_no_info(self, numbers_list=None, goods_art_list=None, token=None):
# 获取商品基础信息，入参为商品的编号
url = "{domain}/api/backend/goods_client/goods_query".format(
domain=settings.DOMAIN
)
data = {"goods_art_list": goods_art_list}
print("url：", url)
print("请求货号：", goods_art_list)
post_headers = {
"Authorization": token,
# "Origin": settings.Headers["Origin"],
# "Host": settings.Headers["Host"],
"Content-Length": "",
"Content-Type": "application/json",
"Accept": "application/json",
}
data = json.dumps(data)
print(post_headers)
print(data)
# post_headers["Content-Length"] = str(len(data))
_s = self.s.post(url=url, data=data, headers=post_headers)
# _s = self.s.get(url=url, params=params, headers=settings.Headers)
response_data = _s.json()
print(response_data)
print("\n")
goods_number_data = {}
# ["", "", "", "", "", "", "", "", "", "", "", ]
if "data" not in response_data:
return {}
for data in response_data["data"]:
goods_number_data[data["goods_art_no"]] = {}
goods_number_data[data["goods_art_no"]]["商品货号"] = data[
"goods_art_no"
].upper()
goods_number_data[data["goods_art_no"]]["款号"] = data[
"goods_number"
].upper()
goods_number_data[data["goods_art_no"]]["商品面料"] = data["fabric"]
goods_number_data[data["goods_art_no"]]["商品内里"] = data["lining"]
goods_number_data[data["goods_art_no"]]["商品鞋底"] = data["sole"]
goods_number_data[data["goods_art_no"]]["鞋垫"] = data["insole"]
goods_number_data[data["goods_art_no"]]["颜色名称"] = data["color"]
goods_number_data[data["goods_art_no"]]["商品标题"] = data["goods_title"]
goods_number_data[data["goods_art_no"]]["商品价格"] = data["retail_price"]
return goods_number_data
def uploadImage(self, local_path: str) -> str:
post_headers = {"Authorization": settings.Authorization}
url = settings.DOMAIN + "/api/upload"
resultData = self.s.post(
url, files={"file": open(local_path, "rb")}, headers=post_headers
).json()
return resultData["data"]["url"]
# ============pixian抠图处理==========================
def dispose_point(self, _type):
# 扣分 sub；add为增加分数，每次操作一分
url = "{domain}/api/ai_image/client/dispose_point".format(
domain=settings.DOMAIN
)
data = {"type": _type}
_s = self.s.post(
url=url, headers=self.post_headers, data=json.dumps(data), timeout=10
)
response_data = _s.json()
return response_data
def send_message(self, text):
# 发送钉钉消息
url = "{domain}/api/ai_image/client/send_message".format(domain=settings.DOMAIN)
data = {"message": text}
_s = self.s.post(
url=url, headers=self.post_headers, data=json.dumps(data), timeout=10
)
response_data = _s.json()
return response_data
def get_cutout_image_times(self):
# 获取抠图剩余次数
url = "{domain}/api/ai_image/client/search_company_balance".format(
domain=settings.DOMAIN
)
_s = self.s.post(url=url, headers=self.post_headers, timeout=10)
response_data = _s.json()
if "data" not in response_data:
return False
else:
return response_data["data"]
def get_key_secret(self):
# 获取抠图剩余次数
url = "{domain}/api/ai_image/client/get_key_serect".format(
domain=settings.DOMAIN
)
_s = self.s.post(url=url, headers=self.post_headers, timeout=10)
response_data = _s.json()
return response_data["data"]