拒绝无效内卷！用QClaw打造打工人专属"情绪-生存-睡眠"Agent

emobin/
├── SKILL.md          # Skill的元数据与触发词定义
├── main.py           # 主逻辑处理
├── philosophy.py     # 哲学升华模块
├── rap_generator.py  # Rap生成模块
└── config.json       # 配置文件

# EmoBin - 情绪垃圾桶与内容升华器

## 描述
接收用户的职场/生活吐槽，将其转化为哲学思考或Rap歌词，实现情绪宣泄与素材积累的双重价值。

## 触发词
- "吐槽" / "骂" / "烦死了" → 进入接收模式
- "升华一下" / "哲学模式" → 调用philosophy.py
- "来段rap" / "diss一下" → 调用rap_generator.py
- "保存到素材库" → 将内容追加到本地markdown文件

## 参数
- content: 用户的原始吐槽内容（必需）
- mode: 处理模式，可选"philosophy"或"rap"（默认philosophy）
- intensity: 毒舌强度等级1-5（用于Rap生成时的攻击性调节）

## 示例
用户输入："吐槽 这个需求又要改，产品经理是不是觉得时间不是成本？ 模式：rap 强度：4"

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
EmoBin - 情绪垃圾桶主模块
基于QClaw SDK v1.x开发
"""

import json
import os
from datetime import datetime
from typing import Literal
import qclaw
from qclaw import Context, Tool

# 初始化上下文
ctx = Context()

class EmoBinProcessor:
    def __init__(self):
        self.storage_path = os.path.expanduser("~/Documents/EmoBin/")
        self.ensure_storage()
        
    def ensure_storage(self):
        """确保存储目录存在"""
        if not os.path.exists(self.storage_path):
            os.makedirs(self.storage_path)
            # 创建素材库主文件
            with open(f"{self.storage_path}/materials.md", "w", encoding="utf-8") as f:
                f.write("# 情绪素材库\n\n> 自动收集的吐槽与升华内容\n\n")

    @Tool(description="分析用户情绪强度并分类")
    def analyze_emotion(self, content: str) -> dict:
        """
        使用本地LLM分析情绪类型与强度
        QClaw会自动路由到配置的模型（DeepSeek/Claude等）
        """
        prompt = f"""
        分析以下吐槽文本的情绪特征：
        文本："{content}"
        
        请返回JSON格式：
        {{
            "emotion_type": "anger/frustration/sarcasm/resignation",
            "intensity": 1-5,
            "keywords": ["关键词1", "关键词2"],
            "core_conflict": "核心矛盾总结（10字以内）"
        }}
        """
        
        # 调用QClaw封装的本地LLM接口
        response = ctx.llm.complete(prompt, temperature=0.3, json_mode=True)
        return json.loads(response.text)

    @Tool(description="将吐槽内容转化为哲学思考")
    def to_philosophy(self, content: str, emotion_data: dict) -> str:
        """
        哲学升华模块：将职场黑话映射到存在主义/后现代理论
        """
        philosophy_prompt = f"""
        你是一位精通存在主义、后现代主义、法兰克福学派的哲学教授。
        请将以下职场吐槽转化为严肃的哲学论述（300-400字）：
        
        原始吐槽：{content}
        核心矛盾：{emotion_data['core_conflict']}
        
        要求：
        1. 引用至少一位哲学家（海德格尔/萨特/福柯/鲍德里亚等）
        2. 使用学术化的黑话体系（异化、规训、景观社会、单向度的人等）
        3. 保持原文的核心诉求，但包装成"现代性困境"的批判
        4. 结尾要有升华：从个人抱怨上升到人类普遍处境
        
        风格参考：让·鲍德里亚《消费社会》的批判语调 + 微博哲学bot的犀利感
        """
        
        result = ctx.llm.complete(philosophy_prompt, temperature=0.8)
        return result.text

    @Tool(description="将吐槽转化为Rap歌词")
    def to_rap(self, content: str, intensity: int, emotion_data: dict) -> str:
        """
        Rap生成模块：带Flow标注的说唱词创作
        """
        flow_styles = {
            1: "Jazz Boombap, 慵懒叙事",
            2: "Trap, 稳定八拍", 
            3: "Drill, 急促进鼓",
            4: "Hardcore, 双韵脚快嘴",
            5: "Death Grips式工业噪音流, 无规则Flow"
        }
        
        rap_prompt = f"""
        你是一位underground battle MC。请将以下职场吐槽改写为Rap Verse（16-20 bars）：
        
        原始素材：{content}
        攻击强度：{intensity}/5
        推荐Flow：{flow_styles.get(intensity, "Trap")}
        
        创作要求：
        1. 每句用[ ]标注Flow技巧（如[停顿]、[三连音]、[加速]）
        2. 使用多音节韵脚（multi-syllable rhymes）
        3. Punchline要狠但保持隐喻（避免直接脏话）
        4. 加入Ad-libs（如"skr"、"ey"）标注
        5. 最后4 bars要有反转（从抱怨转向自我赋权）
        
        输出格式：
        [Intro: 情绪氛围描述]
        [Verse 1]
        歌词行1 [Flow技巧]
        ...
        [Outro]
        """
        
        result = ctx.llm.complete(rap_prompt, temperature=0.9)
        return result.text

    @Tool(description="保存内容到本地素材库")
    def save_to_vault(self, original: str, processed: str, mode: str):
        """
        持久化存储到本地markdown
        QClaw确保所有文件操作在本地沙箱内完成
        """
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
        entry = f"""
