Telegram 自动翻译机器人部署完整教程(DeepLX + Telethon)
- 建站相关
- 2025-10-26
- 330热度
- 0评论
部署 DeepLX(本地翻译引擎)
创建文件 /home/deeplx/docker-compose.yml:
version: '3.8'
services:
deeplx:
image: ghcr.io/owo-network/deeplx:latest
container_name: deeplx
restart: always
ports:
- "1188:1188"
environment:
TZ: Asia/Shanghai
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
启动:
docker compose up -d
验证是否成功:
curl -X POST http://127.0.0.1:1188/translate \
-H "Content-Type: application/json" \
-d '{"text":"Hello world","source_lang":"EN","target_lang":"ZH"}'
应返回:
{"data":"世界你好","code":200}
✅ 表示 DeepLX 已可用。
安装 Python 依赖
python3 -m pip install --break-system-packages aiohttp telethon
🧾 六、创建配置文件 config.json
nano /home/deeplx/config.json
{
"api_id": 123456,
"api_hash": "9834fsfsfsfaqe3rfvdv",
"target_config": {}
}
⚠️ 替换 api_id 与 api_hash 为你在 https://my.telegram.org
获取的开发者参数。(要用国外家宽申请)
创建翻译脚本 translate.py
nano /home/deeplx/translate.py
粘贴以下内容👇(使用本地 DeepLX 接口):
# -*- coding: utf-8 -*-
import asyncio
import json
import logging
import os
import time
import aiohttp
from telethon import events
from telethon.sync import TelegramClient
# ===== 日志配置 =====
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
# ===== 配置加载与保存 =====
def load_config():
if not os.path.exists('config.json'):
logging.error('config.json not found, please create one first.')
exit()
with open('config.json', 'r') as f:
return json.load(f)
def save_config():
cfg['target_config'] = target_config
with open('config.json', 'w') as f:
json.dump(cfg, f, indent=2)
# ===== 初始化配置 =====
cfg = load_config()
api_id = cfg['api_id']
api_hash = cfg['api_hash']
target_config = cfg.get('target_config', {})
# ===== 初始化 Telegram 客户端 =====
client = TelegramClient('chat_translator', api_id, api_hash)
# ===== 本地 DeepLX 翻译接口 =====
DEEPLX_URL = "http://127.0.0.1:1188/translate"
# ===== 翻译函数 =====
async def translate_single(text, source_lang, target_lang, session):
if source_lang == target_lang:
return target_lang, text
payload = {
"text": text,
"source_lang": source_lang,
"target_lang": target_lang
}
start_time = time.time()
async with session.post(DEEPLX_URL, json=payload) as response:
if response.status != 200:
raise Exception(f"翻译失败: HTTP {response.status}")
result = await response.json()
if result.get('code') != 200:
raise Exception(f"翻译失败: {result}")
logging.info(f"翻译 {source_lang} → {target_lang} 成功, 耗时 {time.time() - start_time:.2f}s")
return target_lang, result['data']
async def translate_text(text, source_lang, target_langs):
result = {}
async with aiohttp.ClientSession() as session:
tasks = [translate_single(text, source_lang, lang, session) for lang in target_langs]
for lang, translated in await asyncio.gather(*tasks):
result[lang] = translated
return result
# ===== 命令模式 =====
async def command_mode(event, target_key, text) -> bool:
# 关闭翻译
if text == '.tt-off':
await event.delete()
if target_key in target_config:
del target_config[target_key]
save_config()
logging.info(f"已关闭翻译: {target_key}")
return False
# 启用翻译
if text.startswith('.tt-on,'):
await event.delete()
try:
_, src, targets = text.split(',', 2)
except ValueError:
logging.error("命令格式错误,应为:.tt-on,ZH,EN|RU")
return False
target_config[target_key] = {
"source_lang": src.upper(),
"target_langs": [x.upper() for x in targets.split('|')]
}
save_config()
logging.info(f"已启用翻译: {target_config[target_key]}")
return False
return True
# ===== 监听消息并翻译 =====
@client.on(events.NewMessage(outgoing=True))
async def handle_message(event):
target_key = f"{event.chat_id}.{event.sender_id}"
try:
message = event.message
if not message.text:
return
text = message.text.strip()
if not text:
return
# 命令判断
if text.startswith('.tt-') and not await command_mode(event, target_key, text):
return
# 未启用翻译
if target_key not in target_config:
return
conf = target_config[target_key]
translated_texts = await translate_text(text, conf['source_lang'], conf['target_langs'])
# 重新安排显示顺序:英文在上(主文本),中文在下(代码块)
main_texts = []
code_texts = []
for lang in conf['target_langs']:
if lang == 'EN':
main_texts.append(translated_texts[lang])
else:
code_texts.append(translated_texts[lang])
# 构建输出
if main_texts:
output = '\n'.join(main_texts)
else:
# 如果没有英文,使用第一个目标语言作为主文本
first_lang = conf['target_langs'][0]
output = translated_texts[first_lang]
# 将其从代码文本中移除
if first_lang in [lang for lang in conf['target_langs']]:
code_texts = [translated_texts[lang] for lang in conf['target_langs'][1:]]
# 如果有其他语言(包括中文),放在代码块中
if code_texts:
output += '\n``<code>%s\n</code>``' % '\n'.join(code_texts)
await message.edit(output)
logging.info(f"消息翻译完成: {output[:50]}...")
except Exception as e:
logging.error(f"消息处理异常: {e}")
# ===== 启动客户端 =====
try:
client.start()
logging.info("🤖 Telegram 翻译机器人已启动,使用本地 DeepLX 接口")
client.run_until_disconnected()
finally:
client.disconnect()
logging.info("⚠️ 机器人已断开连接")
保存退出。
首次运行登录 Telegram
cd /home/deeplx
python3 translate.py
会要你输入你的tg手机号,然后要求输入验证码
首次会提示:
Please enter your phone number:
Please enter the code you received:
Signed in successfully as ...
登录成功后按 Ctrl + C 停止。
此时生成文件:
chat_translator.session
后台运行与开机自启
1️⃣ 创建启动脚本
nano /home/deeplx/start_translate.sh
内容:
#!/bin/bash
cd /home/deeplx
nohup python3 translate.py > /home/deeplx/translate.log 2>&1 &
保存:
chmod +x /home/deeplx/start_translate.sh
2️⃣ 设置开机自启与每小时检查
crontab -e
添加:
@reboot /home/deeplx/start_translate.sh
0 * * * * pgrep -f "translate.py" > /dev/null || /home/deeplx/start_translate.sh
含义:
开机后自动启动翻译机器人
每小时检查一次是否运行,如停止则自动重启
使用对应账号在对应聊天群里设置启用/禁用。
指令:
.tt-off 群内直接输入,禁用本账号/频道号在该聊天/群的翻译功能。
.tt-on,zh,zh|en|ru 设置并开启本账号/频道号在该聊天/群的翻译功能。参数依次是:源语言,目标语言列表。
目标语言列表将按顺序显示,第一个语言作为主要语言。
如果在目标语言列表中不包含原语言,则相当于上版本的不保留原内容。
目标语言列表使用|隔开。
以上配置会实时写进配置文件,后续重启脚本都会加载。
该服务跑在国外服务器上效果会更佳!
