如何通过转发元数据（Forwarded From）用 Python 绘制电报频道传播拓扑图

当一条消息在多个 Telegram 频道之间被连续转发时，肉眼通常只能看到当前消息的“Forwarded From”来源，却很难判断内容最早出现在哪里、经过哪些频道扩散，以及哪些节点承担了主要传播作用。

借助 Telegram API、Python 与图分析工具，我们可以提取公开消息中的转发元数据，建立频道之间的有向关系，并绘制一张可交互或可导出的传播拓扑图。

本文使用 Telethon 获取已获授权访问的频道数据，并通过 NetworkX 与 PyVis 完成建图和可视化。整个流程适合公开信息研究、品牌传播复盘和社区内容分析，但不应用于绕过权限、追踪私人身份或批量收集非公开数据。

🧭 先理解 Forwarded From 能告诉你什么

Telegram 消息的转发来源并不是普通文本，而是服务端返回的结构化元数据。在 Telethon 中，它通常可以从消息对象的 fwd_from 属性读取。

如果消息来自公开频道，元数据中可能包含来源频道、原消息编号、原发布时间和作者署名。需要注意的是，Telegram 记录的通常是当前消息声明的直接来源，不一定保留完整的多级转发链。

message.fwd_from
message.fwd_from.from_id
message.fwd_from.channel_post
message.fwd_from.date
message.fwd_from.from_name

例如，频道 C 中的消息显示转发自频道 B，并不代表 API 一定会同时告诉你 B 的消息是否来自频道 A。要还原更长的传播路径，需要分别采集 A、B、C 的相关公开消息，再根据频道标识、原帖编号、时间和内容特征进行连接。

此外，开启内容保护、隐藏发送者、来源已删除或访问权限不足时，转发字段可能为空或只剩文字署名。因此，拓扑图应被视为“可观察传播关系”，而不是完整的信息传播真相。

🛠️ 准备 Telegram API 与 Python 环境

首先登录 Telegram 官方开发者后台，在 API development tools 页面创建应用，获得 api_id 和 api_hash。这两个参数属于敏感凭据，不要上传到 GitHub，也不要写入公开网页。

python -m venv .venv
source .venv/bin/activate
pip install telethon networkx pyvis pandas python-dotenv

Windows PowerShell 可使用 .venv\Scripts\Activate.ps1 激活环境。建议通过环境变量管理凭据，避免在分析脚本中留下可直接登录账号的信息。

TELEGRAM_API_ID=12345678
TELEGRAM_API_HASH=your_api_hash
TELEGRAM_SESSION=topology_session

首次运行 Telethon 时，客户端会要求输入手机号、验证码以及可能存在的两步验证密码。生成的 session 文件相当于登录凭证，应加入 .gitignore 并妥善保存。

📥 提取频道消息与转发关系

拓扑图可以抽象成一个有向图：来源频道是起点，当前频道是终点，每次可识别转发形成一条边。多次发生相同方向的转发时，可以增加边的权重，而不必重复创建节点。

下面的示例只读取你有权访问的目标频道，并限制每个频道的消息数量。频道列表既可以使用公开用户名，也可以使用 Telethon 能够解析的实体标识。

import os
import asyncio
import pandas as pd
from dotenv import load_dotenv
from telethon import TelegramClient
from telethon.tl.types import PeerChannel

load_dotenv()

api_id = int(os.environ["TELEGRAM_API_ID"])
api_hash = os.environ["TELEGRAM_API_HASH"]
session = os.getenv("TELEGRAM_SESSION", "topology_session")

targets = [
    "public_channel_a",
    "public_channel_b",
    "public_channel_c",
]