## {timestamp} | {mode.upper()}模式

**原始吐槽：**
> {original}

**升华内容：**
{processed}

---
"""
        vault_path = f"{self.storage_path}/materials.md"
        
        # 使用QClaw的安全文件操作API
        ctx.file.append(vault_path, entry, encoding="utf-8")
        
        return {"status": "saved", "path": vault_path, "timestamp": timestamp}

# 实例化处理器
processor = EmoBinProcessor()

@qclaw.on_message()
def handle_message(content: str, mode: Literal["philosophy", "rap"] = "philosophy", intensity: int = 3):
    """
    QClaw消息入口函数
    当用户在微信发送"吐槽 xxx"时被触发
    """
    try:
        # 1. 情绪分析
        ctx.send_typing()  # 发送"对方正在输入"状态到微信
        emotion_data = processor.analyze_emotion(content)
        
        # 2. 根据模式处理
        if mode == "philosophy":
            result = processor.to_philosophy(content, emotion_data)
            title = "🧠 哲学升华版"
        else:
            result = processor.to_rap(content, intensity, emotion_data)
            title = "🎤 Rap Battle版"
        
        # 3. 自动保存（可选）
        save_result = processor.save_to_vault(content, result, mode)
        
        # 4. 构造返回消息
        reply = f"""
{title}
{result}

