openclaw-skill-money-idea-generator

Original source / Raw SKILL.md

---
name: openclaw-skill-money-idea-generator
description: "赚钱灵感生成器 | Money Idea Generator. 自动发现 AI 变现机会，生成可落地的赚钱灵感。支持多平台监控：GitHub、抖音、B站、小红书、Twitter。触发词：赚钱灵感、赚钱机会、AI 变现、商业灵感."
metadata:
  openclaw:
    emoji: "💰"
    category: "business"
    tags: ["money", "idea", "business", "AI", "monetization", "startup", "multi-source"]
    requires:
      bins: ["python3", "curl"]
---

# 赚钱灵感生成器

自动发现 AI 变现机会，生成可落地的赚钱灵感。

## 核心功能

### 1. 多平台监控
- GitHub Trending AI 项目
- 抖音热门话题
- B站科技区视频
- 小红书热门笔记
- Twitter AI trending

### 2. 变现潜力分析
- 判断项目是否"能赚钱"
- 分析市场需求和竞争
- 给出潜力评分（高/中/低）

### 3. 灵感生成
- 自动生成赚钱灵感
- 提供具体实现路径
- 估算启动成本和预期收入

### 4. 资产池管理
- 保存灵感到资产池
- 跟踪执行状态
- 记录实际收益

### 5. 个性化推荐
- 根据用户偏好推荐
- 按领域分类（AI 工具、SaaS、内容创作）
- 按难度分级（入门/进阶/专业）

## 使用方式

### 获取今日赚钱灵感

```
给我生成一个赚钱灵感
最近有什么赚钱机会？
```

### 分析特定项目

```
分析这个项目能不能赚钱：https://github.com/xxx/xxx
```

### 获取热门变现机会

```
最近有什么 AI 变现机会？
```

## 变现潜力判断规则

### 高潜力特征
- ✅ 解决明确痛点
- ✅ 技术门槛适中
- ✅ 有付费用户验证
- ✅ 可规模化复制
- ✅ 新增星标 > 100/周

### 低潜力特征
- ❌ 技术门槛太高
- ❌ 市场太小众
- ❌ 已有巨头垄断
- ❌ 需要大量资金
- ❌ 无明确商业模式

## 赚钱灵感分类

| 类别 | 示例 | 启动成本 | 预期收入 |
|------|------|---------|---------|
| **AI 工具开发** | OpenClaw 部署服务 | ¥500 | ¥5,000+/月 |
| **内容创作** | AI 日报订阅 | ¥0 | ¥2,000+/月 |
| **技术服务** | 自动化脚本开发 | ¥0 | ¥3,000+/月 |
| **教育培训** | AI 教程课程 | ¥0 | ¥10,000+/月 |
| **SaaS 产品** | AI Agent 平台 | ¥1,000 | ¥50,000+/月 |

## 配置

在 `config.py` 中配置：
- `GITHUB_TOKEN`: GitHub API Token（可选，提高速率限制）
- `MIN_STARS`: 最小星标数（默认 50）
- `CATEGORIES`: 关注的领域（默认 AI/LLM/Agent）
- `USER_PREFERENCES`: 用户偏好（预算、技能、兴趣）

---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### README.md

```markdown
# 💰 赚钱灵感生成器 | Money Idea Generator

[![OpenClaw Skill](https://img.shields.io/badge/OpenClaw-Skill-blue)](https://github.com/devotion-coding/openclaw-skill-money-idea-generator)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

自动发现 AI 变现机会，生成可落地的赚钱灵感。

## ✨ 功能特点

- 🔍 **多平台监控**：GitHub、抖音、B站、小红书、Twitter
- 📊 **潜力分析**：判断项目是否能赚钱
- 💡 **灵感生成**：生成具体赚钱灵感 + 实现路径
- 📦 **资产池**：保存灵感、跟踪执行、记录收益
- 👤 **个性化**：根据用户偏好推荐

## 🚀 快速开始

### 安装（OpenClaw 用户）

```bash
clawhub install openclaw-skill-money-idea-generator
```

### 手动安装

```bash
# 克隆仓库
git clone https://github.com/devotion-coding/openclaw-skill-money-idea-generator.git

# 进入目录
cd openclaw-skill-money-idea-generator/scripts

# 安装依赖
pip install requests
```

### 运行测试

```bash
python3 main.py
```

## 📖 使用方式

### 获取今日赚钱灵感

```
给我生成一个赚钱灵感
最近有什么赚钱机会？
```

### 分析特定项目

```
分析这个项目能不能赚钱：https://github.com/xxx/xxx
```

### 获取热门变现机会

```
最近有什么 AI 变现机会？
```

## 💵 变现方式

| 方式 | 成本 | 预期收入 | 时间 |
|------|------|----------|------|
| 部署服务 | ¥100-500 | ¥2,000-10,000/月 | 1-3 天 |
| 技术咨询 | ¥0-100 | ¥5,000-20,000/月 | 0.5-2 天 |
| 培训课程 | ¥0-200 | ¥1,000-50,000/月 | 3-7 天 |
| 定制开发 | ¥500-2,000 | ¥5,000-50,000/月 | 7-30 天 |

## 📝 输出示例

```
============================================================
💰 今日赚钱灵感
============================================================
生成时间：2026-02-28 18:26

【灵感 #1】Agent-Reach 培训课程
  描述：制作 Agent-Reach 使用教程和培训课程
  目标用户：学习者, 开发者, 企业员工
  启动成本：¥200
  预期收入：¥25,500/月
  所需时间：3 天
  🔗 https://github.com/Panniantong/Agent-Reach

【灵感 #2】goclaw 培训课程
  描述：制作 goclaw 使用教程和培训课程
  目标用户：学习者, 开发者, 企业员工
  启动成本：¥200
  预期收入：¥25,500/月
  所需时间：3 天
  🔗 https://github.com/nextlevelbuilder/goclaw
```

## ⚙️ 配置

在 `scripts/config.py` 中配置：

```python
# GitHub Token（可选，提高 API 速率限制）
GITHUB_TOKEN = ""

