File size: 7,178 Bytes
a4b70d9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 |
from __future__ import annotations
import uuid
import aiohttp
import random
import asyncio
import json
from ...providers.response import ImageResponse
from ...errors import MissingRequirementsError, NoValidHarFileError
from ...typing import AsyncResult, Messages
from ...requests.raise_for_status import raise_for_status
from ...requests.aiohttp import get_connector
from ...requests import get_nodriver
from ..Copilot import get_headers, get_har_files
from ..base_provider import AsyncGeneratorProvider, ProviderModelMixin
from ..helper import get_random_hex, format_media_prompt
from ... import debug
class MicrosoftDesigner(AsyncGeneratorProvider, ProviderModelMixin):
label = "Microsoft Designer"
url = "https://designer.microsoft.com"
working = True
use_nodriver = True
needs_auth = True
default_image_model = "dall-e-3"
image_models = [default_image_model, "1024x1024", "1024x1792", "1792x1024"]
models = image_models
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
prompt: str = None,
proxy: str = None,
**kwargs
) -> AsyncResult:
image_size = "1024x1024"
if model != cls.default_image_model and model in cls.image_models:
image_size = model
yield await cls.generate(format_media_prompt(messages, prompt), image_size, proxy)
@classmethod
async def generate(cls, prompt: str, image_size: str, proxy: str = None) -> ImageResponse:
try:
access_token, user_agent = readHAR("https://designerapp.officeapps.live.com")
except NoValidHarFileError as h:
debug.log(f"{cls.__name__}: {h}")
try:
access_token, user_agent = await get_access_token_and_user_agent(cls.url, proxy)
except MissingRequirementsError:
raise h
images = await create_images(prompt, access_token, user_agent, image_size, proxy)
return ImageResponse(images, prompt)
async def create_images(prompt: str, access_token: str, user_agent: str, image_size: str, proxy: str = None, seed: int = None):
url = 'https://designerapp.officeapps.live.com/designerapp/DallE.ashx?action=GetDallEImagesCogSci'
if seed is None:
seed = random.randint(0, 10000)
headers = {
"User-Agent": user_agent,
"Accept": "application/json, text/plain, */*",
"Accept-Language": "en-US",
'Authorization': f'Bearer {access_token}',
"AudienceGroup": "Production",
"Caller": "DesignerApp",
"ClientId": "b5c2664a-7e9b-4a7a-8c9a-cd2c52dcf621",
"SessionId": str(uuid.uuid4()),
"UserId": get_random_hex(16),
"ContainerId": "1e2843a7-2a98-4a6c-93f2-42002de5c478",
"FileToken": "9f1a4cb7-37e7-4c90-b44d-cb61cfda4bb8",
"x-upload-to-storage-das": "1",
"traceparent": "",
"X-DC-Hint": "FranceCentral",
"Platform": "Web",
"HostApp": "DesignerApp",
"ReleaseChannel": "",
"IsSignedInUser": "true",
"Locale": "de-DE",
"UserType": "MSA",
"x-req-start": "2615401",
"ClientBuild": "1.0.20241120.9",
"ClientName": "DesignerApp",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "cross-site",
"Pragma": "no-cache",
"Cache-Control": "no-cache",
"Referer": "https://designer.microsoft.com/"
}
form_data = aiohttp.FormData()
form_data.add_field('dalle-caption', prompt)
form_data.add_field('dalle-scenario-name', 'TextToImage')
form_data.add_field('dalle-batch-size', '4')
form_data.add_field('dalle-image-response-format', 'UrlWithBase64Thumbnail')
form_data.add_field('dalle-seed', seed)
form_data.add_field('ClientFlights', 'EnableBICForDALLEFlight')
form_data.add_field('dalle-hear-back-in-ms', 1000)
form_data.add_field('dalle-include-b64-thumbnails', 'true')
form_data.add_field('dalle-aspect-ratio-scaling-factor-b64-thumbnails', 0.3)
form_data.add_field('dalle-image-size', image_size)
async with aiohttp.ClientSession(connector=get_connector(proxy=proxy)) as session:
async with session.post(url, headers=headers, data=form_data) as response:
await raise_for_status(response)
response_data = await response.json()
form_data.add_field('dalle-boost-count', response_data.get('dalle-boost-count', 0))
polling_meta_data = response_data.get('polling_response', {}).get('polling_meta_data', {})
form_data.add_field('dalle-poll-url', polling_meta_data.get('poll_url', ''))
while True:
await asyncio.sleep(polling_meta_data.get('poll_interval', 1000) / 1000)
async with session.post(url, headers=headers, data=form_data) as response:
await raise_for_status(response)
response_data = await response.json()
images = [image["ImageUrl"] for image in response_data.get('image_urls_thumbnail', [])]
if images:
return images
def readHAR(url: str) -> tuple[str, str]:
api_key = None
user_agent = None
for path in get_har_files():
with open(path, 'rb') as file:
try:
harFile = json.loads(file.read())
except json.JSONDecodeError:
# Error: not a HAR file!
continue
for v in harFile['log']['entries']:
if v['request']['url'].startswith(url):
v_headers = get_headers(v)
if "authorization" in v_headers:
api_key = v_headers["authorization"].split(maxsplit=1).pop()
if "user-agent" in v_headers:
user_agent = v_headers["user-agent"]
if api_key is None:
raise NoValidHarFileError("No access token found in .har files")
return api_key, user_agent
async def get_access_token_and_user_agent(url: str, proxy: str = None):
browser, stop_browser = await get_nodriver(proxy=proxy, user_data_dir="designer")
try:
page = await browser.get(url)
user_agent = await page.evaluate("navigator.userAgent", return_by_value=True)
access_token = None
while access_token is None:
access_token = await page.evaluate("""
(() => {
for (var i = 0; i < localStorage.length; i++) {
try {
item = JSON.parse(localStorage.getItem(localStorage.key(i)));
if (item.credentialType == "AccessToken"
&& item.expiresOn > Math.floor(Date.now() / 1000)
&& item.target.includes("designerappservice")) {
return item.secret;
}
} catch(e) {}
}
})()
""")
if access_token is None:
await asyncio.sleep(1)
await page.close()
return access_token, user_agent
finally:
stop_browser() |