---
💾 已自动保存至素材库 | 情绪强度：{'⭐' * emotion_data['intensity']}
📝 核心矛盾：{emotion_data['core_conflict']}
        """
        
        return reply
        
    except Exception as e:
        return f"❌ 情绪处理失败：{str(e)}\n建议：尝试降低毒性或使用更隐喻的表达方式"

if __name__ == "__main__":
    # 本地测试入口
    test_content = "这个需求又要改第八版，产品经理是不是觉得时间不是成本？"
    print(handle_message(test_content, mode="rap", intensity=4))

{
  "dangerous_combos": [
    {
      "ingredients": ["螃蟹", "柿子"],
      "level": 5,
      "reason": "鞣酸与蛋白质结合导致胃结石",
      "absolute_forbidden": true
    },
    {
      "ingredients": ["蜂蜜", "洋葱"],
      "level": 3,
      "reason": "轻微氧化反应，可能导致腹泻",
      "absolute_forbidden": false
    }
  ],
  "toxicity_scale": {
    "1": "完全安全，只是口味猎奇",
    "2": "肠胃敏感者可能不适",
    "3": "建议准备健胃消食片",
    "4": "需要签订自愿参与协议",
    "5": "生化危机级别，严禁尝试"
  }
}

#!/usr/bin/env python3
"""
DarkChef - 暗黑料理生成器
基于QClaw的食材剩余管理与黑暗料理推荐系统
"""

import random
import json
from dataclasses import dataclass
from typing import List, Tuple
import qclaw
from qclaw import Tool, Context

ctx = Context()

@dataclass
class Recipe:
    name: str
    ingredients: List[str]
    steps: List[str]
    toxicity_level: int  # 1-5星
    flavor_profile: str
    survival_tips: str

class DarkChefEngine:
    def __init__(self):
        self.config = self._load_config()
        self.cooking_methods = [
            "爆炒", "乱炖", "微波加热", "空气炸锅", 
            "搅拌机打成泥", "跨界fusion", "分子料理手法"
        ]
        
    def _load_config(self):
        skill_path = ctx.skill.get_path()  # 获取当前Skill目录
        with open(f"{skill_path}/config.json", "r", encoding="utf-8") as f:
            return json.load(f)
    
    def check_safety(self, ingredients: List[str]) -> Tuple[int, str]:
        """
        检查食材组合的安全性
        返回：(毒性等级, 警告信息)
        """
        max_level = 1
        warnings = []
        
        for danger in self.config["dangerous_combos"]:
            if all(ing in ingredients for ing in danger["ingredients"]):
                if danger["absolute_forbidden"]:
                    return 5, f"🚫 致命组合：{'+'.join(danger['ingredients'])} - {danger['reason']}"
                max_level = max(max_level, danger["level"])
                warnings.append(f"⚠️ {danger['reason']}")
        
        return max_level, "; ".join(warnings) if warnings else "✅ 理论上不会致死"

    @Tool(description="根据剩余食材生成暗黑料理方案")
    def generate_recipe(self, ingredients: str, mood: str = "绝望") -> Recipe:
        """
        主生成函数
        ingredients: 逗号分隔的食材列表，如"鸡蛋,酸奶,老干妈"
        mood: 当前心情，影响料理疯狂程度
        """
        ing_list = [i.strip() for i in ingredients.split(",")]
        
        # 1. 安全检查
        base_toxicity, warning = self.check_safety(ing_list)
        
        # 2. 如果检测到致命组合，直接拒绝
        if base_toxicity == 5:
            return Recipe(
                name="生化危机预防协议",
                ingredients=ing_list,
                steps=["立即停止", "分开存放", "点外卖"],
                toxicity_level=5,
                flavor_profile="活着真好",
                survival_tips="请不要尝试任何组合"
            )
        
        # 3. 调用LLM生成创意菜谱
        prompt = f"""
        你是一位疯癫的米其林黑暗料理主厨，擅长用奇怪的食材组合创造"理论上能吃"的食物。
        
        剩余食材：{', '.join(ing_list)}
        当前心情：{mood}（影响疯狂程度）
        安全警告：{warning}
        
        请生成一个料理方案，要求：
        1. 菜名要有meme感或哲学隐喻
        2. 烹饪步骤必须包含至少一个"邪门"操作（如"用咖啡机煮方便面"）
        3. 口味描述要用通感修辞（如"像失恋的周二下午"）
        4. 给出具体的毒性自评（1-5星）和生存建议
        
        返回格式：
        菜名：xxx
        难度：⭐⭐⭐
        毒性：⭐⭐（{base_toxicity}星基准）
        风味档案：xxx
        步骤：
        1. xxx
        2. xxx
        生存贴士：xxx
        """
        
        response = ctx.llm.complete(prompt, temperature=0.95)
        parsed = self._parse_recipe(response.text, ing_list, base_toxicity)
        
        # 4. 自动记录到"生存日志"
        self._log_attempt(parsed)
        
        return parsed
    
    def _parse_recipe(self, text: str, ingredients: List[str], base_toxicity: int) -> Recipe:
        """解析LLM返回的文本为结构化数据"""
        lines = text.strip().split("\n")
        
        name = lines[0].replace("菜名：", "").strip() if lines else "无名暗黑料理"
        steps = []
        survival = "无"
        
        in_steps = False
        for line in lines:
            if line.startswith("步骤："):
                in_steps = True
                continue
            if in_steps and line.strip():
                if line.startswith("生存贴士："):
                    survival = line.replace("生存贴士：", "").strip()
                    break
                steps.append(line.strip())
        
        # 动态调整毒性：如果LLM给出的描述很可怕，加1星
        toxicity_keywords = ["喷射", "住院", "洗胃", "幻觉"]
        bonus = 1 if any(k in text for k in toxicity_keywords) else 0
        final_toxicity = min(5, base_toxicity + bonus)
        
        return Recipe(
            name=name,
            ingredients=ingredients,
            steps=steps,
            toxicity_level=final_toxicity,
            flavor_profile="猎奇向",
            survival_tips=survival
        )
    
    def _log_attempt(self, recipe: Recipe):
        """记录到本地生存日志，用于后续优化"""
        log_entry = {
            "timestamp": ctx.time.now(),
            "recipe": recipe.name,
            "toxicity": recipe.toxicity_level,
            "ingredients": recipe.ingredients
        }
        log_path = f"{ctx.skill.get_path()}/survival_log.jsonl"
        ctx.file.append(log_path, json.dumps(log_entry, ensure_ascii=False) + "\n")

# 初始化引擎
chef = DarkChefEngine()

@qclaw.on_message()
def cook(ingredients: str, style: str = "fusion"):
    """
    QClaw入口：用户发送"今晚吃什么 鸡蛋,酸奶,老干妈"
    """
    try:
        recipe = chef.generate_recipe(ingredients)
        
        # 构造可视化输出
        toxicity_stars = "⭐" * recipe.toxicity_level + "☆" * (5 - recipe.toxicity_level)
        danger_level = ["绿色：可食用", "黄色：勇士专属", "橙色：肠胃挑战", "红色：生死状", "黑色：遗书预备"][recipe.toxicity_level - 1]
        
        reply = f"""
