如何构建一台永不枯竭的 AI 内容创作引擎

1import requests
2import json
3from datetime import datetime
4
5SCRAPE_API_KEY = "your_api_key_here"  # 在 app.scrapecreators.com 获取
6
7BASE_URL = "https://api.scrapecreators.com"
8
9HEADERS = {
10    "x-api-key": SCRAPE_API_KEY,
11    "Content-Type": "application/json"
12}
13
14def get_tiktok_trending(count=30):
15    """拉取 TikTok 热门推荐 — 最有效的信号"""
16    response = requests.get(
17        f"{BASE_URL}/v1/tiktok/get-trending-feed",
18        headers=HEADERS,
19        params={"count": count}
20    )
21    return response.json()
22
23def get_tiktok_profile_videos(username, count=20):
24    """拉取任意竞争对手/灵感账号的热门视频"""
25    response = requests.get(
26        f"{BASE_URL}/v3/tiktok/profile/videos",
27        headers=HEADERS,
28        params={"username": username, "count": count}
29    )
30    return response.json()
31
32def get_tiktok_video_comments(video_url, count=50):
33    """拉取评论 — 受众最真实的声音、痛点、反应"""
34    response = requests.get(
35        f"{BASE_URL}/v1/tiktok/video/comments",
36        headers=HEADERS,
37        params={"video_url": video_url, "count": count}
38    )
39    return response.json()
40
41def get_instagram_trending_reels(count=20):
42    """拉取 Instagram 热门 Reels"""
43    response = requests.get(
44        f"{BASE_URL}/v1/instagram/reels/trending",
45        headers=HEADERS,
46        params={"count": count}
47    )
48    return response.json()
49
50def get_reddit_pain_points(subreddit, sort="hot", limit=25):
51    """拉取 Reddit 帖子 — 未经修饰的受众语言"""
52    response = requests.get(
53        f"{BASE_URL}/v1/reddit/subreddit",
54        headers=HEADERS,
55        params={"subreddit": subreddit, "sort": sort, "limit": limit}
56    )
57    return response.json()
58
59def get_youtube_trending_shorts(count=20):
60    """拉取热门 YouTube Shorts，获取短内容创意"""
61    response = requests.get(
62        f"{BASE_URL}/v1/youtube/shorts/trending",
63        headers=HEADERS,
64        params={"count": count}
65    )
66    return response.json()
67
68def run_daily_capture(niche_accounts, subreddits):
69    """每天全平台抓取"""
70    captured = {
71        "date": datetime.now().isoformat(),
72        "tiktok_trending": get_tiktok_trending(30),
73        "instagram_reels": get_instagram_trending_reels(20),
74        "youtube_shorts": get_youtube_trending_shorts(20),
75        "competitor_videos": [],
76        "reddit_posts": []
77    }
78
79    # 从竞争对手/领域账号拉取
80    for account in niche_accounts:
81        videos = get_tiktok_profile_videos(account, count=10)
82        captured["competitor_videos"].extend(
83            videos.get("videos", [])
84        )
85
86    # 拉取 Reddit 痛点
87    for sub in subreddits:
88        posts = get_reddit_pain_points(sub)
89        captured["reddit_posts"].extend(
90            posts.get("posts", [])
91        )
92
93    # 保存到内容数据库
94    with open(f"capture_{datetime.now().strftime('%Y%m%d')}.json", "w") as f:
95        json.dump(captured, f, indent=2)
96
97    print(f"捕获: {len(captured['competitor_videos'])} 个视频, "
98          f"{len(captured['reddit_posts'])} 个 Reddit 帖子")
99    return captured
100
101# 每天运行
102if __name__ == "__main__":
103    run_daily_capture(
104        niche_accounts=["account1", "account2", "account3"],
105        subreddits=["entrepreneurship", "SideProject", "AItools"]
106    )

