#!/usr/bin/env python3
"""
Blogwatcher Daily - RSS Feed 监控脚本
放在 ~/.hermes/skills/research/blogwatcher-daily/scripts/blogwatcher-daily.py
Usage:
# 扫描 / 抓取
python3 blogwatcher-daily.py # 扫描所有订阅，新增文章存入数据库
python3 blogwatcher-daily.py --update "频道名" # 只更新指定订阅（可用 --list 序号或名称）
python3 blogwatcher-daily.py --update 3 --all # 强制回扫指定订阅最新 10 篇
# 订阅管理
python3 blogwatcher-daily.py --list # 列出所有订阅（按分类分组）
python3 blogwatcher-daily.py --add "频道名" "RSS URL" # 添加订阅（默认分类=未分类）
python3 blogwatcher-daily.py --add "频道名" "RSS URL" --category "AI技术" # 添加订阅并指定分类
# 查看文章
python3 blogwatcher-daily.py --articles "频道名" # 列出该订阅所有文章
python3 blogwatcher-daily.py --articles 3 --unread # 只列该订阅的未读文章
# 标记已读
python3 blogwatcher-daily.py --mark-read # 全库标记为已读
python3 blogwatcher-daily.py --mark-read "频道名" # 只标记指定订阅
# 删除
python3 blogwatcher-daily.py --delete "频道名" # 删除该订阅下所有文章
python3 blogwatcher-daily.py --delete 3 --article-id 2 # 删除该订阅下第 2 篇（1-based，顺序同 --articles）
# 导出
python3 blogwatcher-daily.py --export markdown # 今天的文章
python3 blogwatcher-daily.py --export html --date 2026-08-28 # 指定日期
python3 blogwatcher-daily.py --export txt --limit 20 # 全库最新 20 篇
python3 blogwatcher-daily.py --export markdown --sub "异次元软件世界" --limit 20 # 单订阅最新 20 篇
python3 blogwatcher-daily.py --export markdown --category "AI技术" # 分类全部文章
python3 blogwatcher-daily.py --export markdown --category "AI技术" --date $(date +%F) # 分类今日文章
订阅文件格式（subscriptions.txt）：
新格式（推荐）：`分类|频道名|URL`
旧格式（仍兼容）：`频道名|URL`（自动归为 `未分类`）
订阅标识（SUB）：
可传 --list 输出中的整数序号（如 3），也可传订阅名（精确 → 忽略大小写精确 → 忽略大小写包含）。
"""
import os
import re
import sys
import sqlite3
import argparse
import feedparser
from datetime import date
from html import escape, unescape
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
import ssl
# ========== 配置 ==========
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")
# RSSHub 基础地址（可配置）
RSSHUB_BASE = os.environ.get("RSSHUB_URL", "http://192.168.3.45:1200")
# YouTube channel ID 提取正则
YOUTUBE_CHANNEL_RE = re.compile(r'youtube\.com/channel/([\w-]+)', re.IGNORECASE)
YOUTUBE_FEED_RE = re.compile(r'youtube\.com/feeds/videos\.xml\?channel_id=([\w-]+)', re.IGNORECASE)
YOUTUBE_USER_RE = re.compile(r'youtube\.com/user/([\w-]+)', re.IGNORECASE)
# ========== 数据库 ==========
def get_db():
"""获取数据库连接"""
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS articles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
feed_url TEXT NOT NULL,
channel_title TEXT,
title TEXT NOT NULL,
link TEXT UNIQUE NOT NULL,
description TEXT,
pub_date TEXT,
is_read INTEGER DEFAULT 0,
fetched_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS subscriptions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
feed_url TEXT UNIQUE NOT NULL,
enabled INTEGER DEFAULT 1,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
return conn
def is_article_new(conn, link):
"""检查文章是否已存在"""
cursor = conn.execute("SELECT is_read FROM articles WHERE link = ?", (link,))
row = cursor.fetchone()
return row is None
def save_article(conn, feed_url, channel_title, title, link, description, pub_date):
"""保存文章到数据库"""
try:
conn.execute("""
INSERT OR IGNORE INTO articles
(feed_url, channel_title, title, link, description, pub_date)
VALUES (?, ?, ?, ?, ?, ?)
""", (feed_url, channel_title, title, link, description, pub_date))
conn.commit()
except Exception as e:
print(f" ⚠️ 保存失败: {e}")
def mark_all_read(conn, feed_url=None):
"""标记所有文章为已读"""
if feed_url:
conn.execute("UPDATE articles SET is_read = 1 WHERE feed_url = ?", (feed_url,))
else:
conn.execute("UPDATE articles SET is_read = 1")
conn.commit()
# ========== RSS 获取 ==========
def build_fetch_url(url):
"""根据 URL 类型决定获取方式：
- YouTube channel/user/feed URL → 路由到 RSSHub
- RSSHub URL → 直接使用
- 其他 → 直接访问（绕过 RSSHub /rss/ 路由不稳定）
"""
if url.startswith(RSSHUB_BASE):
return url
# YouTube channel
m = YOUTUBE_CHANNEL_RE.search(url)
if m:
return f"{RSSHUB_BASE}/youtube/channel/{m.group(1)}"
# YouTube feed
m = YOUTUBE_FEED_RE.search(url)
if m:
return f"{RSSHUB_BASE}/youtube/channel/{m.group(1)}"
# YouTube user
m = YOUTUBE_USER_RE.search(url)
if m:
return f"{RSSHUB_BASE}/youtube/user/{m.group(1)}"
return url
def convert_to_stored_url(url):
"""将用户输入的 YouTube URL 转换为 RSSHub 存储格式"""
if url.startswith(RSSHUB_BASE):
return url
m = YOUTUBE_CHANNEL_RE.search(url)
if m:
return f"{RSSHUB_BASE}/youtube/channel/{m.group(1)}"
m = YOUTUBE_FEED_RE.search(url)
if m:
return f"{RSSHUB_BASE}/youtube/channel/{m.group(1)}"
m = YOUTUBE_USER_RE.search(url)
if m:
return f"{RSSHUB_BASE}/youtube/user/{m.group(1)}"
return url
def fetch_rss(url, timeout=15):
"""获取并解析 RSS feed"""
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
fetch_url = build_fetch_url(url)
req = Request(fetch_url, headers={'User-Agent': 'Mozilla/5.0 (compatible; Blogwatcher/1.0)'})
try:
with urlopen(req, timeout=timeout, context=ctx) as response:
content = response.read().decode('utf-8', errors='replace')
return parse_rss(content, fetch_url)
except (URLError, HTTPError, Exception) as e:
print(f" ❌ 获取失败: {e}")
return None, []
def parse_rss(xml_content, fetch_url):
"""解析 RSS/Atom XML（支持 RSS 1.0/2.0/Atom）"""
parsed = feedparser.parse(xml_content, response_headers={'fetch_url': fetch_url})
channel_title = parsed.feed.get('title', 'Unknown') if parsed.feed else 'Unknown'
items = []
for entry in parsed.entries[:20]: # 最多取20篇
# 提取 link
link_text = ''
if hasattr(entry, 'link') and entry.link:
link_text = entry.link
elif hasattr(entry, 'id') and entry.id:
link_text = entry.id
# 提取 title
title_text = entry.get('title', 'No Title') or 'No Title'
# 提取 description/summary
desc_text = ''
if hasattr(entry, 'summary') and entry.summary:
desc_text = re.sub(r'<[^>]+>', '', entry.summary)
desc_text = unescape(desc_text)
desc_text = re.sub(r'\s+', ' ', desc_text).strip()[:200]
elif hasattr(entry, 'description') and entry.description:
desc_text = re.sub(r'<[^>]+>', '', entry.description)
desc_text = unescape(desc_text)
desc_text = re.sub(r'\s+', ' ', desc_text).strip()[:200]
# 提取 pubDate
pub_text = ''
if hasattr(entry, 'published') and entry.published:
pub_text = entry.published
elif hasattr(entry, 'updated') and entry.updated:
pub_text = entry.updated
items.append({
'title': title_text,
'link': link_text,
'description': desc_text,
'pub_date': pub_text
})
return channel_title, items
# ========== 订阅管理 ==========
def load_subscriptions():
"""从文件加载订阅列表
支持两种格式（同一文件内可混用，脚本自动识别）：
- 新格式：`category|name|url[|enabled]`
- 旧格式：`name|url[|enabled]`（分类默认为 `未分类`）
判定规则：若 `parts[1]` 以 `http://` / `https://` 开头，视为旧格式；
否则视为新格式（`parts[2]` 必须是 URL）。
"""
subs = []
if not os.path.exists(SUBSCRIPTIONS_FILE):
return subs
with open(SUBSCRIPTIONS_FILE, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
parts = [p.strip() for p in line.split('|')]
if len(parts) < 2:
continue
if parts[1].startswith(('http://', 'https://')):
category = '未分类'
name = parts[0]
url = parts[1]
enabled_str = parts[2] if len(parts) > 2 else None
elif len(parts) >= 3 and parts[2].startswith(('http://', 'https://')):
category = parts[0]
name = parts[1]
url = parts[2]
enabled_str = parts[3] if len(parts) > 3 else None
else:
continue
enabled = enabled_str != '0' if enabled_str is not None else True
if enabled:
subs.append({'category': category, 'name': name, 'url': url})
return subs
def save_subscription(name, url, category='未分类'):
"""添加订阅到文件"""
subs = load_subscriptions()
for sub in subs:
if sub['url'] == url:
print(f"⚠️ 订阅已存在: {name}")
return False
with open(SUBSCRIPTIONS_FILE, 'a', encoding='utf-8') as f:
f.write(f"{category}|{name}|{url}\n")
print(f"✅ 已添加订阅: [{category}] {name}")
return True
def list_subscriptions():
"""列出所有订阅（按分类分组，全局 1-based 序号保持不变）"""
subs = load_subscriptions()
if not subs:
print("📭 暂无订阅")
return
grouped = {}
order = []
for i, sub in enumerate(subs, 1):
cat = sub.get('category', '未分类')
if cat not in grouped:
grouped[cat] = []
order.append(cat)
grouped[cat].append((i, sub))
print(f"\n📡 当前订阅 ({len(subs)} 个 · {len(order)} 个分类):\n")
for cat in order:
items = grouped[cat]
print(f"── 【{cat}】({len(items)} 个) ──")
for i, sub in items:
print(f" [{i}] {sub['name']}")
print(f" {sub['url']}")
print()
def find_subscriptions_by_category(category):
"""按分类查找所有订阅。
匹配顺序：精确 → 忽略大小写精确 → 忽略大小写子串。
匹配到多个分类时打印警告，仍返回全部结果（导出场景下允许跨相邻分类合并）。
"""
subs = load_subscriptions()
if not subs:
print("❌ 订阅列表为空", file=sys.stderr)
return []
exact = [s for s in subs if s.get('category', '未分类') == category]
if exact:
return exact
lower = category.lower()
exact_ci = [s for s in subs if s.get('category', '未分类').lower() == lower]
if exact_ci:
return exact_ci
substr = [s for s in subs if lower in s.get('category', '未分类').lower()]
if substr:
matched_cats = sorted({s['category'] for s in substr})
if len(matched_cats) > 1:
print(f"⚠️ '{category}' 子串匹配到多个分类: {matched_cats}", file=sys.stderr)
return substr
print(f"❌ 未找到分类: {category}", file=sys.stderr)
all_cats = sorted({s.get('category', '未分类') for s in subs})
print(f" 可用分类: {all_cats}", file=sys.stderr)
return []
def find_subscription(identifier):
"""按序号（--list 中的 1-based 索引）或名称定位订阅。
匹配顺序：数字 → 序号；否则 精确 → 忽略大小写精确 → 忽略大小写包含。
多条匹配时打印候选并返回 None。
"""
subs = load_subscriptions()
if not subs:
print("❌ 订阅列表为空", file=sys.stderr)
return None
if identifier.isdigit():
idx = int(identifier)
if 1 <= idx <= len(subs):
return subs[idx - 1]
for sub in subs:
if sub['name'] == identifier:
return sub
lower = identifier.lower()
exact_ci = [s for s in subs if s['name'].lower() == lower]
if len(exact_ci) == 1:
return exact_ci[0]
if len(exact_ci) > 1:
print(f"❌ 忽略大小写下匹配到多个订阅: {[s['name'] for s in exact_ci]}", file=sys.stderr)
return None
substr = [s for s in subs if lower in s['name'].lower()]
if len(substr) == 1:
return substr[0]
if len(substr) > 1:
print(f"❌ 匹配到多个订阅: {[s['name'] for s in substr]}", file=sys.stderr)
return None
print(f"❌ 未找到订阅: {identifier}", file=sys.stderr)
return None
# ========== 主流程 ==========
def _scan_one(conn, sub, force_all=False):
"""扫描单个订阅（内部辅助）。返回 (channel_title 或 None, 抓取总数, 新增数量)。"""
channel_title, items = fetch_rss(sub['url'])
if items is None:
return None, 0, 0
items_to_save = items[:10] if force_all else [i for i in items[:10] if is_article_new(conn, i['link'])]
for item in items_to_save:
save_article(conn, sub['url'], channel_title, item['title'], item['link'],
item['description'], item['pub_date'])
return channel_title, len(items), len(items_to_save)
def scan_all(force_all=False):
"""扫描所有订阅
force_all: True 则忽略已读状态，每个频道强制抓10篇
"""
print("=" * 50)
print("Blogwatcher Daily Scan")
print("=" * 50)
subs = load_subscriptions()
if not subs:
print("\n📭 暂无订阅，请先添加:")
print(f" python3 {__file__} --add \"频道名\" \"RSS URL\"")
return
conn = get_db()
new_count = 0
print(f"\n📡 开始扫描 {len(subs)} 个订阅...\n")
for sub in subs:
print(f"🔍 扫描: {sub['name']}")
channel_title, total, new_in_feed = _scan_one(conn, sub, force_all=force_all)
if channel_title is None:
print(f" ⏭️ 跳过\n")
continue
new_count += new_in_feed
print(f" ✅ {channel_title}: {total} 篇, 新增 {new_in_feed} 篇\n")
conn.close()
print("-" * 50)
print(f"📊 扫描完成: 共发现 {new_count} 篇新文章\n")
if new_count == 0:
print("📭 今日无新文章")
def scan_subscription(sub, force_all=False):
print("=" * 50)
print(f"Blogwatcher — 更新单个订阅: {sub['name']}")
print("=" * 50)
conn = get_db()
print(f"\n🔍 扫描: {sub['name']}")
channel_title, total, new_in_feed = _scan_one(conn, sub, force_all=force_all)
conn.close()
if channel_title is None:
print(f" ⏭️ 跳过\n")
print("-" * 50)
print("📊 扫描完成: 未获取到新文章")
return
print(f" ✅ {channel_title}: {total} 篇, 新增 {new_in_feed} 篇\n")
print("-" * 50)
print(f"📊 扫描完成: 新增 {new_in_feed} 篇文章")
# ========== 文章管理 ==========
def _fetch_articles_for_feed(conn, feed_url, unread_only=False):
conn.row_factory = sqlite3.Row
query = """
SELECT id, title, link, description, pub_date, is_read, fetched_at
FROM articles
WHERE feed_url = ?
"""
params = [feed_url]
if unread_only:
query += " AND is_read = 0"
query += " ORDER BY fetched_at DESC, id DESC"
return conn.execute(query, params).fetchall()
def list_articles_for_subscription(sub, unread_only=False):
conn = get_db()
rows = _fetch_articles_for_feed(conn, sub['url'], unread_only=unread_only)
conn.close()
label = "未读" if unread_only else "全部"
print(f"\n📄 [{sub['name']}] {label}文章 ({len(rows)} 篇):\n")
if not rows:
print(" (无文章)")
return
for i, row in enumerate(rows, 1):
mark = "🆕" if not row['is_read'] else "📖"
title = (row['title'] or 'Untitled').strip()
print(f" [{i}] {mark} {title}")
if row['link']:
print(f" 🔗 {row['link']}")
if row['pub_date']:
print(f" 📅 {row['pub_date']}")
print()
def delete_articles_for_subscription(sub, article_index=None):
"""删除该订阅下文章。
article_index=None → 删除全部；否则按 --articles 显示顺序删除第 article_index 篇（1-based）。
返回 True 表示删除操作成功（或空库时的幂等 no-op），False 表示失败（如序号越界）。
"""
conn = get_db()
if article_index is None:
count = conn.execute(
"SELECT COUNT(*) FROM articles WHERE feed_url = ?", (sub['url'],)
).fetchone()[0]
if count == 0:
print(f"📭 [{sub['name']}] 无文章可删除")
conn.close()
return True
conn.execute("DELETE FROM articles WHERE feed_url = ?", (sub['url'],))
conn.commit()
conn.close()
print(f"🗑️ 已删除 [{sub['name']}] 全部 {count} 篇文章")
return True
rows = _fetch_articles_for_feed(conn, sub['url'])
if article_index < 1 or article_index > len(rows):
print(f"❌ 无效序号 {article_index}，[{sub['name']}] 当前有 {len(rows)} 篇文章", file=sys.stderr)
conn.close()
return False
victim = rows[article_index - 1]
conn.execute("DELETE FROM articles WHERE id = ?", (victim['id'],))
conn.commit()
conn.close()
print(f"🗑️ 已删除 [{sub['name']}] 第 {article_index} 篇: {(victim['title'] or 'Untitled').strip()}")
return True
# ========== 导出 ==========
def fetch_articles(target_date=None, limit=None, feed_url=None, feed_urls=None):
"""从 articles 表读取；可选按本地日期过滤（date(fetched_at,'localtime')）
+ 可选按单订阅 feed_url 或多订阅 feed_urls 过滤 + 可选限制数量。"""
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
conditions = []
params = []
if target_date:
conditions.append("date(fetched_at, 'localtime') = ?")
params.append(target_date)
if feed_url:
conditions.append("feed_url = ?")
params.append(feed_url)
if feed_urls:
placeholders = ",".join("?" * len(feed_urls))
conditions.append(f"feed_url IN ({placeholders})")
params.extend(feed_urls)
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
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 = [
"",
'',
f"Blogwatcher Daily — {escape(label)}",
"",
"",
f"Blogwatcher Daily — {escape(label)}
",
]
if not articles:
parts.append("无文章
")
return "\n".join(parts) + "\n"
parts.append(f"共 {len(articles)} 篇文章。
")
for channel, items in _group_by_channel(articles).items():
parts.append(f"【{escape(channel)}】
\n")
for it in items:
title = (it["title"] or "Untitled").strip()
link = (it["link"] or "").strip()
parts.append(f' - {escape(title)}')
desc = (it["description"] or "").strip()
if desc:
parts.append(f"
{escape(desc[:200])}
")
parts.append(" ")
parts.append("
")
parts.append("