实现Telegram投稿机器人

资源信息

要实现一个Telegram投稿机器人，你可以参考以下步骤：

1. 创建Telegram机器人：
   • 在Telegram中搜索@BotFather并开始一个对话。
   • 发送/newbot命令来创建一个新的机器人。
   • 按照提示为你的机器人选择一个名字和用户名。
   • BotFather会提供一个唯一的API token，这是你后续开发中需要用到的。
2. 设置开发环境：
   • 选择你的编程语言，比如Python、JavaScript等。
   • 安装必要的依赖库。例如，如果你使用Python，可以安装python-telegram-bot库。
   • 选择一个IDE（集成开发环境），比如Visual Studio Code。
3. 编写机器人代码：
   • 使用Telegram Bot API来实现机器人的功能。你可以编写代码来处理消息、命令和与外部服务的集成。
   • 例如，使用Python的python-telegram-bot库，你可以创建一个简单的回声机器人，它会回复用户发送的每条消息。
4. 运行和测试你的机器人：
   • 在本地运行你的机器人代码，并在Telegram中与其交互以测试功能。
   • 确保你的机器人能够接收和处理投稿请求，并能够将内容发布到目标平台。
5. 部署你的机器人：
   • 为了让你机器人24/7在线，你可以将其部署到云服务平台，如Heroku、AWS等。
6. 添加额外的功能：
   • 根据需要，你可以添加更多功能，比如处理命令、发送通知、执行任务等。

例如，如果你想使用Python创建一个简单的Telegram机器人，你可以按照以下代码示例操作：

import os
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters

def start(update, context):
    update.message.reply_text('Welcome! I\'m your new Telegram bot.')

def echo(update, context):
    update.message.reply_text(update.message.text)

def main():
    TOKEN = os.environ['TELEGRAM_APITOKEN']
    updater = Updater(token=TOKEN, use_context=True)
    dispatcher = updater.dispatcher

    start_handler = CommandHandler('start', start)
    echo_handler = MessageHandler(Filters.text & ~Filters.command, echo)

    dispatcher.add_handler(start_handler)
    dispatcher.add_handler(echo_handler)

    updater.start_polling()
    updater.idle()

if __name__ == '__main__':
    main()

在这个例子中，我们定义了两个处理函数start和echo，分别用于处理/start命令和消息。然后我们创建了一个Updater实例，并添加了这两个处理函数。最后，我们启动了机器人的轮询，并使其进入空闲状态等待消息。

夸克资源投稿机器人

鹏星放在了GitHub上面运行

代码实现

import re
import os
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, filters, ContextTypes

# 初始化机器人
TOKEN = os.getenv("TELEGRAM_TOKEN")
CHANNEL_ID = '@naclyunpan'  # 替换为你的频道 Chat ID

# 临时存储用户投稿内容
user_posts = {}


# 启动命令，提供投稿模板
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    template_message = (
        "请按照以下格式投稿：\n\n"
        "图片\n\n"
        "名称：\n\n描述：\n\n链接：（夸克网盘）\n\n"
        "📁 大小：\n🏷 标签："
    )
    await update.message.reply_text(template_message)


# 处理用户的投稿消息
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # 检查消息和聊天对象是否存在
    if update.message is None or update.message.chat is None:
        await context.bot.send_message(chat_id=update.effective_chat.id,
                                       text="切勿在原文案上编辑，请重新发布。")
        return  # 确保消息存在且有效

    user_id = update.message.from_user.id

    # 检查投稿内容是否包含图片和文字
    if not update.message.photo or not update.message.caption:
        await update.message.reply_text("投稿格式不正确，请按照模板重新投稿。\n\n"
                                        "请按照以下格式投稿：\n\n"
                                        "图片\n\n"
                                        "名称：\n\n描述：\n\n链接：（夸克网盘）\n\n"
                                        "📁 大小：\n🏷 标签：")
        return

    # 获取图片和文字内容
    image = update.message.photo[-1].file_id
    caption = update.message.caption

    # 定义格式正则表达式
    pattern = (
        r"名称：\s*.*\n\n"
        r"描述：\s*.*\n\n"
        r"链接：\s*https:\/\/pan\.quark\.cn\/s\/[^\/]+\n\n"
        r"📁 大小：\s*.*\n"
        r"🏷 标签：\s*.*"
    )

    # 验证格式
    if not re.fullmatch(pattern, caption):
        await update.message.reply_text("投稿格式不正确，请按照模板重新投稿。\n\n"
                                        "请按照以下格式投稿：\n\n"
                                        "图片\n\n"
                                        "名称：\n\n描述：\n\n链接：（夸克网盘）\n\n"
                                        "📁 大小：\n🏷 标签：")
        return

    # 检查链接是否为夸克网盘链接
    quark_link_pattern = r"https:\/\/pan\.quark\.cn\/s\/[^\/]+"
    if not re.search(quark_link_pattern, caption):
        await update.message.reply_text("链接必须是夸克网盘的链接，请重新投稿。")
        return

    # 存储用户投稿内容以便之后编辑
    user_posts[user_id] = {'image': image, 'caption': caption}

    # 提供确认和编辑按钮
    keyboard = [
        [InlineKeyboardButton("编辑", callback_data="edit_post")],
        [InlineKeyboardButton("确认发布", callback_data="confirm_post")]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)

    # 确认并提示用户可以编辑
    await update.message.reply_text("感谢您的投稿！您可以选择编辑或确认发布到频道：", reply_markup=reply_markup)