# 用户偏好
USER_PREFERENCES = {
    'budget': 1000,          # 预算
    'time_available': 2,     # 可用时间（小时/天）
    'skills': ['Python', 'AI'],  # 技能
    'interests': ['SaaS', '工具'],  # 兴趣领域
}
```

## 📁 文件结构

```
openclaw-skill-money-idea-generator/
├── README.md                    # 说明文档
├── SKILL.md                     # OpenClaw 技能元数据
├── LICENSE                      # 开源协议
├── CHANGELOG.md                 # 更新日志
└── scripts/
    ├── config.py                # 配置文件
    ├── github_monitor.py        # GitHub 监控
    ├── idea_analyzer.py         # 灵感分析
    ├── asset_pool.py            # 资产池管理
    ├── multi_source_monitor.py  # 多数据源监控
    └── main.py                  # 主入口
```

## 🌐 支持的数据源

| 平台 | 类型 | 说明 |
|------|------|------|
| GitHub | 项目 | AI/LLM/Agent 热门项目 |
| 抖音 | 热点 | AI赚钱、副业话题 |
| B站 | 视频 | 科技区热门内容 |
| 小红书 | 笔记 | 副业、创业话题 |
| Twitter | 趋势 | AI trending |

## 🤝 贡献

欢迎提交 Issue 和 Pull Request！

1. Fork 本仓库
2. 创建特性分支 (`git checkout -b feature/AmazingFeature`)
3. 提交更改 (`git commit -m 'Add some AmazingFeature'`)
4. 推送到分支 (`git push origin feature/AmazingFeature`)
5. 提交 Pull Request

## 📄 License

[MIT](LICENSE)

## 🙏 致谢

- 灵感来源：《一人企业方法论》
- 数据来源：GitHub API、B站 API

---

**⭐ 如果觉得有用，请给个 Star！**
```

### _meta.json

```json
{
  "owner": "devotion-coding",
  "slug": "openclaw-skill-money-idea-generator",
  "displayName": "Openclaw Skill Money Idea Generator",
  "latest": {
    "version": "0.1.2",
    "publishedAt": 1772276405176,
    "commit": "https://github.com/openclaw/skills/commit/dc3f44930f78a92a4b861109328213037856dac4"
  },
  "history": []
}

```

