准备环境
目的：发送不能转发的链接到TG机器人，服务端自动下载并上载到指定频道
## 1、linux服务器 ## 2、服务器需有python3.10以上环境 pip3 install python-telegram-bot
高亮代码格式有问题，用文本模式看
python机器人后台tgbot.py
高亮代码格式有问题，用文本模式看
from enum import Enum
import os,subprocess,sys,time
from telegram import Update # 获取消息队列的
from telegram.ext import filters, MessageHandler, ApplicationBuilder, CommandHandler, ContextTypes
# bot_token是什么自己谷哥
bot_token = "19011110:AAH9111111uwVelu8gKjgDOH24I1M"
# 回复固定内容
# async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
# # 定义一些行为
# # 向发来 /start 的用户发送消息
# await context.bot.send_message(chat_id=update.effective_chat.id,
# text=f"这是一个转存机器人")
async def transfer(update: Update, context: ContextTypes.DEFAULT_TYPE):
# 定义一些行为
# 接收到的文本信息
msg=update.message['text']
if(msg.find("https://t.me/")==-1):
# await context.bot.send_message(chat_id=update.effective_chat.id,text=r"请发送无法转发的链接")
time.sleep(3)
await context.bot.delete_message(chat_id=update.effective_chat.id,message_id=update.message['message_id'])
else:
#回复信息
await context.bot.send_message(chat_id=update.effective_chat.id,text=r"◎开始下载")
process = subprocess.Popen('/root/tdl.sh '+msg, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True)
with process.stdout:
for line in iter(process.stdout.readline, b''):
# print(line.decode().strip())
if(str(line.decode().strip()).find("◎")!=-1):
await context.bot.send_message(chat_id=update.effective_chat.id,text=line.decode().strip())
exitcode = process.wait()
if __name__ == '__main__':
application = ApplicationBuilder().token(bot_token).build()
transfer_handler = MessageHandler((~filters.COMMAND), transfer)
application.add_handler(transfer_handler)
application.run_polling(allowed_updates=Update.ALL_TYPES)
tdl.sh额外插件
用于下载无法转发的文件，并上载到频道
高亮代码格式有问题，用文本模式看
#!/usr/bin/env bash
export PATH=~/bin:/usr/local/bin:/usr/local/sbin:/usr/bin:/usr/sbin:/sbin:/bin
tdl_dir="/root/tdl"
#检查ROOT权限
check_root() {
[[ $EUID != 0 ]] && echo -e "${Error} 当前非ROOT账号(或没有ROOT权限)，无法继续操作，请更换ROOT账号或使用 ${Green_background_prefix}sudo su${Font_color_suffix} 命令获取临时ROOT权限（执行后可能会提示输入当前账号的密码）。" && exit 1
}
#检查系统版本
check_sys() {
if [[ -f /etc/redhat-release ]]; then
release="centos"
elif cat /etc/issue | grep -q -E -i "debian"; then
release="debian"
elif cat /etc/issue | grep -q -E -i "ubuntu"; then
release="ubuntu"
elif cat /etc/issue | grep -q -E -i "centos|red hat|redhat"; then
release="centos"
elif cat /proc/version | grep -q -E -i "debian"; then
release="debian"
elif cat /proc/version | grep -q -E -i "ubuntu"; then
release="ubuntu"
elif cat /proc/version | grep -q -E -i "centos|red hat|redhat"; then
release="centos"
fi
ARCH=$(uname -m)
[ $(command -v dpkg) ] && dpkgARCH=$(dpkg --print-architecture | awk -F- '{ print $NF }')
}
#安装依赖
Installation_dependency() {
if [[ ${release} = "centos" ]]; then
yum update
yum install -y wget curl nano ca-certificates findutils jq tar gzip dpkg
else
apt-get update
apt-get install -y wget curl nano ca-certificates findutils jq tar gzip dpkg
fi
if [[ ! -s /etc/ssl/certs/ca-certificates.crt ]]; then
wget -qO- git.io/ca-certificates.sh | bash
fi
}
#获取最新版本
check_new_ver() {
tdl_new_ver=$(
{
wget -t2 -T3 -qO- "https://api.github.com/repos/iyear/tdl/releases/latest" ||
wget -t2 -T3 -qO- "https://gh-api.p3terx.com/repos/iyear/tdl/releases/latest"
} | grep -o '"tag_name": ".*"' | head -n 1 | cut -d'"' -f4
)
if [[ -z ${tdl_new_ver} ]]; then
echo -e "${Error} 最新版本获取失败，请手动获取最新版本号[ https://github.com/iyear/tdl/releases ]"
read -e -p "请输入版本号:" tdl_new_ver
[[ -z "${tdl_new_ver}" ]] && echo "取消..." && exit 1
fi
}
#下载
Download_tdl() {
update_dl=$1
if [[ $ARCH == i*86 || $dpkgARCH == i*86 ]]; then
ARCH="i386"
elif [[ $ARCH == "x86_64" || $dpkgARCH == "amd64" ]]; then
ARCH="amd64"
elif [[ $ARCH == "aarch64" || $dpkgARCH == "arm64" ]]; then
ARCH="arm64"
elif [[ $ARCH == "armv7l" || $dpkgARCH == "armhf" ]]; then
ARCH="armhf"
else
echo -e "${Error} 不支持此 CPU 架构。"
exit 1
fi
while [[ $(which tdl_dir) ]]; do
echo -e "${Info} 删除旧版 二进制文件..."
rm -vf $(which tdl_dir)
done
[[ ${update_dl} = "update" ]] && rm -f "${tdl_dir}"
DOWNLOAD_URL="https://github.com/iyear/tdl/releases/download/${tdl_new_ver}/tdl_Linux_${ARCH}.tar.gz"
{
wget -t2 -T3 -O- "${DOWNLOAD_URL}" ||
wget -t2 -T3 -O- "https://gh-acc.p3terx.com/${DOWNLOAD_URL}"
} | tar -zx
[[ ! -e ${tdl_dir} ]] && echo -e "${Error} 主程序安装失败！" && exit 1
chmod -R 777 ${tdl_dir}
echo -e "${Info} 主程序安装完成！"
}
run_tdl() {
str=${1}
if [[ $str = "" ]];then
echo "请输入需下载TG文件的链接，多个连接使用空格分隔"
echo
read -e str
fi
for element in $str;
do
link+=" -u $element"
done
#下载
${tdl_dir} dl -n a t 8 -l 10 -d /root/tdl_down${link}
# echo ${tdl_dir} dl -n a t 8 -l 4 -d /root/tdl_down${link}
echo -e "◎下载完成，开始上传"
#上传到的频道
${tdl_dir} up -n a t 8 -l 10 -p "/root/tdl_down" -c https://t.me/xxx --rm
echo -e "◎上传完成"
echo -e "已删除上传完成的文件"
exit 1
}
Install_tdl() {
if [[ ${1} = "" ]];then
echo "更新tdl,否则直接运行 ? (y/N)"
echo
read -e -p "(默认: n):" unyn
[[ -z ${unyn} ]] && unyn="n"
if [[ ${unyn} == [Yy] ]]; then
check_root
check_sys
echo -e "${Info} 开始安装/配置 依赖..."
Installation_dependency
echo -e "${Info} 开始下载/安装 主程序..."
check_new_ver
Download_tdl "update"
else
echo && echo "运行tdl..." && echo
fi
fi
run_tdl ${1}
}
Install_tdl ${1}
添加服务，挂后台
vi /lib/systemd/system/tgbot.service
[Unit] Description=tgbot [Service] Type=simple User=root Restart=always ExecStart=/usr/bin/python3.10 /root/tgbot.py #路径有空格需要用引号框起来 ExecReload=/bin/kill -HUP $MAINPID KillMode=process [Install] WantedBy=multi-user.target
systemctl enable tgbot.service systemctl start tgbot.service
本文作者为55gY，转载请注明。
本站资源来自互联网收集，仅供用于学习和交流，请勿用于商业用途。如有侵权、不妥之处，请联系站长并出示版权证明以便删除。敬请谅解！