# 处理编辑按钮的回调
async def handle_edit_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    user_id = query.from_user.id

    # 确保回应按钮点击事件
    await query.answer()

    # 检查当前消息内容
    current_text = query.message.text
    new_text = "请发送新的投稿内容，格式与之前相同。"

    # 只有在新内容不同的情况下，才进行编辑
    if current_text != new_text:
        await query.edit_message_text(new_text)
    else:
        await query.answer("消息内容未改变。", show_alert=True)


# 更新投稿内容
async def update_post(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.message.from_user.id

    # 检查用户是否处于编辑状态
    if context.user_data.get('editing'):
        new_text = update.message.text

        # 更新投稿内容
        if user_id in user_posts:
            user_posts[user_id]['caption'] = new_text
            await update.message.reply_text("您的投稿内容已更新！")
            # 清除编辑状态
            context.user_data['editing'] = False
        else:
            await update.message.reply_text("没有找到可编辑的投稿内容。")
    else:
        await update.message.reply_text("请先使用编辑按钮来编辑您的投稿内容。")


# 确认发布到频道的回调
async def handle_confirm_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    user_id = query.from_user.id

    # 检查用户是否有投稿内容
    if user_id in user_posts:
        post_data = user_posts[user_id]
        image = post_data['image']
        caption = post_data['caption']

        # 构建频道消息内容
        channel_message = (
            f"{caption}\n\n"
            f"📢 频道：@naclyunpan"
        )

        # 发送图片和文字到频道
        await context.bot.send_photo(chat_id=CHANNEL_ID, photo=image, caption=channel_message)

        # 回复用户投稿成功，并从临时存储中删除内容
        await query.answer("内容已成功发布到频道！")
        await query.edit_message_text("您的投稿已成功发布到频道。感谢您的支持！")
        del user_posts[user_id]
    else:
        await query.answer("找不到您的投稿内容，无法发送到频道。")


def main():
    # 创建应用程序实例
    application = Application.builder().token(TOKEN).build()

    # 添加命令处理器
    application.add_handler(CommandHandler("start", start))

    # 添加消息处理器，处理首次投稿和更新内容
    application.add_handler(MessageHandler(filters.TEXT | filters.PHOTO, handle_message))

    # 添加编辑和确认回调处理器
    application.add_handler(CallbackQueryHandler(handle_edit_callback, pattern="edit_post"))
    application.add_handler(CallbackQueryHandler(handle_confirm_callback, pattern="confirm_post"))

    # 添加更新消息处理器
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, update_post))

    # 开始轮询更新
    application.run_polling()


if __name__ == '__main__':
    main()

安装依赖包

requirements.txt

anyio==4.6.2.post1
certifi==2024.8.30
exceptiongroup==1.2.2
h11==0.14.0
httpcore==1.0.6
httpx==0.27.2
idna==3.10
python-telegram-bot==21.6
sniffio==1.3.1
style==1.1.0
typing_extensions==4.12.2
update==0.0.1

启动 GitHub Actions

mian.yaml

name: tgBot

on:
  push:  # 当有代码推送时触发
  schedule:
    - cron: '0 */5 * * *'  # 每隔5小时触发一次

jobs:
  run_demo_actions:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.9'

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt

      - name: Run telegram_contribute.py
        env:
          TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
        run: python telegram_contribute.py

资源获取

下载地址：夸克网盘

GitHub：https://github.com/luozhipeng1/tg_contribute

telegram：https://t.me/naclyunpan