1import anthropic
2import json
3
4client = anthropic.Anthropic()  # 使用 ANTHROPIC_API_KEY 环境变量
5
6def extract_content_patterns(captured_data):
7    """
8    将爆款内容数据喂给 Claude。
9    提取钩子、格式、情绪、模板。
10    """
11
12    # 格式化数据以便分析
13    video_sample = []
14    for video in captured_data.get("tiktok_trending", {}).get("videos", [])[:20]:
15        video_sample.append({
16            "description": video.get("desc", ""),
17            "views": video.get("stats", {}).get("playCount", 0),
18            "likes": video.get("stats", {}).get("diggCount", 0),
19            "comments": video.get("stats", {}).get("commentCount", 0),
20            "shares": video.get("stats", {}).get("shareCount", 0),
21        })
22
23    reddit_sample = []
24    for post in captured_data.get("reddit_posts", [])[:15]:
25        reddit_sample.append({
26            "title": post.get("title", ""),
27            "upvotes": post.get("score", 0),
28            "comments": post.get("num_comments", 0),
29        })
30
31    prompt = f"""
32你是一名爆款内容策略师，正在分析社交媒体表现数据。
33
34以下是今天捕获的爆款内容数据：
35
36TIKTOK 热门视频：
37{json.dumps(video_sample, indent=2)}
38
39REDDIT 帖子（受众痛点）：
40{json.dumps(reddit_sample, indent=2)}
41
42分析这些数据并提取：
43
441. 前 5 个钩子
45   - 开头语或文字覆盖模式
46   - 为什么能触发「停下来」的反应
47   - 触发的情绪：好奇 / 恐惧 / 错失恐惧 / 解脱 / 认同
48
492. 前 5 种内容格式
50   - 内容结构（步骤式 / 故事 / 对比 / 列表）
51   - 这种格式为什么在该平台有效
52
533. 前 5 个受众痛点
54   - 来自 Reddit 和视频评论的真实问题
55   - 受众使用的原话（他们的话，不是营销话术）
56
574. 10 个原创内容创意
58   - 基于什么有效 + 痛点
59   - 每个创意：[钩子] + [格式] + [平台] + [为什么有效]
60
61仅以结构化 JSON 输出，不要前言。
62
63JSON 结构：
64{{
65  "hooks": [
66    {{"hook": "...", "emotion": "...", "why_it_works": "..."}}
67  ],
68  "formats": [
69    {{"format": "...", "structure": "...", "why_it_works": "..."}}
70  ],
71  "pain_points": [
72    {{"pain": "...", "exact_language": "...", "frequency": "..."}}
73  ],
74  "content_ideas": [
75    {{
76      "hook": "...",
77      "format": "...",
78      "platform": "...",
79      "why_it_will_work": "...",
80      "estimated_emotion": "..."
81    }}
82  ]
83}}
84"""
85
86    response = client.messages.create(
87        model="claude-sonnet-4-6",
88        max_tokens=2000,
89        messages=[{"role": "user", "content": prompt}]
90    )
91
92    # 解析 JSON 响应
93    raw = response.content[0].text
94    return json.loads(raw)
95
96# 示例输出：
97# {
98#   "hooks": [
99#     {
100#       "hook": "我用 3 个文件替代了整个内容团队",
101#       "emotion": "好奇 + 错失恐惧",
102#       "why_it_works": "身份挑战 + 看似不可能的声明"
103#     }
104#   ],
105#   "pain_points": [
106#     {
107#       "pain": "两周后就没灵感了",
108#       "exact_language": "我再也不知道该发什么了",
109#       "frequency": "高"
110#     }
111#   ]
112# }

