修改blogwatcher-daily脚本

.gitignore

@@ -2,3 +2,5 @@
 
 .DS_Store
 **/.DS_Store
+
+__pycache__/

blogwatcher-daily/SKILL.md

@@ -1,12 +1,12 @@
 ---
 name: blogwatcher-daily
-description: RSS 订阅监控 + 每日笔记生成。使用 RSSHub + feedparser 抓取 31 个订阅，自动去重并存入 SQLite，新文章追加写入 Markdown 笔记。
+description: RSS 订阅监控 + 文章导出。使用 RSSHub + feedparser 抓取 31 个订阅，自动去重并存入 SQLite；可将文章按 txt/markdown/html 三种格式导出。
-version: 1.0
+version: 1.3
 category: custom
-tags: [rss, blog, monitoring, automation]
+tags: [rss, blog, monitoring, automation, export]
 metadata:
   author: Hermes Agent
-  last_updated: 2025-04-19
+  last_updated: 2026-08-29
   platform: macos, ubuntu
   custom_skill_path: /Users/weishen/.hermes/skills/custom/
   installation_note: "blogwatcher-daily 脚本在 Mac mini 本地运行（依赖 feedparser）。需要 RSSHub 服务（http://192.168.3.45:1200）访问 YouTube/Bilibili 等被墙源。"
@@ -14,7 +14,7 @@ metadata:
 
 # Blogwatcher Daily
 
-RSS 订阅监控 + 每日笔记生成自动化。
+RSS 订阅监控自动化，抓取后去重入库，可按 txt / markdown / html 三种格式导出。
 
 ## 依赖
 
@@ -22,7 +22,7 @@ RSS 订阅监控 + 每日笔记生成自动化。
 pip3 install feedparser
 ```
 
-> feedparser 是 Python 最成熟的 RSS 解析库，支持 RSS 1.0/2.0/Atom、任意编码、畸形 XML。
+> feedparser 是 Python 最成熟的 RSS 解析库，支持 RSS 1.0/2.0/Atom、任意编码、畸形 XML。HTML 导出使用 Python 标准库 `html.escape`，无第三方依赖。
 
 ## 实际抓取架构（重要）
 
@@ -48,35 +48,13 @@ curl → 目标网站
 - 已有的 RSSHub URL → 直接使用
 - 其他 → 直接访问原始 URL
 
-```
-YouTube 频道 URL
-    ↓ 脚本自动识别并转为 RSSHub 格式
-http://192.168.3.45:1200/youtube/channel/{id}
-    ↓
-curl → RSSHub → YouTube
-
-普通 RSS（Engadget, Slashdot 等）
-    ↓ 脚本直接访问（绕过 RSSHub /rss/ 路由）
-原始 RSS URL
-    ↓
-curl → 目标网站
-```
-
-**关键发现**：RSSHub 的 `/rss/{url}` 路由不稳定（返回 RSSHub 欢迎页），
-因此普通 RSS 源直接访问，不走 RSSHub 代理。
-
-脚本内部 `build_fetch_url()` 根据 URL 类型自动选择路由：
-- YouTube → RSSHub
-- 已有的 RSSHub URL → 直接使用
-- 其他 → 直接访问原始 URL
-
 ## 目录结构
 
 ```
 ~/.hermes/skills/custom/blogwatcher-daily/
 ├── SKILL.md                    # 本文件
 ├── scripts/
-│   └── blogwatcher-daily.py    # 主脚本
+│   └── blogwatcher-daily.py    # 扫描入库 + 订阅管理 + 文章导出（单文件）
 ├── subscriptions.txt           # 订阅列表（name|URL）
 └── blogwatcher.db              # SQLite 数据库（自动创建）
 ```
@@ -90,7 +68,7 @@ python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py
 ```
 
 - 扫描所有订阅，新增文章存入数据库
-- 生成今日笔记：`~/Workspace/nexus/ishenwei/blogwatcher/YYYY-MM-DD.md`
+- 控制台打印每个订阅新增文章数量及总计
 
 ### 添加订阅
 