### scripts/asset_pool.py

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
资产池管理
保存灵感、跟踪执行、记录收益
"""

import json
import os
from datetime import datetime
from typing import Dict, List, Optional


class AssetPool:
    """资产池管理器"""
    
    def __init__(self, data_dir: str = None):
        self.data_dir = data_dir or os.path.expanduser('~/.openclaw/workspace/memory/money-ideas')
        os.makedirs(self.data_dir, exist_ok=True)
        
        self.ideas_file = os.path.join(self.data_dir, 'ideas.json')
        self.execution_file = os.path.join(self.data_dir, 'executions.json')
        self.revenue_file = os.path.join(self.data_dir, 'revenue.json')
        
        self._load_data()
    
    def _load_data(self):
        """加载数据"""
        self.ideas = self._load_json(self.ideas_file, [])
        self.executions = self._load_json(self.execution_file, [])
        self.revenues = self._load_json(self.revenue_file, [])
    
    def _load_json(self, filepath: str, default):
        """加载 JSON 文件"""
        if os.path.exists(filepath):
            try:
                with open(filepath, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except:
                return default
        return default
    
    def _save_json(self, filepath: str, data):
        """保存 JSON 文件"""
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    
    # === 灵感池 ===
    
    def add_idea(self, idea: Dict) -> str:
        """
        添加灵感到资产池
        
        Args:
            idea: 灵感信息
            
        Returns:
            灵感 ID
        """
        idea_id = f"idea-{datetime.now().strftime('%Y%m%d%H%M%S')}"
        idea['id'] = idea_id
        idea['created_at'] = datetime.now().isoformat()
        idea['status'] = 'pending'  # pending, in_progress, completed, failed
        idea['score'] = 0  # 潜力分数
        idea['tags'] = idea.get('tags', [])
        
        self.ideas.append(idea)
        self._save_json(self.ideas_file, self.ideas)
        
        return idea_id
    
    def get_idea(self, idea_id: str) -> Optional[Dict]:
        """获取灵感"""
        for idea in self.ideas:
            if idea['id'] == idea_id:
                return idea
        return None
    
    def list_ideas(self, status: str = None, limit: int = 10) -> List[Dict]:
        """
        列出灵感
        
        Args:
            status: 状态筛选
            limit: 数量限制
            
        Returns:
            灵感列表
        """
        ideas = self.ideas
        
        if status:
            ideas = [i for i in ideas if i.get('status') == status]
        
        # 按创建时间倒序
        ideas = sorted(ideas, key=lambda x: x.get('created_at', ''), reverse=True)
        
        return ideas[:limit]
    
    def update_idea_status(self, idea_id: str, status: str, notes: str = None):
        """更新灵感状态"""
        for idea in self.ideas:
            if idea['id'] == idea_id:
                idea['status'] = status
                idea['updated_at'] = datetime.now().isoformat()
                if notes:
                    idea['notes'] = notes
                self._save_json(self.ideas_file, self.ideas)
                return True
        return False
    
    # === 执行跟踪 ===
    
    def start_execution(self, idea_id: str) -> str:
        """
        开始执行灵感
        
        Args:
            idea_id: 灵感 ID
            
        Returns:
            执行 ID
        """
        exec_id = f"exec-{datetime.now().strftime('%Y%m%d%H%M%S')}"
        
        execution = {
            'id': exec_id,
            'idea_id': idea_id,
            'started_at': datetime.now().isoformat(),
            'status': 'in_progress',
            'steps': [],
            'logs': [],
        }
        
        self.executions.append(execution)
        self._save_json(self.execution_file, self.executions)
        
        # 更新灵感状态
        self.update_idea_status(idea_id, 'in_progress')
        
        return exec_id
    
    def add_execution_step(self, exec_id: str, step: str, status: str = 'completed'):
        """添加执行步骤"""
        for execution in self.executions:
            if execution['id'] == exec_id:
                execution['steps'].append({
                    'step': step,
                    'status': status,
                    'timestamp': datetime.now().isoformat(),
                })
                self._save_json(self.execution_file, self.executions)
                return True
        return False
    
    def add_execution_log(self, exec_id: str, log: str):
        """添加执行日志"""
        for execution in self.executions:
            if execution['id'] == exec_id:
                execution['logs'].append({
                    'log': log,
                    'timestamp': datetime.now().isoformat(),
                })
                self._save_json(self.execution_file, self.executions)
                return True
        return False
    
    def complete_execution(self, exec_id: str, success: bool = True, notes: str = None):
        """完成执行"""
        for execution in self.executions:
            if execution['id'] == exec_id:
                execution['status'] = 'success' if success else 'failed'
                execution['completed_at'] = datetime.now().isoformat()
                if notes:
                    execution['notes'] = notes
                self._save_json(self.execution_file, self.executions)
                
                # 更新灵感状态
                idea_id = execution['idea_id']
                self.update_idea_status(idea_id, 'completed' if success else 'failed', notes)
                
                return True
        return False
    
    # === 收益记录 ===
    
    def add_revenue(self, idea_id: str, amount: float, source: str, notes: str = None) -> str:
        """
        记录收益
        
        Args:
            idea_id: 灵感 ID
            amount: 金额
            source: 收入来源
            notes: 备注
            
        Returns:
            记录 ID
        """
        rev_id = f"rev-{datetime.now().strftime('%Y%m%d%H%M%S')}"
        
        revenue = {
            'id': rev_id,
            'idea_id': idea_id,
            'amount': amount,
            'source': source,
            'notes': notes,
            'recorded_at': datetime.now().isoformat(),
        }
        
        self.revenues.append(revenue)
        self._save_json(self.revenue_file, self.revenues)
        
        return rev_id
    
    def get_revenue_stats(self, idea_id: str = None) -> Dict:
        """
        获取收益统计
        
        Args:
            idea_id: 灵感 ID（可选，不传则统计全部）
            
        Returns:
            统计数据
        """
        revenues = self.revenues
        
        if idea_id:
            revenues = [r for r in revenues if r['idea_id'] == idea_id]
        
        total = sum(r['amount'] for r in revenues)
        count = len(revenues)
        avg = total / count if count > 0 else 0
        
        # 按来源统计
        by_source = {}
        for r in revenues:
            source = r['source']
            by_source[source] = by_source.get(source, 0) + r['amount']
        
        return {
            'total': total,
            'count': count,
            'average': avg,
            'by_source': by_source,
        }
    
    # === 统计分析 ===
    
    def get_overview(self) -> Dict:
        """获取资产池概览"""
        # 灵感统计
        idea_stats = {
            'total': len(self.ideas),
            'pending': len([i for i in self.ideas if i.get('status') == 'pending']),
            'in_progress': len([i for i in self.ideas if i.get('status') == 'in_progress']),
            'completed': len([i for i in self.ideas if i.get('status') == 'completed']),
            'failed': len([i for i in self.ideas if i.get('status') == 'failed']),
        }
        
        # 成功率
        finished = idea_stats['completed'] + idea_stats['failed']
        success_rate = idea_stats['completed'] / finished if finished > 0 else 0
        
        # 收益统计
        revenue_stats = self.get_revenue_stats()
        
        # 执行统计
        exec_stats = {
            'total': len(self.executions),
            'in_progress': len([e for e in self.executions if e.get('status') == 'in_progress']),
            'success': len([e for e in self.executions if e.get('status') == 'success']),
            'failed': len([e for e in self.executions if e.get('status') == 'failed']),
        }
        
        return {
            'ideas': idea_stats,
            'success_rate': success_rate,
            'revenue': revenue_stats,
            'executions': exec_stats,
        }


# 测试
if __name__ == '__main__':
    pool = AssetPool()
    
    # 测试添加灵感
    idea_id = pool.add_idea({
        'name': 'OpenClaw 部署服务',
        'description': '帮助用户部署 OpenClaw',
        'type': 'deployment_service',
        'tags': ['AI', '自动化'],
    })
    print(f"添加灵感: {idea_id}")
    
    # 测试开始执行
    exec_id = pool.start_execution(idea_id)
    print(f"开始执行: {exec_id}")
    
    # 添加步骤
    pool.add_execution_step(exec_id, '在闲鱼发布服务')
    pool.add_execution_step(exec_id, '收到第一个客户咨询')
    
    # 完成执行
    pool.complete_execution(exec_id, success=True, notes='第一个客户成功交付')
    
    # 记录收益
    pool.add_revenue(idea_id, 299, '闲鱼', 'OpenClaw 部署服务')
    
    # 查看统计
    overview = pool.get_overview()
    print("\n=== 资产池概览 ===")
    print(json.dumps(overview, indent=2, ensure_ascii=False))
```

### scripts/config.py

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
赚钱灵感生成器 - 配置文件
"""

import os

# GitHub API Token（可选，提高速率限制）
GITHUB_TOKEN = os.environ.get('GITHUB_TOKEN', '')

# 监控的项目最小星标数
MIN_STARS = 50

# 最小新增星标/周
MIN_WEEKLY_STARS = 20

# 关注的领域（GitHub 搜索关键词）
CATEGORIES = [
    'AI LLM Agent',
    'AI automation',
    'chatbot GPT',
    'AI agent framework',
    'LLM tools',
]

# 排除的项目（太成熟或无法变现）
EXCLUDE_KEYWORDS = [
    'awesome-',  # 资源列表
    'tutorial',  # 教程
    'course',    # 课程
    'book',      # 书籍
]

