|
|
| import os |
| import sys |
| import cv2 |
| import json |
| import random |
| import time |
| import requests |
| import func_timeout |
| import numpy as np |
| import gradio as gr |
|
|
|
|
| ApiUrl = os.environ['ApiUrl'] |
| OpenId = os.environ['OpenId'] |
| ApiKey = os.environ['ApiKey'] |
| UploadToken = os.environ['UploadToken'] |
|
|
|
|
| proj_dir = os.path.dirname(os.path.abspath(__file__)) |
| data_dir = os.path.join(proj_dir, 'Datas') |
| |
| tmpFolder = "tmp" |
| os.makedirs(tmpFolder, exist_ok=True) |
|
|
|
|
|
|
| def upload_imgs(apiUrl, UploadToken, cloth_image, pose_image): |
| folder = os.path.join(tmpFolder, str(random.randint(0, 100))) |
| os.makedirs(folder, exist_ok=True) |
| pose_path = os.path.join(folder, 'pose.jpg') |
| cloth_path = os.path.join(folder, 'cloth.jpg') |
| cv2.imwrite(pose_path, pose_image[:,:,::-1]) |
| cv2.imwrite(cloth_path, cloth_image[:,:,::-1]) |
|
|
| params = {'token':UploadToken, |
| 'input1':'pose.jpg', |
| 'input2':'cloth.jpg', |
| 'protocol':'https'} |
| session = requests.session() |
| ret = requests.post(f"{apiUrl}/upload", data=json.dumps(params)) |
| if ret.status_code==200: |
| if 'upload1' in ret.json(): |
| data = ret.json() |
| with open(cloth_path, 'rb') as file: |
| headers = {"Content-Type": 'image/jpeg'} |
| response = requests.put(data['upload2'], data=file, headers=headers) |
| if response.status_code == 200: |
| print("上传成功") |
| else: |
| print(f"上传失败，状态码: {response.status_code}, 响应内容: {response.text}") |
| return |
| with open(pose_path, 'rb') as file: |
| response = requests.put(data['upload1'], data=file, headers=headers) |
| if response.status_code == 200: |
| print("上传成功") |
| else: |
| print(f"上传失败，状态码: {response.status_code}, 响应内容: {response.text}") |
| return |
| if os.path.exists(pose_path): os.remove(pose_path) |
| if os.path.exists(cloth_path): os.remove(cloth_path) |
| return {'pose':data['upload1'], 'cloth':data['upload2']} |
|
|
| def publicFastSwap(apiUrl, openId, apiKey, uploads, category, caption): |
| if category=="upper_cloth": |
| category = 1 |
| elif category=="lower_cloth": |
| category = 2 |
| elif category=="dresses": |
| category = 3 |
| elif category=="full_body": |
| category = 4 |
| params = {'openId':OpenId, 'apiKey':ApiKey, |
| 'task_type':"10", 'image':str(uploads['pose']), |
| 'mask':str(uploads['cloth']), |
| 'param1':str(category), 'param2':str(caption), |
| 'param3':'', 'param4':'', 'param5':'', } |
| session = requests.session() |
| ret = requests.post(f"{ApiUrl}/public", data=json.dumps(params)) |
| if ret.status_code==200: |
| if 'id' in ret.json(): |
| print('public task successfully!') |
| return ret.json()['id'] |
|
|
| def getTaskRes(apiUrl, taskId): |
| params = {'id':taskId, 'task_type':"10"} |
| session = requests.session() |
| ret = requests.post(f"{apiUrl}/status", data=json.dumps(params)) |
| if ret.status_code==200: |
| if 'status' in ret.json(): |
| return ret.json() |
| else: |
| print(ret.json(), ret.status_code, 'call status failed') |
| return None |
|
|
| @func_timeout.func_set_timeout(10) |
| def check_func(ip): |
| session = requests.session() |
| ret = requests.get(f"https://webapi-pc.meitu.com/common/ip_location?ip={ip}") |
| for k in ret.json()['data']: |
| nat = ret.json()['data'][k]['nation'] |
| if nat.lower() in Regions.lower(): |
| print(nat, 'invalid') |
| return False |
| else: |
| print(nat, 'valid') |
| return True |
| def check_warp(ip): |
| try: |
| return check_func(ip) |
| except Exception as e: |
| print(e) |
| return True |
|
|
