Lsposed_Modules_Updates_Tracker/ci.py

import json
from configparser import RawConfigParser
from os import sep, mkdir
from os.path import exists

from pyrogram import Client
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from httpx import get
from sqlitedict import SqliteDict

if not exists("data"):
    mkdir("data")
sqlite = SqliteDict(f"data{sep}data.sqlite", encode=json.dumps, decode=json.loads, autocommit=True)
# data.sqlite 结构如下：
# {
#   "module 名称": {
#       "msg_link": str,
#       "subscribes": List[订阅id: int],
#   },
#   "update_time": str,
# }
# 读取配置文件
config = RawConfigParser()
config.read("config.ini")
bot_token: str = ""
admin_id: int = 0
channel_id: int = 0
max_update_file: int = 10
api_id: int = 0
api_hash: str = ""
api_id = config.getint("pyrogram", "api_id", fallback=api_id)
api_hash = config.get("pyrogram", "api_hash", fallback=api_hash)
bot_token = config.get("basic", "bot_token", fallback=bot_token)
admin_id = config.getint("basic", "admin", fallback=admin_id)
channel_id = config.getint("basic", "channel_id", fallback=channel_id)
max_update_file = config.getint("basic", "max_update_file", fallback=max_update_file)
""" Init httpx client """
# 使用自定义 UA
headers = {
    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.132 Safari/537.36"
}


# 自定义类型
class Bot:
    def __init__(self, data: dict):
        self.uid = data["id"]
        self.username = data["username"]
        self.name = data["first_name"]


me = Bot(get(f"https://api.telegram.org/bot{bot_token}/getme").json()["result"])
# 初始化客户端
scheduler = AsyncIOScheduler()
if not scheduler.running:
    scheduler.configure(timezone="Asia/ShangHai")
    scheduler.start()
app = Client("bot", api_id=api_id, api_hash=api_hash, bot_token=bot_token, plugins=dict(root="plugins"))