# 变现潜力评分规则
POTENTIAL_RULES = {
    'high': {
        'min_stars': 100,
        'min_weekly_growth': 50,
        'keywords': ['api', 'sdk', 'framework', 'platform'],
    },
    'medium': {
        'min_stars': 50,
        'min_weekly_growth': 20,
        'keywords': ['tool', 'cli', 'bot'],
    },
    'low': {
        'min_stars': 10,
        'min_weekly_growth': 5,
        'keywords': [],
    }
}

# 赚钱灵感模板
IDEA_TEMPLATES = {
    'deployment_service': {
        'name': '{project_name} 部署服务',
        'description': '帮助用户部署 {project_name}，提供技术支持',
        'target_users': ['技术小白', '企业用户', '创业者'],
        'cost_range': (100, 500),
        'income_range': (2000, 10000),
        'time_range': (1, 3),  # 天
    },
    'consulting': {
        'name': '{project_name} 技术咨询',
        'description': '提供 {project_name} 技术咨询和方案设计',
        'target_users': ['企业', '创业者', '产品经理'],
        'cost_range': (0, 100),
        'income_range': (5000, 20000),
        'time_range': (0.5, 2),
    },
    'training': {
        'name': '{project_name} 培训课程',
        'description': '制作 {project_name} 使用教程和培训课程',
        'target_users': ['学习者', '开发者', '企业员工'],
        'cost_range': (0, 200),
        'income_range': (1000, 50000),
        'time_range': (3, 7),
    },
    'customization': {
        'name': '{project_name} 定制开发',
        'description': '为用户定制开发基于 {project_name} 的功能',
        'target_users': ['企业', '创业者', '产品团队'],
        'cost_range': (500, 2000),
        'income_range': (5000, 50000),
        'time_range': (7, 30),
    },
}

# 用户偏好（从记忆系统读取）
USER_PREFERENCES = {
    'budget': 1000,  # 预算
    'time_available': 2,  # 可用时间（天/周）
    'skills': ['Python', 'AI', 'Web'],  # 技能
    'interests': ['AI工具', '自动化', 'SaaS'],  # 兴趣
}
```

### scripts/github_monitor.py

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
GitHub 热门项目监控
"""

import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional

from config import GITHUB_TOKEN, CATEGORIES, EXCLUDE_KEYWORDS, MIN_STARS


class GitHubMonitor:
    """GitHub 热门项目监控"""
    
    def __init__(self, token: str = GITHUB_TOKEN):
        self.token = token
        self.headers = {'Authorization': f'token {token}'} if token else {}
        self.base_url = 'https://api.github.com'
    
    def search_ai_projects(self, days: int = 7, min_stars: int = MIN_STARS) -> List[Dict]:
        """
        搜索最近热门的 AI 项目
        
        Args:
            days: 最近几天创建的
            min_stars: 最小星标数
            
        Returns:
            项目列表
        """
        results = []
        
        # 计算日期
        since = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
        
        for category in CATEGORIES:
            query = f'{category} created:>{since} stars:>{min_stars}'
            url = f'{self.base_url}/search/repositories'
            params = {
                'q': query,
                'sort': 'stars',
                'order': 'desc',
                'per_page': 20,
            }
            
            try:
                response = requests.get(url, headers=self.headers, params=params, timeout=10)
                if response.status_code == 200:
                    data = response.json()
                    for item in data.get('items', []):
                        # 排除不需要的项目
                        if self._should_exclude(item['full_name'], item.get('description', '')):
                            continue
                        
                        results.append({
                            'name': item['full_name'],
                            'stars': item['stargazers_count'],
                            'language': item.get('language', ''),
                            'description': item.get('description', ''),
                            'url': item['html_url'],
                            'created_at': item['created_at'],
                            'pushed_at': item['pushed_at'],
                            'category': category,
                        })
            except Exception as e:
                print(f"搜索失败 ({category}): {e}")
        
        # 去重并排序
        seen = set()
        unique_results = []
        for item in sorted(results, key=lambda x: x['stars'], reverse=True):
            if item['name'] not in seen:
                seen.add(item['name'])
                unique_results.append(item)
        
        return unique_results[:20]
    
    def get_trending_repos(self, language: str = '', since: str = 'weekly') -> List[Dict]:
        """
        获取 GitHub Trending 项目
        
        Args:
            language: 编程语言
            since: 时间范围 (daily, weekly, monthly)
            
        Returns:
            项目列表
        """
        # 使用非官方 API
        url = f'https://api.gitterapp.com/repositories'
        params = {
            'language': language,
            'since': since,
        }
        
        try:
            response = requests.get(url, params=params, timeout=10)
            if response.status_code == 200:
                data = response.json()
                results = []
                for item in data[:20]:
                    if self._should_exclude(item.get('full_name', ''), item.get('description', '')):
                        continue
                    
                    results.append({
                        'name': item.get('full_name', ''),
                        'stars': item.get('stargazers_count', 0),
                        'language': item.get('language', ''),
                        'description': item.get('description', ''),
                        'url': item.get('html_url', ''),
                        'trending_stars': item.get('trending_stars', 0),
                    })
                return results
        except Exception as e:
            print(f"获取 Trending 失败: {e}")
        
        # 降级：使用搜索 API
        return self.search_ai_projects(days=7)
    
    def _should_exclude(self, name: str, description: str) -> bool:
        """判断是否应该排除"""
        text = f"{name} {description or ''}".lower()
        for keyword in EXCLUDE_KEYWORDS:
            if keyword.lower() in text:
                return True
        return False
    
    def get_repo_details(self, owner: str, repo: str) -> Optional[Dict]:
        """
        获取仓库详情
        
        Args:
            owner: 仓库所有者
            repo: 仓库名
            
        Returns:
            仓库详情
        """
        url = f'{self.base_url}/repos/{owner}/{repo}'
        
        try:
            response = requests.get(url, headers=self.headers, timeout=10)
            if response.status_code == 200:
                data = response.json()
                return {
                    'name': data['full_name'],
                    'stars': data['stargazers_count'],
                    'forks': data['forks_count'],
                    'language': data.get('language', ''),
                    'description': data.get('description', ''),
                    'topics': data.get('topics', []),
                    'license': data.get('license', {}).get('spdx_id', ''),
                    'open_issues': data['open_issues_count'],
                    'watchers': data['watchers_count'],
                    'url': data['html_url'],
                    'created_at': data['created_at'],
                    'pushed_at': data['pushed_at'],
                }
        except Exception as e:
            print(f"获取仓库详情失败: {e}")
        
        return None


# 测试
if __name__ == '__main__':
    monitor = GitHubMonitor()
    
    print('=== 最近热门 AI 项目 ===')
    projects = monitor.search_ai_projects(days=7)
    for i, p in enumerate(projects[:10], 1):
        print(f"{i}. {p['name']} ⭐{p['stars']}")
        print(f"   {p['description'][:50]}...")
        print(f"   🔗 {p['url']}")
        print()
```