1import json
2from datetime import datetime
3from pathlib import Path
4
5class ContentBrain:
6    """
7    你内容机器的记忆层。
8    存储和检索模式、创意和表现数据。
9    每周变得更聪明。
10    """
11
12    def __init__(self, db_path="content_brain.json"):
13        self.db_path = Path(db_path)
14        self.brain = self._load()
15
16    def _load(self):
17        if self.db_path.exists():
18            with open(self.db_path) as f:
19                return json.load(f)
20        return {
21            "hooks": [],
22            "formats": [],
23            "pain_points": [],
24            "ideas_backlog": [],
25            "performance_log": [],
26            "voice_rules": {
27                "always": [],
28                "never": [],
29                "examples": []
30            }
31        }
32
33    def _save(self):
34        with open(self.db_path, "w") as f:
35            json.dump(self.brain, f, indent=2)
36
37    def add_patterns(self, extracted_patterns):
38        """添加今天抓取中提取的模式"""
39        today = datetime.now().isoformat()
40
41        for hook in extracted_patterns.get("hooks", []):
42            hook["added"] = today
43            hook["used_count"] = 0
44            self.brain["hooks"].append(hook)
45
46        for pain in extracted_patterns.get("pain_points", []):
47            pain["added"] = today
48            self.brain["pain_points"].append(pain)
49
50        for idea in extracted_patterns.get("content_ideas", []):
51            idea["added"] = today
52            idea["status"] = "pending"
53            self.brain["ideas_backlog"].append(idea)
54
55        self._save()
56        print(f"添加了 {len(extracted_patterns.get('hooks', []))} 个钩子，"
57              f"{len(extracted_patterns.get('content_ideas', []))} 个创意")
58
59    def get_next_ideas(self, platform=None, count=5):
60        """从 backlog 中拉取下一个未使用的创意"""
61        pending = [i for i in self.brain["ideas_backlog"]
62                   if i["status"] == "pending"]
63
64        if platform:
65            pending = [i for i in pending if i.get("platform") == platform]
66
67        return pending[:count]
68
69    def mark_used(self, idea_hook, performance=None):
70        """标记创意为已使用，并可选记录表现"""
71        for idea in self.brain["ideas_backlog"]:
72            if idea["hook"] == idea_hook:
73                idea["status"] = "used"
74                idea["used_date"] = datetime.now().isoformat()
75                if performance:
76                    idea["performance"] = performance
77                break
78
79        if performance:
80            self.brain["performance_log"].append({
81                "hook": idea_hook,
82                "performance": performance,
83                "date": datetime.now().isoformat()
84            })
85
86        self._save()
87
88    def get_top_hooks(self, emotion=None, limit=10):
89        """从库中获取最佳钩子"""
90        hooks = self.brain["hooks"]
91        if emotion:
92            hooks = [h for h in hooks if h.get("emotion") == emotion]
93        return hooks[:limit]
94
95    def get_backlog_size(self):
96        pending = [i for i in self.brain["ideas_backlog"]
97                   if i["status"] == "pending"]
98        return len(pending)
99
100    def weekly_summary(self):
101        """大脑的总结"""
102        return {
103            "total_hooks": len(self.brain["hooks"]),
104            "ideas_pending": self.get_backlog_size(),
105            "ideas_used": len([i for i in self.brain["ideas_backlog"]
106                               if i["status"] == "used"]),
107            "pain_points_tracked": len(self.brain["pain_points"]),
108            "posts_logged": len(self.brain["performance_log"])
109        }
110
111# 使用
112brain = ContentBrain()
113print(brain.weekly_summary())
114# 30 天后：
115# {
116#   "total_hooks": 147,
117#   "ideas_pending": 83,
118#   "ideas_used": 64,
119#   "pain_points_tracked": 210,
120#   "posts_logged": 64
121# }

