import sys
import subprocess
import os
import asyncio
from datetime import datetime, timedelta, timezone
==================== 0. 面板专属：环境依赖自动安装器 ====================
lib_path = os.path.join(os.path.dirname(os.path.abspath(file)), ‘libs’)
sys.path.insert(0, lib_path)
try:
import telethon
import dotenv
except ImportError:
print(“>>> 首次运行：检测到缺少依赖，正在局部目录(libs)自动安装，请稍候…”)
subprocess.check_call([
sys.executable, “-m”, “pip”, “install”, “-t”, lib_path,
“telethon”, “python-dotenv”
])
print(“>>> 依赖安装完成！准备启动程序…”)
from telethon import TelegramClient, errors
from telethon.tl.types import (
MessageMediaDocument,
DocumentAttributeFilename,
InputMessagesFilterVideo
)
from dotenv import load_dotenv
================== 1. 读取配置 ==================
load_dotenv()
try:
api_id = int(os.getenv(“API_ID”))
api_hash = os.getenv(“API_HASH”)
PHONE_NUMBER = os.getenv(“PHONE_NUMBER”)
TWO_STEP_PASSWORD = os.getenv(“TWO_STEP_PASSWORD”) or None
raw_target = os.getenv("TARGET_CHANNEL")
try:
TARGET_CHANNEL = int(raw_target)
except (ValueError, TypeError):
TARGET_CHANNEL = raw_target
# 🎯 数量限制（0 = 不限制）
TARGET_VIDEO_COUNT = int(os.getenv("SCAN_LIMIT", "0"))
# ⏰ 时间限制
DAYS_LIMIT = int(os.getenv("DAYS_LIMIT", "0"))
HOURS_LIMIT = int(os.getenv("HOURS_LIMIT", "0"))
time_threshold = None
if DAYS_LIMIT > 0:
time_threshold = datetime.now(timezone.utc) - timedelta(days=DAYS_LIMIT)
elif HOURS_LIMIT > 0:
time_threshold = datetime.now(timezone.utc) - timedelta(hours=HOURS_LIMIT)except Exception as e:
print(f”❌ 配置错误: {e}”)
sys.exit(1)
提前创建 session 文件夹，防止面板报错
if not os.path.exists(“session”):
os.makedirs(“session”)
client = TelegramClient(“session/user_session”, api_id, api_hash)
================== 2. 工具函数 ==================
def get_video_info(message):
if not message.media or not isinstance(message.media, MessageMediaDocument):
return False, None, None
doc = message.media.document
if not (doc.mime_type and doc.mime_type.startswith("video/")):
return False, None, None
file_name = "未知文件名"
duration = 0
for attr in doc.attributes:
if isinstance(attr, DocumentAttributeFilename):
file_name = attr.file_name
if hasattr(attr, "duration"):
duration = attr.duration
# ✅ 推荐判重 key（增强版）
key = (doc.size, doc.mime_type, doc.dc_id, duration)
return True, key, file_name================== 3. 主逻辑 ==================
async def main():
print(“🔐 正在登录 Telegram…”)
await client.start(phone=PHONE_NUMBER, password=TWO_STEP_PASSWORD)
print(“✅ 登录成功”)
try:
target = await client.get_entity(TARGET_CHANNEL)
target_name = getattr(target, "title", TARGET_CHANNEL)
except Exception as e:
print(f"❌ 无法获取频道: {e}")
return
print("\n📺 目标频道：", target_name)
if DAYS_LIMIT > 0:
print(f"🕒 时间范围：最近 {DAYS_LIMIT} 天")
elif HOURS_LIMIT > 0:
print(f"🕒 时间范围：最近 {HOURS_LIMIT} 小时")
else:
print("🕒 时间范围：无限制")
if TARGET_VIDEO_COUNT == 0:
print("🎯 视频数量：无限制")
else:
print(f"🎯 视频数量：最多 {TARGET_VIDEO_COUNT} 个")
print("⚙️ 判重策略：size + mime + dc + duration")
print("-" * 40)
seen_keys = set()
duplicates = []
scanned_msgs = 0
found_videos = 0
print("⏳ 扫描中（仅视频）...")
async for msg in client.iter_messages(
target,
limit=None,
filter=InputMessagesFilterVideo
):
scanned_msgs += 1
# ⏰ 时间限制
if time_threshold and msg.date < time_threshold:
print("⏹ 已超出时间范围，停止扫描")
break
is_vid, key, file_name = get_video_info(msg)
if not is_vid:
continue
found_videos += 1
if key in seen_keys:
duplicates.append((msg.id, file_name))
else:
seen_keys.add(key)
# 📊 进度
if found_videos % 20 == 0:
print(f" 已处理 {found_videos} 个视频")
# 🎯 数量限制
if TARGET_VIDEO_COUNT != 0 and found_videos >= TARGET_VIDEO_COUNT:
print("✅ 达到数量限制，停止扫描")
break
print("-" * 40)
print("📊 统计结果")
print(f"扫描消息数：{scanned_msgs}")
print(f"视频数：{found_videos}")
print(f"重复数：{len(duplicates)}")
if not duplicates:
print("✅ 没有重复视频")
return
# 这里会暂停并等待输入
print("✅ 按Y输入重复视频")
confirm = input(f"\n⚠️ 删除 {len(duplicates)} 条重复视频？(y确认): ").strip().lower()
if confirm != "y":
print("🚫 已取消")
return
print("🗑️ 开始删除...")
delete_ids = [d[0] for d in duplicates]
batch_size = 50
for i in range(0, len(delete_ids), batch_size):
batch = delete_ids[i:i + batch_size]
try:
await client.delete_messages(target, batch)
print(f" 已删除 {min(i+batch_size, len(delete_ids))}/{len(delete_ids)}")
await asyncio.sleep(2)
except errors.FloodWaitError as e:
print(f"⏳ FloodWait: 等待 {e.seconds}s")
await asyncio.sleep(e.seconds + 2)
except errors.MessageIdInvalidError:
print("⚠️ 某些消息已不存在，跳过")
except Exception as e:
print(f"❌ 删除失败: {e}")
print("✅ 清理完成")================== 4. 入口 ==================
if name == “main“:
with client:
client.loop.run_until_complete(main())