Lets Go Rss

# Copy this file to `.env` (which is git-ignored) and fill in values you want
# to override. The cron wrapper (scripts/run_update_cron.sh) `source`s this
# automatically before running Python, so every var below is available to
# classifier.py / scrapers.py / rss_engine.py at runtime.

# --- Classifier ------------------------------------------------------------
# Providing ANTHROPIC_API_KEY switches the classifier to the LLM path (Claude
# Haiku 4.5 by default). Without it, the keyword-based SimpleClassifier is
# used — still fine for most content, but less accurate.
# ANTHROPIC_API_KEY=sk-ant-...

# Override the classifier model (default: claude-haiku-4-5-20251001)
# RSS_CLASSIFIER_MODEL=claude-haiku-4-5-20251001

# --- RSSHub ----------------------------------------------------------------
# The skill prefers :1201 (its own managed instance, see rsshub_manager.py)
# and falls back to :1200 (ready-cowork's embedded worker, if that app is
# running). Override either if your topology differs.
# RSSHUB_BASE_URL_PRIMARY=http://127.0.0.1:1201
# RSSHUB_BASE_URL_FALLBACK=http://127.0.0.1:1200

# Port used by rsshub_manager.py for the managed instance (default 1201)
# RSSHUB_MANAGED_PORT=1201

# Explicit paths if `node`/`npm` are not on PATH for cron
# RSSHUB_NODE_BIN=/opt/homebrew/bin/node
# RSSHUB_NPM_BIN=/opt/homebrew/bin/npm

# --- HTTP timeouts & retries (seconds) -------------------------------------
# RSS_HTTP_TIMEOUT=10
# RSS_HTTP_RETRIES=2
# RSS_HTTP_BACKOFF=0.8
# RSS_XHS_TIMEOUT=15
# RSS_YTDLP_TIMEOUT=45

# --- UX --------------------------------------------------------------------
# RSS_NOTIFY_MACOS=0 to disable macOS notifications (e.g. XHS cookie expiry)
# RSS_NOTIFY_MACOS=1

# Parallel fetch workers (default 5)
# RSS_MAX_WORKERS=5

# --- Playwright tier (skill-managed Chromium) ------------------------------
# Fetches Bilibili / XHS / Twitter via our own headless Chromium, bypassing
# the anti-scraping that RSSHub routes hit. Playwright + its Chromium are
# already pip-installable (zero user action for Bilibili, which is public).
# Twitter / XHS additionally require a one-time sign-in (cookies persist):
#     python scripts/lets_go_rss.py --login twitter
#     python scripts/lets_go_rss.py --login xiaohongshu
#
# Comma-separated whitelist. Empty = disabled.
# RSS_PLAYWRIGHT_PLATFORMS=bilibili,xiaohongshu,twitter
# RSS_PLAYWRIGHT_PROFILE=~/.lets-go-rss/browser-profile

