Update forwarder.py

forwarder.py

@@ -45,26 +45,20 @@ def extract_slug_from_url(url):
     match = re.search(r"https?://hdhive\.com/resource/(?:[^/]+/)?([a-f0-9]{32})", url, re.IGNORECASE)
     if match:
         return match.group(1)
-    if re.fullmatch(r"[a-f0-9]{32}", url, re.IGNORECASE):
-        return url
     return None
 
-def get_slug_and_target_entity(message):
-    """返回 (slug, 目标实体)，目标实体是 URL 中包含 HDHive slug 的实体"""
+def get_slug_from_message(message):
     if not message.entities:
-        return None, None
+        return None
     for entity in message.entities:
         url = None
         if hasattr(entity, 'url'):
             url = entity.url
-        elif isinstance(entity, MessageEntityUrl) and message.text:
-            url = message.text[entity.offset:entity.offset+entity.length]
         if url:
             slug = extract_slug_from_url(url)
             if slug:
-                print(f"找到目标实体，URL: {url}, slug: {slug}")
-                return slug, entity
-    return None, None
+                return slug
+    return None
 
 def get_direct_link(slug):
     if slug in slug_cache:
@@ -74,69 +68,160 @@ def get_direct_link(slug):
     if UNLOCK_PASSWORD:
         payload["password"] = UNLOCK_PASSWORD
     try:
-        resp = requests.post(url, json=payload, timeout=10)
+        resp = requests.post(url, json=payload, timeout=15)
         if resp.status_code == 200:
             data = resp.json()
             if data.get("success") and data.get("link"):
                 direct_link = data["link"]
                 slug_cache[slug] = direct_link
                 return direct_link
-            elif data.get("need_password"):
-                print(f"资源 {slug} 需要密码但未提供或密码错误")
-            else:
-                print(f"解锁失败: {data.get('message')}")
-        else:
-            print(f"API 请求失败: {resp.status_code}")
     except Exception as e:
-        print(f"调用解锁 API 出错: {e}")
+        print(f"解锁API错误: {e}")
     return None
 
 async def send_modified_message(client, dest_entity, original_msg):
     if isinstance(original_msg, MessageService):
-        print(f"跳过服务消息 {original_msg.id}")
         return
 
     if not original_msg.text:
         await client.forward_messages(dest_entity, original_msg)
-        print(f"直接转发无文本消息 {original_msg.id}")
         return
 
-    slug, target_entity = get_slug_and_target_entity(original_msg)
+    slug = get_slug_from_message(original_msg)
     if not slug:
         await client.forward_messages(dest_entity, original_msg)
-        print(f"直接转发消息 {original_msg.id}（无 HDHive 链接）")
         return
 
     direct_link = get_direct_link(slug)
     if not direct_link:
         await client.forward_messages(dest_entity, original_msg)
-        print(f"直接转发消息 {original_msg.id}（解锁失败）")
         return
 
-    # 构造新文本：只替换目标实体对应的文本（例如“直达链接”）
-    text = original_msg.text
-    parts = []
-    last = 0
-    for entity in original_msg.entities:
-        start = entity.offset
-        end = start + entity.length
-        if start > last:
-            parts.append(text[last:start])
-        if entity == target_entity:
-            # 获取原实体文本（如“直达链接”）
-            original_text = text[start:end]
-            print(f"替换文本: '{original_text}' -> {direct_link}")
-            parts.append(direct_link)
-        else:
-            parts.append(text[start:end])
-        last = end
-    if last < len(text):
-        parts.append(text[last:])
-    new_text = ''.join(parts)
+    # 直接替换文本中的所有链接为直链
+    new_text = original_msg.text
+    if original_msg.entities:
+        # 从后往前替换，避免偏移问题
+        for entity in reversed(original_msg.entities):
+            url = None
+            if hasattr(entity, 'url'):
+                url = entity.url
+            if url and extract_slug_from_url(url) == slug:
+                start = entity.offset
+                end = start + entity.length
+                new_text = new_text[:start] + direct_link + new_text[end:]
 
     await client.send_message(dest_entity, new_text)
