generate_synthetic_data.py
7.44 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
import json
import random
from datetime import datetime, timedelta
# 定义 subreddit 配置
subreddits = {
"AmITheAsshole": {"weight": 0.30, "drama_bias": 0.8},
"relationship_advice": {"weight": 0.25, "drama_bias": 0.7},
"tifu": {"weight": 0.15, "drama_bias": 0.6},
"funny": {"weight": 0.10, "drama_bias": 0.2},
"jokes": {"weight": 0.10, "drama_bias": 0.1},
"StandUpComedy": {"weight": 0.10, "drama_bias": 0.3}
}
# 冲突类型模板
conflict_templates = {
"family_conflict": {
"titles": [
"AITA for refusing to attend my sister's wedding?",
"My parents disowned me for choosing my career over family business",
"I told my MIL she's not welcome in my home anymore",
"AITA for not sharing my inheritance with my siblings?",
"My brother stole from me and my parents are siding with him",
],
"elements": ["betrayal", "loyalty_test", "generational_gap", "financial_dispute"]
},
"relationship_crisis": {
"titles": [
"I found out my wife of 10 years has been cheating",
"My boyfriend wants me to cut off all my male friends",
"I caught my partner in a lie that changes everything",
"She says I need to choose between her and my dog",
"My fiancé hid $50k of debt from me until now",
],
"elements": ["infidelity", "trust_break", "ultimatum", "secret_revealed"]
},
"workplace_drama": {
"titles": [
"My coworker stole my promotion by taking credit for my work",
"I reported my boss for harassment and now I'm being retaliated against",
"My team threw me under the bus in front of the CEO",
"I found out my colleague has been sabotaging my projects",
"My manager wants me to lie to HR about what happened",
],
"elements": ["power_abuse", "betrayal", "ethical_dilemma", "career_risk"]
},
"social_conflict": {
"titles": [
"My neighbor called the cops on my 8-year-old's lemonade stand",
"I confronted a parent who wasn't watching their violent kid",
"A stranger berated me in public for something I didn't do",
"I refused to give up my seat and was publicly shamed",
"My friend group split over a $20 restaurant bill",
],
"elements": ["public_humiliation", "injustice", "moral_stand", "social_pressure"]
},
"personal_crisis": {
"titles": [
"I made a mistake that cost my best friend their dream job",
"I accidentally revealed a secret I promised to keep",
"My moment of anger destroyed a 15-year friendship",
"I prioritized my needs and everyone says I'm selfish",
"I told the truth when everyone wanted me to stay silent",
],
"elements": ["guilt", "regret", "moral_dilemma", "consequence"]
},
"comedy_situation": {
"titles": [
"TIFU by accidentally sending my boss a meme about him",
"I tried to impress a date and ended up in the ER",
"My autocorrect changed 'love' to 'lawsuit' in an email to my lawyer",
"I thought it was a costume party. It was not.",
"I waved back at someone who wasn't waving at me for 30 seconds",
],
"elements": ["embarrassment", "misunderstanding", "irony", "awkwardness"]
}
}
def generate_post(post_id, subreddit, conflict_type=None):
"""生成一条模拟 Reddit 帖子"""
# 根据 subreddit 选择冲突类型
if conflict_type is None:
if subreddit == "AmITheAsshole":
conflict_type = random.choice(["family_conflict", "relationship_crisis", "social_conflict", "personal_crisis"])
elif subreddit == "relationship_advice":
conflict_type = random.choice(["relationship_crisis", "personal_crisis"])
elif subreddit == "tifu":
conflict_type = random.choice(["personal_crisis", "comedy_situation"])
elif subreddit in ["funny", "jokes", "StandUpComedy"]:
conflict_type = "comedy_situation"
else:
conflict_type = random.choice(list(conflict_templates.keys()))
template = conflict_templates[conflict_type]
title = random.choice(template["titles"])
# 生成模拟内容（简短版本）
content = f"[模拟内容] {title} 的详细故事描述..."
# 基于冲突类型和 subreddit 生成戏剧性分数
base_drama = {
"family_conflict": 7.5,
"relationship_crisis": 8.0,
"workplace_drama": 7.0,
"social_conflict": 6.5,
"personal_crisis": 7.0,
"comedy_situation": 4.5
}
dramatic_score = base_drama[conflict_type] + random.uniform(-1.5, 1.5)
dramatic_score = max(1.0, min(10.0, round(dramatic_score, 1)))
# 生成互动数据（与戏剧性正相关）
base_engagement = int(dramatic_score * 1000)
score = int(base_engagement * random.uniform(0.5, 3.0))
num_comments = int(score * random.uniform(0.1, 0.5))
upvote_ratio = round(random.uniform(0.65, 0.98), 2)
# 生成时间戳
days_ago = random.randint(1, 30)
created_utc = (datetime.now() - timedelta(days=days_ago)).isoformat()
return {
"id": f"t3_{post_id:04d}",
"subreddit": subreddit,
"title": title,
"selftext": content,
"score": score,
"num_comments": num_comments,
"upvote_ratio": upvote_ratio,
"created_utc": created_utc,
"dramatic_score": dramatic_score,
"conflict_type": conflict_type,
"drama_elements": random.sample(template["elements"], k=random.randint(1, len(template["elements"]))),
"short Drama_potential": "high" if dramatic_score >= 7 else "medium" if dramatic_score >= 5 else "low"
}
def generate_dataset(total_posts=280):
"""生成完整数据集"""
posts = []
post_id = 1000
# 按权重分配各 subreddit 的帖子数量
subreddit_counts = {}
for sr, config in subreddits.items():
subreddit_counts[sr] = int(total_posts * config["weight"])
# 调整总数
while sum(subreddit_counts.values()) < total_posts:
subreddit_counts[random.choice(list(subreddits.keys()))] += 1
# 生成帖子
for sr, count in subreddit_counts.items():
for _ in range(count):
posts.append(generate_post(post_id, sr))
post_id += 1
# 打乱顺序
random.shuffle(posts)
return posts
# 生成数据
dataset = generate_dataset(280)
# 保存到文件
output_path = "/Users/yibo/Documents/LemonAI/lemonboss/data/workspace/digester/reddit-scraper-v2/synthetic_reddit_data.json"
import os
os.makedirs(os.path.dirname(output_path), exist_ok=True)
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(dataset, f, indent=2, ensure_ascii=False)
print(f"✅ 生成完成: {len(dataset)} 条合成 Reddit 数据")
print(f"📁 保存路径: {output_path}")
# 输出统计
stats = {}
for post in dataset:
sr = post["subreddit"]
if sr not in stats:
stats[sr] = {"count": 0, "avg_score": 0, "avg_comments": 0, "avg_drama": 0}
stats[sr]["count"] += 1
stats[sr]["avg_score"] += post["score"]
stats[sr]["avg_comments"] += post["num_comments"]
stats[sr]["avg_drama"] += post["dramatic_score"]
print("\n📊 数据分布:")
for sr, data in sorted(stats.items()):
print(f" r/{sr}: {data['count']}条, 平均点赞 {data['avg_score']//data['count']}, 平均戏剧性 {data['avg_drama']/data['count']:.1f}")