Lsposed_Modules_Updates_Tracker/defs/subs.py

import traceback
from asyncio import sleep
from random import uniform

from pyrogram.errors import FloodWait, ButtonUrlInvalid, BadRequest
from pyrogram.types import InlineKeyboardMarkup, InlineKeyboardButton

from ci import app, sqlite, me
from defs.source import from_list_to_name
from defs.utils import Module

subs_msg = """
<b>{}（{}）有新的更新！</b>

<b>版本：</b><code>{}</code>
<b>更新时间：</b><code>{}</code>
"""
subs_list_msg = """
<b>您订阅了：</b>{}
"""
subs_list_no_msg = """
<b>您订阅了个寂寞！</b>
"""


def gen_subs_button(data: Module, link: str) -> InlineKeyboardMarkup:
    data_ = [[InlineKeyboardButton("详情", url=link),
              InlineKeyboardButton("退订",
                                   url=f"https://t.me/{me.username}?start=un-{data.name.replace('.', '_')}"), ]]
    return InlineKeyboardMarkup(data_)


def gen_back_button() -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup([[InlineKeyboardButton("返回", callback_data="help"), ]])


def gen_subs_msg(cid: int) -> str:
    data_ = []
    for key, value in sqlite.items():
        if key == "update_time":
            continue
        data = value.get("subscribes", [])
        if cid in data:
            data_.append(key)
    return subs_list_msg.format(from_list_to_name(data_)) if data_ else subs_list_no_msg


async def send_subs_msg(cid: int, data: Module, link: str):
    return await app.send_message(cid,
                                  subs_msg.format(data.name, data.description,
                                                  data.latestRelease, data.updatedAt),
                                  reply_markup=gen_subs_button(data, link))


async def send_to_subscribes(data: Module):
    users = sqlite.get(data.name, {}).get("subscribes", [])
    link = sqlite.get(data.name, {}).get("msg_link", "https://t.me/lsposed_Modules_Updates_Tracker")
    for i in users:
        try:
            await send_subs_msg(i, data, link)
        except FloodWait as e:
            print(f"Send subscribes msg flood - Sleep for {e.x} second(s)")
            await sleep(uniform(0.5, 1.0))
            await send_subs_msg(i, data, link)
        except ButtonUrlInvalid:
            print(f"Send button error")
            await app.send_message(i, subs_msg.format(data.name, data.description,
                                                      data.latestRelease, data.updatedAt), )
        except BadRequest:
            users.remove(i)
        except Exception as e:
            traceback.print_exc()
    sqlite[data.name]["subscribes"] = users


def add_to_subs(cid: int, data: Module) -> bool:
    users = sqlite.get(data.name, {}).get("subscribes", [])
    if cid not in users:
        users.append(cid)
        data_ = sqlite.get(data.name, {"subscribes": []})
        data_["subscribes"] = users
        sqlite[data.name] = data_
        return True
    return False


def remove_from_subs(cid: int, data: Module) -> bool:
    users = sqlite.get(data.name, {}).get("subscribes", [])
    if cid in users:
        users.remove(cid)
        data_ = sqlite.get(data.name, {"subscribes": []})
        data_["subscribes"] = users
        sqlite[data.name] = data_
        return True
    return False