### 1.3 解决的核心问题

| 问题 | 解决方案 |
|------|---------|
| AI 知识陈旧 | 实时搜索 30 天内的最新讨论 |
| 信息孤岛 | 跨 8 个平台聚合，发现跨平台趋势 |
| 搜索噪音 | 多维度评分算法排序，过滤低质量内容 |
| 缺乏背景 | 发现@handle、subreddit 等社区生态 |
| 预测盲区 | Polymarket 预测市场提供「真金白银」的情绪指标 |

---

## 二、架构分析：技术深度解析

### 2.1 两阶段搜索架构

last30days-skill 采用了**两阶段搜索架构**，这是其与普通搜索工具的核心差异。

#### 第一阶段：广度发现

**Reddit 搜索**（via ScrapeCreators）：

```text

**X/Twitter 搜索**（三条路径，按优先级）：

| 路径 | API | 认证方式 | 特点 |
|------|-----|---------|------|
| 1. 捆绑 Bird 客户端 | Twitter GraphQL | AUTH_TOKEN + CT0 Cookie | 本地运行，无需浏览器 |
| 2. xAI 后备 | api.x.ai | XAI_API_KEY | 无需 Cookie，但需 xAI 账号 |
| 3. OpenAI 后备 | api.openai.com | OPENAI_API_KEY | 仅作为最后备选 |

**YouTube 搜索**：

```text

**Hacker News**（via Algolia）：

```text

**Polymarket 预测市场**（via Gamma）：

```text

#### 第二阶段：智能补充搜索

这是 v2.0 引入的关键创新。

**问题**：第一阶段搜索可能遗漏「当事人自己的帖子」（当事人账号发帖时不带话题关键词）

**解决方案**：

```text

### 2.2 多维度评分算法

**评分公式**（v2.9+）：

```text

| 系数 | 指标 | 权重理由 |
|------|------|---------|
| 0.50 | score（ upvotes/points） | 主要参与度指标 |
| 0.35 | comments | 讨论深度 |
| 0.05 | ratio（顶评/总评） | 社区质量信号 |
| 0.10 | top_comment_score | 优质回复的放大效应 |

**为什么用 log1p 而非原始值？**

- 避免大 V 帖子主导（10k upvotes 和 100 upvotes 差距被压缩）
- 让小众社区的高质量讨论也能浮现

**Polymarket 专用评分**（5 因子加权）：

| 因子 | 权重 | 说明 |
|------|------|------|
| text_relevance | 30% | 与查询的相关性 |
| 24h_volume | 30% | 流动性/市场活跃度 |
| liquidity_depth | 15% | 市场深度 |
| price_movement_velocity | 15% | 价格变动速度 |
| outcome_competitiveness | 10% | 竞争度（接近 50% 说明争议大） |

### 2.3 X Handle 解析机制

**问题**：搜索「Dor Brothers」（一个电影制作团队），关键词搜索找不到他们的帖子，因为他们发帖时不会说「Dor Brothers」。

**解决方案**：

```text

### 2.4 目录结构

```text

---

## 三、使用说明：完整指南

### 3.1 安装

#### 方式一：Claude Code Marketplace（推荐）

```bash
/plugin marketplace add mvanhorn/last30days-skill
/plugin install last30days@last30days-skill
```bash
clawhub install last30days-official
```bash
# 克隆到 Claude Code skills 目录
git clone https://github.com/mvanhorn/last30days-skill.git ~/.claude/skills/last30days

# 或 Codex CLI
git clone https://github.com/mvanhorn/last30days-skill.git ~/.agents/skills/last30days
```bash
mkdir -p ~/.config/last30days
cat > ~/.config/last30days/.env << 'EOF'
# Reddit + TikTok + Instagram（一个密钥覆盖三个平台）
# 获取地址: https://scrapecreators.com
SCRAPECREATORS_API_KEY=your_key_here

