Telegram全自动翻译

没用任何技术含量,纯让gpt帮忙写的。有bug可以问GPT

始皇改版:将Telegram全自动翻译更进一步

免费GPTAPI:https://linux.do/t/topic/17460

效果图


代码如下

import yaml
import logging
import time
import re
from telethon.sync import TelegramClient
from telethon import events
from openai import OpenAI

# 设置日志记录,便于调试和追踪程序运行情况。
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 尝试从YAML配置文件中读取必要的配置信息。
try:
    with open('config.yaml', 'r', encoding='utf-8') as file:
        config = yaml.safe_load(file)
except Exception as e:
    logging.error(f"Failed to load config file: {e}")
    exit(1)

# 提取Telegram和OpenAI的配置信息。
api_id = config['telegram']['api_id']
api_hash = config['telegram']['api_hash']
session_name = config['telegram']['session_name']
api_key = config['openai']['api_key']
base_url = config['openai']['base_url']
target_users = config['target_users']
target_groups = config['target_groups']

# 初始化Telegram客户端和OpenAI客户端。
client = TelegramClient(session_name, api_id, api_hash)
openai_client = OpenAI(api_key=api_key, base_url=base_url)



# 检查字符串是否包含中文字符。
def contains_chinese(text):
    return any('\u4e00' <= character <= '\u9fff' for character in text)

# 检查字符串是否完全不包含中文字符。
def contains_no_chinese(text):
    return all(character < '\u4e00' or character > '\u9fff' for character in text)

# 检查字符串是否包含任何非中文字符。
def contains_non_chinese(text):
    return any(character < '\u4e00' or character > '\u9fff' for character in text)

# 检查字符串是否是纯链接。
def is_pure_url(text):
    url_pattern = r'^\s*http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+\s*$'
    return re.match(url_pattern, text) is not None

# 使用OpenAI的API进行文本翻译的函数。 gpt-3.5-turbo/coze gpt4
def translate_text(text, target_language):
    # 使用全局OpenAI客户端进行翻译。
    response = openai_client.chat.completions.create(
        model="coze",
        messages=[
            {"role": "system", "content": "You are a translation engine, only returning translated answers."},
            {"role": "user", "content": f"Translate the text to {target_language} please do not explain my original text, do not explain the translation results, Do not explain the context.:\n{text}"}
        ]
    )
    translated_text = response.choices[0].message.content
    return translated_text

# 监听新消息事件,进行消息处理。
@client.on(events.NewMessage)
async def handle_message(event):
    if event.chat_id not in target_groups and event.sender_id not in target_users:
        return  # 如果消息不是来自目标用户或群组,则忽略
    start_time = time.time()  # 记录开始时间
    try:
        message = event.message

        # 跳过空文本或非文本消息。
        if not message.text or message.text.strip() == "":
            return

        # 跳过仅包含链接的消息。
        if is_pure_url(message.text):
            return

        # 记录来自target_users的消息和你自己发出去的消息为log。
        if (message.chat_id in target_users or (message.sender_id in target_users and event.is_group)) or message.out:
            logging.info(f"原文: {message.text}")

        # 对特定用户/群聊的消息进行翻译。
        if (message.chat_id in target_users or (message.sender_id in target_users and event.is_group)) and not message.out:
            if contains_non_chinese(message.text):
                # 如果消息只包含链接,跳过翻译。
                if is_pure_url(message.text):
                    pass
                # 如果消息不只包含链接,进行翻译。
                else:
                    translated_text = translate_text(message.text, "zh_CN")  
                    if translated_text and contains_chinese(translated_text):
                        modified_message = f"**{translated_text}**"
                        await client.send_message(entity=message.chat_id, message=modified_message, reply_to=message.id)
                end_time = time.time()  # 记录结束时间
                logging.info(f"翻译总耗时:{end_time - start_time}秒")  # 记录并打印翻译耗时
                logging.info(f"引用并翻译成功,已回复:\n{modified_message}")

        # 对自己在特定群聊发送的中文消息进行英文翻译。
        elif message.out and message.chat_id in target_groups:
            if contains_chinese(message.text):
                translated_text = translate_text(message.text, "English")  
                if translated_text:
                    # 修改原消息,添加英文翻译。
                    modified_message = f"{message.text}\n{translated_text}"
                    await message.edit(modified_message)
                    end_time = time.time()  # 记录结束时间
                    logging.info(f"翻译总耗时:{end_time - start_time}秒")  # 记录并打印翻译耗时
                    logging.info(f"消息编辑成功,新消息:\n{modified_message}")

    except Exception as e:
        # 记录处理消息时发生的异常。
        logging.error(f"Error handling message: {e}")

