CameraMachine/python/service/online_request/module_online_data.py

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):
        self.s = requests.session()
        self.post_headers = {"Authorization": settings.Headers["Authorization"],
                             "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):
        self.post_headers = {"Authorization": settings.Headers["Authorization"],
                             "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):
        # 获取商品基础信息，入参为商品的编号
        url = "{domain}/api/backend/produce/goods/info".format(
            domain=settings.DOMAIN
        )
        if numbers_list:
            data = {
                'numbers': numbers_list
            }
            print("请求编码：", numbers_list)
        else:
            data = {
                'goods_art_nos': goods_art_list
            }
            print("请求货号：", goods_art_list)

        post_headers = {"Authorization": settings.Headers["Authorization"],
                        "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=settings.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": settings.Headers["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']

    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=settings.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):
        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=settings.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']


class GetOnlineDataHLM(OnlineDataRequest):

    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 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(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):
        # 获取商品基础信息，入参为商品的编号
        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": settings.Headers["Authorization"],
                        "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"]

        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"]