### scripts/idea_analyzer.py

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
赚钱灵感分析器
分析项目的变现潜力，生成赚钱灵感
"""

import re
from typing import Dict, List, Optional, Tuple

from config import POTENTIAL_RULES, IDEA_TEMPLATES


class IdeaAnalyzer:
    """赚钱灵感分析器"""
    
    def __init__(self):
        self.templates = IDEA_TEMPLATES
    
    def analyze_potential(self, project: Dict) -> Tuple[str, float]:
        """
        分析项目的变现潜力
        
        Args:
            project: 项目信息
            
        Returns:
            (潜力等级, 潜力分数 0-100)
        """
        score = 0
        
        # 1. 星标数评分（最多 30 分）
        stars = project.get('stars', 0)
        if stars > 1000:
            score += 30
        elif stars > 500:
            score += 25
        elif stars > 100:
            score += 20
        elif stars > 50:
            score += 15
        else:
            score += 10
        
        # 2. 新增星标速度评分（最多 25 分）
        trending_stars = project.get('trending_stars', 0)
        if trending_stars > 100:
            score += 25
        elif trending_stars > 50:
            score += 20
        elif trending_stars > 20:
            score += 15
        elif trending_stars > 10:
            score += 10
        else:
            score += 5
        
        # 3. 技术栈评分（最多 20 分）
        language = project.get('language', '').lower()
        description = project.get('description', '').lower()
        
        # AI/LLM 相关加分
        ai_keywords = ['ai', 'llm', 'gpt', 'chatgpt', 'claude', 'agent', 'openai', 'anthropic']
        if any(k in description for k in ai_keywords):
            score += 15
        
        # 热门语言加分
        hot_languages = ['python', 'typescript', 'rust', 'go']
        if language in hot_languages:
            score += 5
        
        # 4. 变现关键词评分（最多 15 分）
        monetization_keywords = [
            'api', 'sdk', 'cli', 'framework', 'platform', 'tool',
            'automation', 'chatbot', 'assistant', 'dashboard'
        ]
        if any(k in description for k in monetization_keywords):
            score += 15
        elif any(k in description for k in ['library', 'package']):
            score += 10
        
        # 5. 商业模式评分（最多 10 分）
        if 'saas' in description or 'enterprise' in description:
            score += 10
        elif 'api' in description or 'service' in description:
            score += 7
        elif 'tool' in description or 'app' in description:
            score += 5
        
        # 确定潜力等级
        if score >= 70:
            potential = '高'
        elif score >= 50:
            potential = '中'
        else:
            potential = '低'
        
        return potential, score
    
    def generate_ideas(self, project: Dict, user_preferences: Dict = None) -> List[Dict]:
        """
        为项目生成赚钱灵感
        
        Args:
            project: 项目信息
            user_preferences: 用户偏好
            
        Returns:
            赚钱灵感列表
        """
        ideas = []
        project_name = project.get('name', '').split('/')[-1]
        description = project.get('description', '')
        
        # 分析适合的变现方式
        suitable_types = self._analyze_suitable_types(project)
        
        for idea_type in suitable_types:
            template = self.templates.get(idea_type)
            if not template:
                continue
            
            # 根据用户偏好调整
            cost, income, time = self._adjust_for_user(template, user_preferences)
            
            idea = {
                'type': idea_type,
                'name': template['name'].format(project_name=project_name),
                'description': template['description'].format(project_name=project_name),
                'target_users': template['target_users'],
                'cost': cost,
                'expected_income': income,
                'time_needed': time,
                'project': project_name,
                'project_url': project.get('url', ''),
            }
            
            ideas.append(idea)
        
        return ideas
    
    def _analyze_suitable_types(self, project: Dict) -> List[str]:
        """分析项目适合的变现方式"""
        description = project.get('description', '').lower()
        language = project.get('language', '').lower()
        stars = project.get('stars', 0)
        
        suitable = []
        
        # 1. 部署服务 - 大多数项目都适合
        if stars > 50:
            suitable.append('deployment_service')
        
        # 2. 技术咨询 - 复杂项目适合
        if any(k in description for k in ['framework', 'platform', 'architecture']):
            suitable.append('consulting')
        
        # 3. 培训课程 - 有学习门槛的项目适合
        if any(k in description for k in ['tutorial', 'guide', 'learn']) or stars > 100:
            suitable.append('training')
        
        # 4. 定制开发 - 企业级项目适合
        if any(k in description for k in ['enterprise', 'business', 'saas', 'api']):
            suitable.append('customization')
        
        # 默认至少返回部署服务和培训
        if not suitable:
            suitable = ['deployment_service', 'training']
        
        return suitable
    
    def _adjust_for_user(self, template: Dict, preferences: Dict = None) -> Tuple[int, int, int]:
        """根据用户偏好调整成本/收入/时间"""
        if not preferences:
            return (
                template['cost_range'][0],
                template['income_range'][0],
                template['time_range'][0]
            )
        
        # 根据预算调整成本
        budget = preferences.get('budget', 1000)
        cost = min(template['cost_range'][1], max(template['cost_range'][0], budget // 2))
        
        # 根据可用时间调整
        time_available = preferences.get('time_available', 2)
        time_needed = min(template['time_range'][1], max(template['time_range'][0], time_available))
        
        # 根据技能调整收入预期
        skills = preferences.get('skills', [])
        income = template['income_range'][0]
        if any(s.lower() in ['python', 'ai', 'web', 'javascript'] for s in skills):
            income = (template['income_range'][0] + template['income_range'][1]) // 2
        
        return cost, income, time_needed
    
    def rank_ideas(self, ideas: List[Dict], user_preferences: Dict = None) -> List[Dict]:
        """对灵感进行排序"""
        def get_score(idea):
            # 收入越高越好
            income_score = idea['expected_income'] / 1000
            
            # 时间越短越好
            time_score = 10 / max(idea['time_needed'], 1)
            
            # 成本越低越好
            cost_score = 10 / max(idea['cost'] / 100, 1)
            
            return income_score + time_score + cost_score
        
        return sorted(ideas, key=get_score, reverse=True)


# 测试
if __name__ == '__main__':
    analyzer = IdeaAnalyzer()
    
    # 测试项目
    test_project = {
        'name': 'openai/chatgpt',
        'stars': 50000,
        'trending_stars': 150,
        'description': 'Official Python library for the ChatGPT API',
        'language': 'Python',
    }
    
    potential, score = analyzer.analyze_potential(test_project)
    print(f"潜力: {potential} (分数: {score})")
    
    ideas = analyzer.generate_ideas(test_project)
    for idea in ideas:
        print(f"\n灵感: {idea['name']}")
        print(f"  目标用户: {idea['target_users']}")
        print(f"  成本: ¥{idea['cost']}")
        print(f"  预期收入: ¥{idea['expected_income']}/月")
        print(f"  所需时间: {idea['time_needed']} 天")
```

### scripts/main.py

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
赚钱灵感生成器 - 主入口
集成资产池 + 多数据源
"""

