【青龙脚本】Telegram 机器人自动签到方案

分享一套基于青龙脚本的 Telegram 机器人自动签到方案，实现多 TG 机器人定时签到，解放双手，适合自动化玩家。

环境要求

1. Python环境

Python 3.7+
telethon 库

2. Telegram API凭证

API_ID: 从 https://my.telegram.org 获取
API_HASH: 从 https://my.telegram.org 获取

快速开始

第一步：安装依赖

# 进入青龙容器
docker exec -it qinglong bash 
# 安装Telethon 
pip3 install telethon

第二步：上传脚本

# 创建目录
mkdir -p /ql/data/scripts/telegram-pure  
# 上传文件 telegram_checkin_pure_user.py 到该目录

注意目录我是新版青龙 v2.20.1 脚本目录是/ql/data/scripts/

telegram_checkin_prune_user.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Telegram 用户账号批量机器人签到脚本

功能：
1. 支持直接发送命令签到，例如 /sign、/qd、/signin。
2. 支持发送 /start 后，按按钮文本查找并点击签到按钮。
3. 支持根据 successKeywords 判断签到是否成功。
4. 支持 TG_SESSION 字符串会话或本地 telegram_session.session 文件会话。
5. 首次终端运行时，即使没有 BOT_CONFIG_*，也可以正常登录并保存会话文件。

依赖：
    pip install telethon

必要环境变量：
    TG_API_ID
    TG_API_HASH

可选环境变量：
    TG_SESSION

机器人配置示例：
    BOT_CONFIG_1={"name":"按钮签到","botUsername":"checkin_bot_a","checkinCommand":"/start","buttonTexts":["签到"],"successKeywords":["签到成功","已经签到"]}
    BOT_CONFIG_2={"name":"命令签到","botUsername":"checkin_bot_b","checkinCommand":"/sign","successKeywords":["签到成功","打卡成功"]}
