TuKouKouClient/python/services/other/module_online_data.py

# import json
# import time
import copy
import random
import time
import requests
import json
import numpy as np
import os, io
from PIL import Image
from io import BytesIO
Authorization = "Bearer e4ffe8dfe772670ca9c9d234104a7be67a31d689"
Origin = "http://my2.pubdata.cn"
Host = "mybackend2.pubdata.cn"
DOMAIN = "http://mybackend2.pubdata.cn"
Headers = {
    "Authorization": Authorization,
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36",
    "Origin": Origin,
    "Host": Host,
    "Content-Type": "application/json;charset=UTF-8",
}
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 GetOnlineData(object):
    def __init__(self):
        self.s = requests.session()
        self.post_headers = {
            "Authorization": Authorization,
            "Origin": Origin,
            "Host": Host,
            "Content-Length": "0",
            "Content-Type": "application/json",
            "Accept": "application/json",
        }

    def refresh_headers(self):
        self.post_headers = {
            "Authorization": Authorization,
            "Origin": Origin,
            "Host": Host,
            "Content-Length": "0",
            "Content-Type": "application/json",
            "Accept": "application/json",
        }

    def get_key_secret(self):
        # 获取抠图剩余次数
        url = "{domain}/api/ai_image/client/get_key_serect".format(domain=DOMAIN)
        _s = self.s.post(url=url, headers=self.post_headers, timeout=10)
        response_data = _s.json()
        return response_data["data"]

    def get_cutout_image_times(self):
        # 获取抠图剩余次数
        url = "{domain}/api/ai_image/client/search_company_balance".format(
            domain=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 search_progress(self, generate_ids):
        # 查进度
        url = "{domain}/api/ai_image/client/search_progress".format(domain=DOMAIN)
        data = {"generate_ids": generate_ids}
        _s = self.s.post(url=url, headers=self.post_headers, data=json.dumps(data), timeout=60)
        response_data = _s.json()
        return response_data["data"]

    def download_picture(self, url, out_path):
        response = requests.get(url, timeout=30)
        pic = response.content
        if out_path:
            with open(out_path, 'wb') as f:
                f.write(pic)
        else:
            return Image.open(BytesIO(pic))

    def remove_background(self, images_url):
        url = "{domain}/api/ai_image/client/remove_background".format(domain=DOMAIN)
        data = {"images": images_url}
        _s = self.s.post(url=url, headers=self.post_headers, data=json.dumps(data), timeout=30)
        response_data = _s.json()
        return response_data["data"]["generate_ids"], int(response_data["data"]["balance"])

    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=DOMAIN,
        )
        _s = self.s.get(url=url, headers=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 upload_pic(self, file_path, buffer=None):
        if buffer is None:
            root_path, file_name = os.path.split(file_path)
            e = os.path.splitext(file_name)[1]

            __format = {"jpg": "JPEG",
                        "JPG": "JPEG",
                        "JPEG": "JPEG",
                        "jpeg": "JPEG",
                        "png": "PNG",
                        "PNG": "PNG", }

            _format = __format[e[1:]]
            if _format == "JPEG":
                buffer = io.BytesIO()
                im = Image.open(file_path)
                im.save(buffer, format='JPEG')
                buffer.seek(0)
            else:
                with open(file_path, 'rb') as file:
                    buffer = io.BytesIO(file.read())

            files = [
                ('file',
                 (file_path,
                  buffer,
                  'image/{}'.format(_format)))
            ]
        else:
            files = [
                ('file',
                 ("1.jpg",
                  buffer,
                  'image/{}'.format("JPEG")))
            ]

        url = "{domain}/api/backend/upload".format(domain=DOMAIN)

        headers = {
            "Authorization": Authorization,
            "Origin": Origin,
            "Host": Host,
        }

        _s = requests.post(url=url, headers=headers, files=files, timeout=60)
        # print(_s.text)
        response_data = _s.json()
        return response_data["data"]["url"]

    def get_keys(self):
        k = "pxnib99dbchtmdm"
        s = "ub9uj5678gs4m2bnrass1t3tn6ughlk065ianosk06akagolcr2u"
        return (k,s)

    def dispose_point(self,_type):
        # 扣分 sub；add为增加分数，每次操作一分
        url = "{domain}/api/ai_image/client/dispose_point".format(domain=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=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