import json
from datetime import datetime
from typing import Dict, List, Optional

from github_monitor import GitHubMonitor
from idea_analyzer import IdeaAnalyzer
from asset_pool import AssetPool
from multi_source_monitor import MultiSourceMonitor
from config import USER_PREFERENCES


class MoneyIdeaGenerator:
    """赚钱灵感生成器（含资产池 + 多数据源）"""
    
    def __init__(self, user_preferences: Dict = None):
        self.monitor = GitHubMonitor()
        self.multi_monitor = MultiSourceMonitor()  # 多数据源
        self.analyzer = IdeaAnalyzer()
        self.pool = AssetPool()
        self.preferences = user_preferences or USER_PREFERENCES
    
    def generate_daily_ideas(self, count: int = 3, save_to_pool: bool = True, use_multi_source: bool = True) -> List[Dict]:
        """
        生成每日赚钱灵感
        
        Args:
            count: 灵感数量
            save_to_pool: 是否保存到资产池
            use_multi_source: 是否使用多数据源
            
        Returns:
            赚钱灵感列表
        """
        ideas = []
        
        # 1. 使用多数据源
        if use_multi_source:
            print("正在获取多平台热点...")
            hot_data = self.multi_monitor.get_all_hot()
            hot_ideas = self.multi_monitor.generate_ideas_from_hot(hot_data)
            
            # 转换格式
            for hot in hot_ideas[:count]:
                ideas.append({
                    'type': 'hot_topic',
                    'name': hot['title'],
                    'description': hot['description'],
                    'target_users': ['创业者', '自媒体运营者', '开发者'],
                    'cost': 100,
                    'expected_income': 3000,
                    'time_needed': 3,
                    'source': hot['source'],
                    'potential': hot['potential'],
                    'project_url': hot['url'],
                })
        
        # 2. GitHub 项目分析
        print("正在搜索 GitHub 热门 AI 项目...")
        projects = self.monitor.search_ai_projects(days=7, min_stars=50)
        
        if projects:
            print("正在分析变现潜力...")
            analyzed = []
            for project in projects[:20]:
                potential, score = self.analyzer.analyze_potential(project)
                project['potential'] = potential
                project['potential_score'] = score
                analyzed.append(project)
            
            analyzed.sort(key=lambda x: x['potential_score'], reverse=True)
            
            for project in analyzed[:count]:
                project_ideas = self.analyzer.generate_ideas(project, self.preferences)
                ideas.extend(project_ideas)
        
        # 3. 如果没有数据，使用备用灵感
        if not ideas:
            ideas = self._get_fallback_ideas()
        
        # 4. 排序并截取
        ranked_ideas = self.analyzer.rank_ideas(ideas, self.preferences)
        final_ideas = ranked_ideas[:count]
        
        # 5. 保存到资产池
        if save_to_pool:
            for idea in final_ideas:
                self.pool.add_idea(idea)
        
        return final_ideas
    
    def analyze_project(self, project_url: str) -> Optional[Dict]:
        """
        分析特定项目的变现潜力
        
        Args:
            project_url: 项目 URL
            
        Returns:
            分析结果
        """
        # 从 URL 提取 owner/repo
        import re
        match = re.search(r'github\.com/([^/]+)/([^/]+)', project_url)
        if not match:
            return None
        
        owner, repo = match.groups()
        
        # 获取项目详情
        project = self.monitor.get_repo_details(owner, repo)
        if not project:
            return None
        
        # 分析潜力
        potential, score = self.analyzer.analyze_potential(project)
        project['potential'] = potential
        project['potential_score'] = score
        
        # 生成灵感
        ideas = self.analyzer.generate_ideas(project, self.preferences)
        
        return {
            'project': project,
            'potential': potential,
            'potential_score': score,
            'ideas': ideas,
        }
    
    def get_hot_opportunities(self) -> List[Dict]:
        """
        获取当前热门变现机会
        
        Returns:
            变现机会列表
        """
        # 获取 GitHub Trending
        print("正在获取 GitHub Trending...")
        trending = self.monitor.get_trending_repos(language='', since='weekly')
        
        # 筛选 AI 相关项目
        ai_keywords = ['ai', 'llm', 'gpt', 'agent', 'chatbot', 'ml', 'machine learning']
        ai_projects = []
        
        for project in trending:
            desc = project.get('description', '').lower()
            if any(k in desc for k in ai_keywords):
                potential, score = self.analyzer.analyze_potential(project)
                project['potential'] = potential
                project['potential_score'] = score
                ai_projects.append(project)
        
        # 排序
        ai_projects.sort(key=lambda x: x['potential_score'], reverse=True)
        
        return ai_projects[:10]
    
    def _get_fallback_ideas(self) -> List[Dict]:
        """获取备用灵感（当无法获取数据时）"""
        return [
            {
                'type': 'deployment_service',
                'name': 'OpenClaw 部署服务',
                'description': '帮助用户部署 OpenClaw AI 智能体，提供技术支持',
                'target_users': ['自媒体运营者', '电商卖家', '小微企业'],
                'cost': 500,
                'expected_income': 5000,
                'time_needed': 2,
                'project': 'OpenClaw',
                'implementation': [
                    '在闲鱼发布服务（¥199-299）',
                    '准备服务器（阿里云/腾讯云最低配）',
                    '标准化部署流程（2小时/客户）',
                    '提供 7 天技术支持',
                ],
            },
            {
                'type': 'training',
                'name': 'AI Agent 使用教程',
                'description': '制作 AI Agent 使用教程，卖给想学习的人',
                'target_users': ['学习者', '开发者', '企业员工'],
                'cost': 0,
                'expected_income': 1000,
                'time_needed': 3,
                'project': 'AI Agent',
                'implementation': [
                    '录制视频教程（B站/抖音）',
                    '制作文字教程（公众号/知乎）',
                    '设置付费订阅（¥99/月）',
                    '定期更新内容',
                ],
            },
            {
                'type': 'consulting',
                'name': 'AI 自动化咨询服务',
                'description': '为企业提供 AI 自动化转型咨询',
                'target_users': ['企业', '创业者', '产品经理'],
                'cost': 100,
                'expected_income': 10000,
                'time_needed': 1,
                'project': 'AI Automation',
                'implementation': [
                    '制作案例展示文档',
                    '在 LinkedIn/知乎 发布内容',
                    '设置咨询价格（¥500-1000/小时）',
                    '提供方案设计服务',
                ],
            },
        ]


