封装fastapi

python/api.py

@@ -1,42 +1,11 @@
-from flask import Flask, request
-from flask_cors import CORS
-import json
-from jsonschema import validate, ValidationError
+from middleware import *
 import datetime
-from services.Exception import ApiException
-app = Flask(__name__)
-# 配置 CORS，允许所有来源
-CORS(app)
 
-def response(message="success", data=None, code=0):
-    return json.dumps({"message": "", "data": data, "code": code}), 200
 
-@app.errorhandler(ApiException)  # 捕捉全局异常
-def internal_error(exc:ApiException):
-    print("Internal error:",exc.msg)
-    return (
-        json.dumps({"message": str(exc.msg), "code": exc.code}, ensure_ascii=False),
-        200,
-    )
-
-
-@app.errorhandler(404)
-def page_not_found(e):
-    return (
-        json.dumps({"code": 4999, "msg": "请求出错，页面未找到"}, ensure_ascii=False),
-        404,
-    )
-
-@app.route("/", methods=["GET"])
+@app.get("/")
 def index():
-    '''入口文件'''
-    name = request.args.get("name", "World")
-    return response(message="",data=f"Hello, {name}!")
-
-def Validateutils(data,schema):
-    '''数据验证'''
-    try:
-        validate(instance=data, schema=schema)
-    except ValidationError as e:
-        # 处理验证失败的逻辑
-        raise ApiException(e.message)
+    """入口文件"""
+    return {
+        "message": "请求成功",
+        "data": f"Hello, World!",
+    }

python/main.py

@@ -1,7 +1,8 @@
 import argparse
 import signal
 import sys
-from api import app
+from api import *
+import uvicorn
 
 # argparse
 parser = argparse.ArgumentParser(description="Process some integers.")
@@ -22,8 +23,8 @@ signal.signal(signal.SIGINT, signal_handler)
 
 if __name__ == "__main__":
     # 以api方式启动服务会出现警告，请忽略
-    app.run(port=args.port,use_reloader=True)
-
-# 或许flask内置的stdio与node.js stdio有冲突，导致控制台无法显示信息。
-# 如果想要查看控制台输出，请单独启动服务 npm run dev-python
-print("python server is running at port:", args.port)
+    # app.run(port=args.port,use_reloader=True)
+    uvicorn.run(app="main:app", host="127.0.0.1", port=int(args.port), reload=True)
+    # 或许flask内置的stdio与node.js stdio有冲突，导致控制台无法显示信息。
+    # 如果想要查看控制台输出，请单独启动服务 npm run dev-python
+    print("python server is running at port:", args.port)

python/middleware.py

@@ -0,0 +1,41 @@
+from typing import Union,Optional
+from fastapi import FastAPI, Request, Body, Form, Query
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.middleware.cors import CORSMiddleware
+import random
+import os
+from pydantic import BaseModel, validator, conint, constr,Field
+# 关闭文档
+app = FastAPI( redoc_url=None)
+# app.mount("/model", StaticFiles(directory="model"), name="model")
+app.add_middleware(
+        CORSMiddleware,
+        # 允许跨域的源列表，例如 ["http://www.example.org"] 等等，["*"] 表示允许任何源
+        allow_origins=["*"],
+        # 跨域请求是否支持 cookie，默认是 False，如果为 True，allow_origins 必须为具体的源，不可以是 ["*"]
+        allow_credentials=False,
+        # 允许跨域请求的 HTTP 方法列表，默认是 ["GET"]
+        allow_methods=["*"],
+        # 允许跨域请求的 HTTP 请求头列表，默认是 []，可以使用 ["*"] 表示允许所有的请求头
+        # 当然 Accept、Accept-Language、Content-Language 以及 Content-Type 总之被允许的
+        allow_headers=["*"],
+        # 可以被浏览器访问的响应头, 默认是 []，一般很少指定
+        # expose_headers=["*"]
+        # 设定浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600，一般也很少指定
+        # max_age=1000
+)
+
+
+class UnicornException(Exception):
+    def __init__(self, msg: str,code:int=400):
+        self.msg = msg
+        self.code = code
+
+
+@app.exception_handler(UnicornException)
+async def error_throw(request: Request, exc: UnicornException):
+    return JSONResponse(
+        status_code=200, content={"code": exc.code, "error_msg": exc.msg}
+    )

python/services/other/command_func.py