async def collect_forward_edges():
    rows = []

    async with TelegramClient(session, api_id, api_hash) as client:
        for target in targets:
            destination = await client.get_entity(target)
            destination_name = (
                getattr(destination, "title", None)
                or getattr(destination, "username", None)
                or str(destination.id)
            )

            async for message in client.iter_messages(destination, limit=1000):
                forward = message.fwd_from
                if not forward or not isinstance(forward.from_id, PeerChannel):
                    continue

                source_id = forward.from_id.channel_id

                try:
                    source = await client.get_entity(forward.from_id)
                    source_name = (
                        getattr(source, "title", None)
                        or getattr(source, "username", None)
                        or str(source_id)
                    )
                except Exception:
                    source_name = f"channel_{source_id}"

                rows.append({
                    "source_id": source_id,
                    "source": source_name,
                    "target_id": destination.id,
                    "target": destination_name,
                    "forwarded_at": message.date.isoformat(),
                    "original_post_id": forward.channel_post,
                    "original_date": (
                        forward.date.isoformat() if forward.date else None
                    ),
                    "current_message_id": message.id,
                })

                await asyncio.sleep(0.05)

    return pd.DataFrame(rows)

if __name__ == "__main__":
    dataframe = asyncio.run(collect_forward_edges())
    dataframe.to_csv("telegram_forward_edges.csv", index=False)
    print(f"已提取 {len(dataframe)} 条可识别转发关系")

这里使用频道数字 ID 作为稳定标识，以频道标题作为展示名称，因为标题可能重名或随时修改。对于无法解析的来源频道，脚本保留数字 ID，而不是直接丢弃这条关系。

正式采集时应控制请求频率，并处理 Telegram 返回的 FloodWaitError。分析规模较大时，可以定期保存游标和结果，避免中断后从头扫描全部历史消息。

🕸️ 使用 NetworkX 构建传播网络

完成数据采集后，下一步是聚合相同频道之间的转发次数。边权重越大，表示样本时间范围内该来源频道被目标频道转发得越频繁。

import pandas as pd
import networkx as nx

df = pd.read_csv("telegram_forward_edges.csv")

edge_stats = (
    df.groupby(["source_id", "source", "target_id", "target"])
      .size()
      .reset_index(name="weight")
)

graph = nx.DiGraph()

for row in edge_stats.itertuples(index=False):
    graph.add_node(str(row.source_id), label=row.source)
    graph.add_node(str(row.target_id), label=row.target)
    graph.add_edge(
        str(row.source_id),
        str(row.target_id),
        weight=int(row.weight)
    )

pagerank = nx.pagerank(graph, weight="weight")
in_degree = dict(graph.in_degree(weight="weight"))
out_degree = dict(graph.out_degree(weight="weight"))

for node_id in graph.nodes:
    graph.nodes[node_id]["pagerank"] = pagerank.get(node_id, 0)
    graph.nodes[node_id]["in_degree"] = in_degree.get(node_id, 0)
    graph.nodes[node_id]["out_degree"] = out_degree.get(node_id, 0)

入度较高的频道经常接收其他频道内容，可能扮演聚合者角色；出度较高的频道则频繁成为其他目标频道的直接来源。PageRank 可用于识别处于重要传播路径上的节点，但不能直接证明谁是内容原创者。

如果研究目标是追踪单条内容，应进一步使用原帖 ID、发布时间和文本哈希进行分组。仅凭相似文案容易误把转载、同步发布和独立创作视为同一传播链。

🎨 用 PyVis 绘制可交互拓扑图

PyVis 可以将 NetworkX 图转换成浏览器可查看的 HTML 文件，并支持拖动、缩放和悬停提示。节点大小可以映射 PageRank，连线宽度则映射转发次数。

from pyvis.network import Network

net = Network(
    height="760px",
    width="100%",
    directed=True,
    bgcolor="#ffffff",
    font_color="#222222",
    cdn_resources="in_line"
)