"""

import asyncio
import json
import logging
import os
from typing import Any, Dict, List, Optional, Tuple

from telethon import TelegramClient
from telethon.errors import (
    FloodWaitError,
    PeerFloodError,
    UserBannedInChannelError,
)
from telethon.sessions import StringSession


# ==================== 日志配置 ====================

logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] [%(levelname)s] %(message)s",
    datefmt="%Y/%m/%d %H:%M:%S",
)
logger = logging.getLogger(__name__)

SUCCESS_LEVEL = 25
logging.addLevelName(SUCCESS_LEVEL, "SUCCESS")


def _success(self: logging.Logger, message: str, *args: Any, **kwargs: Any) -> None:
    if self.isEnabledFor(SUCCESS_LEVEL):
        self._log(SUCCESS_LEVEL, message, args, **kwargs)


# 给 logger 增加 logger.success(...) 方法。
setattr(logging.Logger, "success", _success)


# ==================== 配置 ====================

class Config:
    API_ID = int(os.getenv("TG_API_ID", "0") or "0")
    API_HASH = os.getenv("TG_API_HASH", "").strip()
    SESSION_STRING = os.getenv("TG_SESSION", "").strip()
    SESSION_FILE = os.getenv("TG_SESSION_FILE", "telegram_session").strip()

    BOTS: List[Dict[str, Any]] = []

    RETRY_TIMES = int(os.getenv("TG_RETRY_TIMES", "3") or "3")
    WAIT_TIME = int(os.getenv("TG_WAIT_TIME", "2") or "2")
    TIMEOUT = int(os.getenv("TG_TIMEOUT", "40") or "40")

    @classmethod
    def load_from_env(cls) -> bool:
        """
        加载 BOT_CONFIG_1 至 BOT_CONFIG_50。

        返回：
            True：至少加载到一个有效机器人配置。
            False：没有读取到有效机器人配置。

        没有配置时不抛异常，便于首次终端登录并生成会话文件。
        """
        cls.BOTS = []

        for index in range(1, 51):
            env_name = "BOT_CONFIG_{}".format(index)
            config_text = os.getenv(env_name, "").strip()

            if not config_text:
                continue

            try:
                raw = json.loads(config_text)
                if not isinstance(raw, dict):
                    raise ValueError("配置必须是 JSON 对象")

                bot = {
                    "name": raw.get("name", "Bot {}".format(index)),
                    "username": str(raw.get("botUsername", "") or "").strip().lstrip("@"),
                    "command": str(raw.get("checkinCommand", "/start") or "/start").strip(),
                    "keywords": raw.get("successKeywords", []) or [],
                    "button_texts": raw.get(
                        "buttonTexts",
                        ["签到", "立即签到", "打卡", "领取", "Check in", "Claim"],
                    ) or [],
                    "enabled": raw.get("enabled", True),
                }

                if not isinstance(bot["keywords"], list):
                    raise ValueError("successKeywords 必须是数组")

                if not isinstance(bot["button_texts"], list):
                    raise ValueError("buttonTexts 必须是数组")

                bot["keywords"] = [
                    str(item).strip()
                    for item in bot["keywords"]
                    if str(item).strip()
                ]
                bot["button_texts"] = [
                    str(item).strip()
                    for item in bot["button_texts"]
                    if str(item).strip()
                ]

                cls.BOTS.append(bot)

            except Exception as exc:
                logger.error("解析 %s 失败: %s", env_name, exc)

        if not cls.BOTS:
            logger.warning(
                "未加载到有效的 BOT_CONFIG_*，本次仅完成 Telegram 登录，不执行签到"
            )
            return False

        logger.info("加载了 %s 个机器人配置", len(cls.BOTS))
        logger.info(
            "机器人列表: %s",
            ", ".join(
                "{}(@{})".format(bot["name"], bot["username"] or "未配置")
                for bot in cls.BOTS
            ),
        )
        return True


# ==================== Telegram 客户端 ====================

class TelegramUserClient:
    def __init__(self) -> None:
        self.client: Optional[TelegramClient] = None

    async def initialize(self) -> None:
        if not Config.API_ID or not Config.API_HASH:
            raise ValueError("缺少 TG_API_ID 或 TG_API_HASH")

        if Config.SESSION_STRING:
            logger.info("使用会话字符串模式（TG_SESSION）")
            self.client = TelegramClient(
                StringSession(Config.SESSION_STRING),
                Config.API_ID,
                Config.API_HASH,
            )
        else:
            logger.info("使用会话文件模式（%s.session）", Config.SESSION_FILE)
            self.client = TelegramClient(
                Config.SESSION_FILE,
                Config.API_ID,
                Config.API_HASH,
            )

        # 首次执行时，Telethon 会在终端提示手机号、验证码和二步验证密码。
        await self.client.start()

        me = await self.client.get_me()
        if me is None:
            raise RuntimeError("Telegram 登录成功后未能读取当前用户信息")

        display_name = " ".join(
            part for part in [getattr(me, "first_name", None), getattr(me, "last_name", None)]
            if part
        ).strip()
        username = getattr(me, "username", None)

        if username:
            logger.info("✅ 用户登录成功: %s (@%s)", display_name or me.id, username)
        else:
            logger.info("✅ 用户登录成功: %s", display_name or me.id)

    async def close(self) -> None:
        if self.client is not None:
            await self.client.disconnect()
            logger.info("客户端已关闭")

    async def click_button_by_text(
        self,
        message: Any,
        targets: List[str],
    ) -> Tuple[bool, Optional[str]]:
        """
        点击包含指定文本的按钮。

        返回：
            (是否点击成功, callback 弹窗或点击结果文本)
        """
        if not message or not getattr(message, "buttons", None):
            return False, None

        normalized_targets = [
            target.lower().strip()
            for target in targets
            if target and target.strip()
        ]

        if not normalized_targets:
            return False, None

        for row_index, row in enumerate(message.buttons):
            for column_index, button in enumerate(row):
                text = str(getattr(button, "text", "") or "").strip()
                if not text:
                    continue

                lowered = text.lower()
                if not any(target in lowered for target in normalized_targets):
                    continue

                logger.info(
                    "准备点击按钮: '%s' (row=%s, col=%s)",
                    text,
                    row_index,
                    column_index,
                )

                result = await message.click(row_index, column_index)
                logger.info("✅ 已点击按钮: '%s'", text)

                callback_message = getattr(result, "message", None)
                if callback_message:
                    return True, str(callback_message)

                raw_text = getattr(result, "raw_text", None)
                if raw_text:
                    return True, str(raw_text)

                return True, None

        return False, None


# ==================== 签到服务 ====================

class CheckinService:
    def __init__(self, telegram: TelegramUserClient) -> None:
        self.telegram = telegram

    @staticmethod
    def _contains_keyword(text: str, keywords: List[str]) -> bool:
        """保持与常见中文签到配置兼容，使用包含匹配。"""
        return any(keyword in text for keyword in keywords)

    async def perform_checkin(
        self,
        bot: Dict[str, Any],
        attempt: int = 1,
    ) -> Dict[str, Any]:
        bot_name = bot["name"]
        bot_username = bot["username"]
        command = bot["command"]
        keywords = bot.get("keywords", [])
        button_texts = bot.get(
            "button_texts",
            ["签到", "立即签到", "打卡", "领取"],
        )

        if not bot_username:
            return {"success": False, "message": "botUsername 为空"}

        if not keywords:
            return {
                "success": False,
                "message": "successKeywords 为空，请配置签到成功判定关键词",
            }

        if self.telegram.client is None:
            return {"success": False, "message": "Telegram 客户端尚未初始化"}

        logger.info(
            "[%s] 第 %s/%s 次尝试",
            bot_name,
            attempt,
            Config.RETRY_TIMES,
        )

        try:
            entity = await self.telegram.client.get_entity(bot_username)

            async with self.telegram.client.conversation(
                entity,
                timeout=Config.TIMEOUT,
            ) as conversation:
                logger.info("[%s] 发送命令: %s", bot_name, command)

                try:
                    await conversation.send_message(command)
                except FloodWaitError as exc:
                    logger.warning(
                        "[%s] 触发 FloodWait，等待 %s 秒后重发",
                        bot_name,
                        exc.seconds,
                    )
                    await asyncio.sleep(exc.seconds)
                    await conversation.send_message(command)

                first_message = await conversation.get_response()
                first_text = str(first_message.raw_text or "").strip()
                logger.info("[%s] 回复1: %s", bot_name, first_text[:300])

                # 直接命令签到：第一条回复已经包含成功关键词。
                if self._contains_keyword(first_text, keywords):
                    logger.success("[%s] ✅ 签到成功（文本命中）", bot_name)
                    return {"success": True, "message": first_text}

                # /start 按钮签到。
                if getattr(first_message, "buttons", None):
                    clicked, callback_text = await self.telegram.click_button_by_text(
                        first_message,
                        button_texts,
                    )

                    if not clicked:
                        return {
                            "success": False,
                            "message": "检测到按钮，但没有匹配到 buttonTexts，请调整配置",
                        }

                    if callback_text:
                        logger.info(
                            "[%s] 回调弹窗/点击结果: %s",
                            bot_name,
                            callback_text,
                        )

                        if self._contains_keyword(callback_text, keywords):
                            logger.success(
                                "[%s] ✅ 签到成功（按钮回调命中）",
                                bot_name,
                            )
                            return {"success": True, "message": callback_text}

                        return {"success": False, "message": callback_text}

                    # 有些机器人点击按钮后不会弹窗，而是再发送一条消息。
                    second_message = await conversation.get_response()
                    second_text = str(second_message.raw_text or "").strip()
                    logger.info("[%s] 回复2: %s", bot_name, second_text[:300])

                    if self._contains_keyword(second_text, keywords):
                        logger.success(
                            "[%s] ✅ 签到成功（按钮后消息命中）",
                            bot_name,
                        )
                        return {"success": True, "message": second_text}

                    return {
                        "success": False,
                        "message": second_text or "按钮点击后未命中 successKeywords",
                    }

                return {
                    "success": False,
                    "message": first_text or "未收到可判定的签到结果",
                }

        except (PeerFloodError, UserBannedInChannelError) as exc:
            logger.error("[%s] 账号受限或被禁言: %s", bot_name, exc)
            return {"success": False, "message": str(exc)}

        except Exception as exc:
            logger.error("[%s] ❌ 第 %s 次失败: %s", bot_name, attempt, exc)

            if attempt < Config.RETRY_TIMES:
                await asyncio.sleep(2)
                return await self.perform_checkin(bot, attempt + 1)

            return {"success": False, "message": str(exc)}

    async def execute(self) -> Dict[str, Any]:
        logger.info("=" * 50)
        logger.info(" 开始批量签到任务")
        logger.info("共 %s 个机器人", len(Config.BOTS))
        logger.info("=" * 50)

        results: List[Dict[str, Any]] = []

        for bot in Config.BOTS:
            if not bot.get("enabled", True):
                logger.info("跳过禁用机器人: %s", bot.get("name"))
                continue

            result = await self.perform_checkin(bot)
            results.append({"bot": bot.get("name"), **result})

            if Config.WAIT_TIME > 0:
                await asyncio.sleep(Config.WAIT_TIME)

        self.print_summary(results)

        # 保持原脚本语义：只要至少一个机器人成功，任务整体返回成功。
        return {
            "success": any(item["success"] for item in results),
            "results": results,
        }

    @staticmethod
    def print_summary(results: List[Dict[str, Any]]) -> None:
        print("\n" + "=" * 50)
        print(" 签到结果汇总")
        print("=" * 50)

        success_count = 0
        failure_count = 0

        for result in results:
            if result["success"]:
                success_count += 1
                print("✅ 成功 {}".format(result["bot"]))
            else:
                failure_count += 1
                print("❌ 失败 {}".format(result["bot"]))
                print("     原因: {}".format(result.get("message", "未知原因")))

        print("=" * 50)
        print(
            "总计: {} | 成功: {} | 失败: {}".format(
                len(results),
                success_count,
                failure_count,
            )
        )
        print("=" * 50)

        if success_count > 0:
            logger.success("批量签到任务完成！")
        else:
            logger.error("❌ 所有签到都失败了")


# ==================== 主程序 ====================

async def main() -> bool:
    telegram = TelegramUserClient()

    try:
        # 必须先初始化 Telegram。
        # 这样首次在终端执行时，即使没有 BOT_CONFIG_*，也可以完成登录并保存 session。
        await telegram.initialize()

        # 登录完成后再读取签到机器人配置。
        has_bot_config = Config.load_from_env()

        if not has_bot_config:
            logger.info("✅ Telegram 登录验证完成，会话文件已保存")
            logger.info("未配置签到机器人，本次不执行签到，程序正常结束")
            return True

        service = CheckinService(telegram)
        result = await service.execute()
        return bool(result.get("success"))

    finally:
        await telegram.close()


if __name__ == "__main__":
    try:
        succeeded = asyncio.run(main())
        raise SystemExit(0 if succeeded else 1)

    except KeyboardInterrupt:
        logger.info("用户中断程序")
        raise SystemExit(0)

    except Exception as exc:
        logger.error("运行错误: %s", exc)
        raise SystemExit(1)

第三步：配置环境变量

在青龙面板 → 环境变量 → 添加：

# 必需配置
TG_API_ID = 12345678          # 你的API ID
TG_API_HASH = abcdef123456    # 你的API Hash

# 机器人配置（示例）

# 先发送 /start 然后点击 签到按钮 当检测到返回successKeywords关键字即判定为成功签到
BOT_CONFIG_1 = {"name":"需要点击按钮签到","botUsername":"checkin_bot_a","checkinCommand":"/start","buttonTexts":["签到"],"successKeywords":["签到成功"]}

# 直接发送 /sigin 完成签到 检测到返回successKeywords关键字打卡成功即判定为签到成功
BOT_CONFIG_2 = {"name":"普通签到","botUsername":"checkin_bot_b","checkinCommand":"/sigin","successKeywords":["打卡成功"]}

BOT_CONFIG_ 详解
{ "name": "名称随意起", "botUsername": "tg机器人的用户名", "checkinCommand": "签到命令", "buttonTexts": [ "签到" ], "successKeywords": [ "签到成功" ] }
buttonTexts 需要点击按钮的文本如果不需要点击按钮签到可不需要此项配置
successKeywords 签到成功返回的关键字

首次运行流程：

终端会显示登录链接
点击链接或复制到浏览器
登录Telegram账号
脚本自动保存会话到 telegram_session.session

# 首次运行（登录）

# 进入脚本目录
cd /ql/data/scripts/telegram-pure

# 首次运行会提示登录 需要输入手机号 和 验证码
# ID和HASH 替换为自己的 虽然上面青龙环境已经配置了这里在终端执行获取不到青龙的环境变量 需要临时指定下
export TG_API_ID=123456
export TG_API_HASH=abcdefg

python3 telegram_checkin_pure_user.py

完整首次登录验证后即可在面板添加定时任务

第五步：添加定时任务

在青龙面板 → 定时任务 → 添加：

名称: Telegram签到
命令: task telegram-pure/telegram_checkin_pure_user.py
定时规则: 30 08 * * * (每天8点半执行)

执行日志示例

## 开始执行... 2026-02-04 08:40:02

[2026/02/04 08:40:04] [INFO] 加载了 3 个机器人配置
[2026/02/04 08:40:04] [INFO] 机器人列表: xxx签到(@xxx), xxx(@xxx), xxx(@xxx)
[2026/02/04 08:40:04] [INFO] 使用会话文件模式（SESSION_FILE）
[2026/02/04 08:40:04] [INFO] Connecting to xxxxxx:443/TcpFull...
[2026/02/04 08:40:04] [INFO] Connection to xxxxxx:443/TcpFull complete!
[2026/02/04 08:40:08] [INFO] ✅ 用户登录成功: xxxx
[2026/02/04 08:40:08] [INFO] ==================================================
[2026/02/04 08:40:08] [INFO] 🚀 开始批量签到任务
[2026/02/04 08:40:08] [INFO] 共 3 个机器人
[2026/02/04 08:40:08] [INFO] ==================================================
[2026/02/04 08:40:08] [INFO] [xxx签到] 第 1/3 次尝试
[2026/02/04 08:40:08] [INFO] [xxx签到] 发送命令: /sign
[2026/02/04 08:40:08] [INFO] [xxx签到] 回复1: 您是今天第 642 位签到的
获得了 2 个xx
已签到 19 天
[2026/02/04 08:40:08] [SUCCESS] [xxx签到] ✅ 签到成功（文本命中）
[2026/02/04 08:40:10] [INFO] [xxx] 第 1/3 次尝试
[2026/02/04 08:40:11] [INFO] [xxx] 发送命令: /qd
[2026/02/04 08:40:13] [INFO] [xxx] 回复1: 签到成功！

此次签到获得积分:6

要试试轰炸吗？您当前可以轰炸43次
您可以尝试发送 /hz
您也可以尝试 /yshz

您也可以尝试 对机器人发送“任务”或点击/rw  ，获取更多积分
[2026/02/04 08:40:13] [SUCCESS] [xxx] ✅ 签到成功（文本命中）
[2026/02/04 08:40:15] [INFO] [xxx] 第 1/3 次尝试
[2026/02/04 08:40:15] [INFO] [xxx] 发送命令: /start
[2026/02/04 08:40:16] [INFO] [xxx] 回复1: ✨ 只有你想见我的时候我们的相遇才有意义

🍉你好鸭 xxx 请选择功能👇
[2026/02/04 08:40:16] [INFO] 准备点击按钮: '🎯 签到' (row=1, col=0)
[2026/02/04 08:40:16] [INFO] ✅ 已点击按钮: '🎯 签到'
[2026/02/04 08:40:16] [INFO] [xxx] 回调弹窗/结果: ⭕ 您今天已经签到过了！签到是无聊的活动哦。
[2026/02/04 08:40:16] [SUCCESS] [xxx] ✅ 签到成功（按钮回调命中）

==================================================
📊 签到结果汇总
==================================================
✅ 成功 xxx签到
✅ 成功 xxx
✅ 成功 xxx
==================================================
总计: 3 | 成功: 3 | 失败: 0
==================================================
[2026/02/04 08:40:18] [SUCCESS] 🎉 批量签到任务完成！
[2026/02/04 08:40:18] [INFO] Disconnecting from xxxxxx:443/TcpFull...
[2026/02/04 08:40:18] [INFO] Disconnection from xxxxxx:443/TcpFull complete!
[2026/02/04 08:40:19] [INFO] 客户端已关闭

## 执行结束... 2026-02-04 08:40:19  耗时 17 秒

重要注意事项

1. 首次运行必须登录

第一次运行会提示输入手机号和验证码
登录成功后会自动保存会话
后续运行无需再次登录

2. 机器人必须与用户私聊过

用户账号发送消息给机器人
机器人回复会发送到用户的私聊
脚本会监听私聊消息验证签到结果
必须确保已与每个机器人私聊过

3. 频率限制

Telegram有发送频率限制
建议设置 WAIT_TIME = 2 或更高
避免短时间内发送过多消息

4. 会话文件安全

telegram_session.session 包含登录凭证
不要分享该文件
定期清理或更新

5. JSON格式要求

使用双引号 "
不要有多余的逗号
可以使用在线JSON验证工具检查

WXY

WXY

【青龙脚本】Telegram 机器人自动签到方案

环境要求

1. Python环境

2. Telegram API凭证

快速开始

第一步：安装依赖

第二步：上传脚本

第三步：配置环境变量

第五步：添加定时任务

执行日志示例

重要注意事项

1. 首次运行必须登录

2. 机器人必须与用户私聊过

3. 频率限制

4. 会话文件安全

5. JSON格式要求