app.py

#!/usr/local/bin/python3
# -*- coding: utf-8 -*-

import uuid
import time
import os
import logging
import logging.handlers
import requests
import gradio as gr
import simplejson as json
import oss2

# LOGGER = logging.getLogger('skybox')
# HANDLER = logging.handlers.RotatingFileHandler(
#     './logs/skybox.log', mode='a', maxBytes=20 * 1000 * 1000, backupCount=5, encoding='utf-8')
# FORMATTER = logging.Formatter('%(asctime)s - %(funcName)s - %(lineno)d - %(levelname)s - %(message)s')
# HANDLER.setFormatter(FORMATTER)
# LOGGER.propagate = 0
# LOGGER.addHandler(HANDLER)
# LOGGER.setLevel(logging.DEBUG)

CSS = '.gradio-container a {color:#b7adf4 !important}'
HEADERS = {'app-key': os.environ['a3'], 'origin': os.environ['a4'], 'referer': os.environ['a4']}


def login(loginId, password):
    try:
        data = {'loginId': loginId, 'password': password}
        cookies = {'720yun_v8_session': str(uuid.uuid4())}

        resp = requests.post(os.environ['a5'], data=data, headers=HEADERS, cookies=cookies)
        if resp.status_code == 200:
            return resp.json()['data']['token']
    except Exception as e:
        # LOGGER.error(e)
        raise gr.Error('登录错误，请确认你的720yun账号是否正确')


def create_panorama(prompt, negative_prompt):
    try:
        data = {
            'api_key': os.environ['a2'],
            'generator': 'stable-skybox',
            'prompt': prompt.strip()[0:598],
            'negative_text': negative_prompt.strip()[0:398]
        }
        resp = requests.post(os.environ['a1'], data=data)
        # LOGGER.info(resp.text)
        if resp.status_code == 200:
            time.sleep(20)
            request_id = resp.json()['request']['id']
            flag = True
            countdown = 50
            while flag:
                countdown -= 1
                if countdown <= 0:
                    raise gr.Error('AI老师罢工了，请稍后重试')

                resp = requests.get(f'{os.environ["a1"]}/{request_id}?api_key={os.environ["a2"]}')
                if resp.status_code == 200:
                    progress_data = resp.json()['request']
                    if progress_data['status'] == 'complete':
                        flag = False
                        return request_id, progress_data['file_url']
                time.sleep(1)
    except Exception as e:
        # LOGGER.error(e)
        print(e)
        raise gr.Error('AI老师罢工了，请稍后重试')


def upload(token, file_id, file_url):
    now = int(time.time())
    data = [{
        'fileId': f'{file_id}-{now}',
        'name': f'AI素材-{now}',
        'watermarked': 0,
        'size': 1024 * 1024 * 6,
        'albumId': 0,
        'exif': {},
        'gps': {},
        'panoId': 0,
        'action': 1
    }]
    HEADERS['app-authorization'] = token

    try:
        resp = requests.post(os.environ['a6'], data={'panos': json.dumps(data)}, headers=HEADERS)
        if resp.status_code == 200:
            resp_data = resp.json()['data'][0]
            access_key_id = resp_data['accessKeyId']
            security_token = resp_data['securityToken']
            accessKey_secret = resp_data['accessKeySecret']
            bucket_name = resp_data['bucketName']
            path = resp_data['path'][1:]
            pano_id = resp_data['panoId']
            task_id = resp_data['taskId']
            expired = resp_data['expired']

            auth = oss2.StsAuth(access_key_id, accessKey_secret, security_token)
            bucket = oss2.Bucket(auth, os.environ['a9'], bucket_name)

            input_stream = requests.get(file_url)
            result = bucket.put_object(f'{path}/{pano_id}.jpg', input_stream)
            if result.status == 200:
                resp = requests.post(f'{os.environ["a6"]}/{task_id}', data={'status': 3, 'expired': expired}, headers=HEADERS)
                if resp.status_code == 200:
                    time.sleep(5)
                    flag = True
                    while flag:
                        pano_ids = [pano['id'] for pano in requests.get(os.environ["a7"], headers=HEADERS).json()['data']]
                        if pano_id not in pano_ids:
                            flag = False
                            break
                        time.sleep(0.5)
                    data = {
                        'name': f'AI全景作品-{now}',
                        'materials': json.dumps([{
                            'type': 1,
                            'id': pano_id
                        }]),
                        'templateId': 2,
                        'publishPlatform': 1,
                        'keywords': 'AI全景',
                        'source': 99
                    }
                    resp = requests.post(os.environ['a8'], data=data, headers=HEADERS)
                    # LOGGER.error(resp.text)
                    if resp.status_code == 200:
                        return resp.json()['data']['tid']
    except Exception as e:
        # LOGGER.error(e)
        print(e)
        raise gr.Error('AI老师罢工了，请稍后重试')


def main(loginId, password, prompt, negative_prompt, state, progress=gr.Progress()):
    try:
        if 'token' not in state:
            state['token'] = login(loginId, password)
        token = state['token']

        file_id, image = create_panorama(prompt, negative_prompt)
        panorama_id = upload(token, file_id, image)
        return f'https://www.720yun.com/vr/{panorama_id}'
    except Exception as e:
        return f'{e}'


with gr.Blocks(css=CSS) as demo:
    session = gr.State({})

    gr.Markdown("""
        # 创造属于你自己的AI全景
        1. 需要你的 **[720yun.com](https://www.720yun.com)** 账号.
        2. 由 **[720yun.com](https://www.720yun.com)** 提供支持.
        3. 描述例子: a beautiful matte painting, northernmost continent, a gigantic square fortress covered by blizzard, epic composition, post apocalyptic, sci-fi, futuristic, fantasy, by Jan Urschel and Sergey Vasnev and Emmanuel Shiu and Michal Karcz, cinematic, cinematic lighting, light effect, epic, octane render, unreal engine
    """)
    with gr.Row():
        with gr.Column():
            login_id = gr.Textbox(label='账号: (必填)', placeholder='720yun 账号')
        with gr.Column():
            password = gr.Textbox(label='密码: (必填)', type='password', placeholder='720yun 密码')
    with gr.Row():
        prompt = gr.Textbox(label='描述: (必填)', lines=4, placeholder='请使用英文输入描述，最多600个字符')
    with gr.Row():
        negative_prompt = gr.Textbox(label='负面描述', lines=2, placeholder='请使用英文输入负面描述，最多400个字符')
    with gr.Row():
        out = gr.Textbox(label='输出')

    btn = gr.Button('运行')
    btn.click(fn=main, inputs=[login_id, password, prompt, negative_prompt, session], outputs=out, show_progress=True)

demo.queue(concurrency_count=3).launch()