🍳 【{recipe.name}】
━━━━━━━━━━━━━━━━━━━━
🧪 食材：{' + '.join(recipe.ingredients)}
⚗️ 毒性评估：{toxicity_stars} ({danger_level})
🎭 风味：{recipe.flavor_profile}

👨‍🍳 烹饪步骤：
{chr(10).join(f"  {i+1}. {step}" for i, step in enumerate(recipe.steps))}

💊 生存贴士：{recipe.survival_tips}

━━━━━━━━━━━━━━━━━━━━
⚠️ 免责声明：本菜谱由AI生成，QClaw及开发者不对任何肠胃不适负责
        """
        
        # 如果毒性过高，追加安全警告
        if recipe.toxicity_level >= 4:
            reply += "\n🚨 警告：检测到高危险操作，建议拍照留念后再食用"
            
        return reply
        
    except Exception as e:
        return f"厨房爆炸了：{str(e)}\n建议：还是点麦当劳吧"

@qclaw.scheduled(cron="0 18 * * 5")  # 每周五晚6点自动提醒
def friday_survival_check():
    """
    定时任务：周五晚上检查冰箱并推送建议
    QClaw支持Cron表达式定时触发
    """
    # 这里可以接入智能冰箱API或让用户提前拍照OCR
    # 简化版：随机推荐一个"周五特调"
    backup_ingredients = ["剩米饭", "鸡蛋", "辣酱", "可乐"]
    recipe = chef.generate_recipe(",".join(backup_ingredients), mood="周末前的绝望")
    
    return f"""
🍽️ 周五生存指南自动推送
检测到您可能面临"不知道吃什么"的 existential crisis。