# 启动客户端并保持运行。
try:
    client.start()
    client.run_until_disconnected()
finally:
    # 断开客户端连接。
    client.disconnect()

配置文件 config代码如下

telegram:
  api_id: 123456789   #https://my.telegram.org/apps 获取到的api_id
  api_hash: "12345678999876543210"     #https://my.telegram.org/apps 获取到的App api_hash  这里双引号别丢
  session_name: "translate"

openai:
  base_url: "https://freeapi.iil.im/v1"
  api_key: "免费的sk-key"   #建议用免费api: https://linux.do/t/topic/17460
  

target_users: #被翻译的人,一行一个, 把对方发送的英文翻译为中文(群内也会翻译他) 
  - 5001717356
  - 1412647738  #AI精神病院的 fofa
  - 805275901  #始皇群的某个老外


target_groups: #翻译自己发送的中文为 中英 ,一行一个, 被添加的人或群,对他们发送消息翻译为中英。
  - -1002103720350  #AI精神院
  - -1001680975844  #始皇群  

依赖安装
pip install pyyaml telethon openai

脚本仓库地址

需要长时间挂机在服务器。不会的问gpt
脚本和config文件放在同一个目录。

第一次登录会要验证码,tg的 需要输入 区号前缀 比如+8618792913567
22222D

更新切换模型命令。
此次更新使用方法
/fymodel 你需要切换的模型
同时除了默认的gpt-4-coze模型,还增加了免费gpt-3.5-turbo
/fymodel gpt-3.5-turbo
大家最新可用gpt-4模型
/fymodel gpt-4 速度超快

import yaml
import logging
import time
import re
from telethon.sync import TelegramClient
from telethon import events
from openai import OpenAI

# 设置日志记录,便于调试和追踪程序运行情况。
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 尝试从YAML配置文件中读取必要的配置信息。
try:
    with open('config.yaml', 'r', encoding='utf-8') as file:
        config = yaml.safe_load(file)
except Exception as e:
    logging.error(f"Failed to load config file: {e}")
    exit(1)

# 提取Telegram和OpenAI的配置信息。
api_id = config['telegram']['api_id']
api_hash = config['telegram']['api_hash']
session_name = config['telegram']['session_name']
api_key = config['openai']['api_key']
base_url = config['openai']['base_url']
target_users = config['target_users']
target_groups = config['target_groups']

# 初始化Telegram客户端和OpenAI客户端。
client = TelegramClient(session_name, api_id, api_hash)
openai_client = OpenAI(api_key=api_key, base_url=base_url)

# 默认使用的模型。
current_model = "coze"


# 检查字符串是否包含中文字符。
def contains_chinese(text):
    return any('\u4e00' <= character <= '\u9fff' for character in text)

# 检查字符串是否完全不包含中文字符。
def contains_no_chinese(text):
    return all(character < '\u4e00' or character > '\u9fff' for character in text)

# 检查字符串是否包含任何非中文字符。
def contains_non_chinese(text):
    return any(character < '\u4e00' or character > '\u9fff' for character in text)

# 检查字符串是否是纯链接。
def is_pure_url(text):
    url_pattern = r'^\s*http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+\s*$'
    return re.match(url_pattern, text) is not None

