diff --git a/sync_script.py b/sync_script.py deleted file mode 100644 index 6e0057a..0000000 --- a/sync_script.py +++ /dev/null @@ -1,182 +0,0 @@ -import os -import logging -import asyncio -import json -import requests -import filecmp -import shutil -import zipfile -from telegram import Bot -from git import Repo -from dotenv import load_dotenv -from telethon.sync import TelegramClient -from telegram.ext import Application - -# 配置日志模块 -logging.basicConfig( - filename='/var/log/telegram_gitea_sync.log', # 日志文件存储路径 - level=logging.ERROR, # 只记录错误级别及以上日志 - format='%(asctime)s - %(levelname)s - %(message)s' # 格式：时间戳、日志级别和消息 -) - -# 加载环境变量 -load_dotenv() - -TELEGRAM_API_ID = os.getenv("TELEGRAM_API_ID") -TELEGRAM_API_HASH = os.getenv("TELEGRAM_API_HASH") -BOT_TOKEN = os.getenv("BOT_TOKEN") -GITEA_TOKEN = os.getenv("GITEA_TOKEN") -GITEA_REPO = os.getenv("GITEA_REPO") -GITEA_BRANCH = os.getenv("GITEA_BRANCH") -TELEGRAM_CHAT_ID = "-1002513646618" - -# 初始化 -bot = Bot(token=BOT_TOKEN) -TEMP_DIR = "/opt/telegram_gitea_sync/tmp" -REPO_DIR = "/opt/telegram_gitea_sync/repo" -os.makedirs(TEMP_DIR, exist_ok=True) - -async def send_telegram_message(chat_id, text): - """ - 异步发送 Telegram 消息 - """ - try: - await bot.send_message(chat_id=chat_id, text=text) - except Exception as e: - print(f"发送消息失败: {e}") - -async def download_latest_file(): - """ - 使用个人账号从 Telegram 下载最新文件 - """ - try: - # 登录 Telegram - client = TelegramClient('session_name', TELEGRAM_API_ID, TELEGRAM_API_HASH) - await client.start(phone="+17165880598") # 替换为您的 Telegram 注册手机号 - - print("成功登录 Telegram") - - # 获取目标频道最新消息 - messages = await client.get_messages("PandaGroovePG", limit=10) - print(f"获取到的消息数量: {len(messages)}") - - # 遍历消息，查找文件 - for message in messages: - if message.file: - file_name = message.file.name - file_path = f"{TEMP_DIR}/{file_name}" - await message.download_media(file_path) - print(f"文件下载完成: {file_path}") - return file_path - - # 未找到文件 - raise FileNotFoundError("未找到可下载的文件") - except Exception as e: - print(f"文件下载失败: {e}") - raise - -async def process_files(file_path): - """ - 解压、递归比对文件，并同步到仓库 - """ - try: - print("开始解压文件...") - shutil.unpack_archive(file_path, TEMP_DIR) # 解压 ZIP 文件 - print(f"文件已解压到临时目录: {TEMP_DIR}") - - # 遍历所有文件和文件夹，逐一比对内容 - for root, dirs, files in os.walk(TEMP_DIR): # 遍历解压后的临时目录 - for file in files: - temp_file_path = os.path.join(root, file) # 解压后的文件路径 - repo_file_path = temp_file_path.replace(TEMP_DIR, REPO_DIR) # 替换路径到仓库 - - # 如果仓库中不存在该文件，直接复制 - if not os.path.exists(repo_file_path): - print(f"新文件发现，复制到仓库: {temp_file_path}") - os.makedirs(os.path.dirname(repo_file_path), exist_ok=True) - shutil.copy(temp_file_path, repo_file_path) - else: - # 文件存在，进行内容比对 - if not filecmp.cmp(temp_file_path, repo_file_path, shallow=False): - print(f"文件已更新，覆盖旧版本: {temp_file_path}") - shutil.copy(temp_file_path, repo_file_path) - else: - print(f"文件未更改，跳过: {temp_file_path}") - - print("所有文件比对完成，准备同步到 Gitea") - await sync_to_gitea() - except Exception as e: - print(f"处理文件时发生错误: {e}") - raise - - -async def process_jsm_json(json_path): - """ - 处理 jsm.json 文件 - """ - try: - with open(json_path, "r", encoding="utf-8") as f: - data = json.load(f) - data["logo"] = "./bj/fyj.gif" - for site in data.get("sites", []): - if site.get("key") == "lf_js_search": - site["proxy"] = "noproxy" - with open(json_path, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, indent=4) - except Exception as e: - await send_telegram_message(TELEGRAM_CHAT_ID, f"处理 jsm.json 文件失败: {e}") - raise - -async def sync_to_gitea(): - """ - 推送更新到 Gitea 仓库 - """ - try: - print(f"检查仓库路径是否存在: {REPO_DIR}") - if not os.path.exists(REPO_DIR): - raise FileNotFoundError(f"仓库路径不存在: {REPO_DIR}") - - repo = Repo(REPO_DIR) - print("成功加载 Gitea 仓库") - - repo.git.add("--all") - print("添加所有修改文件") - - repo.index.commit("自动更新文件") - print("提交修改") - - origin = repo.remote(name="origin") - origin.push() - print("推送更新到远程仓库成功") - except Exception as e: - print(f"同步到 Gitea 失败: {e}") - raise - - -async def main(): - """ - 主流程 - """ - try: - print("开始执行主流程...") # 添加调试信息 - await send_telegram_message(TELEGRAM_CHAT_ID, "开始检测最新文件...") - print("已发送电报通知") # 添加调试信息 - - file_path = await download_latest_file() - print(f"文件下载完成，路径为: {file_path}") # 输出下载文件路径 - - await process_files(file_path) - print("文件处理完成并已同步到 Gitea") # 确认文件处理完成 - - await send_telegram_message(TELEGRAM_CHAT_ID, "同步完成！") - print("任务完成，通知已发送到电报频道") # 确认任务完成 - except Exception as e: - error_message = f"发生错误: {e}" - print(error_message) # 输出错误信息到终端 - await send_telegram_message(TELEGRAM_CHAT_ID, error_message) - -if __name__ == "__main__": - print("脚本已启动，准备执行主逻辑...") # 添加调试信息 - import asyncio - asyncio.run(main()) - print("脚本已完成运行！") # 添加结束时的调试信息