-    print(f"已发送替换后的消息 {original_msg.id}")
 
-# 以下 forward_history, forward_new_messages, get_recent_verification_messages, main 等函数保持不变（与之前提供的完整版一致，这里为了简洁省略，但实际代码中必须包含）
-# 请确保您的代码中包含这些函数，我已经在之前的消息中提供过完整版本。
-# 由于消息长度限制，此处仅给出核心修改，实际部署时请使用我在上一条消息中给出的完整 forwarder.py 并替换其中的 send_modified_message 函数。
+async def forward_history(client, source_entity, dest_entity, limit):
+    print(f"处理历史消息: {source_entity.id}")
+    messages = []
+    async for msg in client.iter_messages(source_entity, limit=limit, reverse=True):
+        messages.append(msg)
+    if not messages:
+        return None
+    for msg in messages:
+        await send_modified_message(client, dest_entity, msg)
+        await asyncio.sleep(0.3)
+    return messages[-1].id
+
+async def forward_new_messages(client, source_entity, dest_entity, last_id):
+    try:
+        async for msg in client.iter_messages(source_entity, limit=1):
+            latest = msg
+            break
+        else:
+            return last_id
+        if latest.id > last_id:
+            new_msgs = []
+            async for msg in client.iter_messages(source_entity, min_id=last_id, max_id=latest.id, reverse=True):
+                new_msgs.append(msg)
+            for msg in new_msgs:
+                await send_modified_message(client, dest_entity, msg)
+                await asyncio.sleep(0.3)
+            return latest.id
+        return last_id
+    except Exception as e:
+        print(f"轮询错误: {e}")
+        return last_id
+
+async def main():
+    print("等待60秒...")
+    await asyncio.sleep(60)
+
+    client = TelegramClient(StringSession(SESSION_STRING), API_ID, API_HASH)
+    await client.start()
+    print("连接成功")
+
+    sources = []
+    for src in SOURCES:
+        try:
+            entity = await client.get_entity(src)
+            sources.append(entity)
+            print(f"加载源: {src}")
+        except Exception as e:
+            print(f"加载失败 {src}: {e}")
+
+    if not sources:
+        print("无可用源")
+        return
+
+    dest = await client.get_entity(DESTINATION)
+    print(f"目标: {DESTINATION}")
+
+    last_ids = await load_last_ids()
+
+    if HISTORY_LIMIT > 0:
+        print(f"转发历史 ({HISTORY_LIMIT}条/源)...")
+        for entity in sources:
+            key = str(entity.id)
+            new_id = await forward_history(client, entity, dest, HISTORY_LIMIT)
+            if new_id:
+                last_ids[key] = new_id
+                await save_last_ids(last_ids)
+        print("历史转发完成")
+
+    for entity in sources:
+        try:
+            @client.on(events.NewMessage(chats=entity))
+            async def handler(event):
+                await send_modified_message(client, dest, event.message)
+            print(f"实时监听: {entity.id}")
+        except Exception as e:
+            print(f"无法监听 {entity.id}: {e}")
+            polling_sources.append(entity)
+
+    for entity in polling_sources:
+        key = str(entity.id)
+        if key not in last_ids:
+            async for msg in client.iter_messages(entity, limit=1):
+                last_ids[key] = msg.id if msg else 0
+                break
+            else:
+                last_ids[key] = 0
+    await save_last_ids(last_ids)
+
+    async def poll():
+        while True:
+            for entity in polling_sources:
+                key = str(entity.id)
+                last = last_ids.get(key, 0)
+                new_last = await forward_new_messages(client, entity, dest, last)
+                if new_last != last:
+                    last_ids[key] = new_last
+                    await save_last_ids(last_ids)
+            await asyncio.sleep(POLL_INTERVAL)
+
+    if polling_sources:
+        asyncio.create_task(poll())
+        print(f"轮询启动, 间隔{POLL_INTERVAL}秒")
+
+    print("监听运行中...")
+    await client.run_until_disconnected()
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print("停止")
+    except Exception as e:
+        print(f"错误: {e}")