1import anthropic
2import requests
3import json
4
5client = anthropic.Anthropic()
6
7KIE_API_KEY = "your_kie_api_key"   # kie.ai
8KIE_BASE_URL = "https://api.kie.ai"
9
10# ── 文字生成 ───────────────────────────────────
11
12def generate_platform_post(idea, platform, voice_rules):
13    """
14    根据创意生成平台原生帖子。
15    使用 Content Brain 中的声音规则。
16    """
17
18    platform_guides = {
19        "twitter": "短小精悍，小写，无标签，每句话之间换行。不超过 280 字符或使用线程格式。",
20        "linkedin": "专业但人性化，个人叙事，第一人称故事加感悟，1300-2000 字符。",
21        "instagram": "视觉优先，轮播格式，第一张幻灯片用粗体钩子，每张不超过 30 字，末尾放保存行动号召。",
22        "tiktok": "45-60 秒视频脚本，前 2 秒出现钩子，展示而非讲述，自然口语。",
23        "reddit": "乐于助人，不推销，真正有用的内容，社区口吻，值得被点赞。"
24    }
25
26    prompt = f"""
27你是一名 {platform} 的爆款内容写手。
28
29创意：
30钩子：{idea['hook']}
31格式：{idea['format']}
32痛点：{idea.get('pain_point', '未指定')}
33要触发的情绪：{idea.get('emotion', '好奇')}
34
35{platform} 的平台规则：
36{platform_guides.get(platform, '请自然地为该平台写作。')}
37
38声音规则（必须遵守）：
39始终：{', '.join(voice_rules.get('always', []))}
40绝不：{', '.join(voice_rules.get('never', []))}
41
42现在写完整的帖子。可直接发布。不要前言。
43"""
44
45    response = client.messages.create(
46        model="claude-sonnet-4-6",
47        max_tokens=1000,
48        messages=[{"role": "user", "content": prompt}]
49    )
50
51    return response.content[0].text
52
53# ── 视觉生成 (KIE.AI) ────────────────────────
54
55def generate_image_kie(prompt_text, style="photorealistic"):
56    """
57    使用 Kie.ai 生成图片。
58    返回图片 URL。
59    """
60    response = requests.post(
61        f"{KIE_BASE_URL}/v1/images/generate",
62        headers={
63            "Authorization": f"Bearer {KIE_API_KEY}",
64            "Content-Type": "application/json"
65        },
66        json={
67            "prompt": prompt_text,
68            "style": style,
69            "width": 1080,
70            "height": 1920,   # 竖屏适合社交媒体
71            "quality": "high"
72        }
73    )
74    data = response.json()
75    return data.get("image_url")
76
77def generate_video_kie(script, voice="natural", duration=60):
78    """
79    使用 Kie.ai 根据脚本生成短视频。
80    返回视频 URL。
81    """
82    response = requests.post(
83        f"{KIE_BASE_URL}/v1/videos/generate",
84        headers={
85            "Authorization": f"Bearer {KIE_API_KEY}",
86            "Content-Type": "application/json"
87        },
88        json={
89            "script": script,
90            "voice": voice,
91            "duration_seconds": duration,
92            "format": "mp4",
93            "aspect_ratio": "9:16"
94        }
95    )
96    data = response.json()
97    return data.get("video_url")
98
99def generate_carousel_visuals(slides, brand_colors=None):
100    """
101    为 Instagram/LinkedIn 生成轮播幻灯片图片。
102    每张幻灯片一张图片。
103    """
104    image_urls = []
105
106    for i, slide in enumerate(slides):
107        visual_prompt = f"""
108创建一张干净、大胆的社交媒体轮播幻灯片。
109
110文字叠层："{slide['text']}"
111幻灯片编号：{i + 1}/{len(slides)}
112风格：简约、高对比度、专业
113背景：深色配浅色文字 或 浅色配深色文字，交替
114字体风格：粗体、无衬线、手机尺寸下易读
115{"品牌色：" + str(brand_colors) if brand_colors else ""}
116
117不要有图库照片感。要有编辑感、现代、让人忍不住想看。
118"""
119        url = generate_image_kie(visual_prompt, style="graphic design")
120        image_urls.append(url)
121        print(f"生成幻灯片 {i+1}: {url}")
122
123    return image_urls
124
125# ── 完整生产流程 ──────────────────────────
126
127def produce_content_batch(brain, platforms=None, ideas_per_platform=2):
128    """
129    从大脑中拉取创意，为每个平台生成内容。
130    返回准备排期的批次。
131    """
132    if platforms is None:
133        platforms = ["twitter", "linkedin", "instagram", "tiktok"]
134
135    voice_rules = brain.brain["voice_rules"]
136    batch = []
137
138    for platform in platforms:
139        ideas = brain.get_next_ideas(platform=platform, count=ideas_per_platform)
140
141        for idea in ideas:
142            print(f"\n生成 {platform} 帖子: {idea['hook'][:50]}...")
143
144            post_text = generate_platform_post(idea, platform, voice_rules)
145
146            produced = {
147                "platform": platform,
148                "idea": idea,
149                "text": post_text,
150                "media": None
151            }
152
153            # 为 Instagram 轮播生成视觉
154            if platform == "instagram" and "carousel" in idea.get("format", "").lower():
155                slides = [
156                    {"text": idea["hook"]},
157                    {"text": "这是为什么它很重要"},
158                    {"text": "步骤 1：从这里开始"},
159                    {"text": "步骤 2：然后做这个"},
160                    {"text": "步骤 3：最后这样"},
161                    {"text": "收藏它。关注获取更多。"}
162                ]
163                produced["media"] = generate_carousel_visuals(slides)
164
165            # 为 TikTok 生成视频
166            elif platform == "tiktok":
167                produced["media"] = generate_video_kie(
168                    script=post_text,
169                    duration=60
170                )
171
172            batch.append(produced)
173            brain.mark_used(idea["hook"])
174
175    return batch
176
177# 运行
178brain = ContentBrain()
179batch = produce_content_batch(brain, platforms=["twitter", "linkedin", "instagram"])
180print(f"\n生成了 {len(batch)} 条内容")