推荐方案：{recipe.name}
毒性：{'⭐' * recipe.toxicity_level}
输入"查看详情"获取完整菜谱，或输入"外卖"放弃治疗。
    """

## 权限声明
- schedule: 需要创建定时任务
- notification: 需要发送系统通知
- wechat: 需要发送微信消息（仅向指定好友）
- file: 需要读写本地睡眠日志

## 安全机制
- 所有消息发送需用户预先配置"监督人"白名单
- 每日最多发送1条惩罚消息，避免骚扰
- 提供"紧急取消"指令（睡前5分钟内有效）

#!/usr/bin/env python3
"""
SleepGuard - 睡眠拖延对抗系统
通过社交压力与经济损失预期倒逼自律
"""

import json
import time
from datetime import datetime, timedelta
from typing import List
import qclaw
from qclaw import Tool, Context, scheduler

ctx = Context()

class SleepGuardian:
    def __init__(self):
        self.config_path = f"{ctx.skill.get_path()}/sleep_config.json"
        self.state_path = f"{ctx.skill.get_path()}/sleep_state.json"
        self.load_config()
        
    def load_config(self):
        """加载用户配置"""
        try:
            with open(self.config_path, "r", encoding="utf-8") as f:
                self.config = json.load(f)
        except FileNotFoundError:
            # 默认配置
            self.config = {
                "bedtime": "23:30",
                "supervisors": [],  # 监督人微信ID列表
                "fine_amount": 50,
                "challenge_msg": "我如果1小时内不回消息就给你转{amount}块",
                "grace_period_minutes": 5
            }
            self.save_config()
    
    def save_config(self):
        with open(self.config_path, "w", encoding="utf-8") as f:
            json.dump(self.config, f, ensure_ascii=False, indent=2)
    
    @Tool(description="设置睡眠监督参数")
    def setup(self, bedtime: str, supervisors: List[str], fine: int = 50):
        """
        初始化配置
        bedtime: "HH:MM"格式
        supervisors: 监督人微信昵称或备注名列表
        """
        self.config.update({
            "bedtime": bedtime,
            "supervisors": supervisors,
            "fine_amount": fine
        })
        self.save_config()
        
        # 注册定时任务：在睡前15分钟发送预警
        bedtime_dt = datetime.strptime(bedtime, "%H:%M")
        warning_time = (bedtime_dt - timedelta(minutes=15)).strftime("%H:%M")
        
        scheduler.daily(warning_time, self.send_bedtime_warning)
        scheduler.daily(bedtime, self.initiate_sleep_challenge)
        
        return f"""
🛡️ 睡眠监督系统已激活
⏰ 就寝时间：{bedtime}
👥 监督人：{', '.join(supervisors)}
💰 违约罚金：{fine}元
⚠️ 系统将在每天{warning_time}发送预警，{bedtime}启动挑战协议
        """
    
    @Tool(description="发送睡前预警")
    def send_bedtime_warning(self):
        """睡前15分钟提醒"""
        ctx.notification.send(
            title="🌙 睡眠预警",
            body=f"还有15分钟就{self.config['bedtime']}了，请准备进入睡眠模式。关闭手机或回复'准备睡觉'以取消挑战。",
            urgency="high"
        )
        return "预警已发送"
    
    @Tool(description="启动睡眠挑战协议")
    def initiate_sleep_challenge(self):
        """
        核心逻辑：发送社死挑战消息给监督人
        """
        # 检查是否已经标记为已睡
        state = self.get_today_state()
        if state.get("status") == "sleeping":
            return "用户已确认入睡，跳过挑战"
        
        # 构造挑战消息
        challenge = self.config["challenge_msg"].format(
            amount=self.config["fine_amount"]
        )
        
        full_msg = f"""
【SleepGuard自动消息】
我是{ctx.user.name}的AI睡眠督察员。
{challenge}
时间戳：{datetime.now().strftime("%H:%M:%S")}
如{self.config['grace_period_minutes']}分钟内未收到回复，视为违约。

（此消息由QClaw自动发送，用于睡眠习惯养成）
        """
        
        # 发送给所有监督人
        sent_count = 0
        for supervisor in self.config["supervisors"]:
            try:
                ctx.wechat.send_message(
                    to=supervisor,
                    content=full_msg,
                    confirm=True  # 需要二次确认，防止滥用
                )
                sent_count += 1
            except Exception as e:
                ctx.log.error(f"发送给{supervisor}失败: {e}")
        
        # 更新状态
        self.update_state("challenge_sent", datetime.now().isoformat())
        
        # 设置延迟检查任务
        scheduler.once(
            delay_minutes=self.config["grace_period_minutes'],
            task=self.check_challenge_result
        )
        
        return f"挑战已发送给{sent_count}位监督人，{self.config['grace_period_minutes']}分钟后检查回复状态"
    
    @Tool(description="检查挑战结果并执行惩罚")
    def check_challenge_result(self):
        """
        检查用户是否在规定时间内回复
        如果没有，发送"社死确认"消息
        """
        state = self.get_today_state()
        
        if state.get("replied_in_time"):
            return "用户已及时回复，挑战通过"
        
        # 用户未回复，发送社死升级消息
        shame_msg = f"""
