diff --git a/easyapi/mirrorUrlApply.py b/easyapi/mirrorUrlApply.py new file mode 100644 index 0000000..c0a3616 --- /dev/null +++ b/easyapi/mirrorUrlApply.py @@ -0,0 +1,156 @@ +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", + }, +] + + +def get_custom_mirrors(): + settings = get_settings() + 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'] + return base_mirrors + + +def replace_mirror_url(): + from urllib.parse import urlparse + + def replace_url(url: str): + u = urlparse(url) + netloc = u.netloc + found = False + user_agent = None + for mirror in get_custom_mirrors(): + 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 + origin_urlopen = urllib.request.urlopen + + def wrap_urlopen(url, **kwargs): + """ + implement of lib urllib + Args: + url: + **kwargs: + + Returns: + + """ + if isinstance(url, str): + found, u, user_agent = replace_url(url) + if found: + url = u.geturl() + data = None + if user_agent is not None: + headers = {'User-Agent': user_agent} + if 'data' in kwargs: + data = kwargs['data'] + url = urllib.request.Request(url, data=data, headers=headers) + + return origin_urlopen.__call__(url, **kwargs) + else: + # url is urllib.request.Request + full_url = url.get_full_url() + found, u, user_agent = replace_url(full_url) + if found: + url.full_url = u.geturl() + if user_agent is not None: + if url.headers is not None: + url.headers['User-Agent'] = user_agent + else: + url.headers = {'User-Agent': user_agent} + + return origin_urlopen.__call__(url, **kwargs) + + 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) + if found: + kwargs['url'] = u.geturl() + elif len(args) >= 3: + url = args[2] + found, u, user_agent = replace_url(url) + 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) + if found: + kwargs['str_or_url'] = u.geturl() + elif len(args) >= 3: + url = args[2] + found, u, user_agent = replace_url(url) + if found: + new_updater = list(args) + new_updater[2] = u.geturl() + args = tuple(new_updater) + + return origin_async_request.__call__(*args, **kwargs) + + urllib.request.urlopen = wrap_urlopen + requests.Session.request = wrap_requests + aiohttp.ClientSession._request = wrap_aiohttp_requests + + +def init(): + try: + replace_mirror_url() + except Exception as e: + print("[easyapi] load mirror url replace patch fail, error: {} ".format(e)) diff --git a/easyapi/settings.py b/easyapi/settings.py new file mode 100644 index 0000000..24a0e0e --- /dev/null +++ b/easyapi/settings.py @@ -0,0 +1,60 @@ +import os +import json +import execution + +extension_folder = os.path.dirname(os.path.realpath(__file__)) +# configDataFilePath = os.path.join(extension_folder, 'config') + +# if not os.path.exists(configDataFilePath): +# os.mkdir(configDataFilePath) + + +def reset_history_size(max_size=execution.MAXIMUM_HISTORY_SIZE, isStart=False): + if not isStart: + set_settings("history_max_size", max_size) + # if not os.path.exists(configDataFilePath): + # os.mkdir(configDataFilePath) + # configFile = os.path.join(configDataFilePath, "easyapi.json") + # with open(configFile, 'w+', encoding="utf-8") as file: + # json.dump({"history_max_size": max_size}, file, indent=2) + # else: + # configFile = os.path.join(configDataFilePath, "easyapi.json") + # if not os.path.exists(configFile): + # with open(configFile, 'w+', encoding="utf-8") as file: + # json.dump({"history_max_size": max_size}, file, indent=2) + # else: + # with open(configFile, 'r+', encoding="UTF-8") as file: + # data = json.load(file) + # if not isStart: + # data['history_max_size'] = max_size + # + # with open(configFile, 'w+', encoding="UTF-8") as file: + # json.dump(data, file, indent=2) + + +def get_settings(file="config/easyapi.json"): + configFile = check_dir(file) + setting = {} + if not os.path.exists(configFile): + with open(configFile, 'w+', encoding="utf-8") as file: + json.dump({}, file, indent=2) + else: + with open(configFile, 'r+', encoding="utf-8") as file: + setting = json.load(file) + + return setting + + +def set_settings(key, value, file="config/easyapi.json"): + configFile = check_dir(file) + setting_json = get_settings(file=file) + setting_json[key] = value + with open(configFile, 'w+', encoding="utf-8") as file: + json.dump(setting_json, file, indent=2) + + +def check_dir(filePath): + configDataFilePath = os.path.join(extension_folder, os.path.dirname(filePath)) + if not os.path.exists(configDataFilePath): + os.mkdir(configDataFilePath) + return os.path.join(configDataFilePath, os.path.basename(filePath))