def format_idea_output(ideas: List[Dict]) -> str:
    """格式化输出灵感"""
    output = []
    output.append("=" * 60)
    output.append("💰 今日赚钱灵感")
    output.append("=" * 60)
    output.append(f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}")
    output.append("")
    
    for i, idea in enumerate(ideas, 1):
        output.append(f"【灵感 #{i}】{idea['name']}")
        output.append(f"  描述：{idea['description']}")
        output.append(f"  目标用户：{', '.join(idea['target_users'])}")
        output.append(f"  启动成本：¥{idea['cost']}")
        output.append(f"  预期收入：¥{idea['expected_income']}/月")
        output.append(f"  所需时间：{idea['time_needed']} 天")
        
        if 'implementation' in idea:
            output.append("  实现路径：")
            for step in idea['implementation']:
                output.append(f"    - {step}")
        
        if 'project_url' in idea:
            output.append(f"  🔗 {idea['project_url']}")
        
        output.append("")
    
    return "\n".join(output)


# 主入口
if __name__ == '__main__':
    generator = MoneyIdeaGenerator()
    
    print("\n=== 测试：生成每日赚钱灵感 ===\n")
    ideas = generator.generate_daily_ideas(count=3)
    print(format_idea_output(ideas))
    
    print("\n=== 测试：分析特定项目 ===\n")
    result = generator.analyze_project('https://github.com/langchain-ai/langchain')
    if result:
        print(json.dumps(result, indent=2, ensure_ascii=False))