# 使用OpenAI的API进行文本翻译的函数。 gpt-3.5-turbo/coze gpt4
def translate_text(text, target_language):
    response = openai_client.chat.completions.create(
        model=current_model,
        messages=[
            {"role": "system", "content": "You are a translation engine, only returning translated answers."},
            {"role": "user", "content": f"Translate the text to {target_language} please do not explain my original text, do not explain the translation results, Do not explain the context.:\n{text}"}
        ]
    )
    translated_text = response.choices[0].message.content
    return translated_text
	
# 处理命令以切换模型。
@client.on(events.NewMessage(pattern='/fymodel'))
async def change_model(event):
    global current_model

    # 仅允许您自己的用户ID执行此命令
    authorized_user_id = 856368022  # 替换为您的Telegram用户ID
    if event.sender_id != authorized_user_id:
        await event.reply("您没有权限使用此命令。")
        return

    # 分割命令文本以提取模型名称。
    parts = event.raw_text.split(maxsplit=1)
    if len(parts) > 1:
        model_name = parts[1]
        current_model = model_name
        await event.reply(f"翻译模型已切换至: {current_model}")
    else:
        await event.reply("请指定要切换的模型名称。")

	

# 监听新消息事件,进行消息处理。
@client.on(events.NewMessage)
async def handle_message(event):
    if event.chat_id not in target_groups and event.sender_id not in target_users:
        return  # 如果消息不是来自目标用户或群组,则忽略
    start_time = time.time()  # 记录开始时间
    try:
        message = event.message

        # 跳过空文本或非文本消息。
        if not message.text or message.text.strip() == "":
            return

        # 跳过仅包含链接的消息。
        if is_pure_url(message.text):
            return

        # 记录来自target_users的消息和你自己发出去的消息为log。
        if (message.chat_id in target_users or (message.sender_id in target_users and event.is_group)) or message.out:
            logging.info(f"原文: {message.text}")

        # 对特定用户/群聊的消息进行翻译。
        if (message.chat_id in target_users or (message.sender_id in target_users and event.is_group)) and not message.out:
            if contains_non_chinese(message.text):
                # 如果消息只包含链接,跳过翻译。
                if is_pure_url(message.text):
                    pass
                # 如果消息不只包含链接,进行翻译。
                else:
                    translated_text = translate_text(message.text, "zh_CN")  
                    if translated_text and contains_chinese(translated_text):
                        modified_message = f"**{translated_text}**"
                        await client.send_message(entity=message.chat_id, message=modified_message, reply_to=message.id)
                end_time = time.time()  # 记录结束时间
                logging.info(f"翻译总耗时:{end_time - start_time}秒")  # 记录并打印翻译耗时
                logging.info(f"引用并翻译成功,已回复:\n{modified_message}")

        # 对自己在特定群聊发送的中文消息进行英文翻译。
        elif message.out and message.chat_id in target_groups:
            if contains_chinese(message.text):
                translated_text = translate_text(message.text, "English")  
                if translated_text:
                    # 修改原消息,添加英文翻译。
                    modified_message = f"{message.text}\n{translated_text}"
                    await message.edit(modified_message)
                    end_time = time.time()  # 记录结束时间
                    logging.info(f"翻译总耗时:{end_time - start_time}秒")  # 记录并打印翻译耗时
                    logging.info(f"消息编辑成功,新消息:\n{modified_message}")

    except Exception as e:
        # 记录处理消息时发生的异常。
        logging.error(f"Error handling message: {e}")

# 启动客户端并保持运行。
try:
    client.start()
    client.run_until_disconnected()
finally:
    # 断开客户端连接。
    client.disconnect()
73 Likes

顶一个

1 Like

软件分享软件开发

蒙大将军good job

1 Like

好东西!

3 Likes

牛逼啊

好东西

如果有微信那就是绝杀

大佬牛逼

搞起来 :sunglasses:

微信= = 也没用到和老外聊天吧

我想学英语

666

佬不去机器学习板块水一贴?那里目前好像还是荒漠:rofl:

感谢分享

快找点项目发

跟着佬们学习

1 Like

如何翻译其他用户的呢

target_users: #被翻译的人,一行一个,

在这里添加你想翻译的人的 id即可

把我的id加进去 看看呢

34 Likes