@@ -115,24 +93,13 @@ python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py \
 python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py --list
 ```
 
-### 仅扫描（不写文件）
-
-```bash
-python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py --scan-only
-```
-
 ### 强制回扫（`--all`）
 
 ```bash
-# 强制抓取每频道10篇（忽略已读状态），写入独立文件
+# 强制抓取每频道10篇（忽略已读状态）
 python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py --all
-
-# 测试模式：只打印，不写文件
-python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py --scan-only --all
 ```
 
-- 输出文件：`~/Workspace/nexus/ishenwei/blogwatcher/all-YYYY-MM-DD.md`（**覆盖模式**，每次运行覆盖）
-- **不混入**每日报告 `YYYY-MM-DD.md`
 - 每频道最多10篇
 - 用途：历史内容回扫、测试订阅状态
 
@@ -142,6 +109,51 @@ python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py -
 python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py --mark-read
 ```
 
+### 导出文章（`--export`）
+
+从 SQLite 里读取文章并输出到 stdout，格式通过 `--export {txt,markdown,html}` 选择：
+
+```bash
+# 今天所有文章 → Markdown
+python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py --export markdown
+
+# 指定日期 → HTML
+python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py \
+  --export html --date 2026-08-28 > digest.html
+
+# 全库最新 20 篇 → 纯文本
+python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py \
+  --export txt --limit 20
+```
+
+**过滤规则**（`--date` 与 `--limit` 的组合语义）
+
+| 参数组合 | 行为 |
+|---------|------|
+| `--export FMT`（默认） | 今天（本地时区）的所有文章 |
+| `--export FMT --date X` | 指定日期 X 的所有文章 |
+| `--export FMT --limit N` | 忽略日期，取**全库最新 N 篇** |
+| `--export FMT --date X --limit N` | 日期 X 的最新 N 篇 |
+
+**参数速查**
+
+| 参数 | 说明 |
+|------|------|
+| `--export {txt,markdown,html}` | 触发导出模式并指定格式 |
+| `--date YYYY-MM-DD` | 目标日期（本地时区，默认今天） |
+| `--limit N` | 文章数上限；单独使用时忽略日期 |
+
+**输出去向**
+
+- 正文 → **stdout**（可直接 `> file.md` 重定向）
+- 进度 → **stderr**（`📊 N 篇文章 → FMT`）
+
+**格式细节**
+
+- `txt`：分频道分组，纯文本无标记，适合终端查看或管道
+- `markdown`：`## 【频道】` + `- [title](link)` + 描述引用块
+- `html`：完整 `<!DOCTYPE html>` 单文件，内嵌简洁 CSS，标题/URL/描述全部经 `html.escape` 转义（防 XSS）
+
 ## 添加订阅示例
 
 ### YouTube 频道
@@ -182,13 +194,12 @@ python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py \
 | 配置项 | 默认值 |
 |--------|--------|
 | RSSHub 地址 | `http://192.168.3.45:1200`（可设置 `RSSHUB_URL` 环境变量覆盖） |
-| 笔记输出目录 | `~/Workspace/nexus/ishenwei/blogwatcher/` |
 | 数据库 | `~/.hermes/skills/custom/blogwatcher-daily/blogwatcher.db` |
 | 订阅列表 | `~/.hermes/skills/custom/blogwatcher-daily/subscriptions.txt` |
 
 ## Cron Job 设置
 
-推荐每天早上 6:00 自动执行：
+推荐每天早上 6:00 自动扫描：
 
 ```bash
 cronjob --create \
@@ -200,13 +211,22 @@ cronjob --create \
 
 执行步骤：
 1. 加载 blogwatcher-daily 技能
-2. 运行：python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py
+2. 运行扫描：python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py
-3. 检查输出，确认写入的笔记文件路径
+3. 汇报本次新增文章数量"
-4. 如果有新文章，简短汇总发给我"
+```
+
+如需**扫描 + 落盘 HTML 摘要**（例如放进 Obsidian vault 或自建 dashboard），可用原生 crontab 串起来：
+
+```cron
+0 6 * * *  python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py && \
+           python3 ~/.hermes/skills/custom/blogwatcher-daily/scripts/blogwatcher-daily.py \
+             --export html > ~/blogwatcher/$(date +\%Y-\%m-\%d).html
 ```
 
 ## 数据流程
 
