青龙面板教程(四)：线报监控系统开发

一.前言

青龙面板挂活动，那就少不了线报，尤其是JD系活动的环境变量。通常线报都是在TG上发布的，由于网络限制的原因，获取也很不方便。市面上也有TG线报系统，不过大多都是付费的，我们完全可以自己开发一个定制的线报系统，监控活动和自动更新环境变量执行任务，解放双手，无需时刻守护。

二.原理介绍

核心是TDLib，TG官方提供了TG数据库库（或简称为 TDLib），这是一种供第三方开发人员使用的工具，可以轻松构建快速、安全且功能丰富的 TG应用程序。

TDLib 负责所有网络实现细节、加密和本地数据存储，以便开发人员可以将更多时间用于设计响应式界面和精美动画。

TDLib 支持所有 TG功能，并使在任何平台上开发 TG应用程序变得轻而易举。它可以在 Android、iOS、Windows、macOS、Linux 和几乎任何其他系统上使用。该库是开源的，几乎与任何编程语言兼容。

简而言之，就是利用了TDLib开发一个TG客户端来接收消息，然后提取我们需要的内容。

三.系统开发

TDLib支持的编程语言很多，为了开发和部署方便，这里选择了Python语言，同时官方也提供了Python+TDLib环境的Docker容器，我们只需要安装好容器，编写我们业务逻辑代码就行。

1.安装环境

宝塔面板——终端，拉取官方python-telegram镜像并建立容器，容器默认命名为py-tg，最大内存为512MB，不映射端口和目录，无特殊要求，输入如下命令即可：

docker run -dit --name py-tg -m 512MB akhmetov/python-telegram:latest

 为了方便操作，我们采用宝塔面板自带的Docker模块进行接下来的操作，宝塔面板——Docker，可以看到刚创建的py-tg容器

至此，环境已搭建完毕。

2.业务代码

利用python-telegram的接口，我们可以实现TG客户端接收消息。在这里只实现自动更新环境变量的功能。

（1）在宝塔面板——Docker中打开py-tg容器的目录，可看到目录结构如下：

 点击进入app目录，里面有个examples文件夹，这是官方提供的例子我们不管，接下来在app目录下创建三个文件：ql.py、config.json和main.py

（2）先编写青龙面板的环境变量提取和导入的逻辑代码，利用到了青龙面板的OpenApi，新建文件，命名为ql.py，代码如下：

import re
import requests
from json import dumps as jsonDumps


class QL:
    def __init__(self, address: str, id: str, secret: str) -> None:
        """
        初始化
        """
        self.address = address
        self.id = id
        self.secret = secret
        self.valid = True
        self.login()

    def handleMessage(self, update: dict) -> None:
        """
        处理消息
        """
        try:
            text = update["message"]["content"]["text"]["text"]
        except:
            return
        envs = re.findall("export \w+=\"[^\"]+\"", text)
        if(len(envs) < 1):
            return
        self.log(f"监测到{len(envs)}个环境变量")
        targets = []
        for env in envs:
            name = env.split(" ", 1)[1].split("=", 1)[0]
            value = env.split(" ", 1)[1].split("=", 1)[1][1:-1]
            targets.append({"name": name, "value": value})
        self.addEnvs(targets)

    def log(self, content: str) -> None:
        """
        日志 这里只简单打印
        """
        print(content)

    def login(self) -> None:
        """
        登录
        """
        url = f"{self.address}/open/auth/token?client_id={self.id}&client_secret={self.secret}"
        try:
            rjson = requests.get(url).json()
            if(rjson['code'] == 200):
                self.auth = f"{rjson['data']['token_type']} {rjson['data']['token']}"
            else:
                self.log(f"登录失败：{rjson['message']}")
        except Exception as e:
            self.valid = False
            self.log(f"登录失败：{str(e)}")

    def addEnvs(self, envs: list) -> bool:
        """
        新建环境变量
        """
        url = f"{self.address}/open/envs"
        headers = {"Authorization": self.auth,
                   "content-type": "application/json"}
        try:
            rjson = requests.post(url, headers=headers,
                                  data=jsonDumps(envs)).json()
            if(rjson['code'] == 200):
                self.log(f"新建环境变量成功：{len(envs)}")
                return True
            else:
                self.log(f"新建环境变量失败：{rjson['message']}")
                return False
        except Exception as e:
            self.log(f"新建环境变量失败：{str(e)}")
            return False

（3）编写配置文件，命名为config.json，代码如下：

{
    "api_id": "xxxx",//my.telegram.org申请得到
    "api_hash": "xxxx",//my.telegram.org申请得到
    "phone": "xxxxx",//TG手机号
    "database_encryption_key": "changeme1234",//默认即可
    "proxy_type": {
        "@type": "proxyTypeSocks5"//代理类型，默认socks5
    },
    "proxy_port": "7891",//代理端口
    "proxy_server": "127.0.0.1",//代理地址
    "ql_address": "http://127.0.0.1:5700",//青龙面板地址
    "ql_id": "C9Wi0E6N9-wS",//青龙面板应用Client_ID
    "ql_secret": "id-ZWzSD_dqr43m6bHF_Y9FN"//青龙面板应用Client_Secret
}

（4）TG客户端，实现消息的接收，文件命名为main.py，代码如下：

from telegram.client import Telegram
from json import loads as jsonLoads
from ql import QL


def getConfig():
    """
    读取配置文件
    """
    file = open("config.json", mode="r")
    return jsonLoads(file.read())


if __name__ == "__main__":
    config = getConfig()
    tg = Telegram(
        api_id=config["api_id"],
        api_hash=config["api_hash"],
        phone=config["phone"],
        database_encryption_key=config["database_encryption_key"],
        proxy_server=config["proxy_server"],
        proxy_port=config["proxy_port"],
        proxy_type=config["proxy_type"],
    )
    ql = QL(config["ql_address"], config["ql_id"], config["ql_secret"])
    tg.login()
    tg.add_message_handler(ql.handleMessage)
    tg.idle()

至此系统开发已结束。

四.系统运行

宝塔面板——Docker，进入py-tg容器的终端，进入/app目录：

 安装python的requests模块，执行命令：

pip3 install requests

运行服务：

python3 main.py

首次运行需要验证身份，在TG客户端上会收到一个验证码，查看并输入即可完成

 当有新消息到达时，会自动识别出环境变量并导入青龙面板：

 这样启动关闭终端系统也会关闭，先Ctrl+c关闭系统，执行以下命令：

python3 main.py &

 这样可以关闭终端系统也能在后台运行了。

五.结语

以上只是个小例子，TDLib支持的编程语言很多，可以自己去官方网站查看相关文档，或者在上面代码的基础上增加更多功能。