110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
from os import sep
|
||
from os.path import exists
|
||
from shutil import copyfile
|
||
from typing import List, Optional
|
||
|
||
from httpx import AsyncClient
|
||
|
||
from ci import sqlite, headers, max_update_file
|
||
from json import load
|
||
from defs.format_time import now_time
|
||
from defs.utils import Module
|
||
|
||
new_modules: List[Module] = []
|
||
new_modules_index: dict = {}
|
||
old_modules: List[Module] = []
|
||
old_modules_index: dict = {}
|
||
if exists(f"data{sep}modules.json"):
|
||
with open(f"data{sep}modules.json", "r", encoding="utf-8") as file:
|
||
temp_data = load(file)
|
||
new_modules.extend(Module(temp) for temp in temp_data)
|
||
if exists(f"data{sep}old_modules.json"):
|
||
with open(f"data{sep}old_modules.json", "r", encoding="utf-8") as file:
|
||
temp_data = load(file)
|
||
old_modules.extend(Module(temp) for temp in temp_data)
|
||
new_modules_index = {i.name: i for i in new_modules}
|
||
old_modules_index = {i.name: i.latestRelease for i in old_modules}
|
||
|
||
|
||
async def update_data() -> None:
|
||
global new_modules, old_modules, new_modules_index
|
||
if exists(f"data{sep}modules.json"):
|
||
copyfile(f"data{sep}modules.json", f"data{sep}old_modules.json")
|
||
async with AsyncClient(timeout=10.0, headers=headers, follow_redirects=True) as client:
|
||
data = await client.get("https://modules.lsposed.org/modules.json")
|
||
modules_data = data.json()
|
||
with open(f"data{sep}modules.json", "w", encoding="utf-8") as f:
|
||
f.write(data.text)
|
||
old_modules = new_modules
|
||
new_modules = []
|
||
for i in modules_data:
|
||
try:
|
||
new_modules.append(Module(i))
|
||
except Exception as e:
|
||
print(e)
|
||
new_modules_index.clear()
|
||
for i in new_modules:
|
||
new_modules_index[i.name] = i
|
||
sqlite["update_time"] = now_time()
|
||
|
||
|
||
def compare() -> List[Module]:
|
||
global old_modules_index
|
||
old_modules_index.clear()
|
||
for i in old_modules:
|
||
old_modules_index[i.name] = i.latestRelease
|
||
data = [i for i in new_modules if i.latestRelease != old_modules_index.get(i.name, "")]
|
||
return [] if len(data) > max_update_file else data
|
||
|
||
|
||
async def download(url: str, name: str, pack_name: str) -> (str, str):
|
||
async with AsyncClient(timeout=10.0, headers=headers, follow_redirects=True) as client:
|
||
content = await client.get(url)
|
||
if content.status_code == 404:
|
||
content = (await client.get(f"https://modules.lsposed.org/module/{pack_name}.json")).json()
|
||
data = Module(content)
|
||
url = None
|
||
if data.releases and data.releases[0].releaseAssets:
|
||
# find arm64 or universal first
|
||
asset = next(
|
||
(i for i in data.releases[0].releaseAssets if "arm64" in i.name or "universal" in i.name),
|
||
data.releases[0].releaseAssets[0],
|
||
)
|
||
url = asset.url
|
||
mime = asset.name.split(".")[-1:][0]
|
||
name = data.name.replace('.', '_') + "-" + data.latestRelease + "." + mime
|
||
if not url:
|
||
raise FileNotFoundError
|
||
content = await client.get(url)
|
||
if content.status_code == 404:
|
||
raise FileNotFoundError
|
||
content = content.content
|
||
with open(f"data{sep}{name}", 'wb') as f:
|
||
f.write(content)
|
||
return f"data{sep}{name}", url
|
||
|
||
|
||
def from_name_to_module(name: str) -> Optional[Module]:
|
||
return new_modules_index.get(name, None)
|
||
|
||
|
||
def from_list_to_name(data: List) -> str:
|
||
data_ = ""
|
||
for i in data:
|
||
name = new_modules_index.get(i, None)
|
||
if isinstance(name, Module):
|
||
data_ += f"\n{name.name}（{name.description}）"
|
||
return data_
|
||
|
||
|
||
def from_keyword_to_module(keyword: str) -> Optional[Module]:
|
||
for value in new_modules:
|
||
data = value.name + value.description + value.url + value.homepageUrl + value.summary + \
|
||
value.sourceUrl
|
||
if value.scope:
|
||
for i in value.scope:
|
||
data += i
|
||
if keyword in data:
|
||
return value
|
||
return None
|