242 lines
7.9 KiB
Python
242 lines
7.9 KiB
Python
from enum import Enum
|
|
|
|
from .settings import get_settings
|
|
import copy
|
|
|
|
mirror_url = [
|
|
{
|
|
"id": "rawgithub",
|
|
"o_url": "raw.githubusercontent.com",
|
|
# "n_url": "raw.gitmirror.com",
|
|
"n_url": "None",
|
|
"u_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36 Edg/123.0.0.0"
|
|
},
|
|
{
|
|
"id": "huggingface",
|
|
"o_url": "huggingface.co",
|
|
# "n_url": "hf-mirror.com"
|
|
"n_url": "None",
|
|
"u_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36 Edg/123.0.0.0",
|
|
},
|
|
{
|
|
"id": "github",
|
|
"o_url": "github.com",
|
|
# "n_url": "mirror.ghproxy.com/https://github.com"
|
|
"n_url": "None",
|
|
"u_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36 Edg/123.0.0.0",
|
|
},
|
|
]
|
|
clone_mirror_url = [
|
|
{
|
|
"id": "clone_github",
|
|
"o_url": "github.com",
|
|
# "n_url": "mirror.ghproxy.com/https://github.com"
|
|
"n_url": "None",
|
|
"u_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36 Edg/123.0.0.0",
|
|
},
|
|
]
|
|
|
|
|
|
class Mirror(Enum):
|
|
DOWN_MODEL = 0
|
|
GIT_CLONE = 1
|
|
|
|
|
|
def get_custom_mirrors(mirror_type=None):
|
|
settings = get_settings()
|
|
if mirror_type is Mirror.GIT_CLONE:
|
|
base_mirrors = copy.deepcopy(clone_mirror_url)
|
|
if settings and 'clone_github_mirror' in settings:
|
|
base_mirrors[0]['n_url'] = settings['clone_github_mirror']
|
|
elif mirror_type is Mirror.DOWN_MODEL:
|
|
base_mirrors = copy.deepcopy(mirror_url)
|
|
if settings and 'huggingface_mirror' in settings:
|
|
base_mirrors[1]['n_url'] = settings['huggingface_mirror']
|
|
if settings and 'rawgithub_mirror' in settings:
|
|
base_mirrors[0]['n_url'] = settings['rawgithub_mirror']
|
|
if settings and 'github_mirror' in settings:
|
|
base_mirrors[2]['n_url'] = settings['github_mirror']
|
|
else:
|
|
base_mirrors = {}
|
|
return base_mirrors
|
|
|
|
|
|
def replace_mirror_url():
|
|
from urllib.parse import urlparse
|
|
|
|
def replace_url(url: str, mirror_type: Mirror = None):
|
|
u = urlparse(url)
|
|
netloc = u.netloc
|
|
found = False
|
|
user_agent = None
|
|
for mirror in get_custom_mirrors(mirror_type):
|
|
if netloc is not None and len(netloc) > 0 and netloc.lower() == mirror['o_url'] and mirror['n_url'] != 'None':
|
|
u = u._replace(netloc=mirror['n_url'])
|
|
print('[easyapi] origin url: {}, use mirror url: {}'.format(url, u.geturl()))
|
|
if 'u_agent' in mirror:
|
|
user_agent = mirror['u_agent']
|
|
found = True
|
|
break
|
|
return found, u, user_agent
|
|
|
|
import urllib.request
|
|
import socket
|
|
# open(self, fullurl, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT)
|
|
origin_urllib_open = urllib.request.OpenerDirector.open
|
|
|
|
def wrap_open(obj, fullurl, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
|
|
"""
|
|
implement of lib urllib
|
|
Args:
|
|
**args: self, fullurl
|
|
**kwargs:
|
|
|
|
Returns:
|
|
|
|
"""
|
|
if isinstance(fullurl, str):
|
|
found, u, user_agent = replace_url(fullurl, Mirror.DOWN_MODEL)
|
|
if found:
|
|
url = u.geturl()
|
|
if user_agent is not None:
|
|
headers = {'User-Agent': user_agent}
|
|
url = urllib.request.Request(url, data=data, headers=headers)
|
|
|
|
return origin_urllib_open.__call__(obj, url, data, timeout)
|
|
else:
|
|
return origin_urllib_open.__call__(obj, fullurl, data, timeout)
|
|
|
|
else:
|
|
# url is urllib.request.Request
|
|
full_url = fullurl.get_full_url()
|
|
found, u, user_agent = replace_url(full_url, Mirror.DOWN_MODEL)
|
|
if found:
|
|
fullurl.full_url = u.geturl()
|
|
if user_agent is not None:
|
|
if fullurl.headers is not None:
|
|
fullurl.headers['User-Agent'] = user_agent
|
|
else:
|
|
fullurl.headers = {'User-Agent': user_agent}
|
|
|
|
return origin_urllib_open.__call__(obj, fullurl, data, timeout)
|
|
|
|
import requests
|
|
origin_request = requests.Session.request
|
|
|
|
def wrap_requests(*args, **kwargs):
|
|
"""
|
|
implement of lib requests
|
|
Args:
|
|
**args: self, method, url
|
|
**kwargs:
|
|
|
|
Returns:
|
|
|
|
"""
|
|
|
|
if 'url' in kwargs:
|
|
url = kwargs['url']
|
|
found, u, user_agent = replace_url(url, Mirror.DOWN_MODEL)
|
|
if found:
|
|
kwargs['url'] = u.geturl()
|
|
elif len(args) >= 3:
|
|
url = args[2]
|
|
found, u, user_agent = replace_url(url, Mirror.DOWN_MODEL)
|
|
if found:
|
|
new_updater = list(args)
|
|
new_updater[2] = u.geturl()
|
|
args = tuple(new_updater)
|
|
|
|
return origin_request.__call__(*args, **kwargs)
|
|
|
|
import aiohttp
|
|
origin_async_request = aiohttp.ClientSession._request
|
|
|
|
def wrap_aiohttp_requests(*args, **kwargs):
|
|
"""
|
|
implement of lib aiohttp
|
|
Args:
|
|
**args: self, method, str_or_url
|
|
**kwargs:
|
|
|
|
Returns:
|
|
|
|
"""
|
|
|
|
if 'str_or_url' in kwargs:
|
|
url = kwargs['str_or_url']
|
|
found, u, user_agent = replace_url(url, Mirror.DOWN_MODEL)
|
|
if found:
|
|
kwargs['str_or_url'] = u.geturl()
|
|
elif len(args) >= 3:
|
|
url = args[2]
|
|
found, u, user_agent = replace_url(url, Mirror.DOWN_MODEL)
|
|
if found:
|
|
new_updater = list(args)
|
|
new_updater[2] = u.geturl()
|
|
args = tuple(new_updater)
|
|
|
|
return origin_async_request.__call__(*args, **kwargs)
|
|
|
|
import git
|
|
origin_git_clone = git.Repo._clone
|
|
|
|
def wrap_git_clone(*args, **kwargs):
|
|
"""
|
|
implement of lib git clone
|
|
Args:
|
|
**args: cls, git, url
|
|
**kwargs:
|
|
|
|
Returns:
|
|
|
|
"""
|
|
|
|
if 'url' in kwargs:
|
|
url = kwargs['url']
|
|
found, u, user_agent = replace_url(url, Mirror.GIT_CLONE)
|
|
if found:
|
|
kwargs['url'] = u.geturl()
|
|
elif len(args) >= 3:
|
|
url = args[2]
|
|
found, u, user_agent = replace_url(url, Mirror.GIT_CLONE)
|
|
if found:
|
|
new_updater = list(args)
|
|
new_updater[2] = u.geturl()
|
|
args = tuple(new_updater)
|
|
|
|
return origin_git_clone.__call__(*args, **kwargs)
|
|
|
|
# urllib.request.urlopen = wrap_urlopen
|
|
urllib.request.OpenerDirector.open = wrap_open
|
|
requests.Session.request = wrap_requests
|
|
aiohttp.ClientSession._request = wrap_aiohttp_requests
|
|
git.Repo._clone = wrap_git_clone
|
|
|
|
# try:
|
|
# manager has been not loaded
|
|
# from ComfyUI-Manager.glob import manager_core
|
|
# wrap_manager_git_clone = manager_core.gitclone_install
|
|
#
|
|
# def wrap_manager_git_clone(files):
|
|
# urls = copy.deepcopy(files)
|
|
# if isinstance(urls, []|list|()):
|
|
# for i in range(len(urls)):
|
|
# url = urls[i]
|
|
# found, u, user_agent = replace_url(url, Mirror.GIT_CLONE)
|
|
# if found:
|
|
# urls[i]=u.geturl()
|
|
#
|
|
# return wrap_manager_git_clone.__call__(urls)
|
|
#
|
|
# manager_core.gitclone_install = wrap_manager_git_clone
|
|
# except Exception as e:
|
|
# print("[easyapi] fail to apply manager clone patch, error: {} ".format(e))
|
|
|
|
|
|
def init():
|
|
try:
|
|
replace_mirror_url()
|
|
except Exception as e:
|
|
print("[easyapi] fail to apply mirror url patch, error: {} ".format(e))
|