1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
| import asyncio
import os
import random
import sqlite3
import re
import logging
from logging.handlers import RotatingFileHandler
from datetime import datetime, timezone
from telethon import TelegramClient, events
from telethon.tl.types import (
MessageMediaDocument,
DocumentAttributeVideo
)
from telethon.errors import (
FloodWaitError,
ChatForwardsRestrictedError,
SecurityError
)
from dotenv import load_dotenv
# ==================== 日志 ====================
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[
RotatingFileHandler(
"bot.log",
maxBytes=5 * 1024 * 1024,
backupCount=5,
encoding="utf-8"
),
logging.StreamHandler()
]
)
logger = logging.getLogger("TGForwardBot")
START_TIME = datetime.now(timezone.utc)
# ==================== 数据库 ====================
class DBManager:
def __init__(self, path):
self.conn = sqlite3.connect(
path,
check_same_thread=False,
isolation_level=None,
timeout=30
)
self.cursor = self.conn.cursor()
self.lock = asyncio.Lock()
self._init()
def _init(self):
self.cursor.execute("PRAGMA journal_mode=WAL;")
self.cursor.execute("PRAGMA synchronous=NORMAL;")
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS video_keys (
video_key TEXT PRIMARY KEY,
target_msg_id INTEGER,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS channel_progress (
channel_id TEXT PRIMARY KEY,
last_msg_id INTEGER
)
""")
self.cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_video_time
ON video_keys(timestamp)
""")
async def video_exists(self, key):
async with self.lock:
self.cursor.execute(
"SELECT 1 FROM video_keys WHERE video_key=?",
(key,)
)
return self.cursor.fetchone() is not None
async def add_video(self, key, msg_id):
async with self.lock:
self.cursor.execute(
"INSERT OR REPLACE INTO video_keys VALUES (?, ?, CURRENT_TIMESTAMP)",
(key, msg_id)
)
async def get_progress(self, cid):
async with self.lock:
self.cursor.execute(
"SELECT last_msg_id FROM channel_progress WHERE channel_id=?",
(cid,)
)
r = self.cursor.fetchone()
return r[0] if r else 0
async def update_progress(self, cid, msg_id):
async with self.lock:
self.cursor.execute(
"INSERT OR REPLACE INTO channel_progress VALUES (?, ?)",
(cid, msg_id)
)
async def today_count(self):
async with self.lock:
self.cursor.execute("""
SELECT COUNT(*) FROM video_keys
WHERE timestamp >= datetime('now','localtime','start of day')
""")
return self.cursor.fetchone()[0]
# ==================== 配置 ====================
load_dotenv()
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")
TARGET_CHANNEL = os.getenv("TARGET_CHANNEL")
ADMIN_ID = int(os.getenv("ADMIN_ID", 0))
def parse_channel(v):
v = v.strip()
return int(v) if v.lstrip("-").isdigit() else v
SOURCE_CHANNELS = [
parse_channel(x)
for x in os.getenv("SOURCE_CHANNELS", "").split(",")
if x.strip()
]
SCAN_LIMIT = int(os.getenv("SCAN_LIMIT", 50))
MAX_CAPTION_LENGTH = int(os.getenv("MAX_CAPTION_LENGTH", 1024))
MIN_FILE_SIZE = int(os.getenv("MIN_FILE_SIZE", 0))
MAX_FILE_SIZE = int(os.getenv("MAX_FILE_SIZE", 0))
MIN_DURATION = int(os.getenv("MIN_DURATION", 0))
MAX_DURATION = int(os.getenv("MAX_DURATION", 0))
MIN_WIDTH = int(os.getenv("MIN_WIDTH", 0))
MIN_HEIGHT = int(os.getenv("MIN_HEIGHT", 0))
# ==================== 客户端 & 队列 ====================
client = TelegramClient("user_session", api_id, api_hash)
forward_queue = asyncio.Queue(maxsize=30)
# ==================== 文本清洗 ====================
AD_PATTERNS = [
r"Поддержать\s+проект[:：]?\s*\n?\s*(?:https?://)?t\.me/boost/\S+",
r"@\w+",
r"https?://\S+",
r"t\.me/\S+",
r"(?:VX|wx|微信)[：:]?\s*\w+",
r"(?:群|频道)[：:]?\s*@\w+",
r"加入频道|点击关注|更多资源",
]
def clean_caption(text: str) -> str:
if not text:
return ""
for p in AD_PATTERNS:
text = re.sub(p, "", text, flags=re.I)
text = re.sub(r"\n\s*\n+", "\n", text).strip()
return text[:MAX_CAPTION_LENGTH] if len(text) >= 5 else ""
# ==================== 视频判断 ====================
def is_video_ok(msg):
if not msg.media or not isinstance(msg.media, MessageMediaDocument):
return False
doc = msg.media.document
if not doc.mime_type or not doc.mime_type.startswith("video"):
return False
if MIN_FILE_SIZE and doc.size < MIN_FILE_SIZE:
return False
if MAX_FILE_SIZE and doc.size > MAX_FILE_SIZE:
return False
v = next((a for a in doc.attributes if isinstance(a, DocumentAttributeVideo)), None)
if not v or getattr(v, "round_message", False):
return False
if MIN_DURATION and v.duration < MIN_DURATION:
return False
if MAX_DURATION and v.duration > MAX_DURATION:
return False
if MIN_WIDTH and v.w < MIN_WIDTH:
return False
if MIN_HEIGHT and v.h < MIN_HEIGHT:
return False
return True
# ==================== 核心：源头级去重 Key ====================
def video_key(msg):
"""
Telethon 下最稳定的视频唯一标识
同一视频在任何频道 / 转发 / 重命名中都一致
"""
return str(msg.media.document.id)
# ==================== 数据库初始化 ====================
db = None
async def init_db_by_target():
global db
entity = await client.get_entity(TARGET_CHANNEL)
DB_FILE = f"bot_data_target_{entity.id}.db"
logger.info(f"🗂 使用数据库 {DB_FILE}")
db = DBManager(DB_FILE)
# ==================== Worker ====================
async def worker(wid: int):
logger.info(f"👷 Worker-{wid} 启动")
while True:
msg, key, cid = await forward_queue.get()
success = False
try:
refreshed = await client.get_messages(int(cid), ids=msg.id)
if not refreshed or not is_video_ok(refreshed):
continue
caption = clean_caption(refreshed.message or "")
sent = await client.send_file(
TARGET_CHANNEL,
refreshed.media,
caption=caption,
silent=True
)
await db.add_video(key, sent.id)
success = True
logger.info(f"✅ Worker-{wid} 转发 {msg.id}")
await asyncio.sleep(random.uniform(2, 5))
except FloodWaitError as e:
wait = e.seconds + random.uniform(3, 8)
logger.warning(f"⏳ FloodWait {wait:.1f}s")
await asyncio.sleep(wait)
except (ChatForwardsRestrictedError, SecurityError):
logger.error(f"⛔ 内容保护 {msg.id}")
except Exception as e:
logger.error(f"❌ Worker-{wid} 错误 {e}", exc_info=True)
finally:
if success:
await db.update_progress(cid, msg.id)
forward_queue.task_done()
async def safe_worker(i):
while True:
try:
await worker(i)
except Exception:
logger.critical(f"🔥 Worker-{i} 崩溃，重启", exc_info=True)
await asyncio.sleep(5)
# ==================== 扫描 ====================
async def scan_channel(ch, sem):
async with sem:
try:
entity = await client.get_entity(ch)
cid = str(entity.id)
last = max(0, (await db.get_progress(cid)) - 1)
logger.info(f"🔍 扫描 {ch} from {last}")
async for msg in client.iter_messages(
entity,
min_id=last,
limit=None if SCAN_LIMIT == 0 else SCAN_LIMIT,
reverse=True
):
if not is_video_ok(msg):
continue
key = video_key(msg)
if await db.video_exists(key):
continue
await forward_queue.put((msg, key, cid))
logger.info(f"📥 入队 {msg.id}")
except Exception as e:
logger.error(f"❌ 扫描失败 {ch}: {e}")
# ==================== 实时监听 ====================
@client.on(events.NewMessage(chats=SOURCE_CHANNELS))
async def realtime(event):
msg = event.message
if not is_video_ok(msg):
return
key = video_key(msg)
if await db.video_exists(key):
return
await forward_queue.put((msg, key, str(event.chat_id)))
logger.info(f"⚡ 实时入队 {msg.id}")
# ==================== 状态 ====================
@client.on(events.NewMessage(pattern="/status"))
async def status(event):
if ADMIN_ID and event.sender_id != ADMIN_ID:
return
up = datetime.now(timezone.utc) - START_TIME
await event.reply(
f"🤖 运行中\n"
f"🎯 Target: {TARGET_CHANNEL}\n"
f"⏱ {str(up).split('.')[0]}\n"
f"📊 今日 {await db.today_count()}\n"
f"📥 队列 {forward_queue.qsize()}"
)
# ==================== 主入口 ====================
async def main():
await client.start(PHONE_NUMBER, TWO_STEP_PASSWORD)
logger.info("🚀 启动成功")
await init_db_by_target()
for i in range(2):
asyncio.create_task(safe_worker(i + 1))
sem = asyncio.Semaphore(1)
await asyncio.gather(*(scan_channel(c, sem) for c in SOURCE_CHANNELS))
await client.run_until_disconnected()
if __name__ == "__main__":
with client:
client.loop.run_until_complete(main())
|