for node_id, attrs in graph.nodes(data=True):
    rank = attrs.get("pagerank", 0)
    label = attrs.get("label", node_id)
    net.add_node(
        node_id,
        label=label,
        title=(
            f"{label}<br>"
            f"PageRank: {rank:.4f}<br>"
            f"接收权重: {attrs.get('in_degree', 0)}<br>"
            f"输出权重: {attrs.get('out_degree', 0)}"
        ),
        size=14 + rank * 180,
        color="#24A1DE"
    )

for source, target, attrs in graph.edges(data=True):
    weight = attrs.get("weight", 1)
    net.add_edge(
        source,
        target,
        value=weight,
        title=f"可识别转发次数：{weight}",
        arrows="to"
    )

net.set_options("""
{
  "physics": {
    "barnesHut": {
      "gravitationalConstant": -24000,
      "springLength": 180,
      "springConstant": 0.03
    },
    "minVelocity": 0.75
  },
  "edges": {
    "smooth": {
      "enabled": true,
      "type": "dynamic"
    }
  },
  "interaction": {
    "hover": true,
    "navigationButtons": true
  }
}
""")

net.write_html("telegram_forward_topology.html", open_browser=False)

打开生成的 telegram_forward_topology.html，即可查看频道之间的传播方向。若节点过多导致画面拥挤，可以先过滤低权重边，或按时间窗口、主题关键词和语言拆分成多个子图。

用于报告时，还可以借助 Matplotlib 输出静态 PNG 或 SVG，但可交互页面更适合探索复杂网络。无论采用哪种展示方式，都应在图例中说明节点、箭头、权重和时间范围的含义。

📊 如何正确解读传播拓扑图

拓扑图中箭头从来源频道指向当前发布频道，粗线代表观察期内出现了更多可识别转发。中心位置通常由布局算法决定，不应仅凭“节点在画面中央”判断其影响力。

判断传播角色时，应同时查看时间顺序、入度、出度、边权重和内容一致性。最早发帖的频道可能是首发源，也可能只是你当前样本中最早被观察到的节点。

更严谨的做法是为每条边保留证据字段，包括当前消息 ID、原帖 ID、两个时间戳和可访问链接。这样既能复核图中的关系，也能在来源信息变化后解释分析依据。

对外发布研究结果时，建议只展示公开频道名称，并对无关个人账号、私密群组和敏感标识进行匿名化。数据最小化不仅能降低隐私风险，也能让结果更聚焦于频道级传播结构。

❓ 常见问题解答（FAQ）

为什么有些转发消息读取不到来源频道？

常见原因包括内容保护、隐藏发送者、来源频道不可访问、原消息删除，以及消息本身采用复制粘贴而非 Telegram 原生转发。此时不能通过猜测补全来源，应将其标记为未知或不可观察。

Forwarded From 能恢复完整的多级转发链吗？

通常不能一次性恢复完整链条，因为当前消息只提供 Telegram 保留的转发来源信息。只有同时采集相关频道历史并匹配消息，才可能逐步拼接出多级路径。

机器人 Bot Token 可以代替用户 API 凭据吗？

多数历史频道分析场景不适合仅使用 Bot Token，因为机器人只能访问其权限范围内的更新和内容。Telethon 用户会话能力更完整，但必须遵守 Telegram 条款以及频道访问权限。

如何避免采集时触发 FloodWait？

应限制频道数量和历史深度，在请求之间增加间隔，并捕获 FloodWaitError 后按服务端要求等待。不要并发扫描大量实体，也不要反复解析相同频道信息。

拓扑图可以证明某个频道是原创源头吗？

不能，拓扑图只能展示采样范围内可观测到的转发关系。要判断原创来源，还需要结合最早发布时间、原帖编号、内容证据、外部网页记录和人工核验。

通过 Telethon 提取元数据、使用 NetworkX 建模，再由 PyVis 可视化，就能把零散的 Forwarded From 信息转换成清晰的频道传播网络。真正可靠的分析不仅依赖代码，还依赖明确的采样边界、可复核证据和对数据缺失的诚实说明。