Lsposed_Modules_Updates_Tracker/defs/source.py

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