{"owner":"tangyoha","repo":"telegram_media_downloader","hasSkills":true,"totalSkillsCount":1,"totalTokensCount":2787,"categories":["anthropic-skill"],"hasMcp":false,"mcpConfig":null,"found":[".claude/skills/pyrogram-telegram-bot-techniques/SKILL.md"],"skills":{".claude/skills/pyrogram-telegram-bot-techniques/SKILL.md":"---\nname: pyrogram-telegram-bot-techniques\ndescription: Pyrogram Telegram机器人开发技巧：解析消息链接、自定义Session超时、避免视频黑屏上传、去除消息广告。Use when working with Pyrogram, parsing Telegram message links, uploading videos, or removing advertisement content from messages.\n---\n\n# Pyrogram Telegram 机器人开发技巧\n\n## 1. 解析 Telegram 消息链接\n\n从 Telegram 链接中提取 chat_id、message_id 和 topic_id。\n\n### Link 数据结构\n\n```python\nfrom dataclasses import dataclass\nfrom typing import Optional, Union\nfrom urllib.parse import parse_qs, urlparse\n\n@dataclass\nclass Link:\n    \"\"\"Telegram Link\"\"\"\n    group_id: Union[str, int, None] = None\n    post_id: Optional[int] = None\n    comment_id: Optional[int] = None\n    topic_id: Optional[int] = None\n```\n\n### 提取链接信息\n\n```python\ndef extract_info_from_link(link: str) -> Link:\n    \"\"\"Extract info from link\"\"\"\n    if link in (\"me\", \"self\"):\n        return Link(group_id=link)\n\n    try:\n        u = urlparse(link)\n        paths = [p for p in u.path.split(\"/\") if p]\n        query = parse_qs(u.query)\n    except ValueError:\n        return Link()\n\n    result = Link()\n\n    if \"comment\" in query:\n        result.group_id = paths[0]\n        result.comment_id = int(query[\"comment\"][0])\n    elif len(paths) == 1 and paths[0] != \"c\":\n        result.group_id = paths[0]\n    elif len(paths) == 2:\n        if paths[0] == \"c\":\n            result.group_id = int(f\"-100{paths[1]}\")\n        else:\n            result.group_id = paths[0]\n            result.post_id = int(paths[1])\n    elif len(paths) == 3:\n        if paths[0] == \"c\":\n            result.group_id = int(f\"-100{paths[1]}\")\n            result.post_id = int(paths[2])\n        else:\n            result.group_id = paths[0]\n            result.topic_id = int(paths[1])\n            result.post_id = int(paths[2])\n    elif len(paths) == 4 and paths[0] == \"c\":\n        result.group_id = int(f\"-100{paths[1]}\")\n        result.topic_id = int(paths[2])\n        result.post_id = int(paths[3])\n\n    return result\n```\n\n### 解析链接并处理评论\n\n```python\nasync def parse_link(client: pyrogram.Client, link_str: str):\n    \"\"\"Parse link - handles both regular messages and comments\"\"\"\n    link = extract_info_from_link(link_str)\n    if link.comment_id:\n        # 评论链接需要获取关联群组的 ID\n        chat = await client.get_chat(link.group_id)\n        if chat:\n            return chat.linked_chat.id, link.comment_id, link.topic_id\n\n    return link.group_id, link.post_id, link.topic_id\n```\n\n**支持的链接格式:**\n- `https://t.me/channel_name/123` - 公开频道消息\n- `https://t.me/c/1234567890/123` - 私有频道消息 (自动转换为 -100 格式)\n- `https://t.me/channel_name/topic_id/123` - 话题消息\n- `https://t.me/channel_name/123?comment=456` - 评论消息\n\n---\n\n## 2. 自定义 Session 超时 (HookSession/HookClient)\n\n继承 Pyrogram 的 Session 和 Client 类来自定义连接超时时间。\n\n```python\nimport pyrogram\n\nclass HookSession(pyrogram.session.Session):\n    \"\"\"Hook Session with custom timeout\"\"\"\n\n    def start_timeout(self: pyrogram.session.Session, start_timeout: int):\n        \"\"\"Set the start timeout for the session.\"\"\"\n        self.START_TIMEOUT = start_timeout\n\n\nclass HookClient(pyrogram.Client):\n    \"\"\"Hook Client with custom session timeout\"\"\"\n\n    START_TIME_OUT = 60  # 默认 60 秒\n\n    def __init__(self, name: str, **kwargs):\n        if \"start_timeout\" in kwargs:\n            value = kwargs.get(\"start_timeout\")\n            if value:\n                self.START_TIME_OUT = value\n            kwargs.pop(\"start_timeout\")\n        super().__init__(name, **kwargs)\n\n    async def connect(self) -> bool:\n        if self.is_connected:\n            raise ConnectionError(\"Client is already connected\")\n\n        await self.load_session()\n\n        # 使用自定义的 HookSession\n        self.session = HookSession(\n            self,\n            await self.storage.dc_id(),\n            await self.storage.auth_key(),\n            await self.storage.test_mode(),\n        )\n        self.session.start_timeout(self.START_TIME_OUT)\n\n        await self.session.start()\n        self.is_connected = True\n        return bool(await self.storage.user_id())\n\n    async def start(self):\n        is_authorized = await self.connect()\n        try:\n            if not is_authorized:\n                await self.authorize()\n\n            if not await self.storage.is_bot() and self.takeout:\n                self.takeout_id = (\n                    await self.invoke(\n                        pyrogram.raw.functions.account.InitTakeoutSession()\n                    )\n                ).id\n\n            await self.invoke(pyrogram.raw.functions.updates.GetState())\n        except (Exception, KeyboardInterrupt):\n            await self.disconnect()\n            raise\n        else:\n            self.me = await self.get_me()\n            await self.initialize()\n            return self\n```\n\n**使用方法:**\n\n```python\nclient = HookClient(\n    \"my_bot\",\n    api_id=API_ID,\n    api_hash=API_HASH,\n    start_timeout=120  # 自定义超时时间为 120 秒\n)\n```\n\n---\n\n## 3. 避免视频上传黑屏\n\n上传视频时必须提供缩略图和视频属性，否则视频可能显示为黑屏。\n\n### 下载视频缩略图\n\n```python\nimport os\nimport time\nimport secrets\nimport asyncio\n\nasync def download_thumbnail(\n    client: pyrogram.Client,\n    temp_path: str,\n    message: pyrogram.types.Message,\n):\n    \"\"\"Downloads the thumbnail of a video message to a temporary file.\"\"\"\n    thumbnail_file = None\n    if message.video.thumbs:\n        thumbnail = message.video.thumbs[0] if message.video.thumbs else None\n        unique_name = os.path.join(\n            temp_path,\n            \"thumbnail\",\n            f\"thumb-{int(time.time())}-{secrets.token_hex(8)}.jpg\",\n        )\n\n        max_attempts = 3\n        for attempt in range(1, max_attempts + 1):\n            try:\n                thumbnail_file = await client.download_media(\n                    thumbnail, file_name=unique_name\n                )\n                # 验证文件大小\n                if os.path.getsize(thumbnail_file) == thumbnail.file_size:\n                    break\n\n                raise ValueError(\n                    f\"Thumbnail file size mismatch: {os.path.getsize(thumbnail_file)}\"\n                    f\" vs expected {thumbnail.file_size}\"\n                )\n            except Exception as e:\n                if attempt == max_attempts:\n                    logger.exception(f\"Failed to download thumbnail: {e}\")\n                else:\n                    await asyncio.sleep(2)\n\n    return thumbnail_file\n```\n\n### 上传视频时使用完整属性\n\n```python\n# 关键：必须提供 thumb、width、height、duration\nthumbnail_file = await download_thumbnail(client, temp_save_path, message)\n\nawait message.reply_video(\n    video_file_path,\n    caption=caption,\n    thumb=thumbnail_file,              # 缩略图\n    width=message.video.width,         # 视频宽度\n    height=message.video.height,       # 视频高度\n    duration=message.video.duration,   # 视频时长\n    parse_mode=pyrogram.enums.ParseMode.HTML,\n)\n```\n\n**注意事项:**\n- 没有缩略图会导致视频显示为黑屏\n- 缺少 width/height 会影响视频预览\n- 上传后记得删除临时缩略图文件\n\n---\n\n## 4. 去除消息中的广告\n\n通过分析消息实体(entities)来提取或过滤广告内容，同时保留原有格式。\n\n### MessageProcessor 类\n\n```python\nimport pyrogram\n\ndef get_utf16_length(text: str) -> int:\n    \"\"\"Get UTF-16 length of text (Telegram uses UTF-16 for entity offsets)\"\"\"\n    return len(text.encode('utf-16-le')) // 2\n\nclass MessageProcessor:\n    \"\"\"Helper class for processing message captions and entities.\"\"\"\n\n    def __init__(self, raw_message, filter_str):\n        self.raw_message = raw_message\n        self.raw_caption = raw_message.caption\n        self.filter_str = filter_str\n        self.raw_filter_str = pyrogram.parser.utils.add_surrogates(filter_str)\n        self.raw_caption_str = pyrogram.parser.utils.add_surrogates(raw_message.caption)\n        self.idx = self.raw_caption_str.find(self.raw_filter_str)\n        self.start_offset = self.idx\n        self.end_offset = self.idx + get_utf16_length(filter_str)\n        self.filtered_entities = []\n\n    def process_entities(self):\n        \"\"\"Process and filter message entities.\"\"\"\n        for entity in self.raw_message.caption_entities:\n            cur_start_offset = entity.offset\n            cur_end_offset = entity.offset + entity.length\n\n            # 检查实体是否在过滤范围内\n            if (\n                (cur_start_offset >= self.start_offset and cur_end_offset <= self.end_offset)\n                or (cur_start_offset < self.start_offset and cur_end_offset > self.start_offset)\n                or (cur_start_offset < self.end_offset and cur_end_offset > self.end_offset)\n            ):\n                self.filtered_entities.append(entity)\n\n        self.filtered_entities.sort(key=lambda x: x.offset)\n\n    def get_total_span(self):\n        \"\"\"Calculate the total span for text extraction.\"\"\"\n        if self.filtered_entities:\n            first_entity = self.filtered_entities[0]\n            last_entity = self.filtered_entities[-1]\n            return (\n                min(self.start_offset, first_entity.offset),\n                max(self.end_offset, last_entity.offset + last_entity.length),\n            )\n        return (self.start_offset, self.end_offset)\n\n    def extract_text(self, total_span):\n        \"\"\"Extract and process text with adjusted entity offsets.\"\"\"\n        text = self.raw_caption[total_span[0]:total_span[1]]\n        for entity in self.filtered_entities:\n            entity.offset -= total_span[0]\n        return pyrogram.parser.Parser.unparse(text, self.filtered_entities, True)\n```\n\n### 处理广告替换\n\n```python\nasync def proc_replace_advertisement(message_link: str, filter_str: str):\n    \"\"\"Process and replace advertisement content in a message.\"\"\"\n    chat_id, message_id, _ = await parse_link(client, message_link)\n    raw_message = await client.get_messages(chat_id, message_id)\n\n    processor = MessageProcessor(raw_message, filter_str)\n    processor.process_entities()\n    total_span = processor.get_total_span()\n    return processor.extract_text(total_span)\n```\n\n**工作原理:**\n1. 解析消息链接获取原始消息\n2. 使用 filter_str 定位要保留的文本区域\n3. 提取该区域内的实体(链接、加粗、斜体等)\n4. 重新组装文本，保留格式但去除广告部分\n\n**使用场景:**\n- 转发消息时去除频道水印/广告\n- 保留原始格式(链接、加粗、斜体等)\n- 批量处理时添加到过滤列表\n\n---\n\n## 完整使用示例\n\n```python\nimport pyrogram\n\n# 1. 使用自定义超时的客户端\nclient = HookClient(\n    \"my_bot\",\n    api_id=API_ID,\n    api_hash=API_HASH,\n    start_timeout=120\n)\n\nasync def main():\n    await client.start()\n\n    # 2. 解析消息链接\n    link = \"https://t.me/channel_name/123\"\n    chat_id, message_id, topic_id = await parse_link(client, link)\n\n    # 3. 获取消息并下载视频\n    message = await client.get_messages(chat_id, message_id)\n\n    if message.video:\n        # 下载缩略图避免黑屏\n        thumbnail = await download_thumbnail(client, \"/tmp\", message)\n\n        await client.send_video(\n            chat_id=destination_chat,\n            video=video_path,\n            thumb=thumbnail,\n            width=message.video.width,\n            height=message.video.height,\n            duration=message.video.duration,\n        )\n\n    # 4. 去除广告\n    if message.caption:\n        clean_caption = await proc_replace_advertisement(\n            link,\n            \"想要的内容部分\"\n        )\n        print(clean_caption)\n\n    await client.stop()\n```\n"},"files":{".claude/skills/pyrogram-telegram-bot-techniques/SKILL.md":"---\nname: pyrogram-telegram-bot-techniques\ndescription: Pyrogram Telegram机器人开发技巧：解析消息链接、自定义Session超时、避免视频黑屏上传、去除消息广告。Use when working with Pyrogram, parsing Telegram message links, uploading videos, or removing advertisement content from messages.\n---\n\n# Pyrogram Telegram 机器人开发技巧\n\n## 1. 解析 Telegram 消息链接\n\n从 Telegram 链接中提取 chat_id、message_id 和 topic_id。\n\n### Link 数据结构\n\n```python\nfrom dataclasses import dataclass\nfrom typing import Optional, Union\nfrom urllib.parse import parse_qs, urlparse\n\n@dataclass\nclass Link:\n    \"\"\"Telegram Link\"\"\"\n    group_id: Union[str, int, None] = None\n    post_id: Optional[int] = None\n    comment_id: Optional[int] = None\n    topic_id: Optional[int] = None\n```\n\n### 提取链接信息\n\n```python\ndef extract_info_from_link(link: str) -> Link:\n    \"\"\"Extract info from link\"\"\"\n    if link in (\"me\", \"self\"):\n        return Link(group_id=link)\n\n    try:\n        u = urlparse(link)\n        paths = [p for p in u.path.split(\"/\") if p]\n        query = parse_qs(u.query)\n    except ValueError:\n        return Link()\n\n    result = Link()\n\n    if \"comment\" in query:\n        result.group_id = paths[0]\n        result.comment_id = int(query[\"comment\"][0])\n    elif len(paths) == 1 and paths[0] != \"c\":\n        result.group_id = paths[0]\n    elif len(paths) == 2:\n        if paths[0] == \"c\":\n            result.group_id = int(f\"-100{paths[1]}\")\n        else:\n            result.group_id = paths[0]\n            result.post_id = int(paths[1])\n    elif len(paths) == 3:\n        if paths[0] == \"c\":\n            result.group_id = int(f\"-100{paths[1]}\")\n            result.post_id = int(paths[2])\n        else:\n            result.group_id = paths[0]\n            result.topic_id = int(paths[1])\n            result.post_id = int(paths[2])\n    elif len(paths) == 4 and paths[0] == \"c\":\n        result.group_id = int(f\"-100{paths[1]}\")\n        result.topic_id = int(paths[2])\n        result.post_id = int(paths[3])\n\n    return result\n```\n\n### 解析链接并处理评论\n\n```python\nasync def parse_link(client: pyrogram.Client, link_str: str):\n    \"\"\"Parse link - handles both regular messages and comments\"\"\"\n    link = extract_info_from_link(link_str)\n    if link.comment_id:\n        # 评论链接需要获取关联群组的 ID\n        chat = await client.get_chat(link.group_id)\n        if chat:\n            return chat.linked_chat.id, link.comment_id, link.topic_id\n\n    return link.group_id, link.post_id, link.topic_id\n```\n\n**支持的链接格式:**\n- `https://t.me/channel_name/123` - 公开频道消息\n- `https://t.me/c/1234567890/123` - 私有频道消息 (自动转换为 -100 格式)\n- `https://t.me/channel_name/topic_id/123` - 话题消息\n- `https://t.me/channel_name/123?comment=456` - 评论消息\n\n---\n\n## 2. 自定义 Session 超时 (HookSession/HookClient)\n\n继承 Pyrogram 的 Session 和 Client 类来自定义连接超时时间。\n\n```python\nimport pyrogram\n\nclass HookSession(pyrogram.session.Session):\n    \"\"\"Hook Session with custom timeout\"\"\"\n\n    def start_timeout(self: pyrogram.session.Session, start_timeout: int):\n        \"\"\"Set the start timeout for the session.\"\"\"\n        self.START_TIMEOUT = start_timeout\n\n\nclass HookClient(pyrogram.Client):\n    \"\"\"Hook Client with custom session timeout\"\"\"\n\n    START_TIME_OUT = 60  # 默认 60 秒\n\n    def __init__(self, name: str, **kwargs):\n        if \"start_timeout\" in kwargs:\n            value = kwargs.get(\"start_timeout\")\n            if value:\n                self.START_TIME_OUT = value\n            kwargs.pop(\"start_timeout\")\n        super().__init__(name, **kwargs)\n\n    async def connect(self) -> bool:\n        if self.is_connected:\n            raise ConnectionError(\"Client is already connected\")\n\n        await self.load_session()\n\n        # 使用自定义的 HookSession\n        self.session = HookSession(\n            self,\n            await self.storage.dc_id(),\n            await self.storage.auth_key(),\n            await self.storage.test_mode(),\n        )\n        self.session.start_timeout(self.START_TIME_OUT)\n\n        await self.session.start()\n        self.is_connected = True\n        return bool(await self.storage.user_id())\n\n    async def start(self):\n        is_authorized = await self.connect()\n        try:\n            if not is_authorized:\n                await self.authorize()\n\n            if not await self.storage.is_bot() and self.takeout:\n                self.takeout_id = (\n                    await self.invoke(\n                        pyrogram.raw.functions.account.InitTakeoutSession()\n                    )\n                ).id\n\n            await self.invoke(pyrogram.raw.functions.updates.GetState())\n        except (Exception, KeyboardInterrupt):\n            await self.disconnect()\n            raise\n        else:\n            self.me = await self.get_me()\n            await self.initialize()\n            return self\n```\n\n**使用方法:**\n\n```python\nclient = HookClient(\n    \"my_bot\",\n    api_id=API_ID,\n    api_hash=API_HASH,\n    start_timeout=120  # 自定义超时时间为 120 秒\n)\n```\n\n---\n\n## 3. 避免视频上传黑屏\n\n上传视频时必须提供缩略图和视频属性，否则视频可能显示为黑屏。\n\n### 下载视频缩略图\n\n```python\nimport os\nimport time\nimport secrets\nimport asyncio\n\nasync def download_thumbnail(\n    client: pyrogram.Client,\n    temp_path: str,\n    message: pyrogram.types.Message,\n):\n    \"\"\"Downloads the thumbnail of a video message to a temporary file.\"\"\"\n    thumbnail_file = None\n    if message.video.thumbs:\n        thumbnail = message.video.thumbs[0] if message.video.thumbs else None\n        unique_name = os.path.join(\n            temp_path,\n            \"thumbnail\",\n            f\"thumb-{int(time.time())}-{secrets.token_hex(8)}.jpg\",\n        )\n\n        max_attempts = 3\n        for attempt in range(1, max_attempts + 1):\n            try:\n                thumbnail_file = await client.download_media(\n                    thumbnail, file_name=unique_name\n                )\n                # 验证文件大小\n                if os.path.getsize(thumbnail_file) == thumbnail.file_size:\n                    break\n\n                raise ValueError(\n                    f\"Thumbnail file size mismatch: {os.path.getsize(thumbnail_file)}\"\n                    f\" vs expected {thumbnail.file_size}\"\n                )\n            except Exception as e:\n                if attempt == max_attempts:\n                    logger.exception(f\"Failed to download thumbnail: {e}\")\n                else:\n                    await asyncio.sleep(2)\n\n    return thumbnail_file\n```\n\n### 上传视频时使用完整属性\n\n```python\n# 关键：必须提供 thumb、width、height、duration\nthumbnail_file = await download_thumbnail(client, temp_save_path, message)\n\nawait message.reply_video(\n    video_file_path,\n    caption=caption,\n    thumb=thumbnail_file,              # 缩略图\n    width=message.video.width,         # 视频宽度\n    height=message.video.height,       # 视频高度\n    duration=message.video.duration,   # 视频时长\n    parse_mode=pyrogram.enums.ParseMode.HTML,\n)\n```\n\n**注意事项:**\n- 没有缩略图会导致视频显示为黑屏\n- 缺少 width/height 会影响视频预览\n- 上传后记得删除临时缩略图文件\n\n---\n\n## 4. 去除消息中的广告\n\n通过分析消息实体(entities)来提取或过滤广告内容，同时保留原有格式。\n\n### MessageProcessor 类\n\n```python\nimport pyrogram\n\ndef get_utf16_length(text: str) -> int:\n    \"\"\"Get UTF-16 length of text (Telegram uses UTF-16 for entity offsets)\"\"\"\n    return len(text.encode('utf-16-le')) // 2\n\nclass MessageProcessor:\n    \"\"\"Helper class for processing message captions and entities.\"\"\"\n\n    def __init__(self, raw_message, filter_str):\n        self.raw_message = raw_message\n        self.raw_caption = raw_message.caption\n        self.filter_str = filter_str\n        self.raw_filter_str = pyrogram.parser.utils.add_surrogates(filter_str)\n        self.raw_caption_str = pyrogram.parser.utils.add_surrogates(raw_message.caption)\n        self.idx = self.raw_caption_str.find(self.raw_filter_str)\n        self.start_offset = self.idx\n        self.end_offset = self.idx + get_utf16_length(filter_str)\n        self.filtered_entities = []\n\n    def process_entities(self):\n        \"\"\"Process and filter message entities.\"\"\"\n        for entity in self.raw_message.caption_entities:\n            cur_start_offset = entity.offset\n            cur_end_offset = entity.offset + entity.length\n\n            # 检查实体是否在过滤范围内\n            if (\n                (cur_start_offset >= self.start_offset and cur_end_offset <= self.end_offset)\n                or (cur_start_offset < self.start_offset and cur_end_offset > self.start_offset)\n                or (cur_start_offset < self.end_offset and cur_end_offset > self.end_offset)\n            ):\n                self.filtered_entities.append(entity)\n\n        self.filtered_entities.sort(key=lambda x: x.offset)\n\n    def get_total_span(self):\n        \"\"\"Calculate the total span for text extraction.\"\"\"\n        if self.filtered_entities:\n            first_entity = self.filtered_entities[0]\n            last_entity = self.filtered_entities[-1]\n            return (\n                min(self.start_offset, first_entity.offset),\n                max(self.end_offset, last_entity.offset + last_entity.length),\n            )\n        return (self.start_offset, self.end_offset)\n\n    def extract_text(self, total_span):\n        \"\"\"Extract and process text with adjusted entity offsets.\"\"\"\n        text = self.raw_caption[total_span[0]:total_span[1]]\n        for entity in self.filtered_entities:\n            entity.offset -= total_span[0]\n        return pyrogram.parser.Parser.unparse(text, self.filtered_entities, True)\n```\n\n### 处理广告替换\n\n```python\nasync def proc_replace_advertisement(message_link: str, filter_str: str):\n    \"\"\"Process and replace advertisement content in a message.\"\"\"\n    chat_id, message_id, _ = await parse_link(client, message_link)\n    raw_message = await client.get_messages(chat_id, message_id)\n\n    processor = MessageProcessor(raw_message, filter_str)\n    processor.process_entities()\n    total_span = processor.get_total_span()\n    return processor.extract_text(total_span)\n```\n\n**工作原理:**\n1. 解析消息链接获取原始消息\n2. 使用 filter_str 定位要保留的文本区域\n3. 提取该区域内的实体(链接、加粗、斜体等)\n4. 重新组装文本，保留格式但去除广告部分\n\n**使用场景:**\n- 转发消息时去除频道水印/广告\n- 保留原始格式(链接、加粗、斜体等)\n- 批量处理时添加到过滤列表\n\n---\n\n## 完整使用示例\n\n```python\nimport pyrogram\n\n# 1. 使用自定义超时的客户端\nclient = HookClient(\n    \"my_bot\",\n    api_id=API_ID,\n    api_hash=API_HASH,\n    start_timeout=120\n)\n\nasync def main():\n    await client.start()\n\n    # 2. 解析消息链接\n    link = \"https://t.me/channel_name/123\"\n    chat_id, message_id, topic_id = await parse_link(client, link)\n\n    # 3. 获取消息并下载视频\n    message = await client.get_messages(chat_id, message_id)\n\n    if message.video:\n        # 下载缩略图避免黑屏\n        thumbnail = await download_thumbnail(client, \"/tmp\", message)\n\n        await client.send_video(\n            chat_id=destination_chat,\n            video=video_path,\n            thumb=thumbnail,\n            width=message.video.width,\n            height=message.video.height,\n            duration=message.video.duration,\n        )\n\n    # 4. 去除广告\n    if message.caption:\n        clean_caption = await proc_replace_advertisement(\n            link,\n            \"想要的内容部分\"\n        )\n        print(clean_caption)\n\n    await client.stop()\n```\n"},"items":[{"name":"SKILL.md","path":".claude/skills/pyrogram-telegram-bot-techniques/SKILL.md","rawUrl":"https://raw.githubusercontent.com/tangyoha/telegram_media_downloader/HEAD/.claude/skills/pyrogram-telegram-bot-techniques/SKILL.md","title":"pyrogram-telegram-bot-techniques","category":"anthropic-skill","format":"markdown","content":"---\nname: pyrogram-telegram-bot-techniques\ndescription: Pyrogram Telegram机器人开发技巧：解析消息链接、自定义Session超时、避免视频黑屏上传、去除消息广告。Use when working with Pyrogram, parsing Telegram message links, uploading videos, or removing advertisement content from messages.\n---\n\n# Pyrogram Telegram 机器人开发技巧\n\n## 1. 解析 Telegram 消息链接\n\n从 Telegram 链接中提取 chat_id、message_id 和 topic_id。\n\n### Link 数据结构\n\n```python\nfrom dataclasses import dataclass\nfrom typing import Optional, Union\nfrom urllib.parse import parse_qs, urlparse\n\n@dataclass\nclass Link:\n    \"\"\"Telegram Link\"\"\"\n    group_id: Union[str, int, None] = None\n    post_id: Optional[int] = None\n    comment_id: Optional[int] = None\n    topic_id: Optional[int] = None\n```\n\n### 提取链接信息\n\n```python\ndef extract_info_from_link(link: str) -> Link:\n    \"\"\"Extract info from link\"\"\"\n    if link in (\"me\", \"self\"):\n        return Link(group_id=link)\n\n    try:\n        u = urlparse(link)\n        paths = [p for p in u.path.split(\"/\") if p]\n        query = parse_qs(u.query)\n    except ValueError:\n        return Link()\n\n    result = Link()\n\n    if \"comment\" in query:\n        result.group_id = paths[0]\n        result.comment_id = int(query[\"comment\"][0])\n    elif len(paths) == 1 and paths[0] != \"c\":\n        result.group_id = paths[0]\n    elif len(paths) == 2:\n        if paths[0] == \"c\":\n            result.group_id = int(f\"-100{paths[1]}\")\n        else:\n            result.group_id = paths[0]\n            result.post_id = int(paths[1])\n    elif len(paths) == 3:\n        if paths[0] == \"c\":\n            result.group_id = int(f\"-100{paths[1]}\")\n            result.post_id = int(paths[2])\n        else:\n            result.group_id = paths[0]\n            result.topic_id = int(paths[1])\n            result.post_id = int(paths[2])\n    elif len(paths) == 4 and paths[0] == \"c\":\n        result.group_id = int(f\"-100{paths[1]}\")\n        result.topic_id = int(paths[2])\n        result.post_id = int(paths[3])\n\n    return result\n```\n\n### 解析链接并处理评论\n\n```python\nasync def parse_link(client: pyrogram.Client, link_str: str):\n    \"\"\"Parse link - handles both regular messages and comments\"\"\"\n    link = extract_info_from_link(link_str)\n    if link.comment_id:\n        # 评论链接需要获取关联群组的 ID\n        chat = await client.get_chat(link.group_id)\n        if chat:\n            return chat.linked_chat.id, link.comment_id, link.topic_id\n\n    return link.group_id, link.post_id, link.topic_id\n```\n\n**支持的链接格式:**\n- `https://t.me/channel_name/123` - 公开频道消息\n- `https://t.me/c/1234567890/123` - 私有频道消息 (自动转换为 -100 格式)\n- `https://t.me/channel_name/topic_id/123` - 话题消息\n- `https://t.me/channel_name/123?comment=456` - 评论消息\n\n---\n\n## 2. 自定义 Session 超时 (HookSession/HookClient)\n\n继承 Pyrogram 的 Session 和 Client 类来自定义连接超时时间。\n\n```python\nimport pyrogram\n\nclass HookSession(pyrogram.session.Session):\n    \"\"\"Hook Session with custom timeout\"\"\"\n\n    def start_timeout(self: pyrogram.session.Session, start_timeout: int):\n        \"\"\"Set the start timeout for the session.\"\"\"\n        self.START_TIMEOUT = start_timeout\n\n\nclass HookClient(pyrogram.Client):\n    \"\"\"Hook Client with custom session timeout\"\"\"\n\n    START_TIME_OUT = 60  # 默认 60 秒\n\n    def __init__(self, name: str, **kwargs):\n        if \"start_timeout\" in kwargs:\n            value = kwargs.get(\"start_timeout\")\n            if value:\n                self.START_TIME_OUT = value\n            kwargs.pop(\"start_timeout\")\n        super().__init__(name, **kwargs)\n\n    async def connect(self) -> bool:\n        if self.is_connected:\n            raise ConnectionError(\"Client is already connected\")\n\n        await self.load_session()\n\n        # 使用自定义的 HookSession\n        self.session = HookSession(\n            self,\n            await self.storage.dc_id(),\n            await self.storage.auth_key(),\n            await self.storage.test_mode(),\n        )\n        self.session.start_timeout(self.START_TIME_OUT)\n\n        await self.session.start()\n        self.is_connected = True\n        return bool(await self.storage.user_id())\n\n    async def start(self):\n        is_authorized = await self.connect()\n        try:\n            if not is_authorized:\n                await self.authorize()\n\n            if not await self.storage.is_bot() and self.takeout:\n                self.takeout_id = (\n                    await self.invoke(\n                        pyrogram.raw.functions.account.InitTakeoutSession()\n                    )\n                ).id\n\n            await self.invoke(pyrogram.raw.functions.updates.GetState())\n        except (Exception, KeyboardInterrupt):\n            await self.disconnect()\n            raise\n        else:\n            self.me = await self.get_me()\n            await self.initialize()\n            return self\n```\n\n**使用方法:**\n\n```python\nclient = HookClient(\n    \"my_bot\",\n    api_id=API_ID,\n    api_hash=API_HASH,\n    start_timeout=120  # 自定义超时时间为 120 秒\n)\n```\n\n---\n\n## 3. 避免视频上传黑屏\n\n上传视频时必须提供缩略图和视频属性，否则视频可能显示为黑屏。\n\n### 下载视频缩略图\n\n```python\nimport os\nimport time\nimport secrets\nimport asyncio\n\nasync def download_thumbnail(\n    client: pyrogram.Client,\n    temp_path: str,\n    message: pyrogram.types.Message,\n):\n    \"\"\"Downloads the thumbnail of a video message to a temporary file.\"\"\"\n    thumbnail_file = None\n    if message.video.thumbs:\n        thumbnail = message.video.thumbs[0] if message.video.thumbs else None\n        unique_name = os.path.join(\n            temp_path,\n            \"thumbnail\",\n            f\"thumb-{int(time.time())}-{secrets.token_hex(8)}.jpg\",\n        )\n\n        max_attempts = 3\n        for attempt in range(1, max_attempts + 1):\n            try:\n                thumbnail_file = await client.download_media(\n                    thumbnail, file_name=unique_name\n                )\n                # 验证文件大小\n                if os.path.getsize(thumbnail_file) == thumbnail.file_size:\n                    break\n\n                raise ValueError(\n                    f\"Thumbnail file size mismatch: {os.path.getsize(thumbnail_file)}\"\n                    f\" vs expected {thumbnail.file_size}\"\n                )\n            except Exception as e:\n                if attempt == max_attempts:\n                    logger.exception(f\"Failed to download thumbnail: {e}\")\n                else:\n                    await asyncio.sleep(2)\n\n    return thumbnail_file\n```\n\n### 上传视频时使用完整属性\n\n```python\n# 关键：必须提供 thumb、width、height、duration\nthumbnail_file = await download_thumbnail(client, temp_save_path, message)\n\nawait message.reply_video(\n    video_file_path,\n    caption=caption,\n    thumb=thumbnail_file,              # 缩略图\n    width=message.video.width,         # 视频宽度\n    height=message.video.height,       # 视频高度\n    duration=message.video.duration,   # 视频时长\n    parse_mode=pyrogram.enums.ParseMode.HTML,\n)\n```\n\n**注意事项:**\n- 没有缩略图会导致视频显示为黑屏\n- 缺少 width/height 会影响视频预览\n- 上传后记得删除临时缩略图文件\n\n---\n\n## 4. 去除消息中的广告\n\n通过分析消息实体(entities)来提取或过滤广告内容，同时保留原有格式。\n\n### MessageProcessor 类\n\n```python\nimport pyrogram\n\ndef get_utf16_length(text: str) -> int:\n    \"\"\"Get UTF-16 length of text (Telegram uses UTF-16 for entity offsets)\"\"\"\n    return len(text.encode('utf-16-le')) // 2\n\nclass MessageProcessor:\n    \"\"\"Helper class for processing message captions and entities.\"\"\"\n\n    def __init__(self, raw_message, filter_str):\n        self.raw_message = raw_message\n        self.raw_caption = raw_message.caption\n        self.filter_str = filter_str\n        self.raw_filter_str = pyrogram.parser.utils.add_surrogates(filter_str)\n        self.raw_caption_str = pyrogram.parser.utils.add_surrogates(raw_message.caption)\n        self.idx = self.raw_caption_str.find(self.raw_filter_str)\n        self.start_offset = self.idx\n        self.end_offset = self.idx + get_utf16_length(filter_str)\n        self.filtered_entities = []\n\n    def process_entities(self):\n        \"\"\"Process and filter message entities.\"\"\"\n        for entity in self.raw_message.caption_entities:\n            cur_start_offset = entity.offset\n            cur_end_offset = entity.offset + entity.length\n\n            # 检查实体是否在过滤范围内\n            if (\n                (cur_start_offset >= self.start_offset and cur_end_offset <= self.end_offset)\n                or (cur_start_offset < self.start_offset and cur_end_offset > self.start_offset)\n                or (cur_start_offset < self.end_offset and cur_end_offset > self.end_offset)\n            ):\n                self.filtered_entities.append(entity)\n\n        self.filtered_entities.sort(key=lambda x: x.offset)\n\n    def get_total_span(self):\n        \"\"\"Calculate the total span for text extraction.\"\"\"\n        if self.filtered_entities:\n            first_entity = self.filtered_entities[0]\n            last_entity = self.filtered_entities[-1]\n            return (\n                min(self.start_offset, first_entity.offset),\n                max(self.end_offset, last_entity.offset + last_entity.length),\n            )\n        return (self.start_offset, self.end_offset)\n\n    def extract_text(self, total_span):\n        \"\"\"Extract and process text with adjusted entity offsets.\"\"\"\n        text = self.raw_caption[total_span[0]:total_span[1]]\n        for entity in self.filtered_entities:\n            entity.offset -= total_span[0]\n        return pyrogram.parser.Parser.unparse(text, self.filtered_entities, True)\n```\n\n### 处理广告替换\n\n```python\nasync def proc_replace_advertisement(message_link: str, filter_str: str):\n    \"\"\"Process and replace advertisement content in a message.\"\"\"\n    chat_id, message_id, _ = await parse_link(client, message_link)\n    raw_message = await client.get_messages(chat_id, message_id)\n\n    processor = MessageProcessor(raw_message, filter_str)\n    processor.process_entities()\n    total_span = processor.get_total_span()\n    return processor.extract_text(total_span)\n```\n\n**工作原理:**\n1. 解析消息链接获取原始消息\n2. 使用 filter_str 定位要保留的文本区域\n3. 提取该区域内的实体(链接、加粗、斜体等)\n4. 重新组装文本，保留格式但去除广告部分\n\n**使用场景:**\n- 转发消息时去除频道水印/广告\n- 保留原始格式(链接、加粗、斜体等)\n- 批量处理时添加到过滤列表\n\n---\n\n## 完整使用示例\n\n```python\nimport pyrogram\n\n# 1. 使用自定义超时的客户端\nclient = HookClient(\n    \"my_bot\",\n    api_id=API_ID,\n    api_hash=API_HASH,\n    start_timeout=120\n)\n\nasync def main():\n    await client.start()\n\n    # 2. 解析消息链接\n    link = \"https://t.me/channel_name/123\"\n    chat_id, message_id, topic_id = await parse_link(client, link)\n\n    # 3. 获取消息并下载视频\n    message = await client.get_messages(chat_id, message_id)\n\n    if message.video:\n        # 下载缩略图避免黑屏\n        thumbnail = await download_thumbnail(client, \"/tmp\", message)\n\n        await client.send_video(\n            chat_id=destination_chat,\n            video=video_path,\n            thumb=thumbnail,\n            width=message.video.width,\n            height=message.video.height,\n            duration=message.video.duration,\n        )\n\n    # 4. 去除广告\n    if message.caption:\n        clean_caption = await proc_replace_advertisement(\n            link,\n            \"想要的内容部分\"\n        )\n        print(clean_caption)\n\n    await client.stop()\n```\n","frontmatter":{"name":"pyrogram-telegram-bot-techniques","description":"Pyrogram Telegram机器人开发技巧：解析消息链接、自定义Session超时、避免视频黑屏上传、去除消息广告。Use when working with Pyrogram, parsing Telegram message links, uploading videos, or removing advertisement content from messages."},"isInternal":false,"tokens":2787,"sizeBytes":11847}],"systemPromptSnippet":"<yakaai_skills repo=\"tangyoha/telegram_media_downloader\">\n<!-- File: .claude/skills/pyrogram-telegram-bot-techniques/SKILL.md (Tokens: ~2787 | Category: anthropic-skill) -->\n---\nname: pyrogram-telegram-bot-techniques\ndescription: Pyrogram Telegram机器人开发技巧：解析消息链接、自定义Session超时、避免视频黑屏上传、去除消息广告。Use when working with Pyrogram, parsing Telegram message links, uploading videos, or removing advertisement content from messages.\n---\n\n# Pyrogram Telegram 机器人开发技巧\n\n## 1. 解析 Telegram 消息链接\n\n从 Telegram 链接中提取 chat_id、message_id 和 topic_id。\n\n### Link 数据结构\n\n```python\nfrom dataclasses import dataclass\nfrom typing import Optional, Union\nfrom urllib.parse import parse_qs, urlparse\n\n@dataclass\nclass Link:\n    \"\"\"Telegram Link\"\"\"\n    group_id: Union[str, int, None] = None\n    post_id: Optional[int] = None\n    comment_id: Optional[int] = None\n    topic_id: Optional[int] = None\n```\n\n### 提取链接信息\n\n```python\ndef extract_info_from_link(link: str) -> Link:\n    \"\"\"Extract info from link\"\"\"\n    if link in (\"me\", \"self\"):\n        return Link(group_id=link)\n\n    try:\n        u = urlparse(link)\n        paths = [p for p in u.path.split(\"/\") if p]\n        query = parse_qs(u.query)\n    except ValueError:\n        return Link()\n\n    result = Link()\n\n    if \"comment\" in query:\n        result.group_id = paths[0]\n        result.comment_id = int(query[\"comment\"][0])\n    elif len(paths) == 1 and paths[0] != \"c\":\n        result.group_id = paths[0]\n    elif len(paths) == 2:\n        if paths[0] == \"c\":\n            result.group_id = int(f\"-100{paths[1]}\")\n        else:\n            result.group_id = paths[0]\n            result.post_id = int(paths[1])\n    elif len(paths) == 3:\n        if paths[0] == \"c\":\n            result.group_id = int(f\"-100{paths[1]}\")\n            result.post_id = int(paths[2])\n        else:\n            result.group_id = paths[0]\n            result.topic_id = int(paths[1])\n            result.post_id = int(paths[2])\n    elif len(paths) == 4 and paths[0] == \"c\":\n        result.group_id = int(f\"-100{paths[1]}\")\n        result.topic_id = int(paths[2])\n        result.post_id = int(paths[3])\n\n    return result\n```\n\n### 解析链接并处理评论\n\n```python\nasync def parse_link(client: pyrogram.Client, link_str: str):\n    \"\"\"Parse link - handles both regular messages and comments\"\"\"\n    link = extract_info_from_link(link_str)\n    if link.comment_id:\n        # 评论链接需要获取关联群组的 ID\n        chat = await client.get_chat(link.group_id)\n        if chat:\n            return chat.linked_chat.id, link.comment_id, link.topic_id\n\n    return link.group_id, link.post_id, link.topic_id\n```\n\n**支持的链接格式:**\n- `https://t.me/channel_name/123` - 公开频道消息\n- `https://t.me/c/1234567890/123` - 私有频道消息 (自动转换为 -100 格式)\n- `https://t.me/channel_name/topic_id/123` - 话题消息\n- `https://t.me/channel_name/123?comment=456` - 评论消息\n\n---\n\n## 2. 自定义 Session 超时 (HookSession/HookClient)\n\n继承 Pyrogram 的 Session 和 Client 类来自定义连接超时时间。\n\n```python\nimport pyrogram\n\nclass HookSession(pyrogram.session.Session):\n    \"\"\"Hook Session with custom timeout\"\"\"\n\n    def start_timeout(self: pyrogram.session.Session, start_timeout: int):\n        \"\"\"Set the start timeout for the session.\"\"\"\n        self.START_TIMEOUT = start_timeout\n\n\nclass HookClient(pyrogram.Client):\n    \"\"\"Hook Client with custom session timeout\"\"\"\n\n    START_TIME_OUT = 60  # 默认 60 秒\n\n    def __init__(self, name: str, **kwargs):\n        if \"start_timeout\" in kwargs:\n            value = kwargs.get(\"start_timeout\")\n            if value:\n                self.START_TIME_OUT = value\n            kwargs.pop(\"start_timeout\")\n        super().__init__(name, **kwargs)\n\n    async def connect(self) -> bool:\n        if self.is_connected:\n            raise ConnectionError(\"Client is already connected\")\n\n        await self.load_session()\n\n        # 使用自定义的 HookSession\n        self.session = HookSession(\n            self,\n            await self.storage.dc_id(),\n            await self.storage.auth_key(),\n            await self.storage.test_mode(),\n        )\n        self.session.start_timeout(self.START_TIME_OUT)\n\n        await self.session.start()\n        self.is_connected = True\n        return bool(await self.storage.user_id())\n\n    async def start(self):\n        is_authorized = await self.connect()\n        try:\n            if not is_authorized:\n                await self.authorize()\n\n            if not await self.storage.is_bot() and self.takeout:\n                self.takeout_id = (\n                    await self.invoke(\n                        pyrogram.raw.functions.account.InitTakeoutSession()\n                    )\n                ).id\n\n            await self.invoke(pyrogram.raw.functions.updates.GetState())\n        except (Exception, KeyboardInterrupt):\n            await self.disconnect()\n            raise\n        else:\n            self.me = await self.get_me()\n            await self.initialize()\n            return self\n```\n\n**使用方法:**\n\n```python\nclient = HookClient(\n    \"my_bot\",\n    api_id=API_ID,\n    api_hash=API_HASH,\n    start_timeout=120  # 自定义超时时间为 120 秒\n)\n```\n\n---\n\n## 3. 避免视频上传黑屏\n\n上传视频时必须提供缩略图和视频属性，否则视频可能显示为黑屏。\n\n### 下载视频缩略图\n\n```python\nimport os\nimport time\nimport secrets\nimport asyncio\n\nasync def download_thumbnail(\n    client: pyrogram.Client,\n    temp_path: str,\n    message: pyrogram.types.Message,\n):\n    \"\"\"Downloads the thumbnail of a video message to a temporary file.\"\"\"\n    thumbnail_file = None\n    if message.video.thumbs:\n        thumbnail = message.video.thumbs[0] if message.video.thumbs else None\n        unique_name = os.path.join(\n            temp_path,\n            \"thumbnail\",\n            f\"thumb-{int(time.time())}-{secrets.token_hex(8)}.jpg\",\n        )\n\n        max_attempts = 3\n        for attempt in range(1, max_attempts + 1):\n            try:\n                thumbnail_file = await client.download_media(\n                    thumbnail, file_name=unique_name\n                )\n                # 验证文件大小\n                if os.path.getsize(thumbnail_file) == thumbnail.file_size:\n                    break\n\n                raise ValueError(\n                    f\"Thumbnail file size mismatch: {os.path.getsize(thumbnail_file)}\"\n                    f\" vs expected {thumbnail.file_size}\"\n                )\n            except Exception as e:\n                if attempt == max_attempts:\n                    logger.exception(f\"Failed to download thumbnail: {e}\")\n                else:\n                    await asyncio.sleep(2)\n\n    return thumbnail_file\n```\n\n### 上传视频时使用完整属性\n\n```python\n# 关键：必须提供 thumb、width、height、duration\nthumbnail_file = await download_thumbnail(client, temp_save_path, message)\n\nawait message.reply_video(\n    video_file_path,\n    caption=caption,\n    thumb=thumbnail_file,              # 缩略图\n    width=message.video.width,         # 视频宽度\n    height=message.video.height,       # 视频高度\n    duration=message.video.duration,   # 视频时长\n    parse_mode=pyrogram.enums.ParseMode.HTML,\n)\n```\n\n**注意事项:**\n- 没有缩略图会导致视频显示为黑屏\n- 缺少 width/height 会影响视频预览\n- 上传后记得删除临时缩略图文件\n\n---\n\n## 4. 去除消息中的广告\n\n通过分析消息实体(entities)来提取或过滤广告内容，同时保留原有格式。\n\n### MessageProcessor 类\n\n```python\nimport pyrogram\n\ndef get_utf16_length(text: str) -> int:\n    \"\"\"Get UTF-16 length of text (Telegram uses UTF-16 for entity offsets)\"\"\"\n    return len(text.encode('utf-16-le')) // 2\n\nclass MessageProcessor:\n    \"\"\"Helper class for processing message captions and entities.\"\"\"\n\n    def __init__(self, raw_message, filter_str):\n        self.raw_message = raw_message\n        self.raw_caption = raw_message.caption\n        self.filter_str = filter_str\n        self.raw_filter_str = pyrogram.parser.utils.add_surrogates(filter_str)\n        self.raw_caption_str = pyrogram.parser.utils.add_surrogates(raw_message.caption)\n        self.idx = self.raw_caption_str.find(self.raw_filter_str)\n        self.start_offset = self.idx\n        self.end_offset = self.idx + get_utf16_length(filter_str)\n        self.filtered_entities = []\n\n    def process_entities(self):\n        \"\"\"Process and filter message entities.\"\"\"\n        for entity in self.raw_message.caption_entities:\n            cur_start_offset = entity.offset\n            cur_end_offset = entity.offset + entity.length\n\n            # 检查实体是否在过滤范围内\n            if (\n                (cur_start_offset >= self.start_offset and cur_end_offset <= self.end_offset)\n                or (cur_start_offset < self.start_offset and cur_end_offset > self.start_offset)\n                or (cur_start_offset < self.end_offset and cur_end_offset > self.end_offset)\n            ):\n                self.filtered_entities.append(entity)\n\n        self.filtered_entities.sort(key=lambda x: x.offset)\n\n    def get_total_span(self):\n        \"\"\"Calculate the total span for text extraction.\"\"\"\n        if self.filtered_entities:\n            first_entity = self.filtered_entities[0]\n            last_entity = self.filtered_entities[-1]\n            return (\n                min(self.start_offset, first_entity.offset),\n                max(self.end_offset, last_entity.offset + last_entity.length),\n            )\n        return (self.start_offset, self.end_offset)\n\n    def extract_text(self, total_span):\n        \"\"\"Extract and process text with adjusted entity offsets.\"\"\"\n        text = self.raw_caption[total_span[0]:total_span[1]]\n        for entity in self.filtered_entities:\n            entity.offset -= total_span[0]\n        return pyrogram.parser.Parser.unparse(text, self.filtered_entities, True)\n```\n\n### 处理广告替换\n\n```python\nasync def proc_replace_advertisement(message_link: str, filter_str: str):\n    \"\"\"Process and replace advertisement content in a message.\"\"\"\n    chat_id, message_id, _ = await parse_link(client, message_link)\n    raw_message = await client.get_messages(chat_id, message_id)\n\n    processor = MessageProcessor(raw_message, filter_str)\n    processor.process_entities()\n    total_span = processor.get_total_span()\n    return processor.extract_text(total_span)\n```\n\n**工作原理:**\n1. 解析消息链接获取原始消息\n2. 使用 filter_str 定位要保留的文本区域\n3. 提取该区域内的实体(链接、加粗、斜体等)\n4. 重新组装文本，保留格式但去除广告部分\n\n**使用场景:**\n- 转发消息时去除频道水印/广告\n- 保留原始格式(链接、加粗、斜体等)\n- 批量处理时添加到过滤列表\n\n---\n\n## 完整使用示例\n\n```python\nimport pyrogram\n\n# 1. 使用自定义超时的客户端\nclient = HookClient(\n    \"my_bot\",\n    api_id=API_ID,\n    api_hash=API_HASH,\n    start_timeout=120\n)\n\nasync def main():\n    await client.start()\n\n    # 2. 解析消息链接\n    link = \"https://t.me/channel_name/123\"\n    chat_id, message_id, topic_id = await parse_link(client, link)\n\n    # 3. 获取消息并下载视频\n    message = await client.get_messages(chat_id, message_id)\n\n    if message.video:\n        # 下载缩略图避免黑屏\n        thumbnail = await download_thumbnail(client, \"/tmp\", message)\n\n        await client.send_video(\n            chat_id=destination_chat,\n            video=video_path,\n            thumb=thumbnail,\n            width=message.video.width,\n            height=message.video.height,\n            duration=message.video.duration,\n        )\n\n    # 4. 去除广告\n    if message.caption:\n        clean_caption = await proc_replace_advertisement(\n            link,\n            \"想要的内容部分\"\n        )\n        print(clean_caption)\n\n    await client.stop()\n```\n</yakaai_skills>"}