1import requests
2from datetime import datetime, timedelta
3import json
4
5POSTIZ_API_KEY = "your_postiz_api_key"  # app.postiz.com
6POSTIZ_BASE = "https://api.postiz.com"
7
8POSTING_SCHEDULE = {
9    "twitter":   {"times": ["08:00", "12:00", "17:00"], "days": "daily"},
10    "linkedin":  {"times": ["07:30", "12:00"],           "days": "weekdays"},
11    "instagram": {"times": ["11:00", "19:00"],           "days": "daily"},
12    "tiktok":    {"times": ["08:00", "13:00", "20:00"], "days": "daily"},
13}
14
15def get_next_slot(platform):
16    """计算平台的下一个可用发布时间"""
17    schedule = POSTING_SCHEDULE.get(platform, {})
18    times = schedule.get("times", ["09:00"])
19    now = datetime.now()
20
21    for time_str in times:
22        hour, minute = map(int, time_str.split(":"))
23        slot = now.replace(hour=hour, minute=minute, second=0)
24        if slot > now:
25            return slot.isoformat()
26
27    # 第二天第一个时段
28    tomorrow = now + timedelta(days=1)
29    first_time = times[0]
30    hour, minute = map(int, first_time.split(":"))
31    return tomorrow.replace(hour=hour, minute=minute, second=0).isoformat()
32
33def schedule_post_postiz(content_item, integration_id):
34    """
35    通过 Postiz API 安排单条帖子。
36    处理文本 + 媒体上传。
37    """
38    headers = {
39        "Authorization": f"Bearer {POSTIZ_API_KEY}",
40        "Content-Type": "application/json"
41    }
42
43    scheduled_at = get_next_slot(content_item["platform"])
44
45    payload = {
46        "content": content_item["text"],
47        "scheduledAt": scheduled_at,
48        "integrationId": integration_id,
49        "status": "schedule"
50    }
51
52    # 如果生成了媒体则附加
53    if content_item.get("media"):
54        if isinstance(content_item["media"], list):
55            # 多张图片（轮播）
56            payload["media"] = [{"url": url} for url in content_item["media"]]
57        else:
58            # 单个视频或图片
59            payload["media"] = [{"url": content_item["media"]}]
60
61    response = requests.post(
62        f"{POSTIZ_BASE}/posts",
63        headers=headers,
64        json=payload
65    )
66
67    result = response.json()
68    print(f"已安排 {content_item['platform']} 帖子，发布时间：{scheduled_at}")
69    print(f"帖子 ID：{result.get('id')}")
70    return result
71
72def schedule_full_batch(content_batch, platform_integrations):
73    """
74    通过 Postiz 安排整周的内容批次。
75    platform_integrations：将平台名称映射到 Postiz 集成 ID 的字典
76    """
77    scheduled = []
78
79    for item in content_batch:
80        platform = item["platform"]
81        integration_id = platform_integrations.get(platform)
82
83        if not integration_id:
84            print(f"未找到 {platform} 的集成 ID — 跳过")
85            continue
86
87        result = schedule_post_postiz(item, integration_id)
88        scheduled.append({
89            "platform": platform,
90            "hook": item["idea"]["hook"],
91            "scheduled_at": result.get("scheduledAt"),
92            "post_id": result.get("id")
93        })
94
95    print(f"\n已安排 {len(scheduled)} 条帖子，跨 {len(set(s['platform'] for s in scheduled))} 个平台")
96    return scheduled
97
98# ── 反馈循环 ─────────────────────────────────────
99
100def fetch_post_performance(post_id):
101    """帖子发布后获取表现指标"""
102    response = requests.get(
103        f"{POSTIZ_BASE}/posts/{post_id}/analytics",
104        headers={"Authorization": f"Bearer {POSTIZ_API_KEY}"}
105    )
106    return response.json()
107
108def update_brain_with_performance(brain, scheduled_posts, days_old=3):
109    """
110    帖子运行 3 天后，将表现数据拉回大脑。
111    这是系统学习的关键。
112    """
113    cutoff = datetime.now() - timedelta(days=days_old)
114    updated = 0
115
116    for post in scheduled_posts:
117        scheduled_time = datetime.fromisoformat(post.get("scheduled_at", ""))
118
119        if scheduled_time < cutoff:
120            perf = fetch_post_performance(post["post_id"])
121
122            brain.mark_used(
123                idea_hook=post["hook"],
124                performance={
125                    "views": perf.get("impressions", 0),
126                    "likes": perf.get("likes", 0),
127                    "shares": perf.get("shares", 0),
128                    "comments": perf.get("comments", 0),
129                    "platform": post["platform"]
130                }
131            )
132            updated += 1
133
134    print(f"已将 {updated} 条帖子的表现数据更新到大脑")
135
136# ── 每周报告 ─────────────────────────────────────
137
138def weekly_report(brain):
139    """本周机器学到了什么"""
140    summary = brain.weekly_summary()
141
142    print("\n━━━━━━━━━━━━━━━━━━")
143    print("每周机器报告")
144    print("━━━━━━━━━━━━━━━━━━")
145    print(f"总存储的钩子：            {summary['total_hooks']}")
146    print(f"待办创意数：              {summary['ideas_pending']}")
147    print(f"本周使用的创意：          {summary['ideas_used']}")
148    print(f"跟踪的痛点：              {summary['pain_points_tracked']}")
149    print(f"有数据的帖子：            {summary['posts_logged']}")
150    print("━━━━━━━━━━━━━━━━━━\n")