+### 扫描入库（默认命令）
+
 1. **加载订阅**：从 `subscriptions.txt` 读取所有 name|URL 对
 2. **URL 路由**：`build_fetch_url()` 自动判断：
    - YouTube → 转为 `http://192.168.3.45:1200/youtube/channel/{id}`
@@ -216,9 +236,14 @@ cronjob --create \
 4. **解析**：feedparser 提取 title、link、description、pub_date
 5. **去重**：按 link 去重，已存在则跳过
 6. **存储**：新文章存入 SQLite
-7. **输出**：**追加写入** Markdown 笔记（同一文件多次扫描不会覆盖，仅追加新链接）
+7. **输出**：控制台打印每个订阅新增数量与总计（不生成文件）
-   - 读取已有文件，用正则提取现有链接 → 去重
+
-   - 新文章追加 `## 📦 新增 N 篇` 分隔块
+### 导出（`--export`）
+
+1. **查询**：`fetch_articles()` 按 `date(fetched_at, 'localtime')` 过滤 + 可选 `LIMIT N`
+2. **分组**：`_group_by_channel()` 按 `channel_title` 分桶
+3. **格式化**：`FORMATTERS[fmt]` 分发到 `format_txt` / `format_markdown` / `format_html`
+4. **输出**：正文写 stdout，`📊 N 篇文章 → FMT` 进度写 stderr
 
 ## 已知问题
 
@@ -236,4 +261,5 @@ cronjob --create \
 - 每次最多取每源 10 篇（TED Talks 等除外，取 20 篇）
 - 数据库自动创建，无需手动初始化
 - OPML 文件可用 Python 解析后批量导入（参考 /wiki-ingest 流程中的 OPML 解析代码）
-- **Markdown 追加模式**：同一天多次扫描不会覆盖旧内容，仅追加新链接（去重基于 URL）
+- 导出的"今天"按**本地时区**（SQLite `date(fetched_at, 'localtime')`），不是 UTC
+- HTML 导出使用 `html.escape` 转义标题/链接/描述，可安全嵌入网页；CSS 内嵌，单文件独立可用

blogwatcher-daily/scripts/blogwatcher-daily.py

@@ -4,20 +4,23 @@ Blogwatcher Daily - RSS Feed 监控脚本
 放在 ~/.hermes/skills/research/blogwatcher-daily/scripts/blogwatcher-daily.py
 
 Usage:
-  python3 blogwatcher-daily.py              # 扫描所有订阅，写入今日笔记
+  python3 blogwatcher-daily.py                        # 扫描所有订阅，新增文章存入数据库
-  python3 blogwatcher-daily.py --list       # 列出所有订阅
+  python3 blogwatcher-daily.py --list                 # 列出所有订阅
   python3 blogwatcher-daily.py --add "频道名" "RSS URL"  # 添加订阅
