utils.py

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 = "https://selfitcamera.site/public"
OpenId = 'ovB-x639B8QwdfF7kQYS9QKdK6u8'
ApiKey = 'be0b28ba433b6245ce10b7c341f'
OssUrl = "https://selfit-deploy-1256039085.cos.accelerate.myqcloud.com/"
Regions = "IndiaKorea"
TOKEN = os.environ['TOKEN']


proj_dir = os.path.dirname(os.path.abspath(__file__))
data_dir = os.path.join(proj_dir, 'Datas')
# data_dir = "Datas"
tmpFolder = "tmp"
os.makedirs(tmpFolder, exist_ok=True)


def get_cloth_examples(hr=0):
    cloth_dir = os.path.join(data_dir, 'ClothImgs')
    examples = []
    files = sorted(os.listdir(cloth_dir))
    hr_clothes = [507, 508]
    for f in files:
        if '.jpg' not in f and '.png' not in f:
            continue

        cloth_id = f.split(".")[0]
        if int(cloth_id) in hr_clothes and hr==0:
            continue
        if int(cloth_id) not in hr_clothes and hr==1:
            continue

        cloth_path = os.path.join(cloth_dir, f)
        examples.append(cloth_path)
    examples = examples[::-1]
    return examples

def get_pose_examples():
    pose_dir = os.path.join(data_dir, 'PoseImgs')
    examples = []
    for f in os.listdir(pose_dir):
        if '.jpg' not in f and '.png' not in f:
            continue
        pose_id = f.split(".")[0]
        pose_path = os.path.join(pose_dir, f)
        examples.append(pose_path)
    return examples

def get_result_example(cloth_id, pose_id):
    result_dir = os.path.join(data_dir, 'ResultImgs')
    res_path = os.path.join(result_dir, f"{cloth_id}_{pose_id}.jpg")
    return res_path

def get_tips():
    path1 = OssUrl+'ClothData/Publics/PoseGuide/tip1.jpg'
    path2 = OssUrl+'ClothData/Publics/PoseGuide/tip2.jpg'
    return path1, path2
    
def getAllInfs(apiUrl, openId, apiKey, clientIp):
    params = {'openId':openId, 'apiKey':apiKey, 'ipId':clientIp}
    session = requests.session()
    ret = requests.get(f"{apiUrl}/api/all_infs", params=params)
    res = []
    if ret.status_code==200:
        if 'data' in ret.json():
            records = ret.json()['data']['records']
            for record in records:
                res.append({'pose':OssUrl+record['body_url']+"?thumbnail/768x768>", 
                    'res':OssUrl+record['showUrl']+"?thumbnail/768x768>",
                    'state':int(record['state'])})
                # res.append({'pose':OssUrl+record['body_url'], 
                #     'res':OssUrl+record['showUrl']})
    return res


def upload_pose_img(apiUrl, openId, apiKey, clientIp, timeId, img):
    fileName = clientIp.replace(".", "")+str(timeId)+".jpg"
    local_path = os.path.join(tmpFolder, fileName)
    img = cv2.imread(img)
    cv2.imwrite(os.path.join(tmpFolder, fileName), img)

    json_data = {
        "token": "c0e69e5d129b11efa10c525400b75156",  # 使用 apiKey 作为 token
        "input1": fileName,
        "input2": "",
        "protocol": "",
        "cloud": "ali"
    }
    
    session = requests.session()
    ret = requests.post(
        f"https://selfitapi.uk/upload",
        headers={'Content-Type': 'application/json'},
        json=json_data
    )
    
    res = ""
    if ret.status_code==200:
        if 'upload1' in ret.json():
            upload_url = ret.json()['upload1']
            headers = {'Content-Type': 'image/jpeg'}
            response = session.put(upload_url, data=open(local_path, 'rb').read(), headers=headers)
            print(response.status_code)
            if response.status_code == 200:
                res = upload_url
    if os.path.exists(local_path):
        os.remove(local_path)
    return res


def publicClothSwap(image, clothId, is_hr=0):
    json_data = {
        "image": image,
        "task_type": "11", 
        "param1": str(clothId),
        "param2": "",
        "param3": "",
        "param4": str(is_hr),
        "delete_if_complete": "0"
    }
    
    headers = {
        'Authorization': f'Bearer {TOKEN}',
        'Content-Type': 'application/json'
    }
    
    ret = requests.post(
        'https://selfitapi.uk/public_advton',
        headers=headers,
        json=json_data
    )
    
    if ret.status_code == 200:
        response = ret.json()
        if 'mid_result' in response and 'id' in response:
            return {'mid_result': response['mid_result'], 'id': response['id'], "msg": response['msg']}
        print(response)
        return None
    else:
        return None

def getInfRes(taskId):
    headers = {
        'Content-Type': 'application/json'
    }
    json_data = {
        'id': taskId
    }
    ret = requests.post(
        'https://selfitapi.uk/status_advton',
        headers=headers,
        json=json_data
    )
    if ret.status_code == 200:
        response = ret.json()
        if 'status' in response:
            return response
        print(response)
        return None
    else:
        return None

@func_timeout.func_set_timeout(10)
def check_region(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 in Regions:
            print(nat, 'invalid')
            return False
        else:
            print(nat, 'valid')
    return True
def check_region_warp(ip):
    try:
        return check_region(ip)
    except Exception as e:
        print(e)
        return True