```

### scripts/multi_source_monitor.py

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
多数据源监控
整合 GitHub、抖音、B站、小红书、Twitter
"""

import json
import subprocess
from datetime import datetime
from typing import Dict, List, Optional


class MultiSourceMonitor:
    """多数据源监控器"""
    
    def __init__(self):
        self.sources = ['github', 'douyin', 'bilibili', 'xiaohongshu', 'twitter']
    
    def get_all_hot(self) -> Dict[str, List[Dict]]:
        """
        获取所有平台热点
        
        Returns:
            各平台热点数据
        """
        results = {}
        
        # GitHub 热门
        results['github'] = self._get_github_trending()
        
        # 抖音热门
        results['douyin'] = self._get_douyin_hot()
        
        # B站热门
        results['bilibili'] = self._get_bilibili_hot()
        
        # 小红书热门
        results['xiaohongshu'] = self._get_xiaohongshu_hot()
        
        # Twitter AI trending
        results['twitter'] = self._get_twitter_ai()
        
        return results
    
    def _get_github_trending(self) -> List[Dict]:
        """获取 GitHub Trending AI 项目"""
        from github_monitor import GitHubMonitor
        monitor = GitHubMonitor()
        projects = monitor.search_ai_projects(days=7, min_stars=50)
        
        results = []
        for p in projects[:10]:
            results.append({
                'source': 'github',
                'title': p.get('name', ''),
                'description': p.get('description', ''),
                'url': p.get('url', ''),
                'metrics': {
                    'stars': p.get('stars', 0),
                },
                'tags': ['AI', 'GitHub'],
            })
        
        return results
    
    def _get_douyin_hot(self) -> List[Dict]:
        """获取抖音热门"""
        results = []
        
        try:
            # 使用 bird CLI 获取抖音数据
            # 热门话题：AI、赚钱、副业、创业
            topics = ['AI赚钱', '副业', '创业', 'AI工具']
            
            for topic in topics:
                try:
                    # 这里可以调用抖音 MCP 或其他方式
                    # 暂时返回示例数据
                    results.append({
                        'source': 'douyin',
                        'title': f'{topic}相关热门视频',
                        'description': f'抖音{topic}话题热门内容',
                        'url': f'https://www.douyin.com/search/{topic}',
                        'metrics': {},
                        'tags': [topic, '抖音'],
                    })
                except Exception as e:
                    print(f"抖音数据获取失败: {e}")
        
        except Exception as e:
            print(f"抖音数据获取失败: {e}")
        
        return results[:5]
    
    def _get_bilibili_hot(self) -> List[Dict]:
        """获取 B站热门"""
        results = []
        
        try:
            # B站科技区热门
            import requests
            
            url = 'https://api.bilibili.com/x/web-interface/ranking/v2'
            params = {'rid': 36, 'type': 'all'}  # 36 = 科技区
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
                'Referer': 'https://www.bilibili.com/',
            }
            
            response = requests.get(url, params=params, headers=headers, timeout=10)
            if response.status_code == 200:
                data = response.json()
                for item in data.get('data', {}).get('list', [])[:10]:
                    results.append({
                        'source': 'bilibili',
                        'title': item.get('title', ''),
                        'description': item.get('desc', '')[:100],
                        'url': f"https://www.bilibili.com/video/{item.get('bvid', '')}",
                        'metrics': {
                            'play': item.get('stat', {}).get('view', 0),
                            'like': item.get('stat', {}).get('like', 0),
                        },
                        'tags': ['B站', '科技'],
                    })
        except Exception as e:
            print(f"B站数据获取失败: {e}")
        
        return results
    
    def _get_xiaohongshu_hot(self) -> List[Dict]:
        """获取小红书热门"""
        results = []
        
        # 小红书 API 需要登录，暂时使用搜索方式
        topics = ['AI赚钱', '副业', '创业', '自媒体']
        
        for topic in topics:
            results.append({
                'source': 'xiaohongshu',
                'title': f'{topic}相关热门笔记',
                'description': f'小红书{topic}话题热门内容',
                'url': f'https://www.xiaohongshu.com/search_result?keyword={topic}',
                'metrics': {},
                'tags': [topic, '小红书'],
            })
        
        return results[:5]
    
    def _get_twitter_ai(self) -> List[Dict]:
        """获取 Twitter AI trending"""
        results = []
        
        try:
            # 使用 bird CLI 搜索
            result = subprocess.run(
                ['bird', 'search', 'AI trending', '--limit', '10'],
                capture_output=True,
                text=True,
                timeout=30
            )
            
            if result.returncode == 0:
                # 解析 bird 输出
                lines = result.stdout.strip().split('\n')
                for line in lines[:10]:
                    if line.strip():
                        results.append({
                            'source': 'twitter',
                            'title': line[:100],
                            'description': line,
                            'url': 'https://twitter.com/search?q=AI',
                            'metrics': {},
                            'tags': ['AI', 'Twitter'],
                        })
        except Exception as e:
            print(f"Twitter 数据获取失败: {e}")
        
        # 备用数据
        if not results:
            results.append({
                'source': 'twitter',
                'title': 'AI trending on Twitter',
                'description': 'Twitter AI trending 搜索',
                'url': 'https://twitter.com/search?q=AI%20trending',
                'metrics': {},
                'tags': ['AI', 'Twitter'],
            })
        
        return results
    
    def filter_ai_related(self, items: List[Dict]) -> List[Dict]:
        """筛选 AI 相关内容"""
        ai_keywords = [
            'ai', 'artificial intelligence', '人工智能',
            'llm', 'gpt', 'chatgpt', 'claude',
            'agent', '智能体', '自动化',
            'machine learning', 'deep learning',
            'nlp', 'computer vision',
        ]
        
        filtered = []
        for item in items:
            text = f"{item.get('title', '')} {item.get('description', '')}".lower()
            if any(k in text for k in ai_keywords):
                filtered.append(item)
        
        return filtered
    
    def generate_ideas_from_hot(self, hot_data: Dict) -> List[Dict]:
        """从热点数据生成赚钱灵感"""
        ideas = []
        
        for source, items in hot_data.items():
            for item in items:
                # 简单判断是否有变现潜力
                potential = self._analyze_potential(item)
                if potential > 0:
                    ideas.append({
                        'source': source,
                        'title': item.get('title', ''),
                        'description': item.get('description', ''),
                        'url': item.get('url', ''),
                        'potential': potential,
                        'tags': item.get('tags', []),
                    })
        
        # 按潜力排序
        ideas.sort(key=lambda x: x['potential'], reverse=True)
        
        return ideas[:10]
    
    def _analyze_potential(self, item: Dict) -> int:
        """分析变现潜力（0-100）"""
        score = 0
        
        title = item.get('title', '').lower()
        desc = item.get('description', '').lower()
        text = f"{title} {desc}"
        
        # 关键词加分
        money_keywords = ['赚钱', '副业', '创业', '变现', '商业化', '盈利']
        for k in money_keywords:
            if k in text:
                score += 20
        
        ai_keywords = ['ai', '人工智能', 'gpt', 'agent', '自动化']
        for k in ai_keywords:
            if k in text:
                score += 15
        
        tool_keywords = ['工具', '平台', '系统', '软件', 'app']
        for k in tool_keywords:
            if k in text:
                score += 10
        
        # 数据源加分
        source = item.get('source', '')
        if source == 'github':
            score += 15  # GitHub 项目更容易变现
        elif source == 'bilibili':
            score += 10  # B站科技区有参考价值
        
        return min(score, 100)


# 测试
if __name__ == '__main__':
    monitor = MultiSourceMonitor()
    
    print("=== 获取多平台热点 ===\n")
    hot_data = monitor.get_all_hot()
    
    for source, items in hot_data.items():
        print(f"【{source.upper()}】")
        for i, item in enumerate(items[:3], 1):
            print(f"  {i}. {item['title'][:50]}")
        print()
    
    print("=== 生成赚钱灵感 ===\n")
    ideas = monitor.generate_ideas_from_hot(hot_data)
    for i, idea in enumerate(ideas[:5], 1):
        print(f"【灵感 #{i}】{idea['title'][:40]}")
        print(f"  来源: {idea['source']} | 潜力: {idea['potential']}")
        print()
```