-  python3 blogwatcher-daily.py --scan-only  # 只扫描不写入文件
+  python3 blogwatcher-daily.py --mark-read            # 标记所有文章为已读
-  python3 blogwatcher-daily.py --mark-read  # 标记所有文章为已读
+  python3 blogwatcher-daily.py --export markdown      # 把今天的文章导出为 Markdown
+  python3 blogwatcher-daily.py --export html --date 2026-08-28
+  python3 blogwatcher-daily.py --export txt --limit 20
 """
 
 import os
 import re
+import sys
 import sqlite3
 import argparse
 import feedparser
-from datetime import datetime
+from datetime import date
-from html import unescape
+from html import escape, unescape
 from urllib.request import Request, urlopen
 from urllib.error import URLError, HTTPError
 import ssl
@@ -27,7 +30,6 @@ SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
 SKILL_DIR = os.path.dirname(SCRIPT_DIR)  # .../research/blogwatcher-daily
 DB_PATH = os.path.join(SKILL_DIR, "blogwatcher.db")
 SUBSCRIPTIONS_FILE = os.path.join(SKILL_DIR, "subscriptions.txt")
-OUTPUT_DIR = os.path.expanduser("~/Workspace/nexus/ishenwei/blogwatcher")
 
 # RSSHub 基础地址（可配置）
 RSSHUB_BASE = os.environ.get("RSSHUB_URL", "http://192.168.3.45:1200")
@@ -235,99 +237,10 @@ def list_subscriptions():
         print(f"  [{i}] {sub['name']}")
         print(f"      {sub['url']}\n")
 
-# ========== Markdown 输出 ==========
-def generate_markdown(results, date_str=None):
-    """生成 Markdown 格式"""
-    if date_str is None:
-        date_str = datetime.now().strftime("%Y-%m-%d")
-    
-    md = f"# Blogwatcher Daily - {date_str}\n\n"
-    md += f"_生成时间: {datetime.now().strftime('%H:%M:%S')}_\n\n"
-    md += "---\n\n"
-    
-    if not results:
-        md += "_今日无新文章_\n"
-        return md
-    
-    # 按频道分组
-    channels = {}
-    for item in results:
-        ch = item['channel']
-        if ch not in channels:
-            channels[ch] = []
-        channels[ch].append(item)
-    
-    for channel, items in channels.items():
-        md += f"## 【{channel}】\n\n"
-        for item in items:
-            md += f"- [{item['title']}]({item['link']})\n"
-            if item['description']:
-                md += f"  {item['description'][:150]}...\n"
-            md += "\n"
-        md += "---\n\n"
-    
-    return md
-
-def save_daily_report(new_results, date_str=None):
-    """保存每日报告（追加模式，避免覆盖）"""
-    if date_str is None:
-        date_str = datetime.now().strftime("%Y-%m-%d")
-    
-    os.makedirs(OUTPUT_DIR, exist_ok=True)
-    filepath = os.path.join(OUTPUT_DIR, f"{date_str}.md")
-    
-    # 读取已有内容，提取已有链接避免重复
-    existing_links = set()
-    if os.path.exists(filepath):
-        with open(filepath, 'r') as f:
-            existing_content = f.read()
-        # 简单提取已有链接（用于去重），去掉末尾的 ) 等标点
-        import re as _re
-        raw_links = _re.findall(r'https?://\S+', existing_content)
-        existing_links = set()
-        for l in raw_links:
-            # 去掉末尾的 ) 等非 URL 字符
-            while l and l[-1] in '),;:':
-                l = l[:-1]
-            existing_links.add(l)
-    else:
-        existing_content = ""
-    
-    # 如果没有新结果，且文件已存在，直接返回
-    if not new_results and existing_content:
-        return filepath
-    
-    # 生成新内容的 header
-    new_md = ""
-    new_articles = [a for a in new_results if a['link'] not in existing_links]
-    
-    if new_articles:
-        new_md += f"\n## 📦 新增 {len(new_articles)} 篇 ({datetime.now().strftime('%H:%M:%S')})\n\n"
-        channels = {}
-        for item in new_articles:
-            ch = item['channel']
-            if ch not in channels:
-                channels[ch] = []
-            channels[ch].append(item)
-        for channel, items in channels.items():
-            new_md += f"### 【{channel}】\n\n"
-            for item in items:
-                new_md += f"- [{item['title']}]({item['link']})\n"
-                if item['description']:
-                    new_md += f"  {item['description'][:150]}...\n"
-                new_md += "\n"
-    
-    # 追加写入
-    with open(filepath, 'a') as f:
-        f.write(new_md)
-    
-    return filepath
-
 # ========== 主流程 ==========
-def scan_all(force_all=False, write_file=True):
+def scan_all(force_all=False):
     """扫描所有订阅
     force_all: True 则忽略已读状态，每个频道强制抓10篇