【SleepGuard违约通知】
时间：{datetime.now().strftime("%H:%M")}
状态：{ctx.user.name}未在规定时间内证明已入睡
处罚：请监督执行{self.config['fine_amount']}元转账
备注：熬夜伤身，金钱伤心，望周知 🌚
        """
        
        for supervisor in self.config["supervisors"]:
            ctx.wechat.send_message(supervisor, shame_msg)
        
        # 记录违约
        self.log_violation()
        
        return "违约处理完成"
    
    @Tool(description="用户回复验证")
    def verify_sleep(self, reply_code: str = ""):
        """
        用户收到挑战后回复此指令
        """
        state = self.get_today_state()
        
        if state.get("status") != "challenge_sent":
            return "当前没有待处理的睡眠挑战"
        
        # 检查时间是否在宽限期内
        challenge_time = datetime.fromisoformat(state["challenge_time"])
        if datetime.now() - challenge_time > timedelta(minutes=self.config["grace_period_minutes"]):
            return "⏰ 超时！宽限期已过，违约记录已生成"
        
        self.update_state("replied_in_time", True)
        self.update_state("status", "sleeping")
        
        # 通知监督人解除警报
        for sup in self.config["supervisors"]:
            ctx.wechat.send_message(sup, f"✅ {ctx.user.name}已确认进入睡眠模式，今晚的社死危机解除")
        
        return "🎉 验证成功！祝你好梦。明早见。"
    
    def get_today_state(self):
        """获取今日状态"""
        try:
            with open(self.state_path, "r") as f:
                all_states = json.load(f)
            today = datetime.now().strftime("%Y-%m-%d")
            return all_states.get(today, {})
        except:
            return {}
    
    def update_state(self, key, value):
        """更新状态"""
        today = datetime.now().strftime("%Y-%m-%d")
        try:
            with open(self.state_path, "r") as f:
                all_states = json.load(f)
        except:
            all_states = {}
        
        if today not in all_states:
            all_states[today] = {}
        
        all_states[today][key] = value
        
        with open(self.state_path, "w") as f:
            json.dump(all_states, f)
    
    def log_violation(self):
        """记录违约历史"""
        log_path = f"{ctx.skill.get_path()}/violations.jsonl"
        entry = {
            "date": datetime.now().isoformat(),
            "amount": self.config["fine_amount"],
            "supervisors_count": len(self.config["supervisors"])
        }
        ctx.file.append(log_path, json.dumps(entry) + "\n")

# 初始化
guardian = SleepGuardian()

@qclaw.on_message()
def handle_sleep_cmd(command: str, *args):
    """
    命令分发
    支持：
    - "设置睡眠 23:30 监督人A,监督人B 50"
    - "准备睡觉"（验证回复）
    - "查看违约记录"
    """
    if command.startswith("设置睡眠"):
        # 解析：设置睡眠 HH:MM 监督人列表 金额
        parts = command.split()
        if len(parts) < 3:
            return "格式错误。示例：设置睡眠 23:30 张三,李四 50"
        
        time_str = parts[1]
        supervisors = parts[2].split(",")
        fine = int(parts[3]) if len(parts) > 3 else 50
        
        return guardian.setup(time_str, supervisors, fine)
    
    elif "睡觉" in command or "睡了" in command:
        return guardian.verify_sleep()
    
    elif "违约" in command:
        # 简单统计
        log_path = f"{ctx.skill.get_path()}/violations.jsonl"
        try:
            with open(log_path, "r") as f:
                lines = f.readlines()
            total = len(lines)
            total_fine = sum(json.loads(l)["amount"] for l in lines)
            return f"📊 历史违约{total}次，累计应罚{total_fine}元（赶紧给监督人转账吧）"
        except:
            return "暂无违约记录，继续保持！"
    
    else:
        return "可用指令：\n1. 设置睡眠 [时间] [监督人] [金额]\n2. 准备睡觉（收到挑战后回复）\n3. 查看违约记录"

if __name__ == "__main__":
    # 测试设置
    print(guardian.setup("23:30", ["TestUser"], 50))