# X/Twitter 搜索（推荐方式）
# 登录 x.com，打开浏览器 DevTools，复制 auth_token 和 ct0 Cookie
AUTH_TOKEN=your_auth_token
CT0=your_ct0_token

# xAI 后备（可选，没有 Cookie 时使用）
XAI_API_KEY=xai-your_key

# Bluesky（可选）
BSKY_HANDLE=your.bsky.social
BSKY_APP_PASSWORD=xxxx-xxxx-xxxx

# Web 搜索 API（可选，增加网络搜索能力）
PARALLEL_API_KEY=your_key          # Parallel AI（推荐）
BRAVE_API_KEY=your_key              # Brave Search（免费额度 2000/月）
OPENROUTER_API_KEY=your_key         # OpenRouter/Perplexity Sonar Pro
EOF
chmod 600 ~/.config/last30days/.env
```bash
# 在项目根目录创建
mkdir -p .claude
cat > .claude/last30days.env << 'EOF'
# 这里填项目特定的 API 密钥
SCRAPECREATORS_API_KEY=project_specific_key
EOF
```bash
python3 scripts/last30days.py --diagnose
```text
=== Source Availability ===
✓ ScrapeCreators API: configured
✓ X Search (Bird): working
✓ Bluesky: configured
✓ YouTube (yt-dlp): found in PATH
✓ HN Algolia: available (no key needed)
✓ Polymarket Gamma: available (no key needed)
```bash
/last30days [主题]
```bash
/last30days Claude Code best practices
/last30days AI Agent frameworks 2026
/last30days 最新的开源LLM进展
```bash
/last30days [主题] for [目标工具]
```bash
/last30days prompting techniques for ChatGPT
/last30days nano banana pro prompting
/last30days remotion animations for Claude Code
```bash
cp variants/open/SKILL.md ~/.claude/skills/last30days/SKILL.md
```bash
# 添加关注主题
last30 watch 竞争对手公司 每周
last30 watch AI视频工具 每月
last30 watch Y Combinator 热门公司 每年4月和9月
```bash
# 手动运行所有关注主题
last30 run all

# 查看累积知识
last30 what have you found about AI video?
```bash
# crontab 示例：每周一早上 9 点运行
0 9 * * 1 cd ~/projects && last30 run all >> ~/logs/last30.log 2>&1

# 或 macOS launchd
```text
Query: /last30days prompting techniques for legal questions in ChatGPT

Research Output:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

The dominant theme is hallucination prevention...

Key patterns discovered:
1. Hallucination prevention clauses
2. Role assignment (paralegal, issue-spotter)
3. Structured output requirements
4. Epistemic humility enforcement
5. Scope limitation

