用Dify搭建多平台热点聚合与自动摘要生成流水线

代码不长，直接贴：

import json
from datetime import datetime, timezone, timedelta
from typing import Any, List, Dict

SH_TZ = timezone(timedelta(hours=8))

def parse_update_time(raw: Any) -> str:
    if not raw:
        return ""
    s = str(raw)
    # ISO 带毫秒: 2024-01-01T12:00:00.123Z
    for fmt in ("%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ"):
        try:
            dt = datetime.strptime(s, fmt).replace(tzinfo=timezone.utc)
            return dt.astimezone(SH_TZ).strftime("%Y-%m-%d %H:%M:%S")
        except Exception:
            pass
    # 时间戳（秒或毫秒）
    try:
        ts = int(s)
        if len(s) == 13:
            ts = ts / 1000
        dt = datetime.fromtimestamp(ts, tz=timezone.utc)
        return dt.astimezone(SH_TZ).strftime("%Y-%m-%d %H:%M:%S")
    except Exception:
        return s

def extract_articles(data: Any) -> List[Dict]:
    if isinstance(data, dict) and "articles" in data:
        return data.get("articles", [])
    if isinstance(data, list):
        out = []
        for item in data:
            if isinstance(item, dict):
                out.extend(item.get("articles", []))
        return out
    return []

def build_table_from_arg(arg1: Any) -> List[List[str]]:
    # arg1 可以是 JSON 字符串 / dict / list
    if isinstance(arg1, str):
        try:
            parsed = json.loads(arg1)
        except Exception:
            parsed = arg1
    else:
        parsed = arg1
    # 支持包装 {"arg1": ...} 的情况
    data = parsed.get("arg1", parsed) if isinstance(parsed, dict) else parsed
    articles = extract_articles(data) if data else []
    table = [["标题", "热门评分", "新闻链接-手机端", "新闻链接-PC端", "更新时间"]]
    for item in articles:
        title = item.get("title", "") if isinstance(item, dict) else ""
        hot_score = item.get("hot_score", "")
        mobile_link = item.get("links", {}).get("mobile", "") if isinstance(item, dict) else ""
        pc_link = item.get("links", {}).get("pc", "") if isinstance(item, dict) else ""
        update_time_raw = item.get("metadata", {}).get("update_time", "") if isinstance(item, dict) else ""
        update_time = parse_update_time(update_time_raw)
        table.append([title, hot_score, mobile_link, pc_link, update_time])
    return table

# 在工作流里你可以把返回值转成 JSON 字符串或直接作为 node 输出
def main(arg1):
    try:
        tbl = build_table_from_arg(arg1)
        return {"result": tbl}
    except Exception as e:
        return {"error": str(e)}

不同平台的分支都调这个函数就行——只要数据里有 articles 结构。如果某个平台格式特殊，再单独写个 adapter 转换成标准 articles。

变量聚合器

这个节点把各平台解析结果和一些元数据（抓取时间、平台名、爬虫状态、top N 等）打包成一个整体上下文，喂给 LLM。我常用的字段有：

• fetch_time（格式 2025-12-23 14:30:00）
• platforms（["juejin","bilibili"]）
• article_table（上一步的表格）
• summary_prompt_template（提示词模板）
• top_n（候选量，比如 5）

接入 LLM

1. 右上角头像 → 设置 → 模型供应商，选 OpenAI-API-compatible。

2. 填好 API Key、Endpoint 等。

3. 在'模型调用'节点里选 deepseek-v3.2，调整 temperature、max_tokens，System 提示可以写成：'你是一位清晰、专业的内容编辑，给出要点式总结并加入发布时间'。

用户提示可以这样：

请基于以下表格生成一段不超过 300 字的热闻摘要，要求： 
1. 概述每个平台上最热的 3 条内容（用 bullet 列出） 
2. 每条后面标注平台和抓取时间 
3. 最后给出 1 条行动建议（如转发、深入追踪或忽略） 
表格数据：<插入 article_table> 
抓取时间：<fetch_time>

输出与交付

把模型生成的摘要和解析后的表格拼成完整 Markdown——开头显式带上抓取时间，比如'抓取时间：2025-12-23 14:30:00'，保证可追溯。然后过 Markdown 转换器，输出 PDF 方便存档和分发。

总结

这套流水线跑下来，有几个地方我觉得最关键：

• 平台差异尽量封在解析层，工作流主体不动；
• 时间统一成 UTC+8，避免混淆；
• 提示词模板作为变量传递，改起来方便，生成结果也更可控。

把掘金、B 站、微博、头条的热闻汇总成一份有观点的简报，自动化后省心不少。你可以在这个骨架上加入更多平台，或者调自己的模型参数。