【青龙脚本】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 纯用户账号批量签到脚本
- 兼容直接命令签到：/sign /qd /signin ...
- 兼容 /start 后点击按钮“签到”（支持 callback 弹窗）
- 成功判定完全由 successKeywords 决定（请配置好）

依赖：
pip install telethon

环境变量：
TG_API_ID
TG_API_HASH

机器人配置（最多 50 个）：
BOT_CONFIG_1 = {"name":"需要点击按钮签到","botUsername":"checkin_bot_a","checkinCommand":"/start","buttonTexts":["签到"],"successKeywords":["签到成功"]}

BOT_CONFIG_2 = {"name":"普通签到","botUsername":"checkin_bot_b","checkinCommand":"/sigin","successKeywords":["打卡成功"]}
"""

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

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


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

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

logging.SUCCESS = 25
logging.addLevelName(logging.SUCCESS, 'SUCCESS')

def success(self, message, *args, **kwargs):
    if self.isEnabledFor(logging.SUCCESS):
        self._log(logging.SUCCESS, message, args, **kwargs)

logging.Logger.success = success


# ==================== 配置区域 ====================

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

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

    RETRY_TIMES = 3  # 重试次数
    WAIT_TIME = 2    # 每次请求间隔（秒）
    TIMEOUT = 40     # 超时时间（秒）

    @classmethod
    def load_from_env(cls):
        cls.BOTS = []

        for i in range(1, 51):
            config_str = os.getenv(f'BOT_CONFIG_{i}')
            if not config_str:
                continue
            try:
                cfg = json.loads(config_str)
                cls.BOTS.append({
                    "name": cfg.get("name", f"Bot {i}"),
                    "username": (cfg.get("botUsername", "") or "").lstrip("@"),
                    "command": cfg.get("checkinCommand", "/start"),
                    "keywords": cfg.get("successKeywords", []),
                    "button_texts": cfg.get("buttonTexts", ["签到", "立即签到", "打卡", "领取", "Check in", "Claim"]),
                    "enabled": cfg.get("enabled", True),
                })
            except Exception as e:
                logger.error(f"解析 BOT_CONFIG_{i} 失败: {e}")

        if not cls.BOTS:
            raise ValueError("未加载到任何机器人配置，请设置 BOT_CONFIG_1...")

        logger.info(f"加载了 {len(cls.BOTS)} 个机器人配置")
        logger.info("机器人列表: " + ", ".join([f"{b['name']}(@{b['username']})" for b in cls.BOTS]))


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

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

    async def initialize(self):
        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("使用会话文件模式（SESSION_FILE）")
            self.client = TelegramClient(
                Config.SESSION_FILE,
                Config.API_ID,
                Config.API_HASH
            )

        await self.client.start()
        me = await self.client.get_me()
        logger.info(f"✅ 用户登录成功: {me.first_name} (@{me.username})")

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

    async def click_button_by_text(self, message, targets: List[str]):
        """
        点击匹配按钮并返回 (clicked: bool, result_text: Optional[str])
        - 支持 callback 弹窗（BotCallbackAnswer.message）
        - 支持点击后返回编辑消息/消息对象（raw_text）
        """
        if not message or not getattr(message, "buttons", None):
            return False, None

        targets_l = [t.lower() for t in (targets or [])]

        for r, row in enumerate(message.buttons):
            for c, btn in enumerate(row):
                text = (getattr(btn, "text", "") or "").strip()
                if not text:
                    continue
                tl = text.lower()
                if any(t in tl for t in targets_l):
                    logger.info(f"准备点击按钮: '{text}' (row={r}, col={c})")
                    res = await message.click(r, c)
                    logger.info(f"✅ 已点击按钮: '{text}'")

                    # callback alert（你 okEmby 就是这种）
                    if hasattr(res, "message") and res.message:
                        return True, res.message

                    # 有些情况 click 返回 Message（编辑后的消息）
                    if hasattr(res, "raw_text") and res.raw_text:
                        return True, res.raw_text

                    return True, None

        return False, None


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

class CheckinService:
    def __init__(self, tg: TelegramUserClient):
        self.tg = tg

    async def perform_checkin(self, bot, attempt=1):
        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 为空（请配置用于判定成功的关键词）"}

        logger.info(f"[{bot_name}] 第 {attempt}/{Config.RETRY_TIMES} 次尝试")

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

            async with self.tg.client.conversation(entity, timeout=Config.TIMEOUT) as conv:
                # 1) 发送命令（必须 conv.send_message，否则会报 No message was sent previously）
                logger.info(f"[{bot_name}] 发送命令: {command}")
                try:
                    await conv.send_message(command)
                except FloodWaitError as e:
                    logger.warning(f"[{bot_name}] FloodWait: 等待 {e.seconds}s 后重发")
                    await asyncio.sleep(e.seconds)
                    await conv.send_message(command)

                # 2) 第一条回复
                msg1 = await conv.get_response()
                text1 = (msg1.raw_text or "").strip()
                logger.info(f"[{bot_name}] 回复1: {text1[:300]}")

                # 直接命中关键词（适用于 /sign /qd /signin 等）
                if any(k in text1 for k in keywords):
                    logger.success(f"[{bot_name}] ✅ 签到成功（文本命中）")
                    return {"success": True, "message": text1}

                # 3) /start 类：有按钮就点（不做兜底文本）
                if getattr(msg1, "buttons", None):
                    clicked, cb_text = await self.tg.click_button_by_text(msg1, button_texts)
                    if not clicked:
                        return {"success": False, "message": "检测到按钮但未匹配到 buttonTexts（请调整 buttonTexts）"}

                    # 3.1 优先判定 callback 弹窗 / 点击返回文本
                    if cb_text:
                        logger.info(f"[{bot_name}] 回调弹窗/结果: {cb_text}")
                        if any(k in cb_text for k in keywords):
                            logger.success(f"[{bot_name}] ✅ 签到成功（按钮回调命中）")
                            return {"success": True, "message": cb_text}
                        else:
                            return {"success": False, "message": cb_text}

                    # 3.2 没有弹窗文本，再等第二条消息（有些机器人会发消息）
                    msg2 = await conv.get_response()
                    text2 = (msg2.raw_text or "").strip()
                    logger.info(f"[{bot_name}] 回复2: {text2[:300]}")

                    if any(k in text2 for k in keywords):
                        logger.success(f"[{bot_name}] ✅ 签到成功（按钮后消息命中）")
                        return {"success": True, "message": text2}

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

                # 4) 无按钮且未命中关键词
                return {"success": False, "message": text1 or "未收到可判定结果"}

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

        except Exception as e:
            logger.error(f"[{bot_name}] ❌ 第 {attempt} 次失败: {e}")
            if attempt < Config.RETRY_TIMES:
                await asyncio.sleep(2)
                return await self.perform_checkin(bot, attempt + 1)
            return {"success": False, "message": str(e)}

    async def execute(self):
        logger.info("=" * 50)
        logger.info("🚀 开始批量签到任务")
        logger.info(f"共 {len(Config.BOTS)} 个机器人")
        logger.info("=" * 50)

        results = []
        for bot in Config.BOTS:
            if not bot.get("enabled", True):
                logger.info(f"跳过禁用机器人: {bot.get('name')}")
                continue

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

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

        self.print_summary(results)
        return {"success": any(x["success"] for x in results), "results": results}

    def print_summary(self, results):
        print("\n" + "=" * 50)
        print("📊 签到结果汇总")
        print("=" * 50)

        succ = 0
        fail = 0
        for r in results:
            status = "✅ 成功" if r["success"] else "❌ 失败"
            print(f"{status} {r['bot']}")
            if r["success"]:
                succ += 1
            else:
                fail += 1
                print(f"     原因: {r.get('message')}")

        print("=" * 50)
        print(f"总计: {len(results)} | 成功: {succ} | 失败: {fail}")
        print("=" * 50)

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


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

async def main():
    Config.load_from_env()

    tg = TelegramUserClient()
    await tg.initialize()

    service = CheckinService(tg)
    result = await service.execute()

    await tg.close()
    raise SystemExit(0 if result.get("success") else 1)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("用户中断程序")
        raise SystemExit(0)
    except Exception as e:
        logger.error(f"运行错误: {e}")
        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格式要求