Research Stats:
• 10 Reddit threads (1,200+ upvotes)
• 15 X posts (5,000+ likes)
• 8 YouTube videos (200K+ views)

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
```text
Query: /last30days anthropic odds

Research Output:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

Key findings:
• Pentagon standoff: 22% ban probability ( Polymarket)
• Best AI model (Feb): 98% Anthropic ( Polymarket)
• IPO before OpenAI: 64% YES
• $500B+ valuation: 87% YES

Research Stats:
• 25 X posts (218 likes)
• 13 YouTube videos (719K views)
• 6 HN stories (48 points)
• 11 Polymarket markets

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
```text
┌─────────────────────────────────────────────────────────────┐
│                    last30days 核心架构                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  SKILL.md ← Claude Code 的指令入口                           │
│    ↓                                                        │
│  last30days.py ← 主脚本，协调各模块                          │
│    ↓                                                        │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                    搜索层                             │   │
│  │  • RedditSearch (ScrapeCreators)                     │   │
│  │  • TwitterSearch (Bird/xAI)                          │   │
│  │  • YouTubeSearch (yt-dlp)                           │   │
│  │  • HNSearch (Algolia)                               │   │
│  │  • PolymarketSearch (Gamma)                          │   │
│  │  • BlueskySearch (AT Protocol)                       │   │
│  │  • WebSearch (Parallel/Brave/OpenRouter)             │   │
│  └─────────────────────────────────────────────────────┘   │
│    ↓                                                        │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                    评分层                             │   │
│  │  • TextSimilarityEngine (双向匹配 + 同义词扩展)        │   │
│  │  • EngagementScorer (参与度归一化)                    │   │
│  │  • CrossPlatformDetector (跨平台热点检测)             │   │
│  │  • PolymarketScorer (5因子加权)                       │   │
│  └─────────────────────────────────────────────────────┘   │
│    ↓                                                        │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                    合成层                             │   │
│  │  • LLMSynthesizer (调用 LLM 生成报告)                 │   │
│  │  • CitationManager (引用管理)                         │   │
│  │  • FormatRenderer (Markdown/JSON 输出)                │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘
```python
# scripts/lib/sources/mysource.py
from typing import List, Dict, Any
from .base import SearchSource

class MySource(SearchSource):
    """自定义搜索源示例"""
    
    name = "mysource"
    api_endpoint = "https://api.mysource.com/search"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    def search(
        self, 
        query: str, 
        days: int = 30,
        max_results: int = 20
    ) -> List[Dict[str, Any]]:
        """执行搜索并返回结果"""
        # 1. 调用 API
        response = self._call_api(query, days, max_results)
        
        # 2. 解析响应
        results = self._parse(response)
        
        # 3. 标准化格式
        return [self._normalize(r) for r in results]
    
    def _normalize(self, raw: Dict) -> Dict[str, Any]:
        """标准化为统一格式"""
        return {
            "source": self.name,
            "title": raw.get("title"),
            "url": raw.get("url"),
            "published_at": raw.get("published_at"),
            "engagement": {
                "score": raw.get("upvotes", 0),
                "comments": raw.get("comment_count", 0),
            },
            "author": raw.get("author"),
            "raw": raw,  # 保留原始数据供后续处理
        }
```python
# scripts/last30days.py

from .sources.mysource import MySource

# 在搜索函数中注册
def get_all_sources(config: Config) -> Dict[str, SearchSource]:
    return {
        # 从配置加载已有源...
        "mysource": MySource(config.get("MYSOURCE_API_KEY")),
    }
```python
# 在 scorer.py 中
class MultiSignalScorer:
    def __init__(self):
        self.source_weights = {
            # 从配置加载已有权重...
            "mysource": 0.15,  # 自定义源的权重
        }
    
    def score(self, results: List[Dict]) -> List[Dict]:
        for result in results:
            source = result["source"]
            base = result.get("engagement", {}).get("score", 0)
            
            # 应用源特定权重
            weight = self.source_weights.get(source, 0.10)
            result["final_score"] = base * weight
            
        return sorted(results, key=lambda x: x["final_score"], reverse=True)
```python
# scripts/lib/scorer.py

class CustomScorer:
    """自定义评分器示例"""
    
    def __init__(
        self,
        score_weight: float = 0.50,
        comment_weight: float = 0.35,
        ratio_weight: float = 0.05,
        top_comment_weight: float = 0.10,
    ):
        self.weights = {
            "score": score_weight,
            "comment": comment_weight,
            "ratio": ratio_weight,
            "top_comment": top_comment_weight,
        }
    
    def score(self, item: Dict) -> float:
        import math
        
        score = item.get("engagement", {}).get("score", 0)
        comments = item.get("engagement", {}).get("comments", 0)
        top_comment = item.get("engagement", {}).get("top_comment_score", 0)
        ratio = top_comment / (score + 1)  # 避免除零
        
        return (
            self.weights["score"] * math.log1p(score) +
            self.weights["comment"] * math.log1p(comments) +
            self.weights["ratio"] * (ratio * 10) +
            self.weights["top_comment"] * math.log1p(top_comment)
        )