-    write_file: True 则写入文件
     """
     print("=" * 50)
     print("Blogwatcher Daily Scan")
@@ -375,32 +288,149 @@ def scan_all(force_all=False, write_file=True):
     print("-" * 50)
     print(f"📊 扫描完成: 共发现 {new_count} 篇新文章\n")
     
-    if all_new_articles and write_file:
+    if not all_new_articles:
-        if force_all:
-            # --all 模式：写入独立文件（不追加日常报告）
-            md = generate_markdown(all_new_articles)
-            filepath = os.path.join(OUTPUT_DIR, f"all-{datetime.now().strftime('%Y-%m-%d')}.md")
-            with open(filepath, 'w') as f:
-                f.write(md)
-            print(f"📝 已写入（force-all 模式）: {filepath}")
-        else:
-            filepath = save_daily_report(all_new_articles)
-            print(f"📝 已写入: {filepath}")
-    elif not all_new_articles:
         print("📭 今日无新文章")
     
     return all_new_articles
 
+# ========== 导出 ==========
+def fetch_articles(target_date=None, limit=None):
+    """从 articles 表读取；可选按本地日期过滤（date(fetched_at,'localtime')）+ 可选限制数量"""
+    if not os.path.exists(DB_PATH):
+        print(f"❌ 数据库不存在: {DB_PATH}", file=sys.stderr)
+        return []
+
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+
+    where = ""
+    params = []
+    if target_date:
+        where = "WHERE date(fetched_at, 'localtime') = ?"
+        params.append(target_date)
+
+    query = f"""
+        SELECT channel_title, title, link, description, pub_date, fetched_at
+        FROM articles
+        {where}
+        ORDER BY fetched_at DESC, channel_title
+    """
+    if limit:
+        query += " LIMIT ?"
+        params.append(limit)
+
+    cursor = conn.execute(query, params)
+    rows = [dict(r) for r in cursor.fetchall()]
+    conn.close()
+    return rows
+
+def _group_by_channel(articles):
+    grouped = {}
+    for a in articles:
+        ch = a["channel_title"] or "Unknown"
+        grouped.setdefault(ch, []).append(a)
+    return grouped
+
+def format_txt(articles, label):
+    lines = [f"Blogwatcher Daily — {label}", "=" * 50, ""]
+    if not articles:
+        lines.append("(无文章)")
+        return "\n".join(lines) + "\n"
+    lines.append(f"共 {len(articles)} 篇文章")
+    lines.append("")
+    for channel, items in _group_by_channel(articles).items():
+        lines.append(f"[{channel}]")
+        for it in items:
+            title = (it["title"] or "Untitled").strip()
+            link = (it["link"] or "").strip()
+            lines.append(f"  - {title}")
+            if link:
+                lines.append(f"    {link}")
+            desc = (it["description"] or "").strip()
+            if desc:
+                lines.append(f"    {desc[:200]}")
+        lines.append("")
+    return "\n".join(lines) + "\n"
+
+def format_markdown(articles, label):
+    md = f"# Blogwatcher Daily — {label}\n\n"
+    if not articles:
+        return md + "_无文章_\n"
+    md += f"共 **{len(articles)}** 篇文章。\n\n---\n\n"
+    for channel, items in _group_by_channel(articles).items():
+        md += f"## 【{channel}】\n\n"
+        for it in items:
+            title = (it["title"] or "Untitled").strip()
+            link = (it["link"] or "").strip()
+            md += f"- [{title}]({link})\n"
+            desc = (it["description"] or "").strip()
+            if desc:
+                md += f"  > {desc[:200]}\n"
+        md += "\n"
+    return md
+
+def format_html(articles, label):
+    parts = [
+        "<!DOCTYPE html>",
+        '<html lang="zh-CN"><head><meta charset="utf-8">',
+        f"<title>Blogwatcher Daily — {escape(label)}</title>",
+        "<style>body{font-family:-apple-system,sans-serif;max-width:800px;"
+        "margin:2em auto;padding:0 1em;line-height:1.5}"
+        "h2{border-bottom:1px solid #ddd;padding-bottom:0.3em;margin-top:2em}"
+        "blockquote{color:#555;border-left:3px solid #ccc;margin:0.3em 0 0.6em;"
+        "padding:0.2em 1em}"
+        "li{margin-bottom:0.5em}</style>",
+        "</head><body>",
+        f"<h1>Blogwatcher Daily — {escape(label)}</h1>",
+    ]
+    if not articles:
+        parts.append("<p><em>无文章</em></p></body></html>")
+        return "\n".join(parts) + "\n"
+    parts.append(f"<p>共 <strong>{len(articles)}</strong> 篇文章。</p><hr>")
+    for channel, items in _group_by_channel(articles).items():
+        parts.append(f"<h2>【{escape(channel)}】</h2>\n<ul>")
+        for it in items:
+            title = (it["title"] or "Untitled").strip()
+            link = (it["link"] or "").strip()
+            parts.append(f'  <li><a href="{escape(link)}">{escape(title)}</a>')
+            desc = (it["description"] or "").strip()
+            if desc:
+                parts.append(f"  <blockquote>{escape(desc[:200])}</blockquote>")
+            parts.append("  </li>")
+        parts.append("</ul>")
+    parts.append("</body></html>")
+    return "\n".join(parts) + "\n"
+
+FORMATTERS = {
+    "txt": format_txt,
+    "markdown": format_markdown,
+    "html": format_html,
+}
+
+def export_articles(fmt, target_date, limit):
+    articles = fetch_articles(target_date, limit)
+    if target_date:
+        label = target_date
+    elif limit:
+        label = f"最新 {limit} 篇"
+    else:
+        label = "全部"
+    print(f"📊 {len(articles)} 篇文章 → {fmt}", file=sys.stderr)
+    sys.stdout.write(FORMATTERS[fmt](articles, label))
+
 # ========== CLI ==========
 def main():
     parser = argparse.ArgumentParser(description='Blogwatcher Daily RSS 监控脚本')
     parser.add_argument('--list', '-l', action='store_true', help='列出所有订阅')
     parser.add_argument('--add', nargs=2, metavar=('NAME', 'URL'), help='添加订阅')
-    parser.add_argument('--scan-only', action='store_true', help='仅扫描不写入文件')
     parser.add_argument('--mark-read', action='store_true', help='标记所有为已读')
-    parser.add_argument('--date', '-d', help='指定日期 (YYYY-MM-DD)')
     parser.add_argument('--rsshub', help='设置 RSSHub 地址')
-    parser.add_argument('--all', action='store_true', help='忽略已读状态，强制抓取每频道最新10篇（写入 all-YYYY-MM-DD.md，不追加日常报告）')
+    parser.add_argument('--all', action='store_true', help='忽略已读状态，强制抓取每频道最新10篇')
+    parser.add_argument('--export', choices=['txt', 'markdown', 'html'],
+                        help='导出文章为指定格式，写到 stdout（可搭配 --date/--limit）')
+    parser.add_argument('--date', help='导出的目标日期 YYYY-MM-DD（默认今天，仅 --export 有效）')
+    parser.add_argument('--limit', type=int,
+                        help='导出的文章数上限；单独使用时忽略日期取全库最新 N 篇（仅 --export 有效）')
 
     args = parser.parse_args()
     
@@ -426,11 +456,17 @@ def main():
         conn.close()
         print("✅ 已标记所有文章为已读")
     
-    elif args.scan_only:
+    elif args.export:
-        scan_all(force_all=args.all, write_file=False)
+        if args.date:
+            target_date = args.date
+        elif args.limit:
+            target_date = None
+        else:
+            target_date = date.today().isoformat()
+        export_articles(args.export, target_date, args.limit)
     
     else:
-        scan_all(force_all=args.all, write_file=True)
+        scan_all(force_all=args.all)
 
 if __name__ == "__main__":
     main()