Reddit Scraper Skill

Skill Metadata

skill_name: reddit_scraper
version: 1.0.0
type: content_extraction
domain: social_media / short_drama_materials
target_score: 75
elite_score: 90
current_score: 80.6
status: enabled
benchmark_date: 2026-03-25
test_results:
  accuracy: 21.0/25 (84.0%)
  filtering: 25.0/30 (83.3%)
  extraction: 21.6/30 (72.0%)
  format: 13.0/15 (86.7%)

Description

抓取 Reddit 高赞帖子，识别戏剧性内容，提取结构化短剧素材。

Capabilities

1. Reddit 内容抓取 (RSS/API)
2. 戏剧性内容识别与评分
3. 短剧素材结构化提取
4. 标准化 JSON 输出

Input Schema

{
  "type": "object",
  "required": ["subreddit"],
  "properties": {
    "subreddit": {
      "type": "string",
      "enum": ["AmITheAsshole", "relationship_advice", "funny"],
      "description": "目标子版块"
    },
    "filter": {
      "type": "object",
      "properties": {
        "min_upvotes": { "type": "integer", "default": 1000 },
        "min_comments": { "type": "integer", "default": 50 },
        "time_range": { "type": "string", "enum": ["day", "week", "month"], "default": "week" }
      }
    },
    "limit": {
      "type": "integer",
      "default": 10,
      "maximum": 50
    }
  }
}

Output Schema

{
  "type": "object",
  "properties": {
    "results": {
      "type": "array",
      "items": {
        "$ref": "#/definitions/DramaMaterial"
      }
    },
    "metadata": {
      "type": "object",
      "properties": {
        "scraped_count": { "type": "integer" },
        "filtered_count": { "type": "integer" },
        "avg_dramatic_score": { "type": "number" },
        "execution_time_ms": { "type": "integer" }
      }
    }
  }
}

Execution Logic

Step 1: 内容抓取

def fetch_posts(subreddit, limit=50, time_range="week"):
    url = f"https://www.reddit.com/r/{subreddit}/hot.json"
    params = {"limit": limit, "t": time_range}
    headers = {"User-Agent": "DramaScraper/1.0"}
    response = requests.get(url, params=params, headers=headers)
    return response.json()["data"]["children"]

Step 2: 基础过滤

def basic_filter(post, config):
    data = post["data"]
    return (
        data["ups"] >= config["min_upvotes"] and
        data["num_comments"] >= config["min_comments"] and
        data["upvote_ratio"] >= 0.65 and
        len(data.get("selftext", "")) >= 300
    )

Step 3: 戏剧性评分

def calculate_dramatic_score(post):
    data = post["data"]

    # Engagement ratio (25%)
    engagement_score = min(data["num_comments"] / data["ups"] * 10, 10) * 0.25

    # Content length (15%)
    text_length = len(data.get("selftext", ""))
    length_score = 10 if 500 <= text_length <= 3000 else 5
    length_score *= 0.15

    # Emotion density (30%)
    emotion_words = ["angry", "furious", "heartbroken", "devastated", "shocked",
                     "betrayed", "disappointed", "frustrated", "hate", "love",
                     "生气", "愤怒", "伤心", "失望", "震惊"]
    text = data["title"] + " " + data.get("selftext", "")
    emotion_count = sum(1 for w in emotion_words if w.lower() in text.lower())
    emotion_score = min(emotion_count / len(text.split()) * 100, 10) * 0.30

    # Conflict keywords (30%)
    conflict_categories = {
        "family": ["父母", "婆婆", "老公", "离婚", "family", "parents", "mother-in-law", "divorce"],
        "workplace": ["老板", "同事", "裁员", "boss", "coworker", "layoff"],
        "relationship": ["出轨", "分手", "cheating", "breakup", "affair"],
        "moral": ["自私", "背叛", "selfish", "betrayal", "lying"]
    }
    conflict_hits = sum(
        1 for cat in conflict_categories.values()
        if any(kw in text.lower() for kw in cat)
    )
    conflict_score = min(conflict_hits * 2.5, 10) * 0.30

    return round(engagement_score + length_score + emotion_score + conflict_score, 1)

Step 4: 素材提取

def extract_drama_elements(post):
    data = post["data"]
    text = data.get("selftext", "")

    # Setting detection
    locations = {
        "home": ["home", "house", "apartment", "room", "bed", "kitchen", "家"],
        "office": ["office", "work", "company", "boss", "meeting", "公司"],
        "public": ["restaurant", "park", "street", "store", "餐厅"],
        "event": ["wedding", "party", "dinner", "婚礼"]
    }
    detected_location = "other"
    for loc, keywords in locations.items():
        if any(kw in text.lower() for kw in keywords):
            detected_location = loc
            break

    # Character extraction
    characters = []
    roles = {
        "I": "protagonist",
        "my wife": "supporting",
        "my husband": "supporting",
        "my boyfriend": "supporting",
        "my girlfriend": "supporting",
        "my mom": "supporting",
        "my dad": "supporting",
        "my boss": "antagonist",
        "MIL": "antagonist"
    }

    # Conflict type detection
    conflict_types = []
    if any(w in text.lower() for w in ["divorce", "cheat", "affair", "breakup"]):
        conflict_types.append("relationship_crisis")
    if any(w in text.lower() for w in ["boss", "work", "job", "fired"]):
        conflict_types.append("workplace_drama")
    if any(w in text.lower() for w in ["mom", "dad", "family", "sister", "brother"]):
        conflict_types.append("family_conflict")

    return {
        "setting": {
            "location": detected_location,
            "time": "inferred_from_context",
            "mood": "dramatic"
        },
        "characters": characters,
        "conflict": {
            "type": conflict_types[0] if conflict_types else "interpersonal",
            "trigger": "extracted_from_text",
            "escalation": "extracted_from_text",
            "climax": "extracted_from_text"
        },
        "dialogue_highlights": extract_quotes(text),
        "turning_point": "identified_from_narrative",
        "emotional_arc": "analyzed_from_text"
    }

Step 5: JSON 生成

def generate_output(post, dramatic_score):
    data = post["data"]
    return {
        "source": {
            "platform": "reddit",
            "subreddit": data["subreddit"],
            "post_id": data["id"],
            "url": f"https://reddit.com{data['permalink']}",
            "scraped_at": datetime.now().isoformat()
        },
        "engagement": {
            "upvotes": data["ups"],
            "comments": data["num_comments"],
            "upvote_ratio": data["upvote_ratio"],
            "dramatic_score": dramatic_score
        },
        "content": {
            "title": data["title"],
            "body": data.get("selftext", "")[:2000],
            "drama_tags": detect_drama_tags(data),
            "conflict_type": detect_conflict_type(data)
        },
        "drama_elements": extract_drama_elements(post),
        "script_potential": {
            "hook_strength": min(dramatic_score, 10),
            "viral_elements": detect_viral_elements(data),
            "estimated_duration": estimate_duration(data),
            "genre_fit": detect_genre(data)
        }
    }

Error Handling

错误类型	处理策略
Rate Limit (429)	指数退避，最大重试3次
内容删除	标记为 unavailable，跳过
网络超时	重试，超时时间递增
解析错误	记录错误，继续处理下一个

Performance Metrics

• 平均响应时间:
• 成功率: > 95%
• 戏剧性识别准确率: > 80%

Dependencies

• requests
• json
• datetime
• re (正则表达式)

Usage Example

{
  "skill": "reddit_scraper",
  "input": {
    "subreddit": "AmITheAsshole",
    "filter": {
      "min_upvotes": 2000,
      "time_range": "week"
    },
    "limit": 10
  }
}