本案例利用 crawl4ai 爬虫工具抓取 AI新闻资讯 网站的最新资讯，由 LLM 大语言模型提炼生成新闻摘要，最后合成语音进行播报；涉及python编程、crawl4ai 爬虫工具、文本转音频、Dify自定义工具、提示词等技术。详细过程如下：
一、本地服务程序编写
1、文本转语音服务
详细过程请关注我，查看相关文档。
2、crawl4ai 网站抓取服务
使用python代码编辑器PyCharm编写如下代码，实现crawl4ai 爬虫工具抓取 AI新闻资讯 网站的最新资讯接口服务。
import json
import requests
from bs4 import BeautifulSoup
from fastapi import FastAPI, Query
import logging
import re
import asyncio
from datetime import datetime
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
# 获取新闻列表页面的所有新闻URL
def get_news_urls():
url = "https://www.aibase.com/zh/news"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
}
try:
response = requests.get(url, headers=headers, timeout=15)
news_urls = []
if response.status_code == 200:
soup = BeautifulSoup(response.text, 'html.parser')
# 查找所有新闻卡片
news_cards = soup.select('div.relative.flex.flex-col.gap-4')
for card in news_cards:
link_tag = card.find('a', href=True)
if link_tag:
link = link_tag['href']
# 确保是有效的新闻链接
if link.startswith('/news/') and len(link.split('/')) >= 3:
full_url = f"https://www.aibase.com/zh{link}"
if full_url not in news_urls:
news_urls.append(full_url)
# 如果没找到足够的链接，尝试备选方法
if len(news_urls) < 4:
logger.warning("使用备选方法查找新闻链接...")
all_links = soup.find_all('a', href=True)
for link in all_links:
href = link['href']
if href and '/news/' in href and len(href.split('/')) >= 3:
if not href.startswith('http'):
href = f"https://www.aibase.com/zh{href}"
if href not in news_urls:
news_urls.append(href)
logger.info(f"找到 {len(news_urls)} 条有效新闻链接")
return news_urls
else:
logger.error(f"请求失败，状态码: {response.status_code}")
return []
except Exception as e:
logger.error(f"获取新闻列表时出错: {str(e)}")
return []
# 直接使用BeautifulSoup提取新闻内容
def extract_article_with_bs(url):
logger.info(f"使用BeautifulSoup提取文章: {url}")
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
}
try:
response = requests.get(url, headers=headers, timeout=30)
if response.status_code != 200:
logger.error(f"请求失败，状态码: {response.status_code}")
return None
soup = BeautifulSoup(response.text, 'html.parser')
# 提取标题
title_tag = soup.select_one('h1.md\\:text-4xl.text-3xl, h1.text-3xl, h1')
title = title_tag.get_text(strip=True) if title_tag else "无标题"
# 提取发布日期
date_tag = soup.select_one('div.flex.items-center.flex-wrap.text-sm.text-surface-500 span:last-child')
date_text = date_tag.get_text(strip=True) if date_tag else ""
# 尝试解析日期文本
try:
# 从文本中提取日期部分（如 "Jun 20, 2025"）
match = re.search(r'[A-Za-z]{3}\s\d{1,2},\s\d{4}', date_text)
if match:
pub_date = datetime.strptime(match.group(), '%b %d, %Y').strftime('%Y-%m-%d')
else:
pub_date = date_text
except Exception:
pub_date = date_text
# 提取内容 - 主要方法
content_div = soup.select_one('div.leading-8.text-\\[\\#242424\\].post-content')
# 备选方法
if not content_div:
content_div = soup.select_one('div.post-content')
if not content_div:
content_div = soup.select_one('div[class*="post-content"]')
content_html = ""
if content_div:
# 清理不需要的元素
for tag in content_div.select('.spamTxt'):
tag.decompose()
content_html = str(content_div)
else:
logger.warning("未找到内容区域")
return {
"title": title,
"publication_date": pub_date,
"content": content_html
}
except Exception as e:
logger.error(f"提取文章时出错: {str(e)}", exc_info=True)
return None
# 异步提取单个新闻文章的数据
async def extract_ai_news_article(url):
logger.info(f"提取新闻文章: {url}")
# 使用BeautifulSoup直接提取
article_data = extract_article_with_bs(url)
if article_data:
logger.info(f"成功提取新闻: {article_data.get('title', '无标题')}")
return article_data
else:
logger.warning(f"未提取到有效数据: {url}")
return None
# 主函数：获取所有新闻URL并逐一提取详细数据
async def fetch_news(limit: int = 5):
# 获取所有新闻URL
news_urls = get_news_urls()
logger.info(f"共找到 {len(news_urls)} 条新闻链接")
if not news_urls:
return [], "未找到新闻链接"
# 限制新闻数量
news_urls = news_urls[:min(limit, len(news_urls))]
news_data_list = []
newsdetail = ""
# 创建任务列表
tasks = []
for url in news_urls:
tasks.append(extract_ai_news_article(url))
# 并行执行所有任务
results = await asyncio.gather(*tasks)
# 处理结果
for index, news_data in enumerate(results, start=1):
if news_data:
# 添加到新闻数据列表
news_data_list.append(news_data)
# 从HTML中提取纯文本用于摘要
content = news_data.get("content", "")
if content:
soup = BeautifulSoup(content, 'html.parser')
text_content = soup.get_text(separator=' ', strip=True)
# 简化内容
if len(text_content) > 500:
text_content = text_content[:500] + "..."
else:
text_content = "无法提取内容"
newsdetail += f"新闻{index}: {text_content}；\n"
else:
newsdetail += f"新闻{index}: 提取失败；\n"
return news_data_list, newsdetail
# FastAPI 接口
@app.get("/news/")
async def get_news(limit: int = Query(5, description="返回的新闻数量")):
try:
news_data, newsdetail = await fetch_news(limit)
return {
"news": news_data,
"newsdetail": newsdetail
}
except Exception as e:
logger.error(f"API处理错误: {str(e)}", exc_info=True)
return {"error": "处理请求时出错", "details": str(e)}
# 运行 FastAPI 应用
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8086)
建议使用postman进行接口测试，测试成功后正式进入DIFY工作流编程。
二、DIFY工作流
1、创建文本转音频自定义工具
详细过程请关注我，查看相关文档。
2、创建应用
打开浏览器登录DIFY平台，点击“工作室”菜单，创建空白应用，选择“Chatflow”，填写“应用名称”，选择图标，点击创建。
3、节点配置
本工作流由“开始”、"调用服务获取最新AI资讯“、“提炼新闻内容摘要”、“合并摘要和详情”、“文本转语音”、”整理语音URL“和“直接回复”几个节点实现。
（1）开始
此处添加一个新闻数量的输入字段，用于确定需要抓取的新闻数量。
（2）调用服务获取最新AI资讯
使用“代码执行”工具调用“crawl4ai 网站抓取服务”抓取指定数量的新闻资讯，并分别以数字和字符串形式返回新闻URL和新闻详情。
python代码如下：
import requests
import json
def main(arg1: str) -> dict:
try:
# 构造请求URL和参数
url = 'http://192.168.1.12:8086/news/'
limit = arg1
# 发送GET请求
response = requests.get(url, params={'limit': limit})
# 检查响应状态码
if response.status_code == 200:
# 请求成功，处理结果
result = response.json()
# 提取新闻数据和新闻详情字符串
news_list = result.get('news', [])
newsdetail = result.get('newsdetail', "")
# 确保 news_list 是一个列表
if not isinstance(news_list, list):
return {"error": "服务端返回的新闻数据格式不正确，'news' 字段应为列表。"}
# 格式化新闻数据（如果需要）
formatted_news = []
for news_item in news_list:
if isinstance(news_item, dict): # 如果是字典，直接添加
formatted_news.append(news_item)
elif isinstance(news_item, str): # 如果是字符串，尝试解析为字典
try:
news_dict = json.loads(news_item) # 使用 json.loads 解析字符串
formatted_news.append(news_dict)
except Exception as e:
print(f"解析新闻数据时出错: {e}")
else:
print("无效的新闻数据格式")
# 返回格式化的新闻数据和新闻详情字符串
return {"news": formatted_news, "newsdetail": newsdetail}
else:
# 请求失败，返回错误信息
return {"error": f"请求失败，状态码: {response.status_code}"}
except requests.exceptions.RequestException as e:
# 捕获请求异常
return {"error": f"请求出错: {str(e)}"}
（3）提炼新闻内容摘要
使用LLM工具对新闻内容进行总结归纳形成文章摘要。
模型使用DeepSeek-V3，SYSTEM提示词如下：
你是一名专业的新闻内容编辑助理。你的任务是对提供的新闻文章进行清晰、简洁、结构化的总结，严格遵循以下格式要求： **总结格式（必须严格遵守）：** 1. **摘要：** * 用一段话概括文章的核心内容、主要事件或结论。 * **字数限制：** 控制在 150 字以内。 * **要求：** 语言精炼，重点突出，避免细节罗列。 2. **文章要点：** * 🏷 使用**数字序号 (1., 2., 3., ...)** 清晰列出文章的关键信息点。 * 每个要点应简洁明了，反映文章的一个主要事实、论点、数据或重要细节。 * **数量：** 根据文章信息量列出 3-8 个核心要点。 * **要求：** 直接陈述要点，避免使用项目符号 (`*`, `-`) 或其他格式。仅使用数字序号。 **重要格式规则：** * **禁止加粗：** 所有标题（“摘要：”、“文章要点：”）和内容**均不得使用 `**` 或其他方式进行加粗**。 * **标题格式：** 使用简单的文本标题 `摘要：` 和 `文章要点：` 即可，不加任何修饰符号（如 `##`, `**`, `__`）。 * **结构清晰：** 严格按“摘要：”段落 + “文章要点：”数字序号列表的顺序输出。 * **直接输出：** 省略任何引言、结语或解释性语句（如“好的，以下是根据文章总结的内容：”或“总结完毕。”）。直接输出规定格式的内容。 * **内容要求：** 总结需忠实于原文核心信息，保持客观中立，不添加个人观点或推测。 **请严格遵循以上要求处理接下来的新闻文章内容。**
USER提示词：
文本内容：{{#1737530544739.news#}}
（4）合并摘要和详情
使用“模版转换”工具对文章摘要和新闻详情进行合并。
（5）文本转语音
使用自定义“文本转语音”工具将上一节点合并后的文本转换成音频文件。
（6）整理语音URL
使用“代码执行”工具，对生成的音频URL进行格式处理，便于后续使用。
python代码：
def main(arg1: str) -> str:
# 首先解析外层的 JSON 字符串
data = json.loads(arg1)
filename=data['filename']
url=data['audio_url']
markdown_result = f"<audio controls><source src='{url}' type='audio/mpeg'>{filename}</audio>"
return {"result": markdown_result}
（7）直接回复
直接输出合并后的新闻摘要和新闻详情，同时输出语音即可。
4、预览验证
点击预览，新闻条目数量选2，查看结果：
