#!/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()