AI 科技热点日报助手

每小时自动采集国内 AI 热点新闻，通过 AI 摘要整理后生成 Markdown 日报，并提供 Web 界面按年月日树形目录查看历史日报。

技术栈

• Python 3.11+
• Flask（Web 服务 + API）
• feedparser（RSS 采集）
• requests（HTTP 请求）
• schedule（定时任务）
• 通义千问 API（AI 摘要）

安装依赖

pip install flask feedparser requests schedule markdown

项目结构

ai-daily/├── README.md├── config.json          # 配置文件（AI key、RSS 源等）├── main.py              # 入口：启动定时任务 + Web 服务├── collector.py         # 数据采集（RSS + 网页）├── ai_processor.py      # AI 摘要、分类、打分├── report_generator.py  # 生成 Markdown 文件├── web_server.py        # Flask Web 服务├── templates/│   └── index.html       # H5 前端页面└── reports/             # 生成的日报存储目录    └── 2026/        └── 04/            └── 30/                ├── 20260430_080000.md                └── 20260430_090000.md

运行

cd ysptest/ai-dailypython main.py

启动后：

• 每小时自动采集一次并生成日报
• Web 界面访问: http://localhost:5001
• 日报文件保存在 reports/ 目录下

配置说明 (config.json)

{"ai": {"api_key": "你的通义千问API Key","base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1","model": "qwen-plus"  },"sources": [    {"name": "36氪AI", "type": "rss", "url": "https://36kr.com/feed"},    {"name": "机器之心", "type": "rss", "url": "https://www.jiqizhixin.com/rss"},    {"name": "量子位", "type": "rss", "url": "https://www.qbitai.com/feed"}  ],"schedule_interval_minutes": 1,"web_port": 5001,"max_articles_per_source": 10}

-- coding: utf-8 --"""Web 服务模块：Flask 提供 API + H5 页面查看历史日报"""import osimport jsonfrom flask import Flask, render_template, jsonify, send_from_directoryapp = Flask(name, template_folder="templates", static_folder="reports")def get_reports_tree(reports_dir="reports"):"""扫描 reports 目录，生成年/月/日树形结构"""tree = {}if not os.path.exists(reports_dir):return treefor year in sorted(os.listdir(reports_dir), reverse=True):year_path = os.path.join(reports_dir, year)if not os.path.isdir(year_path):continuetree[year] = {}for month in sorted(os.listdir(year_path), reverse=True):month_path = os.path.join(year_path, month)if not os.path.isdir(month_path):continuetree[year][month] = {}for day in sorted(os.listdir(month_path), reverse=True):day_path = os.path.join(month_path, day)if not os.path.isdir(day_path):continuefiles = sorted([f for f in os.listdir(day_path) if f.endswith(".md")], reverse=True)tree[year][month][day] = filesreturn tree@app.route("/")def index():return render_template("index.html")@app.route("/api/tree")def api_tree():"""返回日报树形目录"""tree = get_reports_tree()return jsonify(tree)@app.route("/api/report////")def api_report(year, month, day, filename):"""返回指定日报的 Markdown 内容"""filepath = os.path.join("reports", year, month, day, filename)if os.path.exists(filepath):with open(filepath, "r", encoding="utf-8") as f:return jsonify({"content": f.read(), "filename": filename})return jsonify({"error": "文件不存在"}), 404@app.route("/api/latest")def api_latest():"""返回最新一篇日报"""tree = get_reports_tree()for year in sorted(tree.keys(), reverse=True):for month in sorted(tree[year].keys(), reverse=True):for day in sorted(tree[year][month].keys(), reverse=True):files = tree[year][month][day]if files:filepath = os.path.join("reports", year, month, day, files[0])with open(filepath, "r", encoding="utf-8") as f:return jsonify({"content": f.read(), "filename": files[0], "path": f"{year}/{month}/{day}/{files[0]}"})return jsonify({"content": "暂无日报", "filename": ""})def start_web(port=5001):"""启动 Web 服务"""app.run(host="0.0.0.0", port=port, debug=False)

-- coding: utf-8 --"""AI 处理模块：调用通义千问对新闻进行摘要、分类、打分"""import requestsimport jsondef process_articles(articles, ai_config):"""调用 AI 对文章列表进行摘要和分类"""if not articles:return {"summary": "暂无数据", "categories": {}}if not ai_config.get("api_key"):# 无 AI key 时，简单按来源分组返回return _fallback_process(articles)# 构建 promptarticle_text = "\n".join([f"{i+1}. [{a['source']}] {a['title']}" for i, a in enumerate(articles[:30])])prompt = f"""以下是今天采集到的国内AI科技热点新闻标题列表：{article_text}请完成以下任务：用2-3句话总结今天AI领域的整体趋势将这些新闻按以下类别分类：大模型、AI应用、自动驾驶、芯片硬件、AI创业、其他每条新闻给出一句话摘要（15字以内）和热度评分（1-5星）请用以下JSON格式返回：{{"trend_summary": "今日趋势总结...","categories": {{"大模型": [{{"index": 1, "brief": "一句话摘要", "stars": 4}}],"AI应用": [...],...}}}}"""try:    headers = {"Authorization": f"Bearer {ai_config['api_key']}", "Content-Type": "application/json"}    data = {        "model": ai_config.get("model", "qwen-plus"),        "messages": [            {"role": "system", "content": "你是一个AI科技新闻编辑，擅长分类和摘要。只返回JSON，不要其他内容。"},            {"role": "user", "content": prompt}        ],        "max_tokens": ai_config.get("max_tokens", 4000)    }    resp = requests.post(f"{ai_config['base_url']}/chat/completions", headers=headers, json=data, timeout=60)    if resp.status_code == 200:        content = resp.json()["choices"][0]["message"]["content"]        # 尝试解析 JSON        content = content.strip()        if content.startswith("```"):            content = content.split("\n", 1)[1].rsplit("```", 1)[0]        result = json.loads(content)        result["_articles"] = articles        return result    else:        print(f"[AI] API 错误: {resp.status_code}{resp.text[:200]}")        return _fallback_process(articles)except Exception as e:    print(f"[AI] 处理失败: {e}")    return _fallback_process(articles)def _fallback_process(articles):"""无 AI 时的降级处理：按来源分组"""categories = {}for a in articles:src = a["source"]if src not in categories:categories[src] = []categories[src].append({"index": len(categories[src]) + 1, "brief": a["title"][:20], "stars": 3})return {"trend_summary": f"共采集到 {len(articles)} 条AI相关新闻","categories": categories,"_articles": articles}