@@ -0,0 +1,15 @@
+import hashlib
+
+
+def get_file_md5(file_path):
+    # 打开文件并读取其中的数据
+    with open(file_path, 'rb') as f:
+        data = f.read()
+    # 计算MD5值并输出结果
+    md5_value = hashlib.md5(data).hexdigest()
+    return md5_value
+
+
+if __name__ == '__main__':
+    print(get_file_md5(
+        r"D:\MyDocuments\PythonCode\MyPython\red_dragonfly\deal_excel\haocaitong_data_completion\2024年春夏季线下商品资料表初步模版11.2.xlsx"))

python/services/other/log.py

@@ -0,0 +1,71 @@
+import logging
+import sys
+import os
+from logging.handlers import RotatingFileHandler
+import threading
+
+
+class MyLogger(object):
+    instance = None
+    init_flag = None
+
+    def __init__(self, name=""):
+        """此处设计为，如果已经存在实例时，不再执行初始化"""
+        if self.init_flag:
+            return
+        else:
+            self.init_flag = True
+
+        self.logger_lock = threading.Lock()
+        self.name = name
+        self.init()
+
+    def init(self):
+        # 创建日志器
+        self.logger = logging.getLogger("图片抠图工具")
+        self.logger.setLevel("DEBUG")
+
+        if not os.path.exists("{}\log".format(os.getcwd())):
+            os.mkdir("{}\log".format(os.getcwd()))
+
+        # 定义日志处理器
+        cl = logging.StreamHandler()  # 将日志输出到终端
+        fl = RotatingFileHandler("log\logging_text.log", mode="a", encoding="utf-8", maxBytes=10 * 1024 * 1024,
+                                 backupCount=5)  # 输出日志到文本
+
+        # 定义初始化格式 # 定义两个日志布局
+        fmt1 = ("%(asctime)s---%(name)s---%(message)s")
+        fmt2 = ("%(asctime)s---%(lineno)d---%(name)s---%(message)s")
+
+        # 创建初始化器
+        a = logging.Formatter(fmt=fmt1)
+        b = logging.Formatter(fmt=fmt2)
+
+        # 添加处理器格式 将上面创建的初始化器添加
+        cl.setFormatter(a)
+        fl.setFormatter(b)
+
+        # 将处理器添加到日志器
+        self.logger.addHandler(cl)
+        self.logger.addHandler(fl)
+
+        sys.excepthook = self.handle_exception
+
+    # 函数，用来记录系统报错
+    def handle_exception(self, exc_type, exc_value, exc_traceback):
+        if issubclass(exc_type, KeyboardInterrupt):
+            sys.__excepthook__(exc_type, exc_value, exc_traceback)
+            return
+        self.logger.error("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))
+
+    def __new__(cls, *args, **kwargs):
+        """如果当前没有实例时，调用父类__new__方法，生成示例，有则返回保存的内存地址。"""
+        if not cls.instance:
+            cls.instance = super().__new__(cls)
+        return cls.instance
+
+# logger.debug("debug message")
+# logger.info("debug message")
+# logger.warning("warning message")
+# logger.error("error message")
+# logger.critical("critical message")

python/services/other/module_aes.py

@@ -0,0 +1,74 @@
+from Crypto.Cipher import AES
+from binascii import b2a_hex, a2b_hex
+import wmi
+
+
+# pip install pycryptodome
+# pip install wmi
+
+class Aes(object):
+    # 如果text不足16位的倍数就用空格补足为16位
+    def add_to_16(self, text):
+        if len(text.encode('utf-8')) % 16:
+            add = 16 - (len(text.encode('utf-8')) % 16)
+        else:
+            add = 0
+        text = text + ('\0' * add)
+        return text.encode('utf-8')
+
+    # 加密函数
+    def encrypt(self, text, key, iv):
+        mode = AES.MODE_CBC
+        text = self.add_to_16(text)
+        cryptos = AES.new(key=key.encode('utf-8'), mode=mode, iv=iv.encode('utf-8'))
+        cipher_text = cryptos.encrypt(text)
+        # 因为AES加密后的字符串不一定是ascii字符集的，输出保存可能存在问题，所以这里转为16进制字符串
+        return b2a_hex(cipher_text)
+
+    # 解密后，去掉补足的空格用strip() 去掉
+    def decrypt(self, text, key, iv):
+        mode = AES.MODE_CBC
+        cryptos = AES.new(key=key.encode('utf-8'), mode=mode, iv=iv.encode('utf-8'))
+        plain_text = cryptos.decrypt(a2b_hex(text))
+        return bytes.decode(plain_text).rstrip('\0')
+
+    def get_computer_key(self):
+        c = wmi.WMI()
+        system_info = c.Win32_ComputerSystem()[0]
+        mac_address = c.Win32_NetworkAdapterConfiguration(IPEnabled=True)[0].MACAddress
+        cpu_id = c.Win32_Processor()[0].ProcessorId
+        machine_code = system_info.Manufacturer + '' + system_info.Model + '' + mac_address + '' + cpu_id
+        machine_code = machine_code.replace(" ", "").replace(":", "")
+        return machine_code
+
+    def get_authorization(self, key_text, key, iv):
+        machine_code = self.get_computer_key()
+        try:
+            text = self.decrypt(key_text, key, iv)  # 解密
+        except:
+            return "false"
+        try:
+            _machine_code, authorization = text.split("_")
+            if _machine_code != machine_code:
+                return "false"
+        except:
+            return "false"
+        return authorization
+
+    def get_key(self, authorization, key, iv):
+        machine_code = self.get_computer_key()
+        key = self.encrypt("{}_{}".format(machine_code, authorization), key, iv)  # 加密
+
+        return key.decode()
+
+
+if __name__ == '__main__':
+    print(Aes().get_computer_key())
+    # e = Aes().encrypt("hellosadsasdasdas1111+ world", "8E5EC2AC2191167DF9B753BA93A1E7B1", "0112030405060701")  # 加密
+    # d = Aes().decrypt(e, "8E5EC2AC2191167DF9B753BA93A1E7B1", "0112030405060701")  # 解密
+    c = Aes().get_key("heldasdasdsdsadsadsadsadsadsadsdsadsadasdasds", "8E5EC2AC2191167DF9B753BA93A1E7B1", "0112030405060701")
+    print(c)
+    print(
+        Aes().get_authorization(c, "8E5EC2AC2191167DF9B753BA93A1E7B1", "0112030405060701"))
+    # print("加密:", e)
+    # print("解密:", d)

python/services/other/module_online_data.py

@@ -0,0 +1,188 @@
+# import json
+# import time
+import copy
+import random
+import time
+import requests
+import settings
+import json
+import numpy as np
+import os, io
+from PIL import Image
+from io import BytesIO
+
+
+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": settings.Headers["Authorization"],
+                             "Origin": settings.Headers["Origin"],
+                             "Host": settings.Headers["Host"],
+                             "Content-Length": "0",
+                             "Content-Type": "application/json",
+                             "Accept": "application/json"}
+
+    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 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"]
+
+    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 search_progress(self, generate_ids):
+        # 查进度
+        url = "{domain}/api/ai_image/client/search_progress".format(
+            domain=settings.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=settings.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=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 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=settings.DOMAIN
+        )
+
+        headers = {"Authorization": settings.Headers["Authorization"],
+                   "Origin": settings.Headers["Origin"],
+                   "Host": settings.Headers["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=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

python/services/other/module_setting.py

@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+import time
+import sys
+import os
+
+import settings
+from UI.set_image_mode import Ui_Form
+from PySide6.QtWidgets import *
+from PySide6.QtCore import *
+
+
+class ToSetting(QWidget, Ui_Form):
+    signal_data = Signal(dict)
+
+    def __init__(self, windows):
+        super().__init__()
+        self.windows = windows
+        self.setupUi(self)
+        self.setFixedSize(self.width(), self.height())
+        self.init()
+        self.setWindowModality(Qt.ApplicationModal)
+        self.show()
+
+    def init(self):
+        self.pushButton.clicked.connect(self.run)
+        # 回填数据
+
+        self.lineEdit.setText(",".join(settings.is_fall_dir))
+
+        self.lineEdit_2.setText(",".join(settings.is_fall_file))
+
+    def run(self):
+        is_fall_dir = self.lineEdit.text()
+        if not is_fall_dir:
+            is_fall_dir = "已扣图"
+        else:
+            if "已扣图" not in is_fall_dir:
+                is_fall_dir += ",已扣图"
+
+        is_fall_dir = is_fall_dir.replace("，", ",")
+        is_fall_dir = is_fall_dir.replace(",,", ",")
+
+        settings.set_config(data_dict={"is_fall_dir": is_fall_dir}, section="basicSetup")
+
+        if is_fall_dir:
+            is_fall_dir = is_fall_dir.split(",")
+            settings.is_fall_dir = [x for x in is_fall_dir if x]
+        else:
+            settings.is_fall_dir = []
+
+
+        is_fall_file = self.lineEdit_2.text()
+        is_fall_file = is_fall_file.replace("，", ",")
+        settings.set_config(data_dict={"is_fall_file": is_fall_file}, section="basicSetup")
+
+        if is_fall_file:
+            is_fall_file = is_fall_file.split(",")
+            settings.is_fall_file = [x for x in is_fall_file if x]
+        else:
+            settings.is_fall_file = []
+
+        self.close()

python/services/other/pic.py

@@ -0,0 +1,26 @@
+from PIL import Image
+
+
+class Picture:
+    def __init__(self, in_path, im=None):
+        if im:
+            self.im = im
+        else:
+            self.im = Image.open(in_path)
+        self.x, self.y = self.im.size
+
+    def save_img(self, outpath, quality=90):
+        # self.im = self.im.convert("RGB")
+        self.im.save(outpath, quality=quality)
+
+    def resize(self, width):
+        re_x = int(width)
+        re_y = int(self.y * re_x / self.x)
+        self.im = self.im.resize((re_x, re_y), Image.BICUBIC)
+        self.x, self.y = self.im.size
+
+    def resize_by_heigh(self, heigh):
+        re_y = int(heigh)
+        re_x = int(self.x * re_y / self.y)
+        self.im = self.im.resize((re_x, re_y), Image.BICUBIC)
+        self.x, self.y = self.im.size

python/services/other/remove_bg_ali.py

@@ -0,0 +1,270 @@
+import copy
+import json
+import os
+from PIL import Image
+from alibabacloud_imageseg20191230.client import Client as imageseg20191230Client
+from alibabacloud_imageseg20191230.models import SegmentCommodityAdvanceRequest
+from alibabacloud_imageseg20191230 import models as imageseg_20191230_models
+from alibabacloud_tea_util.models import RuntimeOptions
+from alibabacloud_tea_openapi import models as open_api_models
+from alibabacloud_tea_openapi.models import Config
+from alibabacloud_tea_util import models as util_models
+import requests
+from io import BytesIO
+import cv2
+import numpy as np
+from other.module_online_data import GetOnlineData
+
+
+# todo 获取密钥
+# key_id,key_secret = GetOnlineData().get_cutout_image_config()
+# 惠利玛的key
+AccessKeyId = "LTAI5tBdDVT9Wc5idJXdGHjw"
+AccessKeySecret = "bCSotQ7eAztOxx6AqHwJJPsb0hkECe"
+
+
+# https://help.aliyun.com/zh/viapi/developer-reference/python?spm=a2c4g.11186623.0.i0#task-2252575
+# pip install alibabacloud_goodstech20191230
+# pip install alibabacloud_tea_openapi
+# pip install alibabacloud_tea_util
+
+
+class Segment(object):
+    def __init__(self):
+        self.client = self.create_client()
+
+    def get_no_bg_common(self, file_path):
+        # 初始化RuntimeObject
+        runtime_option = RuntimeOptions()
+        try:
+            # 场景一：文件在本地
+            img = open(file_path, "rb")
+            # 使用完成之后记得调用img.close()关闭流
+            # 场景二，使用任意可访问的url
+            # url = 'https://viapi-test-bj.oss-cn-beijing.aliyuncs.com/viapi-3.0domepic/ocr/RecognizeBankCard/yhk1.jpg'
+            # img = io.BytesIO(urlopen(url).read())
+            # 4、初始化Request，这里只是以RecognizeBankCard为例，其他能力请使用相应能力对应的类
+            request = SegmentCommodityAdvanceRequest()
+            request.image_urlobject = img
+
+            # 5、调用api，注意，recognize_bank_card_advance需要更换为相应能力对应的方法名。方法名是根据能力名称按照一定规范形成的，如能力名称为SegmentCommonImage，对应方法名应该为segment_common_image_advance。
+            response = self.client.segment_common_image_advance(request, runtime_option)
+            # 获取整体结果
+            # print(response.body)
+            img.close()
+            return response.body
+            # 获取单个字段，这里只是一个例子，具体能力下的字段需要看具体能力的文档
+            # print(response.body.data.card_number)
+            # tips: 可通过response.body.__dict__查看属性名称
+        except Exception as error:
+            # 获取整体报错信息
+            print("error", error)
+            return None
+            # 获取单个字段
+            # print(error.code)
+            # tips: 可通过error.__dict__查看属性名称
+
+    def get_no_bg_goods(self, file_path=None, _im=None):
+        # file_path = r"D:\MyDocuments\PythonCode\MyPython\red_dragonfly\deal_pics\change_color_2\test\_MG_9061.jpg"
+        # file_path_1 = r"D:\MyDocuments\PythonCode\MyPython\red_dragonfly\deal_pics\change_color_2\test\_MG_9061_resize.png"
+        # if file_path:
+        #     img = open(file_path, 'rb')
+        # if _im:
+        # https://blog.csdn.net/weixin_43411585/article/details/107780941
+        im = _im
+        # im.save(file_path)
+        img = BytesIO()
+        im.save(img, format="JPEG")  # format: PNG or JPEG
+        img.seek(0)  # rewind to the start
+
+        # img = img_byte.getvalue()  # im对象转为二进制流
+        # with open(file_path, "wb") as binary_file:
+        #     binary_file.write(im.tobytes())
+
+        # file_path = r"D:\MyDocuments\PythonCode\MyPython\red_dragonfly\deal_pics\change_color_2\test\1.png"
+        # img = open(file_path, 'rb')
+
+        request = imageseg_20191230_models.SegmentCommodityAdvanceRequest()
+        request.image_urlobject = img
+        client = self.create_client()
+        # 5、调用api，注意，recognize_bank_card_advance需要更换为相应能力对应的方法名。方法名是根据能力名称按照一定规范形成的，如能力名称为SegmentCommonImage，对应方法名应该为segment_common_image_advance。
+        runtime = util_models.RuntimeOptions()
+        response = client.segment_commodity_advance(request, runtime)
+        # img.close()
+        # print("1111111111111", response.body)
+        return response.body
+
+    def create_client(self):
+        """
+        使用AK&SK初始化账号Client
+        @param access_key_id:
+        @param access_key_secret:
+        @return: Client
+        @throws Exception
+        """
+        config = open_api_models.Config(
+            # 必填，您的 AccessKey ID,
+            access_key_id=AccessKeyId,
+            # 必填，您的 AccessKey Secret,
+            access_key_secret=AccessKeySecret,
+        )
+        # 访问的域名
+        config.endpoint = f"imageseg.cn-shanghai.aliyuncs.com"
+        return imageseg20191230Client(config)
+
+
+class Picture:
+    def __init__(self, in_path, im=None):
+        if im:
+            self.im = im
+        else:
+            self.im = Image.open(in_path)
+        self.x, self.y = self.im.size
+        # print(self.x, self.y)
+
+    def save_img(self, outpath, quality=90):
+        # self.im = self.im.convert("RGB")
+        self.im.save(outpath, quality=quality)
+
+    def resize(self, width):
+        re_x = int(width)
+        re_y = int(self.y * re_x / self.x)
+        self.im = self.im.resize((re_x, re_y), Image.BICUBIC)
+        self.x, self.y = self.im.size
+
+    def resize_by_heigh(self, heigh):
+        re_y = int(heigh)
+        re_x = int(self.x * re_y / self.y)
+        self.im = self.im.resize((re_x, re_y), Image.BICUBIC)
+        self.x, self.y = self.im.size
+
+
+class RemoveBgALi(object):
+    def __init__(self):
+        self.segment = Segment()
+
+    def get_image_cut(self, file_path, out_file_path=None, original_im=None):
+        if original_im:
+            original_pic = Picture(in_path=None, im=original_im)
+        else:
+            original_pic = Picture(file_path)
+
+        if original_pic.im.mode != "RGB":
+            print("抠图图片不能是PNG")
+            return None
+
+        new_pic = copy.copy(original_pic)
+        after_need_resize = False
+        if new_pic.x > new_pic.y:
+            if new_pic.x > 2000:
+                after_need_resize = True
+                new_pic.resize(2000)
+        else:
+            if new_pic.y > 2000:
+                after_need_resize = True
+                new_pic.resize_by_heigh(heigh=2000)
+
+        # new_pic.im.show()
+        body = self.segment.get_no_bg_goods(file_path=None, _im=new_pic.im)
+        body = eval(str(body))
+        try:
+            image_url = body["Data"]["ImageURL"]
+        except BaseException as e:
+            print("阿里抠图错误：", e)
+            # todo 处理失败，需要删除过程图片
+            return None
+        # 字节流转PIL对象
+        response = requests.get(image_url)
+        pic = response.content
+        _img_im = Image.open(BytesIO(pic))  # 阿里返回的抠图结果 已转PIL对象
+        # 原图更大，则需要执行CV处理
+        if after_need_resize:
+            # 将抠图结果转成mask
+            # _img_im = Image.open(_path)
+            # 将抠图结果放大到原始图大小
+            _img_im = _img_im.resize(original_pic.im.size)
+            new_big_mask = Image.new("RGB", _img_im.size, (0, 0, 0))
+            white = Image.new("RGB", _img_im.size, (255, 255, 255))
+            new_big_mask.paste(white, mask=_img_im.split()[3])
+
+            # ---------制作选区缩小的mask
+            # mask = cv2.imread(mask_path)
+            # mask = cv2.cvtColor(mask, cv2.COLOR_BGR2GRAY)
+            mask = cv2.cvtColor(
+                np.asarray(new_big_mask), cv2.COLOR_BGR2GRAY
+            )  # 将PIL 格式转换为 CV对象
+            mask[mask != 255] = 0
+            # 黑白反转
+            # mask = 255 - mask
+            # 选区缩小10
+            kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (10, 10))
+            erode_im = cv2.morphologyEx(mask, cv2.MORPH_ERODE, kernel)
+
+            # -------再进行抠图处理
+            mask = Image.fromarray(
+                cv2.cvtColor(erode_im, cv2.COLOR_GRAY2RGBA)
+            )  # CV 对象转 PIL
+            transparent_im = Image.new("RGBA", original_pic.im.size, (0, 0, 0, 0))
+            # original_pic.im.show()
+            # mask.show()
+            transparent_im.paste(original_pic.im, (0, 0), mask.convert("L"))
+            # transparent_im.show()
+            # 上述抠图结果进行拼接
+            _img_im.paste(transparent_im, (0, 0), transparent_im)
+            # _img_im.show("11111111111111111111111")
+        if out_file_path:
+            _img_im.save(out_file_path)
+        return _img_im
+
+    def get_image_cut1(self, file_path, out_file_path=None):
+        original_pic = Picture(file_path)
+        new_pic = copy.copy(original_pic)
+        if new_pic.x > 2000:
+            new_pic.resize(2000)
+        # new_pic.im.show()
+        body = self.segment.get_no_bg_goods(file_path=out_file_path, _im=new_pic.im)
+        body = eval(str(body))
+        try:
+            image_url = body["Data"]["ImageURL"]
+        except BaseException as e:
+            print("阿里抠图错误：", e)
+            # todo 处理失败，需要删除过程图片
+            return None
+        # 字节流转PIL对象
+        response = requests.get(image_url)
+        pic = response.content
+        _img_im = Image.open(BytesIO(pic))  # 阿里返回的抠图结果 已转PIL对象
+
+        if original_pic.x > 2000:
+            # 原图更大，则需要执行CV处理
+            # _img_im.show()
+            # 对mask进行放大，然后进行抠图处理
+            print("对mask进行放大，然后进行抠图处理")
+            transparent_im = Image.new("RGBA", original_pic.im.size, (0, 0, 0, 0))
+            # original_pic.im.show()
+            # mask.show()
+            _img_im = _img_im.resize((original_pic.x, original_pic.y))
+            # _img_im.show()
+            transparent_im.paste(original_pic.im, (0, 0), mask=_img_im)
+            # transparent_im.show()
+            # transparent_im.show()
+            _img_im = transparent_im
+            # 上述抠图结果进行拼接
+            # _img_im.paste(transparent_im, (0, 0), transparent_im)
+            pass
+
+        _img_im.save(out_file_path)
+        return _img_im
+
+    def download_picture(self, url, out_path):
+        response = requests.get(url)
+        pic = response.content
+        with open(out_path, "wb") as f:
+            f.write(pic)
+
+
+if __name__ == "__main__":
+    r = RemoveBgALi()
+    path = r"D:\MyDocuments\PythonCode\MyPython\red_dragonfly\deal_pics\auto_capture_V2\IPC\test\171112057820408.png"
+    out_path = "{}._no_bg-out.png".format(path)
+    r.get_image_cut(path, out_file_path=out_path)

python/services/other/rsa.py

@@ -0,0 +1,50 @@
+import base64
+import time
+import json
+
+from Crypto.Cipher import PKCS1_v1_5 as PKCS1_cipher
+from Crypto.PublicKey import RSA
+
+PUBLIC_KEY = '-----BEGIN PUBLIC KEY-----\n\
+MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDElMhAUOwyo78M97jDGlYvl1FZ\n\
+JZV+/EKZ3G7VDO1T5uUNnXWqgxSSwobbd384/npxvfFznoeJ+PaM/fXFNeOGKL9N\n\
+GrswS/jTmSdIUdoMQVbGuzrGUK1liK+L9VGxxCmbvJ1f+6iiOHUwZAx6AXA2Mk2s\n\
+8LVtXFCoM5cZWG4GiQIDAQAB\n\
+-----END PUBLIC KEY-----'
+
+
+class RSAService:
+    def __int__(self):
+        pass
+
+    def encrypt_public(self, phone) -> str:
+        """
+        使用公钥加密电话号码。
+
+        该方法首先构建一个包含电话号码、时间戳和设备信息的数据字典，
+        然后将该字典转换为JSON字符串，并将其编码为UTF-8字节流。
+        接着，使用RSA公钥对这些数据进行加密，并将加密后的数据使用Base64进行编码，
+        最后返回加密后的字符串。
+
+        参数:
+        phone -- 需要加密的电话号码。
+
+        返回:
+        加密后的电话号码字符串。
+        """
+        # 构建数据字典，包含电话号码、当前时间和设备信息
+        data = {"phone": phone, "time": int(time.time() * 1000), "device": "python"}
+        # 将数据字典转换为JSON字符串，并编码为UTF-8字节流
+        msg = json.dumps(data).encode("utf8")
+        # 使用RSA公钥创建加密器
+        cipher = PKCS1_cipher.new(RSA.importKey(PUBLIC_KEY))
+        # 使用RSA公钥加密数据，并使用Base64编码加密后的数据
+        encrypt_text = base64.b64encode(cipher.encrypt(bytes(msg)))
+        # 返回加密后的数据字符串
+        return encrypt_text.decode('utf-8')
+
+
+if __name__ == '__main__':
+    rsaService = RSAService()
+    msg = rsaService.encrypt_public(phone='13323232323')
+    print(msg)

python/services/remove_bg_pixian.py

@@ -0,0 +1,230 @@
+import copy
+import os
+from PIL import Image
+from other.remove_bg_ali import RemoveBgALi
+import requests
+from io import BytesIO
+
+
+class Segment(object):
+    def __init__(self):
+        self.k = "pxnib99dbchtmdm"
+        self.s = "ub9uj5678gs4m2bnrass1t3tn6ughlk065ianosk06akagolcr2u"
+
+    def get_no_bg_goods2(self, file_path=None, _im=None):
+        im = _im
+        img = BytesIO()
+        try:
+            im.save(img, format='JPEG')  # format: PNG or JPEG
+        except:
+            im.save(img, format='PNG')  # format: PNG or JPEG
+        img.seek(0)  # rewind to the start
+
+        # img = "https://ossimg.valimart.net/uploads/vali_ai/20241011/172864207036098.png"
+
+        response = requests.post(
+            'http://47.76.110.118:27777/api/v2/remove-background',
+            files={'image.url': img},
+            data={
+                # TODO: Add more upload options here
+            },
+            auth=(self.k, self.s)
+        )
+        # print(response.content)
+        if response.status_code == requests.codes.ok:
+            return response.content, ""
+        else:
+            return None, response.content
+
+    def get_no_bg_goods_by_url(self, url):
+        response = requests.post(
+            'https://api.pixian.ai/api/v2/remove-background',
+            data={
+                'image.url': url
+            },
+            auth=(self.k, self.s)
+        )
+
+        if response.status_code == requests.codes.ok:
+            return response.content, ""
+        else:
+            print("response.status_code:", response.status_code)
+            return None, response.content
+
+    def get_no_bg_goods(self, file_path=None, _im=None, key=None):
+        im = _im
+        img = BytesIO()
+        try:
+            im.save(img, format='JPEG')  # format: PNG or JPEG
+        except:
+            im.save(img, format='PNG')  # format: PNG or JPEG
+        img.seek(0)  # rewind to the start
+
+        if key:
+            # todo 切换key
+            auth = key
+            # auth = (self.k, self.s)
+        else:
+            auth = (self.k, self.s)
+
+
+        try:
+            response = requests.post(
+                'https://api.pixian.ai/api/v2/remove-background',
+                files={'image': img},
+                data={
+                    # TODO: Add more upload options here
+                },
+                auth=auth,
+                timeout=40
+            )
+        except BaseException as e:
+            data = {"im": None,
+                    "status_code": "time_out",
+                    "message":"{}".format(e)
+                    }
+            return data
+
+        # print(response.content)
+
+        data = {"im": None,
+                "status_code": response.status_code, }
+
+        if response.status_code == requests.codes.ok:
+            data["im"] = Image.open(BytesIO(response.content))
+
+        return data
+
+
+class Picture:
+    def __init__(self, in_path, im=None):
+        if im:
+            self.im = im
+        else:
+            self.im = Image.open(in_path)
+        self.x, self.y = self.im.size
+        # print(self.x, self.y)
+
+    def save_img(self, outpath, quality=90):
+        # self.im = self.im.convert("RGB")
+        self.im.save(outpath, quality=quality)
+
+    def resize(self, width):
+        re_x = int(width)
+        re_y = int(self.y * re_x / self.x)
+        self.im = self.im.resize((re_x, re_y), Image.BICUBIC)
+        self.x, self.y = self.im.size
+
+    def resize_by_heigh(self, heigh):
+        re_y = int(heigh)
+        re_x = int(self.x * re_y / self.y)
+        self.im = self.im.resize((re_x, re_y), Image.BICUBIC)
+        self.x, self.y = self.im.size
+
+
+class RemoveBgPiXian(object):
+    def __init__(self):
+        self.segment = Segment()
+        self.r = RemoveBgALi()
+
+    def direct_matting_image(self, image):
+        # todo 不能超过32,000,000尺寸的数据
+        x, y = image.size
+        f = False
+        if x * y > 32000000:
+            r = 32000000 / x * y
+            image = image.resize(size=(int(x * r), int(y * r)))
+            f = True
+        pic, _ = self.segment.get_no_bg_goods(file_path=None, _im=image)
+        if not pic:
+            return None, _
+        _img_im = Image.open(BytesIO(pic))  # 阿里返回的抠图结果 已转PIL对象
+        if f:
+            _img_im = _img_im.resize(size=(x, y))
+        return _img_im, ""
+
+    def run_by_image_url(self, url):
+        pic, _ = self.segment.get_no_bg_goods_by_url(url)
+        if pic is not None:
+            _img_im = Image.open(BytesIO(pic))
+            return _img_im, None
+        else:
+            return None, _
+
+    def run_by_image_im(self, im, key):
+        return self.segment.get_no_bg_goods(_im=im, key=key)
+
+    def get_image_cut(self, file_path, out_file_path=None, original_im=None, image_preprocessing=False, is_test=False):
+        if original_im:
+            original_pic = Picture(in_path=None, im=original_im)
+        else:
+            original_pic = Picture(file_path)
+
+        if original_pic.im.mode != "RGB":
+            print("抠图图片不能是PNG")
+            return False, {"data": "抠图图片不能是PNG"}
+
+        if is_test:
+            cut_image = self.r.get_image_cut(file_path=None, out_file_path=None, original_im=original_pic.im)
+            if out_file_path:
+                cut_image.save(out_file_path)
+            return True, {}
+
+        if image_preprocessing:
+            cut_image = self.r.get_image_cut(file_path=None, out_file_path=None, original_im=original_pic.im)
+            image_deal_info = {}
+            x1, y1, x2, y2 = cut_image.getbbox()
+            image_deal_info["鞋子原始位置"] = (x1, y1, x2, y2)
+            o_w, o_h = cut_image.size
+            image_deal_info["鞋子原始抠图后大小"] = (o_w, o_h)
+            # 扩边处理
+            _w, _h = x2 - x1, y2 - y1
+            out_px = 0.06
+            _w, _h = int(out_px * _w), int(out_px * _h)
+            n_x1, n_y1, n_x2, n_y2 = x1 - _w, y1 - _h, x2 + _w, y2 + _h
+            if n_x1 < 0:
+                n_x1 = 0
+            if n_y1 < 0:
+                n_y1 = 0
+            if n_x2 > o_w:
+                n_x2 = o_w
+            if n_y2 > o_h:
+                n_y2 = o_h
+
+            image_deal_info["抠图扩边后位置"] = (n_x1, n_y1, n_x2, n_y2)
+            cut_image = original_pic.im.crop(image_deal_info["抠图扩边后位置"])
+            image_deal_info["抠图扩边后图片大小"] = cut_image.size
+            image_deal_info["原始图片大小"] = (original_pic.x, original_pic.y)
+
+            # 使用pixian进行抠图
+            second_cut_image, _ = self.direct_matting_image(image=cut_image)
+            if not second_cut_image:
+                return False, {"data": _}
+
+            if second_cut_image.size != image_deal_info["抠图扩边后图片大小"]:
+                print("图片尺寸还原")
+                second_cut_image = second_cut_image.resize(image_deal_info["抠图扩边后图片大小"])
+
+            # 创建空白图片并粘贴回去
+            _img_im = Image.new(mode="RGBA", size=image_deal_info["原始图片大小"], color=(0, 0, 0, 0))
+            _img_im.paste(second_cut_image, box=(image_deal_info["抠图扩边后位置"][0], image_deal_info["抠图扩边后位置"][1]))
+        else:
+            _img_im = self.direct_matting_image(image=original_pic.im)
+            pass
+
+        if out_file_path:
+            _img_im.save(out_file_path)
+        return True, {}
+
+    def download_picture(self, url, out_path):
+        response = requests.get(url)
+        pic = response.content
+        with open(out_path, 'wb') as f:
+            f.write(pic)
+
+
+if __name__ == '__main__':
+    r = RemoveBgPiXian()
+    path = r"C:\Users\gymmc\Desktop\白底部分品类45度\测试\eva_keai_1.png"
+    out_path = "{}._no_bg-out.png".format(path)
+    r.get_image_cut(path, out_file_path=out_path)