```python
# scripts/lib/renderer.py

class MarkdownRenderer:
    def __init__(self, template: str = "default"):
        self.template = template
    
    def render(self, results: List[Dict], synthesis: str) -> str:
        sections = []
        
        # 标题
        sections.append(f"# Research Report: {results[0]['query']}\n")
        
        # 关键发现
        sections.append("## Key Findings\n")
        sections.append(synthesis)
        
        # 按来源分组
        sections.append("\n## Source Breakdown\n")
        by_source = self._group_by_source(results)
        for source, items in by_source.items():
            sections.append(f"### {source} ({len(items)} results)\n")
            for item in items[:5]:  # 每源最多 5 条
                sections.append(
                    f"- [{item['title']}]({item['url']}) "
                    f"(score: {item['engagement']['score']})\n"
                )
        
        # 统计数据
        sections.append("\n## Research Stats\n")
        sections.append(self._render_stats(results))
        
        return "".join(sections)
```python
# scripts/integrations/slack.py

import requests
from typing import List, Dict

class SlackIntegration:
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
    
    def send_report(
        self, 
        title: str, 
        summary: str, 
        results: List[Dict]
    ):
        blocks = [
            {
                "type": "header",
                "text": {"type": "plain_text", "text": title}
            },
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": summary}
            },
            {"type": "divider"},
        ]
        
        # 添加 Top 3 结果
        for item in results[:3]:
            blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"• <{item['url']}|{item['title']}> ({item['engagement']['score']} points)"
                }
            })
        
        requests.post(
            self.webhook_url,
            json={"blocks": blocks}
        )
```python
# scripts/integrations/notion.py

from notion_client import Client

class NotionIntegration:
    def __init__(self, token: str, database_id: str):
        self.notion = Client(auth=token)
        self.database_id = database_id
    
    def create_research_page(
        self,
        title: str,
        summary: str,
        results: List[Dict],
        tags: List[str]
    ):
        properties = {
            "Title": {"title": [{"text": {"content": title}}]},
            "Tags": {"multi_select": [{"name": t} for t in tags]},
            "Summary": {"rich_text": [{"text": {"content": summary[:2000]}}]},
        }
        
        children = [
            {
                "object": "block",
                "type": "heading_2",
                "heading_2": {"rich_text": [{"text": {"content": "Top Results"}}]}
            }
        ]
        
        for item in results[:10]:
            children.append({
                "object": "block",
                "type": "bulleted_list_item",
                "bulleted_list_item": {
                    "rich_text": [
                        {"text": {"content": f"{item['title']} — "}},
                        {"text": {"link": {"url": item["url"]}, "content": "Link"}}
                    ]
                }
            })
        
        self.notion.pages.create(
            parent={"database_id": self.database_id},
            properties=properties,
            children=children
        )
```bash
# 运行所有测试
pytest tests/ -v

# 运行特定源的测试
pytest tests/test_reddit.py -v

# 运行带覆盖率
pytest tests/ --cov=scripts --cov-report=html
```python
# tests/test_mysource.py

import pytest
from scripts.sources.mysource import MySource

class TestMySource:
    @pytest.fixture
    def source(self):
        return MySource(api_key="test_key")
    
    def test_search_returns_results(self, source):
        results = source.search("test query", days=7, max_results=10)
        assert len(results) <= 10
        assert all("title" in r for r in results)
        assert all("url" in r for r in results)
    
    def test_normalize_standardizes_format(self, source):
        raw = {
            "title": "Test Post",
            "url": "https://example.com/post",
            "published_at": "2026-03-28",
            "upvotes": 100,
            "comment_count": 50,
        }
        normalized = source._normalize(raw)
        
        assert normalized["source"] == "mysource"
        assert normalized["engagement"]["score"] == 100
        assert normalized["engagement"]["comments"] == 50
```bash
sudo "/Applications/Python 3.12/Install Certificates.command"