Sync OpCore Simplify modules

This commit is contained in:
Hoang Hong Quan
2025-01-14 15:53:34 +07:00
parent 2391ab85e0
commit 2dfcc0edc8
2 changed files with 58 additions and 27 deletions

View File

@@ -5,51 +5,66 @@ import random
class Github:
def __init__(self):
self.utils = utils.Utils()
self.headers = {
"User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
self.fetcher = resource_fetcher.ResourceFetcher(self.headers)
self.fetcher = resource_fetcher.ResourceFetcher()
def get_latest_release(self, owner, repo):
url = "https://github.com/{}/{}/releases".format(owner, repo)
response = self.fetcher.fetch_and_parse_content(url)
body = ""
tag_name = None
assets = []
for line in response.splitlines():
if "<a" in line and "href=\"" in line and "/releases/tag/" in line and not tag_name:
tag_name = line.split("/releases/tag/")[1].split("\"")[0]
elif "<div" in line and "body-content" in line:
body = response.split(line.split(">", 1)[0], 1)[1].split("</div>", 1)[0][1:]
break
if not response:
raise ValueError("Failed to fetch release information from GitHub.")
tag_name = self._extract_tag_name(response)
body = self._extract_body_content(response)
release_tag_url = "https://github.com/{}/{}/releases/expanded_assets/{}".format(owner, repo, tag_name)
response = self.fetcher.fetch_and_parse_content(release_tag_url)
if not response:
raise ValueError("Failed to fetch expanded assets information from GitHub.")
assets = self._extract_assets(response)
return {
"body": body,
"assets": assets
}
def _extract_tag_name(self, response):
for line in response.splitlines():
if "<a" in line and "href=\"" in line and "/releases/tag/" in line:
return line.split("/releases/tag/")[1].split("\"")[0]
return None
def _extract_body_content(self, response):
for line in response.splitlines():
if "<div" in line and "body-content" in line:
return response.split(line.split(">", 1)[0], 1)[1].split("</div>", 1)[0][1:]
return ""
def _extract_assets(self, response):
assets = []
for line in response.splitlines():
if "<a" in line and "href=\"" in line and "/releases/download" in line:
download_link = line.split("href=\"", 1)[1].split("\"", 1)[0]
if "tlwm" in download_link or ("tlwm" not in download_link and "DEBUG" not in download_link.upper()):
asset_data = response.split(line)[1].split("</div>", 2)[1]
try:
asset_id = "".join(char for char in asset_data.split("datetime=\"")[-1].split("\"")[0][::-1] if char.isdigit())[:9]
except:
asset_id = "".join(random.choices('0123456789', k=9))
asset_id = self._generate_asset_id(asset_data)
assets.append({
"product_name": self.extract_asset_name(download_link.split("/")[-1]),
"id": int(asset_id),
"url": "https://github.com" + download_link
})
return {
"body": body,
"assets": assets
}
return assets
def _generate_asset_id(self, asset_data):
try:
return "".join(char for char in asset_data.split("datetime=\"")[-1].split("\"")[0][::-1] if char.isdigit())[:9]
except:
return "".join(random.choices('0123456789', k=9))
def extract_asset_name(self, file_name):
end_idx = len(file_name)

View File

@@ -1,3 +1,4 @@
import ssl
import os
import json
import plistlib
@@ -11,9 +12,24 @@ else:
from urllib2 import urlopen, Request
class ResourceFetcher:
def __init__(self, headers={}):
self.request_headers = headers
def __init__(self, headers=None):
self.request_headers = headers or {
"User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36'
}
self.buffer_size = 16 * 1024
self.ssl_context = self.create_ssl_context()
def create_ssl_context(self):
try:
cafile = ssl.get_default_verify_paths().openssl_cafile
if not os.path.exists(cafile):
import certifi
cafile = certifi.where()
ssl_context = ssl.create_default_context(cafile=cafile)
except Exception as e:
print("SSL Context Creation Error: {}".format(e))
ssl_context = ssl._create_unverified_context()
return ssl_context
def is_connected(self, timeout=5):
socket.create_connection(("github.com", 443), timeout=timeout)
@@ -22,7 +38,7 @@ class ResourceFetcher:
self.is_connected()
try:
return urlopen(Request(resource_url, headers=self.request_headers))
return urlopen(Request(resource_url, headers=self.request_headers), context=self.ssl_context)
except Exception as e:
pass