# Runtime generated files
assets/rss_database.db
assets/rss_database.db-shm
assets/rss_database.db-wal
assets/*.xml
assets/*.md
assets/subscriptions.opml
assets/.update.lock
*.xml
*.opml
latest_update.md
summary.md

# Python
__pycache__/
*.pyc
*.pyo

# Node (skill-owned RSSHub install)
node_modules/
package-lock.json
package.json

# RSSHub worker runtime state
assets/.rsshub.pid
assets/.rsshub.log
assets/.last_run_health.json

# OpenClaw push runtime logs
logs/

# Environment overrides
.env

# OS
.DS_Store
assets/last_digest.json

httpx>=0.27.0
yt-dlp
python-dateutil

#!/usr/bin/env python3
"""
Chrome session bridge for lets-go-rss.

Goal:
- Reuse the user's real Chrome session when remote debugging is enabled.
- Keep the implementation skill-local and portable across machines.
- Avoid hard-coding machine-specific paths beyond standard Chrome defaults.

Design:
- Discover DevToolsActivePort from standard Chrome profile locations.
- Connect over CDP via Playwright.
- Reuse an existing browser context if available.
- Open a temporary tab, extract page text / DOM payload, then close the tab.

Notes:
- This does NOT keep a process permanently running by itself yet.
- It is a bridge layer that allows the skill to connect to an already-enabled
  user Chrome session in a portable way.
"""

from __future__ import annotations

import os
from pathlib import Path
from typing import Callable, Any, Optional


class ChromeSessionUnavailable(RuntimeError):
    pass


class ChromeSessionBridge:
    def __init__(self):
        self.chrome_name = os.environ.get("RSS_CHROME_NAME", "Google/Chrome")

    def _candidate_devtools_files(self) -> list[Path]:
        home = Path.home()
        candidates = [
            home / "Library/Application Support/Google/Chrome/DevToolsActivePort",  # macOS Chrome stable
            home / "Library/Application Support/Chromium/DevToolsActivePort",       # macOS Chromium
            home / ".config/google-chrome/DevToolsActivePort",                      # Linux Chrome
            home / ".config/chromium/DevToolsActivePort",                           # Linux Chromium
        ]
        custom = os.environ.get("RSS_CHROME_DEVTOOLS_FILE")
        if custom:
            candidates.insert(0, Path(custom).expanduser())
        return candidates

    def discover_ws_url(self) -> str:
        for path in self._candidate_devtools_files():
            if not path.exists():
                continue
            try:
                lines = path.read_text(encoding="utf-8").splitlines()
                if len(lines) >= 2:
                    port = lines[0].strip()
                    ws_path = lines[1].strip()
                    if port and ws_path:
                        return f"ws://127.0.0.1:{port}{ws_path}"
            except Exception:
                continue
        raise ChromeSessionUnavailable(
            "Chrome remote debugging not available. Open Chrome, enable remote debugging, then retry."
        )

    def with_page(self, url: str, callback: Callable[[Any], Any], wait_ms: int = 6000) -> Any:
        try:
            from playwright.sync_api import sync_playwright
        except Exception as e:
            raise ChromeSessionUnavailable(f"Playwright unavailable: {e}")

        ws_url = self.discover_ws_url()

        with sync_playwright() as p:
            browser = p.chromium.connect_over_cdp(ws_url)
            context = browser.contexts[0] if browser.contexts else browser.new_context()
            page = context.new_page()
            try:
                page.goto(url, wait_until="domcontentloaded", timeout=30000)
                page.wait_for_timeout(wait_ms)
                return callback(page)
            finally:
                try:
                    page.close()
                except Exception:
                    pass
                try:
                    browser.close()
                except Exception:
                    pass

    def extract_basic_page_state(self, url: str, wait_ms: int = 6000) -> dict:
        def _cb(page):
            body = ""
            try:
                body = page.locator("body").inner_text()[:4000]
            except Exception:
                pass
            return {
                "url": page.url,
                "title": page.title(),
                "body": body,
            }
        return self.with_page(url, _cb, wait_ms=wait_ms)

"""
LLM-based content classification module
Classifies RSS items into categories using AI, with keyword-based hot fallback.
"""

import os
import json
import time
from typing import List, Dict, Any, Optional


CATEGORIES = ["科技", "人文", "设计", "娱乐", "其他"]

# Strengthened Chinese keyword table — covers common Bilibili/Xiaohongshu/Weibo content
SIMPLE_KEYWORDS: Dict[str, List[str]] = {
    "科技": [
        # 通用
        "技术", "编程", "代码", "算法", "开发", "开源", "软件", "硬件", "芯片",
        "人工智能", "AI", "机器学习", "深度学习", "大模型", "LLM", "GPT", "Claude",
        "量子", "航天", "火箭", "科学", "物理", "生物", "化学", "实验", "研究",
        "数据库", "云", "服务器", "网络", "安全", "漏洞", "加密", "区块链", "比特币",
        "Linux", "macOS", "Windows", "iOS", "Android", "Docker", "Kubernetes",
        "前端", "后端", "全栈", "API", "SDK", "框架", "库", "GitHub", "Git",
        "评测", "测评", "性能", "基准", "跑分", "配置", "升级", "安装",
        # 英文
        "tech", "code", "programming", "software", "hardware", "algorithm",
        "machine learning", "deep learning", "neural", "model", "compute",
    ],
    "人文": [
        "文学", "历史", "哲学", "社会", "文化", "人文", "思想", "书籍", "阅读", "读书",
        "诗", "诗歌", "散文", "小说", "经典", "古籍", "古代", "近代", "现代史",
        "宗教", "信仰", "伦理", "心理", "心理学", "社会学", "人类学", "政治",
        "传统", "民俗", "考古", "博物", "博物馆", "文物", "遗产",
        "literature", "history", "philosophy", "culture", "society", "politics",
    ],
    "设计": [
        "设计", "UI", "UX", "交互", "平面", "产品设计", "工业设计", "字体", "排版",
        "品牌", "VI", "logo", "标志", "海报", "插画", "配色", "色彩",
        "艺术", "摄影", "视觉", "美学", "审美", "灵感", "灵感来源", "案例",
        "手绘", "素描", "水彩", "油画", "装置", "雕塑", "空间", "建筑", "室内",
        "design", "art", "photography", "visual", "typography", "illustration",
        "architecture", "interior", "brand", "poster",
    ],
    "娱乐": [
        # 游戏
        "游戏", "手游", "端游", "主机", "Steam", "Switch", "PS5", "Xbox",
        "原神", "王者荣耀", "英雄联盟", "LOL", "吃鸡", "PUBG", "我的世界", "Minecraft",
        "攻略", "通关", "速通", "直播", "主播",
        # 影视
        "电影", "影视", "剧集", "电视剧", "综艺", "动漫", "番剧", "动画",
        "漫威", "DC", "奥斯卡", "票房", "导演", "演员", "明星", "偶像",
        "解说", "影评", "剪辑", "预告片",
        # 音乐 / 体育 / 生活
        "音乐", "歌曲", "MV", "演唱会", "乐队", "歌手", "专辑",
        "体育", "足球", "篮球", "NBA", "世界杯", "奥运", "电竞",
        "美食", "做饭", "菜谱", "探店", "吃播",
        "穿搭", "时尚", "美妆", "护肤", "口红", "化妆",
        "旅行", "旅游", "vlog", "日常", "生活", "搞笑", "段子",
        "game", "gaming", "movie", "film", "music", "entertainment",
        "sport", "vlog", "fashion", "makeup", "food",
    ],
}


def _notify(msg: str) -> None:
    """Uniform error surfacing — prefixed so it's greppable in cron logs."""
    print(f"  ⚠️  [classifier] {msg}", flush=True)


def _classify_error_kind(exc: BaseException) -> str:
    """Bucket an anthropic/HTTP error into a short kind tag."""
    name = type(exc).__name__.lower()
    msg = str(exc).lower()
    if "ratelimit" in name or "rate_limit" in msg or "429" in msg:
        return "rate_limit"
    if "auth" in name or "401" in msg or "403" in msg or "invalid x-api-key" in msg or "api key" in msg:
        return "auth"
    if "timeout" in name or "timeout" in msg or "connect" in name or "network" in msg:
        return "network"
    if "notfound" in name or "404" in msg or "model" in msg and "not found" in msg:
        return "model_missing"
    return "other"


class ContentClassifier:
    """Classifies content using Claude API, with SimpleClassifier as hot fallback."""

    DEFAULT_MODEL = os.environ.get(
        "RSS_CLASSIFIER_MODEL", "claude-haiku-4-5-20251001"
    )

    def __init__(self, api_key: Optional[str] = None, model: Optional[str] = None):
        self.api_key = api_key or os.environ.get("ANTHROPIC_API_KEY")
        if not self.api_key:
            raise ValueError("ANTHROPIC_API_KEY environment variable is required")

        try:
            import anthropic
        except ImportError:
            raise ImportError("anthropic package not installed. Run: pip install anthropic")

        self._anthropic = anthropic
        self.client = anthropic.Anthropic(api_key=self.api_key)
        self.model = model or self.DEFAULT_MODEL
        self.categories = CATEGORIES
        self._simple = SimpleClassifier()  # hot fallback
        self._consecutive_errors = 0

        self.system_prompt = """你是一个内容分类专家。你的任务是将提供的内容分类到以下类别之一:

- 科技: 技术、编程、科学、AI、软件、硬件等
- 人文: 文学、历史、哲学、社会、文化等
- 设计: UI/UX、平面设计、产品设计、艺术、摄影等
- 娱乐: 游戏、影视、音乐、综艺、体育、美食、时尚、生活等
- 其他: 不属于以上类别的内容

请只返回类别名称,不要添加任何解释。如果内容跨越多个类别,选择最主要的一个。"""

    def _fallback(self, title: str, description: str, reason: str) -> str:
        """LLM → SimpleClassifier → 其他 三级回退。"""
        cat = self._simple.classify_item(title, description)
        _notify(f"LLM classification failed ({reason}); keyword fallback → {cat}")
        return cat

    def classify_item(self, title: str, description: str = "") -> str:
        """Classify a single item. Never silently returns 其他 on LLM errors —
        always goes through keyword fallback first."""
        content = f"标题: {title}\n"
        if description:
            content += f"描述: {description[:500]}"

        try:
            message = self.client.messages.create(
                model=self.model,
                max_tokens=50,
                temperature=0,
                system=self.system_prompt,
                messages=[{"role": "user", "content": content}],
            )
            self._consecutive_errors = 0
            category = message.content[0].text.strip()

            if category in self.categories:
                return category
            for cat in self.categories:
                if cat in category:
                    return cat
            # LLM returned something unparseable — treat as soft failure
            return self._fallback(title, description, f"unparseable response: {category[:40]!r}")

        except self._anthropic.RateLimitError as e:
            self._consecutive_errors += 1
            # Adaptive sleep; let the caller's batch loop retry on next item
            sleep = min(5.0, 1.0 * self._consecutive_errors)
            _notify(f"rate_limit — sleeping {sleep:.1f}s")
            time.sleep(sleep)
            return self._fallback(title, description, "rate_limit")
        except Exception as e:
            self._consecutive_errors += 1
            return self._fallback(title, description, f"{_classify_error_kind(e)}: {e}")

    def classify_batch(self, items: List[Dict[str, Any]], batch_size: int = 5) -> List[Dict[str, Any]]:
        """Classify multiple items with rate limiting."""
        classified_items = []
        for i in range(0, len(items), batch_size):
            batch = items[i:i + batch_size]
            for item in batch:
                item["category"] = self.classify_item(
                    item.get("title", ""),
                    item.get("description", ""),
                )
                classified_items.append(item)
            if i + batch_size < len(items):
                time.sleep(0.4)
        return classified_items


class SimpleClassifier:
    """Keyword-based classifier — also used as hot fallback for ContentClassifier."""

    def __init__(self):
        self.keywords = SIMPLE_KEYWORDS

    def classify_item(self, title: str, description: str = "") -> str:
        text = (title + " " + description).lower()
        scores = {cat: 0 for cat in self.keywords}
        for category, keywords in self.keywords.items():
            for keyword in keywords:
                if keyword.lower() in text:
                    scores[category] += 1
        max_score = max(scores.values())
        if max_score > 0:
            return max(scores, key=scores.get)
        return "其他"


def get_classifier(use_llm: bool = True) -> Any:
    """Factory function to get appropriate classifier."""
    if use_llm:
        try:
            return ContentClassifier()
        except ValueError:
            print("⚠️  ANTHROPIC_API_KEY not found, falling back to keyword-based classifier")
            return SimpleClassifier()
    return SimpleClassifier()

import re
import sqlite3
import threading
import time
from typing import List, Optional, Dict, Any
from datetime import datetime
from email.utils import parsedate_to_datetime
import json

_ISO_DATE_RE = re.compile(r'^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}')

# Module-level lock: serialises ALL SQLite opens / writes so that
# Playwright's Chromium sub-process can never hold a WAL shm-lock
# while another thread tries to open the same database.
_DB_LOCK = threading.Lock()


class RSSDatabase:
    def __init__(self, db_path: str = "rss_database.db"):
        self.db_path = db_path
        self.init_database()

    def _connect(self, retries: int = 3):
        """Create a SQLite connection with WAL mode, busy timeout, and retry."""
        last_err = None
        for attempt in range(retries):
            try:
                with _DB_LOCK:
                    conn = sqlite3.connect(self.db_path, timeout=30)
                    conn.execute("PRAGMA journal_mode=WAL")
                    conn.execute("PRAGMA busy_timeout=30000")
                return conn
            except sqlite3.OperationalError as e:
                last_err = e
                if attempt < retries - 1:
                    time.sleep(0.5 * (attempt + 1))
        raise last_err

    def init_database(self):
        """Initialize database schema + idempotent migrations."""
        with self._connect() as conn:
            cursor = conn.cursor()

            # Subscriptions table
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS subscriptions (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    url TEXT UNIQUE NOT NULL,
                    platform TEXT NOT NULL,
                    title TEXT,
                    description TEXT,
                    added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    last_updated TIMESTAMP,
                    active INTEGER DEFAULT 1
                )
            """)

            # Items table - stores all fetched items
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS items (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    item_id TEXT UNIQUE NOT NULL,
                    subscription_id INTEGER NOT NULL,
                    title TEXT,
                    description TEXT,
                    link TEXT,
                    category TEXT,
                    pub_date TIMESTAMP,
                    fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    metadata TEXT,
                    FOREIGN KEY (subscription_id) REFERENCES subscriptions(id)
                )
            """)

            # Create indices for faster lookups
            cursor.execute("CREATE INDEX IF NOT EXISTS idx_item_id ON items(item_id)")
            cursor.execute("CREATE INDEX IF NOT EXISTS idx_subscription_id ON items(subscription_id)")
            cursor.execute("CREATE INDEX IF NOT EXISTS idx_category ON items(category)")

            # --- Migrations (idempotent) ---
            self._migrate_subscriptions(cursor)

            conn.commit()

    @staticmethod
    def _column_names(cursor, table: str) -> set:
        cursor.execute(f"PRAGMA table_info({table})")
        return {row[1] for row in cursor.fetchall()}

    def _migrate_subscriptions(self, cursor) -> None:
        cols = self._column_names(cursor, "subscriptions")
        if "consecutive_failures" not in cols:
            cursor.execute(
                "ALTER TABLE subscriptions ADD COLUMN consecutive_failures INTEGER DEFAULT 0"
            )
        if "last_error" not in cols:
            cursor.execute(
                "ALTER TABLE subscriptions ADD COLUMN last_error TEXT"
            )
        if "last_error_kind" not in cols:
            cursor.execute(
                "ALTER TABLE subscriptions ADD COLUMN last_error_kind TEXT"
            )
        if "last_success_at" not in cols:
            cursor.execute(
                "ALTER TABLE subscriptions ADD COLUMN last_success_at TIMESTAMP"
            )

    def record_fetch_outcome(self, subscription_id: int, *,
                             success: bool, error: Optional[str] = None,
                             error_kind: Optional[str] = None) -> None:
        """Update a subscription's health signals after a fetch attempt.

        - On success: clear consecutive_failures, clear last_error, bump
          last_success_at.
        - On failure: increment consecutive_failures, record error + kind.
        """
        now = datetime.now().isoformat()
        with self._connect() as conn:
            cur = conn.cursor()
            if success:
                cur.execute(
                    "UPDATE subscriptions "
                    "SET consecutive_failures=0, last_error=NULL, last_error_kind=NULL, "
                    "    last_success_at=?, last_updated=? "
                    "WHERE id=?",
                    (now, now, subscription_id),
                )
            else:
                cur.execute(
                    "UPDATE subscriptions "
                    "SET consecutive_failures=COALESCE(consecutive_failures, 0) + 1, "
                    "    last_error=?, last_error_kind=?, last_updated=? "
                    "WHERE id=?",
                    (error, error_kind, now, subscription_id),
                )
            conn.commit()

    def add_subscription(self, url: str, platform: str, title: str = "", description: str = "") -> int:
        """Add a new subscription"""
        with self._connect() as conn:
            cursor = conn.cursor()
            try:
                cursor.execute("""
                    INSERT INTO subscriptions (url, platform, title, description)
                    VALUES (?, ?, ?, ?)
                """, (url, platform, title, description))
                conn.commit()
                return cursor.lastrowid
            except sqlite3.IntegrityError:
                # Subscription already exists, return existing id
                cursor.execute("SELECT id FROM subscriptions WHERE url = ?", (url,))
                return cursor.fetchone()[0]

    def get_subscriptions(self, active_only: bool = True) -> List[Dict[str, Any]]:
        """Get all subscriptions"""
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()

            if active_only:
                cursor.execute("SELECT * FROM subscriptions WHERE active = 1")
            else:
                cursor.execute("SELECT * FROM subscriptions")

            return [dict(row) for row in cursor.fetchall()]

    def item_exists(self, item_id: str) -> bool:
        """Check if an item has already been fetched"""
        with self._connect() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT 1 FROM items WHERE item_id = ?", (item_id,))
            return cursor.fetchone() is not None

    @staticmethod
    def _normalize_date(date_str: Optional[str]) -> str:
        """Normalize any date format to ISO 8601 (YYYY-MM-DDTHH:MM:SS)."""
        if not date_str:
            return datetime.now().isoformat()
        # Already ISO 8601? (e.g. 2026-02-11T09:00:00)
        if _ISO_DATE_RE.match(date_str):
            return date_str
        # Try RFC 822 (e.g. 'Wed, 11 Feb 2026 02:07:30 GMT')
        try:
            dt = parsedate_to_datetime(date_str)
            return dt.isoformat()
        except Exception:
            pass
        # Fallback: store as-is
        return date_str

    def add_item(self, item_id: str, subscription_id: int, title: str,
                 description: str = "", link: str = "", category: str = "",
                 pub_date: Optional[str] = None, metadata: Optional[Dict] = None) -> bool:
        """Add a new item. Uses INSERT OR IGNORE for thread-safe dedup."""
        normalized_date = self._normalize_date(pub_date)

        with self._connect() as conn:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT OR IGNORE INTO items (item_id, subscription_id, title,
                    description, link, category, pub_date, metadata)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """, (item_id, subscription_id, title, description, link, category,
                  normalized_date,
                  json.dumps(metadata) if metadata else None))
            conn.commit()
            return cursor.rowcount > 0

    def get_items_by_category(self, category: Optional[str] = None,
                              since: Optional[str] = None) -> List[Dict[str, Any]]:
        """Get items, optionally filtered by category and date"""
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()

            query = "SELECT * FROM items WHERE 1=1"
            params = []

            if category:
                query += " AND category = ?"
                params.append(category)

            if since:
                query += " AND fetched_at >= ?"
                params.append(since)

            query += " ORDER BY pub_date DESC"

            cursor.execute(query, params)
            return [dict(row) for row in cursor.fetchall()]

    def get_new_items_since(self, since: str) -> List[Dict[str, Any]]:
        """Get all items fetched since a specific timestamp"""
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute("""
                SELECT i.*, s.platform, s.url as subscription_url
                FROM items i
                JOIN subscriptions s ON i.subscription_id = s.id
                WHERE i.fetched_at >= ?
                ORDER BY i.category, i.pub_date DESC
            """, (since,))
            return [dict(row) for row in cursor.fetchall()]

    def get_latest_per_subscription(self) -> list:
        """Get the single latest item for each subscription, sorted by pub_date DESC."""
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute("""
                SELECT i.*, s.platform, s.url as subscription_url, s.title as subscription_title
                FROM items i
                JOIN subscriptions s ON i.subscription_id = s.id
                WHERE i.id = (
                    SELECT i2.id FROM items i2
                    WHERE i2.subscription_id = i.subscription_id
                    ORDER BY i2.pub_date DESC
                    LIMIT 1
                )
                ORDER BY i.pub_date DESC
            """)
            return [dict(row) for row in cursor.fetchall()]

    def update_subscription_timestamp(self, subscription_id: int):
        """Update the last_updated timestamp for a subscription"""
        with self._connect() as conn:
            cursor = conn.cursor()
            cursor.execute("""
                UPDATE subscriptions
                SET last_updated = CURRENT_TIMESTAMP
                WHERE id = ?
            """, (subscription_id,))
            conn.commit()

    def update_subscription_title(self, subscription_id: int, title: str):
        """Auto-update subscription title from feed channel name."""
        with self._connect() as conn:
            cursor = conn.cursor()
            cursor.execute("""
                UPDATE subscriptions SET title = ?
                WHERE id = ? AND (title LIKE '%Subscription%' OR title = '')
            """, (title, subscription_id))
            conn.commit()

    def get_all_items(self) -> List[Dict[str, Any]]:
        """Get all items for RSS feed generation"""
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute("""
                SELECT i.*, s.platform
                FROM items i
                JOIN subscriptions s ON i.subscription_id = s.id
                ORDER BY i.pub_date DESC
                LIMIT 1000
            """)
            return [dict(row) for row in cursor.fetchall()]

#!/usr/bin/env python3
"""
Let's Go RSS - Main entry point
Lightweight RSS subscription manager for multiple platforms.
"""

import sys
import os
from pathlib import Path

# Resolve directories relative to this script
SKILL_DIR = Path(__file__).parent.parent
SCRIPTS_DIR = SKILL_DIR / "scripts"
ASSETS_DIR = SKILL_DIR / "assets"

# Add scripts directory to Python path
sys.path.insert(0, str(SCRIPTS_DIR))

# Ensure assets directory exists
ASSETS_DIR.mkdir(exist_ok=True)


def ensure_dependencies():
    """Check and install dependencies if needed."""
    setup_script = SCRIPTS_DIR / "setup.py"
    if not setup_script.exists():
        return True

    import subprocess
    result = subprocess.run(
        [sys.executable, str(setup_script)],
        capture_output=True, text=True
    )
    if result.returncode != 0:
        print("⚠️  Setup check completed with warnings")
        return False
    return True


def print_cached_status() -> int:
    """Print cached report directly without importing engine dependencies."""
    report_path = ASSETS_DIR / "latest_update.md"
    if report_path.exists():
        print(report_path.read_text(encoding="utf-8"))
        return 0
    print("⚠️ 尚无缓存报告。请先运行 --update 生成。")
    return 0


def run_doctor(auto_fix: bool = False) -> int:
    """Print a health snapshot and optionally auto-fix what can be fixed.

    Auto-fixable: stale rsshub pidfile, unhealthy :1201 (restart worker).
    Not auto-fixable (printed as actionables): XHS cookies expired, missing
    ANTHROPIC_API_KEY.
    """
    import json
    import shutil
    import subprocess as _sp
    import urllib.request

    sys.path.insert(0, str(SCRIPTS_DIR))

    print("# Let's Go RSS — Doctor\n")

    # Runtime deps
    print("## 运行时")
    for name in ("python3", "node", "npx", "yt-dlp"):
        where = shutil.which(name)
        print(f"- {name}: {where or '❌ MISSING'}")
    try:
        import httpx  # noqa: F401
        print("- httpx (py): OK")
    except Exception as e:
        print(f"- httpx (py): ❌ {e}")
    try:
        import anthropic  # noqa: F401
        print("- anthropic (py): OK")
    except Exception as e:
        print(f"- anthropic (py): ⚠️ {e}")

    key = os.environ.get("ANTHROPIC_API_KEY")
    if key:
        print(f"- ANTHROPIC_API_KEY: ✅ present (len={len(key)})")
    else:
        env_path = SKILL_DIR / ".env"
        hint = f" (put it in {env_path})" if not env_path.exists() else ""
        print(f"- ANTHROPIC_API_KEY: ⚠️ missing — classifier will use keyword fallback{hint}")

    # RSSHub
    print("\n## RSSHub")
    for label, url in (("primary (:1201, managed)", "http://127.0.0.1:1201/healthz"),
                       ("fallback (:1200, ready-cowork)", "http://127.0.0.1:1200/healthz")):
        try:
            with urllib.request.urlopen(url, timeout=1.5) as r:
                print(f"- {label}: ✅ {r.status}")
        except Exception as e:
            print(f"- {label}: ❌ {e}")

    # Sub health
    try:
        from database import RSSDatabase
        db = RSSDatabase(str(ASSETS_DIR / "rss_database.db"))
        subs = db.get_subscriptions()
        worst = sorted(
            [s for s in subs if (s.get("consecutive_failures") or 0) > 0],
            key=lambda s: -(s.get("consecutive_failures") or 0),
        )[:5]
        print(f"\n## 源健康 ({len(subs)} 总)")
        if worst:
            print("最不健康的源:")
            for s in worst:
                print(f"- {s.get('title') or s['platform']}: "
                      f"fails={s.get('consecutive_failures')} "
                      f"kind={s.get('last_error_kind')} "
                      f"error={(s.get('last_error') or '')[:80]}")
        else:
            print("全部健康 ✅")
    except Exception as e:
        print(f"⚠️ failed to read DB: {e}")

    # XHS cookies
    xhs_cookie = Path(os.path.expanduser("~/.mcp/rednote/cookies.json"))
    print("\n## XHS cookies")
    if xhs_cookie.exists():
        import time as _t
        age_days = (_t.time() - xhs_cookie.stat().st_mtime) / 86400
        if age_days > 14:
            print(f"- ⚠️ {xhs_cookie} is {age_days:.0f} days old — consider `npx rednote-mcp init`")
        else:
            print(f"- ✅ {xhs_cookie} ({age_days:.1f} days old)")
    else:
        print(f"- ❌ {xhs_cookie} not present — run `npx rednote-mcp init`")

    # Playwright managed browser
    print("\n## Playwright (skill-managed Chromium)")
    try:
        import playwright  # noqa: F401
        print("- playwright (py): OK")
    except Exception:
        print("- playwright (py): ❌ not installed — `pip install playwright && python -m playwright install chromium`")
    pw_platforms = os.environ.get("RSS_PLAYWRIGHT_PLATFORMS", "").strip()
    print(f"- RSS_PLAYWRIGHT_PLATFORMS: {pw_platforms or '(unset — Playwright tier disabled)'}")
    pw_profile = Path(os.environ.get(
        "RSS_PLAYWRIGHT_PROFILE",
        str(Path.home() / ".lets-go-rss" / "browser-profile"),
    ))
    if pw_profile.exists():
        cookies_sqlite = pw_profile / "Default" / "Network" / "Cookies"
        size = cookies_sqlite.stat().st_size if cookies_sqlite.exists() else 0
        print(f"- profile: {pw_profile} (cookies db {size} B)")
    else:
        print(f"- profile: {pw_profile} (not yet initialised)")
    print("  login commands: `python scripts/lets_go_rss.py --login {twitter,xiaohongshu}`")

    # Auto-fix pass
    if auto_fix:
        print("\n## Auto-fix")
        mgr = str(SCRIPTS_DIR / "rsshub_manager.py")
        # Restart managed RSSHub if unhealthy
        try:
            r = _sp.run([sys.executable, mgr, "status"],
                        capture_output=True, text=True, timeout=5)
            info = json.loads(r.stdout or "{}")
            if not info.get("healthy"):
                print("- managed rsshub unhealthy → restart")
                _sp.run([sys.executable, mgr, "restart"], timeout=120)
            else:
                print("- managed rsshub healthy ✅ (no action)")
        except Exception as e:
            print(f"- rsshub check failed: {e}")

        # Clear stale update.lock if no pid is alive
        lock = ASSETS_DIR / ".update.lock"
        if lock.exists():
            try:
                txt = lock.read_text()
                # Best-effort: remove if it's clearly stale (> 2h old)
                age = _t.time() - lock.stat().st_mtime
                if age > 7200:
                    lock.unlink()
                    print(f"- cleared stale update.lock ({age/60:.0f} min old)")
            except Exception:
                pass

    return 0


def main():
    """Main entry point — delegates to rss_engine with correct paths."""
    # Fast path: --status should not require runtime scraping dependencies (e.g. httpx).
    if "--status" in sys.argv:
        sys.exit(print_cached_status())

    # Fast path: --overview reads full_overview.md
    if "--overview" in sys.argv:
        overview_path = ASSETS_DIR / "full_overview.md"
        if overview_path.exists():
            print(overview_path.read_text(encoding="utf-8"))
        else:
            print("⚠️ 尚无全量概览。请先运行 --update --digest 生成。")
        sys.exit(0)

    # Fast path: --doctor / --auto-fix — diagnostics, no scraping.
    if "--doctor" in sys.argv or "--auto-fix" in sys.argv:
        sys.exit(run_doctor(auto_fix=("--auto-fix" in sys.argv)))

    # Fast path: --login <platform> — end-to-end flow:
    # 1. opens visible Chromium for one-time sign-in
    # 2. verifies login by fetching an existing subscription
    # 3. writes RSS_PLAYWRIGHT_PLATFORMS=<...,platform> to .env
    if "--login" in sys.argv:
        idx = sys.argv.index("--login")
        platform = sys.argv[idx + 1] if idx + 1 < len(sys.argv) else ""
        if not platform or platform.startswith("-"):
            print("Usage: python scripts/lets_go_rss.py --login <twitter|xiaohongshu|bilibili>")
            sys.exit(2)
        sys.path.insert(0, str(SCRIPTS_DIR))
        from playwright_adapter import run_login_flow
        db_path = str(ASSETS_DIR / "rss_database.db")
        sys.exit(run_login_flow(platform, skill_dir=SKILL_DIR, db_path=db_path))

    # Skip setup for --skip-setup (cron jobs)
    skip_setup = '--skip-setup' in sys.argv
    if not skip_setup:
        if not ensure_dependencies():
            print("⚠️  Dependency setup had warnings, continuing anyway...")

    # Remove --skip-setup from argv so argparse doesn't complain
    sys.argv = [a for a in sys.argv if a != '--skip-setup']

    # Pass absolute db_path to engine instead of os.chdir()
    db_path = str(ASSETS_DIR / "rss_database.db")
    os.environ.setdefault("RSS_ASSETS_DIR", str(ASSETS_DIR))

    from rss_engine import main as rss_main
    rss_main(db_path=db_path)


if __name__ == "__main__":
    main()

"""
Skill-owned Playwright Chromium — scrapes platforms that anti-scrape RSSHub.

Why this exists
---------------
Bilibili / Twitter / Xiaohongshu all 风控-block server-side scrapers (RSSHub
routes get 503 / JSON error envelopes). A real browser with a normal
reputation gets through. We launch our own headless Chromium via Playwright
(already a pip dep + its Chromium is already cached in
~/Library/Caches/ms-playwright/), using a persistent user-data-dir so
login state (when needed) sticks between runs.

Zero user intervention required for:
  * Bilibili user videos       — public data, no login
  * Bilibili user dynamic      — public for most users
(Twitter timeline / XHS notes do require a one-time login via
 `python scripts/lets_go_rss.py --login <platform>`.)

Enable in .env:
    RSS_PLAYWRIGHT_PLATFORMS=bilibili,xiaohongshu,twitter

Concurrency note
----------------
Playwright's sync API is strict single-thread (the BrowserContext belongs
to the thread that created it). rss_engine uses a ThreadPoolExecutor for
parallel fetches, so we route ALL Playwright work through a single
dedicated worker thread via a ThreadPoolExecutor(max_workers=1). Other
fetch threads submit jobs and block on the returned Future.
"""

from __future__ import annotations

import hashlib
import os
import re
import sys
import threading
import time
from concurrent.futures import Future, ThreadPoolExecutor
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional


# ---------------------------------------------------------------------------
# Config / env
# ---------------------------------------------------------------------------

PROFILE_DIR = Path(os.environ.get(
    "RSS_PLAYWRIGHT_PROFILE",
    str(Path.home() / ".lets-go-rss" / "browser-profile"),
))
DEFAULT_UA = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)


def enabled_platforms() -> set:
    raw = os.environ.get("RSS_PLAYWRIGHT_PLATFORMS", "").strip()
    if not raw:
        return set()
    return {p.strip().lower() for p in raw.split(",") if p.strip()}


def is_platform_enabled(platform: str) -> bool:
    return platform.lower() in enabled_platforms()


# ---------------------------------------------------------------------------
# Dedicated single-thread worker — owns Playwright + the BrowserContext
# ---------------------------------------------------------------------------
#
# Sync Playwright objects are bound to their creator thread, so all calls
# must run on the SAME thread. We use an Executor(max_workers=1) as that
# thread, with thread-local initialization that spins up Playwright on
# first use. Other threads submit jobs via `_submit()` and wait on the
# returned Future.

_worker_init_lock = threading.Lock()
_worker_executor: Optional[ThreadPoolExecutor] = None
_thread_local = threading.local()


def _worker_init():
    """Initializer running inside the worker thread itself — creates
    Playwright + a persistent BrowserContext bound to this thread."""
    try:
        from playwright.sync_api import sync_playwright
    except ImportError as e:
        _thread_local.init_error = (
            "playwright is not installed. Run: pip install playwright && "
            "python -m playwright install chromium"
        )
        return

    PROFILE_DIR.mkdir(parents=True, exist_ok=True)
    try:
        pw = sync_playwright().start()
        ctx = pw.chromium.launch_persistent_context(
            user_data_dir=str(PROFILE_DIR),
            headless=True,
            viewport={"width": 1280, "height": 800},
            user_agent=DEFAULT_UA,
            args=[
                "--disable-blink-features=AutomationControlled",
                "--disable-features=IsolateOrigins,site-per-process",
            ],
            ignore_default_args=["--enable-automation"],
        )
        ctx.add_init_script(_STEALTH_INIT_JS)
        _thread_local.pw = pw
        _thread_local.ctx = ctx
        _thread_local.init_error = None
    except Exception as e:
        _thread_local.init_error = f"playwright init failed: {e}"


def _ensure_executor() -> ThreadPoolExecutor:
    global _worker_executor
    with _worker_init_lock:
        if _worker_executor is None:
            _worker_executor = ThreadPoolExecutor(
                max_workers=1, thread_name_prefix="rss-pw",
                initializer=_worker_init,
            )
            import atexit
            atexit.register(_teardown)
    return _worker_executor


def _teardown():
    global _worker_executor
    if _worker_executor is None:
        return
    # Submit a close job on the worker thread
    try:
        _worker_executor.submit(_close_context).result(timeout=5)
    except Exception:
        pass
    _worker_executor.shutdown(wait=False, cancel_futures=True)
    _worker_executor = None


def _close_context():
    ctx = getattr(_thread_local, "ctx", None)
    pw = getattr(_thread_local, "pw", None)
    try:
        if ctx is not None:
            ctx.close()
    except Exception:
        pass
    try:
        if pw is not None:
            pw.stop()
    except Exception:
        pass


def _run_on_worker(job_fn: Callable[[Any], Any], *, page_timeout: float = 30.0):
    """Execute `job_fn(page)` on the worker thread using the shared context.

    The worker thread owns Playwright; this helper creates a fresh Page,
    passes it to the callback, and closes it afterwards. Errors propagate
    back to the caller through the Future.
    """
    def _run():
        if getattr(_thread_local, "init_error", None):
            raise RuntimeError(_thread_local.init_error)
        ctx = getattr(_thread_local, "ctx", None)
        if ctx is None:
            raise RuntimeError("playwright worker not initialised")
        page = ctx.new_page()
        page.set_default_navigation_timeout(int(page_timeout * 1000))
        try:
            return job_fn(page)
        finally:
            try:
                page.close()
            except Exception:
                pass

    exec_ = _ensure_executor()
    fut = exec_.submit(_run)
    # Cap wait at page_timeout + a margin; lets a stuck call fail loudly
    return fut.result(timeout=page_timeout + 10.0)


# ---------------------------------------------------------------------------
# Item mapping helpers
# ---------------------------------------------------------------------------

def _item_id(platform: str, seed: str) -> str:
    return f"{platform}_{hashlib.md5(seed.encode()).hexdigest()[:12]}"


def _clip(s: Any, n: int = 500) -> str:
    if not s:
        return ""
    s = str(s)
    return s[:n]


# ---------------------------------------------------------------------------
# Platform: Bilibili — public, no login needed
# ---------------------------------------------------------------------------

def fetch_bilibili_user(uid: str, timeout: float = 25.0) -> List[Dict[str, Any]]:
    """Fetch a Bilibili UP's recent videos.

    Strategy: visit space.bilibili.com/<uid>, block on the XHR response to
    /x/space/(wbi/)?arc/search (contains vlist). Bilibili's anti-bot
    happily serves real cookies to a legit browser, so no login is needed.
    """
    def _is_arc_search(response) -> bool:
        u = response.url
        return ("/x/space/wbi/arc/search" in u
                or "/x/space/arc/search" in u)

    def _job(page):
        with page.expect_response(_is_arc_search, timeout=int(timeout * 1000)) as info:
            page.goto(
                f"https://space.bilibili.com/{uid}",
                wait_until="domcontentloaded",
            )
        response = info.value
        if response.status != 200:
            raise RuntimeError(f"arc/search returned HTTP {response.status}")
        data = response.json()
        if not isinstance(data, dict) or data.get("code") != 0:
            code = data.get("code") if isinstance(data, dict) else None
            msg = data.get("message") if isinstance(data, dict) else None
            raise RuntimeError(f"arc/search code={code} msg={msg}")
        return (data.get("data", {}).get("list", {}).get("vlist") or [])

    # Bilibili's anti-bot occasionally 412s — retry once after a cool-down
    # the browser state usually clears in 2-3s.
    vlist = None
    last_err: Optional[Exception] = None
    for attempt in range(2):
        try:
            vlist = _run_on_worker(_job, page_timeout=timeout)
            if vlist:
                break
        except Exception as e:
            last_err = e
            msg = str(e).lower()
            # Only retry on transient-looking errors
            if "412" not in msg and "timeout" not in msg and "timed out" not in msg:
                raise
        if attempt == 0:
            time.sleep(3.0)

    if not vlist:
        if last_err:
            raise last_err
        raise RuntimeError("Bilibili arc/search returned empty vlist")

    items: List[Dict[str, Any]] = []
    for v in vlist[:20]:
        bvid = v.get("bvid") or ""
        link = f"https://www.bilibili.com/video/{bvid}" if bvid else ""
        title = v.get("title") or ""
        # Bilibili `created` is unix seconds
        created = v.get("created")
        pub_date = ""
        if created:
            try:
                from datetime import datetime as _dt
                pub_date = _dt.fromtimestamp(int(created)).isoformat()
            except Exception:
                pass
        items.append({
            "item_id": _item_id("bilibili", bvid or link or title),
            "title": title,
            "description": _clip(v.get("description")),
            "link": link,
            "pub_date": pub_date,
            "metadata": {
                "source": "playwright:bilibili",
                "_channel_title": v.get("author") or "",
                "duration": v.get("length") or "",
                "play": v.get("play") or 0,
            },
        })
    return items


# ---------------------------------------------------------------------------
# Platform: Xiaohongshu — requires one-time login (note: platform constraint)
# ---------------------------------------------------------------------------

def fetch_xhs_user(user_id: str, timeout: float = 25.0) -> List[Dict[str, Any]]:
    """Fetch an XHS user's recent notes via our managed profile.

    XHS renders the profile's notes list server-side into DOM (no user_posted
    XHR any more as of 2025+). We parse `section.note-item` cards directly.
    If the profile is not logged in, XHS redirects to captcha/login — raised
    so the error_kind is tagged `auth`.
    """
    def _job(page):
        page.goto(
            f"https://www.xiaohongshu.com/user/profile/{user_id}",
            wait_until="domcontentloaded",
        )
        final_url = page.url
        if "captcha" in final_url or "login" in final_url:
            raise RuntimeError(
                "XHS profile not logged in — run: "
                "python scripts/lets_go_rss.py --login xiaohongshu"
            )
        # Let the note grid render (lazy + image loading doesn't matter to us)
        try:
            page.wait_for_selector("section.note-item", timeout=8000)
        except Exception:
            pass
        # Extract note cards in a single evaluate (robust to DOM churn)
        return page.evaluate("""() => {
            const out = [];
            const cards = document.querySelectorAll('section.note-item');
            for (const c of cards) {
                const a = c.querySelector("a[href*='/explore/']") ||
                          c.querySelector("a[href*='/user/profile/']");
                if (!a) continue;
                const href = a.getAttribute('href') || '';
                const m = href.match(/\\/explore\\/([0-9a-f]+)|\\/profile\\/[^/]+\\/([0-9a-f]+)/i);
                const noteId = m ? (m[1] || m[2]) : '';
                if (!noteId) continue;
                // Title: first non-empty innerText line that isn't the pinned badge
                const lines = (c.innerText || '').split('\\n').map(s => s.trim()).filter(Boolean);
                const title = lines.find(l => l !== '置顶') || '';
                out.push({ note_id: noteId, title });
                if (out.length >= 20) break;
            }
            return out;
        }""")

    raw = _run_on_worker(_job, page_timeout=timeout)
    if not raw:
        raise RuntimeError("XHS DOM extraction returned no notes")

    items: List[Dict[str, Any]] = []
    for n in raw:
        note_id = n.get("note_id", "")
        title = (n.get("title") or "").strip()
        if not note_id:
            continue
        link = f"https://www.xiaohongshu.com/explore/{note_id}"
        items.append({
            "item_id": _item_id("xiaohongshu", note_id),
            "title": title,
            "description": "",
            "link": link,
            "pub_date": "",
            "metadata": {"source": "playwright:xiaohongshu"},
        })
    return items


# ---------------------------------------------------------------------------
# Platform: Twitter/X — requires one-time login
# ---------------------------------------------------------------------------

def fetch_twitter_user(username: str, timeout: float = 25.0) -> List[Dict[str, Any]]:
    """Fetch an X/Twitter user's recent tweets via our managed profile.

    Twitter requires login to view timelines as of 2024+. After one-time
    `--login twitter`, cookies persist in the profile.
    """
    username = username.lstrip("@")
    captured: Dict[str, Any] = {"tweets": []}

    tweet_pattern = re.compile(r"status/(\d+)")

    def _on_response(response):
        try:
            url = response.url
            if "UserTweets" in url and response.status == 200:
                data = response.json()
                # Twitter GraphQL: dig through timeline instructions
                entries = _extract_tweet_entries(data)
                if entries:
                    captured["tweets"].extend(entries)
        except Exception:
            pass

    def _job(page):
        page.on("response", _on_response)
        page.goto(f"https://x.com/{username}", wait_until="domcontentloaded")
        final = page.url
        if "login" in final or "/i/flow/login" in final:
            raise RuntimeError(
                "Twitter not logged in — run: "
                "python scripts/lets_go_rss.py --login twitter"
            )
        deadline = time.time() + 10.0
        while time.time() < deadline and not captured["tweets"]:
            page.wait_for_timeout(500)
        return captured["tweets"]

    tweets = _run_on_worker(_job, page_timeout=timeout)
    if not tweets:
        raise RuntimeError("Twitter XHR capture failed (timeline empty or blocked)")

    items: List[Dict[str, Any]] = []
    seen_ids: set = set()
    for t in tweets[:30]:
        tid = t.get("rest_id") or t.get("id_str")
        if not tid or tid in seen_ids:
            continue
        seen_ids.add(tid)
        text = t.get("full_text") or ""
        link = f"https://x.com/{username}/status/{tid}"
        items.append({
            "item_id": _item_id("twitter", tid),
            "title": (text[:120] + "…") if len(text) > 120 else text,
            "description": _clip(text),
            "link": link,
            "pub_date": t.get("created_at") or "",
            "metadata": {"source": "playwright:twitter"},
        })
        if len(items) >= 20:
            break
    return items


def _extract_tweet_entries(payload: Any) -> List[Dict[str, Any]]:
    """Walk Twitter's GraphQL timeline JSON, yielding tweet result dicts.
    Layout is nested: data.user.result.timeline_v2.timeline.instructions[].entries[].
    Each entry contains content.itemContent.tweet_results.result with rest_id + legacy.full_text.
    """
    out: List[Dict[str, Any]] = []
    try:
        instructions = (
            payload.get("data", {})
            .get("user", {})
            .get("result", {})
            .get("timeline_v2", {})
            .get("timeline", {})
            .get("instructions", [])
        )
    except AttributeError:
        return out

    for ins in instructions:
        for entry in ins.get("entries", []) or []:
            content = entry.get("content", {}) or {}
            item_content = content.get("itemContent") or {}
            tw = (item_content.get("tweet_results") or {}).get("result") or {}
            if not tw:
                continue
            legacy = tw.get("legacy") or {}
            out.append({
                "rest_id": tw.get("rest_id") or legacy.get("id_str"),
                "id_str": legacy.get("id_str"),
                "full_text": legacy.get("full_text") or "",
                "created_at": legacy.get("created_at") or "",
            })
    return out


# ---------------------------------------------------------------------------
# Interactive login — opens a visible browser for the user to sign in
# ---------------------------------------------------------------------------

LOGIN_URLS = {
    "twitter": "https://x.com/i/flow/login",
    "xiaohongshu": "https://www.xiaohongshu.com/",
    "bilibili": "https://passport.bilibili.com/login",
}


_STEALTH_INIT_JS = """
// Hide classic automation signals that sites like x.com check on keystroke.
Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
// Fill in plugins / languages to match a real browser session
Object.defineProperty(navigator, 'plugins', { get: () => [1,2,3,4,5] });
Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN','zh','en-US','en'] });
// Provide a believable chrome.runtime object
window.chrome = window.chrome || { runtime: {}, app: {}, csi: function(){}, loadTimes: function(){} };
// Permissions.query('notifications') must behave like a real browser
const origQuery = navigator.permissions && navigator.permissions.query;
if (origQuery) {
  navigator.permissions.query = (params) =>
    params && params.name === 'notifications'
      ? Promise.resolve({ state: Notification.permission })
      : origQuery.call(navigator.permissions, params);
}
"""


def _open_login_window(platform: str) -> None:
    """Open a visible real Chrome pointed at the platform's login page.
    Blocks until the user closes the window. Cookies persist in our profile.

    Uses `channel='chrome'` (the user's real Google Chrome binary) when
    available, falling back to bundled Chromium. Real Chrome is dramatically
    harder for sites like x.com to fingerprint as automation. We also strip
    the `--enable-automation` launch flag and inject a stealth init script
    (navigator.webdriver / plugins / languages / chrome.runtime).
    """
    _teardown()  # release any headless context first

    from playwright.sync_api import sync_playwright
    PROFILE_DIR.mkdir(parents=True, exist_ok=True)

    def _launch(pw, channel: Optional[str]):
        kwargs = dict(
            user_data_dir=str(PROFILE_DIR),
            headless=False,
            viewport={"width": 1280, "height": 800},
            user_agent=DEFAULT_UA,
            args=[
                "--disable-blink-features=AutomationControlled",
                "--disable-features=IsolateOrigins,site-per-process",
            ],
            ignore_default_args=["--enable-automation"],
        )
        if channel:
            kwargs["channel"] = channel
        return pw.chromium.launch_persistent_context(**kwargs)

    with sync_playwright() as pw:
        ctx = None
        last_err = None
        # Prefer real Chrome, then stable Chrome channels, then bundled Chromium
        for channel in ("chrome", "chrome-beta", "chrome-dev", None):
            try:
                ctx = _launch(pw, channel)
                if channel:
                    print(f"[login] launched real Chrome (channel={channel})")
                else:
                    print("[login] launched bundled Chromium (no real Chrome found)")
                break
            except Exception as e:
                last_err = e
                continue
        if ctx is None:
            raise RuntimeError(f"could not launch any Chrome channel: {last_err}")

        ctx.add_init_script(_STEALTH_INIT_JS)
        page = ctx.new_page()
        page.goto(LOGIN_URLS[platform])
        print(f"[login] opened {LOGIN_URLS[platform]}")
        print("[login] sign in, then close the Chromium window to continue.")
        try:
            page.wait_for_event("close", timeout=0)
        except Exception:
            pass
        ctx.close()


def _pick_test_user(platform: str, db_path: str) -> Optional[str]:
    """Return a user_id/username from an existing DB subscription of this
    platform, for post-login verification. Returns None if no sub exists."""
    import sqlite3
    try:
        conn = sqlite3.connect(db_path)
        cur = conn.cursor()
        cur.execute(
            "SELECT url FROM subscriptions WHERE platform=? LIMIT 1",
            (platform,),
        )
        row = cur.fetchone()
        conn.close()
    except Exception:
        return None
    if not row:
        return None
    url = row[0]
    # Reuse the scraper's own extract_user_id logic
    try:
        sys.path.insert(0, str(Path(__file__).parent))
        from scrapers import ScraperFactory
        scraper = ScraperFactory.get_scraper(platform)
        if scraper is None:
            return None
        return scraper.extract_user_id(url)
    except Exception:
        return None


def _verify_platform(platform: str, db_path: str) -> Dict[str, Any]:
    """Run a test fetch against an existing subscription of this platform.

    Returns {"ok": bool, "detail": str, "items": int}."""
    user_id = _pick_test_user(platform, db_path)
    if not user_id:
        return {"ok": False, "detail": "no existing subscription to test against", "items": 0}

    fetchers = {
        "bilibili": fetch_bilibili_user,
        "xiaohongshu": fetch_xhs_user,
        "twitter": fetch_twitter_user,
    }
    fn = fetchers.get(platform)
    if not fn:
        return {"ok": False, "detail": f"no fetcher for {platform}", "items": 0}

    try:
        items = fn(user_id)
    except Exception as e:
        return {"ok": False, "detail": str(e)[:150], "items": 0}
    return {"ok": bool(items), "detail": f"fetched user={user_id}", "items": len(items)}


def _enable_platform_in_env(platform: str, env_path: Path) -> bool:
    """Idempotently add `platform` to RSS_PLAYWRIGHT_PLATFORMS in .env.

    Creates .env if missing. Returns True if the file was written."""
    lines: List[str] = []
    if env_path.exists():
        lines = env_path.read_text(encoding="utf-8").splitlines()

    # Find existing active line (not a comment)
    key = "RSS_PLAYWRIGHT_PLATFORMS"
    found_idx = -1
    current_list: List[str] = []
    for i, line in enumerate(lines):
        stripped = line.lstrip()
        if stripped.startswith("#"):
            continue
        if stripped.startswith(f"{key}="):
            found_idx = i
            raw = stripped.split("=", 1)[1].strip().strip('"').strip("'")
            current_list = [p.strip() for p in raw.split(",") if p.strip()]
            break

    if platform in current_list:
        return False  # already enabled; no write needed

    # If no existing value, seed with the cron default so we don't silently
    # demote bilibili (which run_update_cron.sh enables by default).
    if not current_list:
        current_list = ["bilibili"]

    new_list = current_list + [p for p in [platform] if p not in current_list]
    new_line = f"{key}={','.join(new_list)}"
    if found_idx >= 0:
        lines[found_idx] = new_line
    else:
        if lines and lines[-1].strip():
            lines.append("")
        lines.append(f"# Added by --login {platform} verification flow")
        lines.append(new_line)

    env_path.parent.mkdir(parents=True, exist_ok=True)
    env_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return True


def run_login_flow(platform: str, *, skill_dir: Path, db_path: str,
                   verify: bool = True, enable: bool = True) -> int:
    """End-to-end: open browser → user logs in → verify → persist in .env.

    This is the one-stop "set up platform X" entry point for an agent to
    orchestrate. Returns 0 on success, nonzero on failure.
    """
    platform = platform.lower()
    if platform not in LOGIN_URLS:
        print(f"Unknown platform: {platform!r}. Supported: {list(LOGIN_URLS)}",
              file=sys.stderr)
        return 2

    print(f"\n[{platform}] step 1/3: opening Chromium for you to sign in …")
    _open_login_window(platform)

    if not verify:
        print(f"[{platform}] step 2/3: skipped (verify=False)")
    else:
        print(f"[{platform}] step 2/3: verifying login by fetching a real sub …")
        result = _verify_platform(platform, db_path)
        if not result["ok"]:
            print(f"[{platform}] ❌ verification failed: {result['detail']}")
            print(f"[{platform}] cookies may not have been saved, or platform")
            print(f"[{platform}] still flags us as non-logged-in. Try again.")
            return 3
        print(f"[{platform}] ✅ verified: got {result['items']} items ({result['detail']})")

    if not enable:
        print(f"[{platform}] step 3/3: skipped (enable=False)")
        return 0

    env_path = skill_dir / ".env"
    changed = _enable_platform_in_env(platform, env_path)
    if changed:
        print(f"[{platform}] ✅ step 3/3: enabled in {env_path}")
    else:
        print(f"[{platform}] step 3/3: already enabled in {env_path} (no change)")
    return 0


def login_platform(platform: str) -> int:
    """Back-compat wrapper — opens the browser but does NOT verify/enable.
    Prefer `run_login_flow` for the end-to-end path."""
    platform = platform.lower()
    if platform not in LOGIN_URLS:
        print(f"Unknown platform: {platform!r}. Supported: {list(LOGIN_URLS)}",
              file=sys.stderr)
        return 2
    _open_login_window(platform)
    print(f"[login] done — cookies saved to {PROFILE_DIR}")
    return 0

"""
Markdown report generator
Creates formatted latest_update.md with categorized content
"""

import os
from typing import List, Dict, Any
from datetime import datetime
from collections import defaultdict


class MarkdownReportGenerator:
    """Generate markdown reports for RSS updates"""

    def __init__(self):
        self.categories = ["科技", "人文", "设计", "娱乐", "其他"]
        self.platform_emojis = {
            "bilibili": "📺",
            "xiaohongshu": "📕",
            "weibo": "📱",
            "youtube": "🎬",
            "vimeo": "🎥",
            "behance": "🎨",
            "douyin": "🎵",
            "twitter": "🐦",
            "zsxq": "⭐",
        }

    def generate_update_report(self, new_items: List[Dict[str, Any]],
                               output_path: str = "latest_update.md",
                               digest: bool = False) -> str:
        """Generate latest update report.
        
        Args:
            digest: If True, generate delta-only report (changed accounts only).
                    If no new items, output a single-line "无更新".
        """

        if not new_items:
            content = self._generate_empty_report()
        elif digest:
            content = self._generate_delta_report(new_items, output_dir=os.path.dirname(output_path) or ".")
        else:
            content = self._generate_full_report(new_items)

        with open(output_path, "w", encoding="utf-8") as f:
            f.write(content)

        return output_path

    def generate_full_overview(self, all_items: List[Dict[str, Any]],
                               output_path: str = "full_overview.md") -> str:
        """Generate full overview report — all accounts, latest 1 item each.
        
        This file is for querying (--overview), NOT for push notifications.
        """
        content = self._generate_overview_report(all_items)
        with open(output_path, "w", encoding="utf-8") as f:
            f.write(content)
        return output_path

    def _build_snapshot(self, items: List[Dict[str, Any]], output_dir: str) -> tuple:
        """Build current snapshot and load previous one for comparison.
        
        Returns: (by_account, changed_keys, current_snapshot)
        """
        import json
        from collections import OrderedDict

        # Group by subscription_url (= per account), keep newest
        by_account = OrderedDict()
        for item in items:
            key = item.get("subscription_url", item.get("platform", "unknown"))
            if key not in by_account:
                by_account[key] = item

        # Load previous digest snapshot
        snapshot_path = os.path.join(output_dir, "last_digest.json")
        prev_snapshot = {}
        try:
            with open(snapshot_path, "r", encoding="utf-8") as f:
                prev_snapshot = json.load(f)  # {sub_url: item_id}
        except (FileNotFoundError, json.JSONDecodeError):
            pass

        # Determine which accounts have new content
        changed_keys = set()
        current_snapshot = {}
        for sub_url, item in by_account.items():
            item_id = item.get("item_id", "")
            current_snapshot[sub_url] = item_id
            if item_id != prev_snapshot.get(sub_url):
                changed_keys.add(sub_url)

        # Save current snapshot for next comparison
        with open(snapshot_path, "w", encoding="utf-8") as f:
            json.dump(current_snapshot, f, ensure_ascii=False)

        return by_account, changed_keys, current_snapshot

    def _format_account_line(self, item: Dict[str, Any], tag: str = "") -> List[str]:
        """Format a single account entry for reports."""
        platform = item.get("platform", "").lower()
        emoji = self.platform_emojis.get(platform, "🔗")
        title = item.get("title", "Untitled")
        link = item.get("link", "")
        sub_title = item.get("subscription_title", "")
        account = sub_title if sub_title and "Subscription" not in sub_title else ""
        name = account or platform.title()

        # Format pub_date if available
        pub_date_str = ""
        raw_date = item.get("pub_date", "")
        if raw_date:
            try:
                from dateutil import parser as dateparser
                dt = dateparser.parse(raw_date)
                pub_date_str = dt.strftime("%m-%d %H:%M")
            except Exception:
                pub_date_str = raw_date[:10] if len(raw_date) >= 10 else ""

        date_suffix = f"  {pub_date_str}" if pub_date_str else ""
        lines = [f"{tag}{emoji} {name}{date_suffix}"]

        if link:
            lines.append(f"   [{title}]({link})")
        else:
            lines.append(f"   {title}")
        lines.append("")
        return lines

    def _generate_delta_report(self, items: List[Dict[str, Any]],
                               output_dir: str = ".") -> str:
        """Generate delta-only report — ONLY accounts with new content.

        This is the file used for push notifications (Feishu, etc.).
        Only changed accounts appear; unchanged accounts are omitted.
        """
        by_account, changed_keys, _ = self._build_snapshot(items, output_dir)

        now = datetime.now().strftime("%Y-%m-%d %H:%M")
        new_count = len(changed_keys)

        if not new_count:
            return "这会rss么的更新"

        header = f"白，rss有更新了！\n\n📡 RSS 增量更新 | {now} | {new_count} 个账号有新内容"
        lines = [header, ""]

        # Show ONLY changed accounts
        for sub_url, item in by_account.items():
            if sub_url in changed_keys:
                lines.extend(self._format_account_line(item, tag="🆕 "))

        return "\n".join(lines)

    def _generate_overview_report(self, items: List[Dict[str, Any]]) -> str:
        """Generate full overview report — ALL accounts, no change markers.

        This file is for user queries (--overview), not for push.
        """
        from collections import OrderedDict

        by_account = OrderedDict()
        for item in items:
            key = item.get("subscription_url", item.get("platform", "unknown"))
            if key not in by_account:
                by_account[key] = item

        now = datetime.now().strftime("%Y-%m-%d %H:%M")
        total = len(by_account)
        header = f"📡 RSS 全量概览 | {now} | {total} 个订阅"
        lines = [header, ""]

        for sub_url, item in by_account.items():
            lines.extend(self._format_account_line(item))

        return "\n".join(lines)

    def _generate_empty_report(self) -> str:
        """Generate report when no new items"""
        return f"""# RSS 更新报告

**生成时间**: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}

**更新状态**: 无新内容

本次更新未发现新内容。

---
*Generated by Universal RSS Engine*
"""

    def _generate_full_report(self, new_items: List[Dict[str, Any]]) -> str:
        """Generate full report with categorized items"""

        # Group items by category
        categorized = defaultdict(list)
        for item in new_items:
            category = item.get("category", "其他")
            categorized[category].append(item)

        # Sort categories
        sorted_categories = []
        for cat in self.categories:
            if cat in categorized:
                sorted_categories.append(cat)

        # Generate markdown
        lines = [
            "# RSS 更新报告",
            "",
            f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            "",
            f"**新增内容**: {len(new_items)} 条",
            "",
        ]

        # Table of contents
        lines.append("## 目录")
        lines.append("")
        for category in sorted_categories:
            count = len(categorized[category])
            lines.append(f"- [{category}](#{category}) ({count}条)")
        lines.append("")
        lines.append("---")
        lines.append("")

        # Content by category
        for category in sorted_categories:
            items = categorized[category]
            lines.append(f"## {category}")
            lines.append("")
            lines.append(f"*共 {len(items)} 条新内容*")
            lines.append("")

            for item in items:
                lines.extend(self._format_item(item))
                lines.append("")

            lines.append("---")
            lines.append("")

        # Statistics
        lines.append("## 统计信息")
        lines.append("")
        lines.append("| 分类 | 数量 |")
        lines.append("|------|------|")
        for category in sorted_categories:
            lines.append(f"| {category} | {len(categorized[category])} |")
        lines.append("")

        # Platform statistics
        platform_stats = defaultdict(int)
        for item in new_items:
            platform = item.get("platform", "unknown")
            platform_stats[platform] += 1

        lines.append("### 平台分布")
        lines.append("")
        lines.append("| 平台 | 数量 |")
        lines.append("|------|------|")
        for platform, count in sorted(platform_stats.items(), key=lambda x: x[1], reverse=True):
            emoji = self.platform_emojis.get(platform, "🔗")
            lines.append(f"| {emoji} {platform.title()} | {count} |")
        lines.append("")

        lines.append("---")
        lines.append("*Generated by Universal RSS Engine*")

        return "\n".join(lines)

    def _format_item(self, item: Dict[str, Any]) -> List[str]:
        """Format a single item for markdown"""
        lines = []

        # Platform emoji
        platform = item.get("platform", "").lower()
        emoji = self.platform_emojis.get(platform, "🔗")

        # Title and link
        title = item.get("title", "Untitled")
        link = item.get("link", "")

        if link:
            lines.append(f"### {emoji} [{title}]({link})")
        else:
            lines.append(f"### {emoji} {title}")

        lines.append("")

        # Description
        description = item.get("description", "")
        if description:
            # Limit description length
            desc_preview = description[:200] + "..." if len(description) > 200 else description
            lines.append(f"> {desc_preview}")
            lines.append("")

        # Metadata
        metadata_parts = []

        # Platform
        if platform:
            metadata_parts.append(f"**平台**: {platform.title()}")

        # Date
        pub_date = item.get("pub_date", "")
        if pub_date:
            try:
                if isinstance(pub_date, str):
                    dt = datetime.fromisoformat(pub_date.replace("Z", "+00:00"))
                    formatted_date = dt.strftime("%Y-%m-%d %H:%M")
                    metadata_parts.append(f"**发布时间**: {formatted_date}")
            except:
                pass

        if metadata_parts:
            lines.append(" | ".join(metadata_parts))
            lines.append("")

        return lines

    def _build_health_section(self, subscriptions: List[Dict[str, Any]],
                               out_dir: str) -> List[str]:
        """Build the 🩺 health section: current-run status + stale sources."""
        import json

        # Load last run health if present
        run_health = None
        health_path = os.path.join(out_dir, ".last_run_health.json")
        try:
            with open(health_path, "r", encoding="utf-8") as f:
                run_health = json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            pass

        # Compute staleness from last_success_at (preferred) or last_updated.
        # last_success_at only advances on actual success, so it accurately
        # flags sources that have been failing for a while.
        stale_threshold_days = 3
        now = datetime.now()
        stale_rows = []  # (days_since, sub)
        for sub in subscriptions:
            signal = sub.get("last_success_at") or sub.get("last_updated")
            if not signal:
                stale_rows.append((None, sub))
                continue
            try:
                dt = datetime.fromisoformat(str(signal).replace("Z", "+00:00"))
                if dt.tzinfo is not None:
                    dt = dt.replace(tzinfo=None)
                days = (now - dt).days
                if days >= stale_threshold_days:
                    stale_rows.append((days, sub))
            except Exception:
                stale_rows.append((None, sub))

        # Nothing to show
        if not run_health and not stale_rows:
            return []

        out = ["## 🩺 健康度", ""]

        if run_health:
            per_source = run_health.get("per_source", {})
            total = len(per_source)
            ok = sum(1 for s in per_source.values() if s.get("status") == "ok")
            no_new = sum(1 for s in per_source.values() if s.get("status") == "no_new")
            errors = sum(1 for s in per_source.values() if s.get("status") == "error")
            elapsed = run_health.get("elapsed_sec", "?")
            new_count = run_health.get("total_new", 0)
            out.append(
                f"**本次运行** (耗时 {elapsed}s)"
                f":✅ {ok} 正常 / → {no_new} 无新增 / ❌ {errors} 失败"
                f"  (共 {total} 源,本轮新增 {new_count} 条)"
            )
            out.append("")

            failed = [
                (url, info) for url, info in per_source.items()
                if info.get("status") == "error"
            ]
            if failed:
                out.append("**失败源**:")
                out.append("")
                for url, info in failed:
                    platform = info.get("platform", "").lower()
                    emoji = self.platform_emojis.get(platform, "🔗")
                    name = info.get("title") or platform.title()
                    err_short = (info.get("error") or "").split("\n")[0][:120]
                    out.append(f"- {emoji} **{name}** — {err_short}")
                    out.append(f"  `{url}`")
                out.append("")

        # Stale sources (>= 3 days)
        if stale_rows:
            # Sort: longest stale first; "None" (never updated) at the bottom
            stale_rows.sort(key=lambda x: (x[0] is None, -(x[0] or 0)))
            out.append(f"**陈旧源** (超过 {stale_threshold_days} 天未成功更新):")
            out.append("")
            for days, sub in stale_rows:
                platform = (sub.get("platform") or "").lower()
                emoji = self.platform_emojis.get(platform, "🔗")
                name = sub.get("title") or platform.title()
                age_str = "从未更新" if days is None else f"{days} 天前"
                out.append(f"- {emoji} **{name}** — 上次成功: {age_str}")
            out.append("")

        out.append("---")
        out.append("")
        return out

    def generate_summary_report(self, db, output_path: str = "summary.md") -> str:
        """Generate overall summary report"""

        subscriptions = db.get_subscriptions()
        all_items = db.get_all_items()

        lines = [
            "# RSS 订阅总览",
            "",
            f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            "",
        ]

        # Health section (🩺) — reads .last_run_health.json from same dir
        health_lines = self._build_health_section(
            subscriptions, os.path.dirname(output_path) or "."
        )
        if health_lines:
            lines.extend(health_lines)

        # Subscription statistics
        lines.append("## 订阅统计")
        lines.append("")
        lines.append(f"**总订阅数**: {len(subscriptions)}")
        lines.append(f"**总内容数**: {len(all_items)}")
        lines.append("")

        # Subscriptions by platform
        platform_subs = defaultdict(int)
        for sub in subscriptions:
            platform_subs[sub.get("platform", "unknown")] += 1

        lines.append("### 按平台分布")
        lines.append("")
        lines.append("| 平台 | 订阅数 |")
        lines.append("|------|--------|")
        for platform, count in sorted(platform_subs.items(), key=lambda x: x[1], reverse=True):
            emoji = self.platform_emojis.get(platform, "🔗")
            lines.append(f"| {emoji} {platform.title()} | {count} |")
        lines.append("")

        # Category statistics
        category_items = defaultdict(int)
        for item in all_items:
            category_items[item.get("category", "其他")] += 1

        lines.append("### 按分类分布")
        lines.append("")
        lines.append("| 分类 | 内容数 |")
        lines.append("|------|--------|")
        for category in self.categories:
            if category in category_items:
                lines.append(f"| {category} | {category_items[category]} |")
        lines.append("")

        # Subscriptions list
        lines.append("## 订阅列表")
        lines.append("")

        # Group by platform
        platform_groups = defaultdict(list)
        for sub in subscriptions:
            platform_groups[sub.get("platform", "unknown")].append(sub)

        for platform in sorted(platform_groups.keys()):
            emoji = self.platform_emojis.get(platform, "🔗")
            lines.append(f"### {emoji} {platform.title()}")
            lines.append("")

            for sub in platform_groups[platform]:
                title = sub.get("title") or sub.get("url", "")
                url = sub.get("url", "")
                last_updated = sub.get("last_updated", "从未更新")

                lines.append(f"- **{title}**")
                lines.append(f"  - URL: `{url}`")
                lines.append(f"  - 最后更新: {last_updated}")
                lines.append("")

        lines.append("---")
        lines.append("*Generated by Universal RSS Engine*")

        content = "\n".join(lines)

        with open(output_path, "w", encoding="utf-8") as f:
            f.write(content)

        return output_path