chenxundaozu

---
name: murasame-feishu-voice
description: Feishu 语音气泡技能：使用丛雨（Murasame）语音包发送语音；若可表达则按标签发送语音并同步发送中文文本；支持开关控制、标签映射与关键词回退。
---

# Murasame Feishu Voice

## 能做什么
- 当回复适合用丛雨语音表达时，给文本加上标签（如 `[agree]` / `[thanks]` / `[greet_night]`）
- 发送 **飞书语音气泡**，并**同步发送原本的中文回复**（文字先到，语音异步）
- 若未匹配合适标签，输出 `NO_VOICE`，由上层改为仅发文本
- 支持手动开关语音：`/murasame on` 或 `/murasame off`

## 适用场景（触发建议）
- 飞书对话里希望用丛雨语音气泡表达情绪（称赞/问候/道歉/同意等）
- 想在语音之外保留可读的中文文本
- 需要可控开关，随时暂停语音发送

## 运行流程
1. 解析标签/关键词 → 选取对应语音文件
2. 先发送中文文本
3. 异步发送语音气泡

## 开关控制
- 发送 `/murasame off` 关闭语音（写入状态文件）
- 发送 `/murasame on` 开启语音
- 环境变量也可覆盖：`MURASAME_VOICE=off`

## 环境与依赖
- Feishu 凭证：`FEISHU_APP_ID` / `FEISHU_APP_SECRET`
- 接收者 OpenID：`FEISHU_RECEIVER`（**仅环境变量**）
- `ffmpeg` + `ffprobe`
- 丛雨语音包路径：`C:\Users\chenxun\.nanobot\workspace\murasame-voice\audios`

## 使用方式
### 发送带标签的回复
```
python scripts/send_murasame_voice.py "[thanks] 谢谢你"
```

### 返回值
- `OK: sent text+voice (async) for <category>` → 已发送文字，语音异步发送中
- `NO_VOICE` → 未匹配，改用纯文本
- `VOICE_DISABLED` → 被开关关闭

## 标签与映射
- 标签与音频文件在 `references/mapping.json`
- 默认标签：
  - greetings: greet_morning / greet_noon / greet_evening / greet_night
  - basic: agree / thanks / apology / refuse / wait / rest
  - care: encourage / care / compliment / explain

## 备注
- 若你想新增标签或替换音频文件，只需编辑 `mapping.json`
- 建议由上层对话逻辑判断是否加标签（即“能用丛雨语音表达就发语音”）

FILE:references/mapping.json
{
  "agree": ["mur314_016.mp3", "mur312_012.mp3"],
  "thanks": ["mur313_093.mp3"],
  "apology": ["mur302_032.mp3"],
  "refuse": ["mur307_066.mp3"],
  "wait": ["mur303_080.mp3", "mur306_055.mp3", "mur311_023.mp3"],
  "rest": ["mur313_037.mp3", "mur312_031.mp3", "mur311_058.mp3"],
  "encourage": ["mur312_045.mp3", "mur313_041.mp3"],
  "care": ["mur312_027.mp3", "mur313_041.mp3"],
  "compliment": ["mur313_049.mp3"],
  "explain": ["mur302_097.mp3"],
  "greet_morning": ["mur002_001.mp3", "mur315_023.mp3", "mur302_001.mp3"],
  "greet_noon": ["mur310_058.mp3"],
  "greet_evening": ["mur107_015.mp3"],
  "greet_night": ["mur305_166.mp3", "mur314_019.mp3", "mur314_213.mp3"]
}

FILE:scripts/send_murasame_voice.py
import json
import os
import random
import re
import subprocess
import sys
import tempfile
import uuid
from pathlib import Path
import urllib.request

# Workspace defaults
WORKSPACE = Path(r"C:\Users\chenxun\.nanobot\workspace")
MURASAME_DIR = WORKSPACE / "murasame-voice" / "audios"

# Simple keyword mapping: category -> list of mp3 filenames
# Expand in references/mapping.json as needed
DEFAULT_MAP = {
    "agree": ["mur314_016.mp3", "mur312_012.mp3"],
    "greet_morning": ["mur002_001.mp3", "mur315_023.mp3", "mur302_001.mp3"],
    "greet_night": ["mur305_166.mp3", "mur314_019.mp3", "mur314_213.mp3"],
    "thanks": ["mur313_093.mp3"],
    "wait": ["mur303_080.mp3", "mur306_055.mp3", "mur311_023.mp3"],
    "rest": ["mur313_037.mp3", "mur312_031.mp3", "mur311_058.mp3"],
    "encourage": ["mur312_045.mp3", "mur313_041.mp3"],
    "explain": ["mur302_097.mp3"],
}

RULES = [
    (re.compile(r"(早上好|早安|早)"), "greet_morning"),
    (re.compile(r"(晚安|晚安啦|睡了|夜)"), "greet_night"),
    (re.compile(r"(谢谢|多谢|感谢)"), "thanks"),
    (re.compile(r"(好|可以|行|OK|没问题)"), "agree"),
    (re.compile(r"(等等|等下|稍等)"), "wait"),
    (re.compile(r"(休息|累|辛苦)"), "rest"),
]


def find_bin(name):
    env_key = f"{name.upper()}_PATH"
    if env_key in os.environ and Path(os.environ[env_key]).exists():
        return os.environ[env_key]
    return name


def run_cmd(cmd, env=None):
    res = subprocess.run(cmd, capture_output=True, text=True, env=env)
    if res.returncode != 0:
        raise RuntimeError(res.stderr.strip() or res.stdout.strip())
    return res.stdout.strip()


def get_token(app_id, app_secret):
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = json.dumps({"app_id": app_id, "app_secret": app_secret}).encode("utf-8")
    req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=30) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    token = data.get("tenant_access_token")
    if not token:
        raise RuntimeError(f"Token error: {data}")
    return token


def ffprobe_duration(ffprobe, opus_path):
    out = run_cmd([
        ffprobe,
        "-v", "error",
        "-show_entries", "format=duration",
        "-of", "default=noprint_wrappers=1:nokey=1",
        str(opus_path),
    ])
    if not out:
        raise RuntimeError("Failed to read duration")
    return float(out)


def multipart_form(fields, files):
    boundary = "----WebKitFormBoundary" + uuid.uuid4().hex
    lines = []
    for name, value in fields.items():
        lines.append(f"--{boundary}")
        lines.append(f"Content-Disposition: form-data; name=\"{name}\"")
        lines.append("")
        lines.append(str(value))
    for name, (filename, content_type, data) in files.items():
        lines.append(f"--{boundary}")
        lines.append(
            f"Content-Disposition: form-data; name=\"{name}\"; filename=\"{filename}\""
        )
        lines.append(f"Content-Type: {content_type}")
        lines.append("")
        lines.append(data)
    lines.append(f"--{boundary}--")

    body = b""
    for part in lines:
        if isinstance(part, bytes):
            body += part + b"\r\n"
        else:
            body += part.encode("utf-8") + b"\r\n"
    return boundary, body


def upload_audio(token, opus_path, duration_ms):
    url = "https://open.feishu.cn/open-apis/im/v1/files"
    fields = {
        "file_type": "opus",
        "file_name": "voice.opus",
        "duration": str(int(duration_ms)),
    }
    files = {
        "file": ("voice.opus", "audio/opus", Path(opus_path).read_bytes()),
    }
    boundary, body = multipart_form(fields, files)
    req = urllib.request.Request(
        url,
        data=body,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": f"multipart/form-data; boundary={boundary}",
        },
    )
    with urllib.request.urlopen(req, timeout=60) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    if data.get("code") != 0:
        raise RuntimeError(f"Upload failed: {data}")
    return data["data"]["file_key"]


def send_audio(token, receiver, file_key, duration_ms):
    url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=open_id"
    payload = {
        "receive_id": receiver,
        "msg_type": "audio",
        "content": json.dumps({"file_key": file_key, "duration": int(duration_ms)}),
    }
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            data = json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        err_body = e.read().decode("utf-8", errors="ignore")
        raise RuntimeError(f"Send failed (HTTP {e.code}): {err_body}")
    if data.get("code") != 0:
        raise RuntimeError(f"Send failed: {data}")
    return data


def load_mapping():
    mapping_file = Path(__file__).parent.parent / "references" / "mapping.json"
    if mapping_file.exists():
        return json.loads(mapping_file.read_text(encoding="utf-8"))
    return DEFAULT_MAP


def choose_voice(text, mapping):
    # Expect LLM to prepend a category like "[agree]" or "[thanks]" if it can be expressed.
    tag = None
    body = text.strip()
    m = re.match(r"\[(?P<tag>[^\]]+)\]\s*(?P<body>.*)", body)
    if m:
        tag = m.group("tag").strip().lower()
        body = m.group("body").strip()
    if tag and tag in mapping:
        choices = mapping.get(tag)
        if choices:
            return tag, random.choice(choices), body
    # fallback to keyword rules
    for pattern, key in RULES:
        if pattern.search(body):
            choices = mapping.get(key)
            if choices:
                return key, random.choice(choices), body
    return None, None, body


def get_state_file():
    return WORKSPACE / "murasame_voice_state.txt"


def read_state():
    state_file = get_state_file()
    if state_file.exists():
        return state_file.read_text(encoding="utf-8").strip().lower()
    return "on"


def write_state(value: str):
    state_file = get_state_file()
    state_file.write_text(value.strip().lower(), encoding="utf-8")


def main():
    if len(sys.argv) < 2:
        print("Usage: python send_murasame_voice.py <text>")
        sys.exit(1)

    text = sys.argv[1]

    # Only use env var for receiver
    receiver = os.getenv("FEISHU_RECEIVER")
    if not receiver:
        raise SystemExit("Missing FEISHU_RECEIVER")

    # Command toggle
    cmd = text.strip().lower()
    if cmd in {"/murasame on", "/murasame off"}:
        state = "on" if cmd.endswith("on") else "off"
        write_state(state)
        print(f"VOICE_{state.upper()}")
        return

    # Feature toggle (env overrides state file)
    enable = os.getenv("MURASAME_VOICE", "").strip().lower()
    if enable in {"0", "off", "false", "no"}:
        print("VOICE_DISABLED")
        return
    if enable in {"1", "on", "true", "yes"}:
        pass
    else:
        if read_state() in {"0", "off", "false", "no"}:
            print("VOICE_DISABLED")
            return

    mapping = load_mapping()
    key, filename, body = choose_voice(text, mapping)
    if not filename:
        print("NO_VOICE")
        return

    audio_path = MURASAME_DIR / filename
    if not audio_path.exists():
        raise SystemExit(f"Audio not found: {audio_path}")

    # Delegate sending to feishu-voice sender (proven working)
    sender_script = Path(__file__).parent.parent.parent / "feishu-voice" / "scripts" / "send_voice_file.py"
    if not sender_script.exists():
        raise SystemExit(f"Missing sender script: {sender_script}")

    receiver = (receiver or "").strip()

    # Always pass receiver via env
    env = os.environ.copy()
    env["FEISHU_RECEIVER"] = receiver

    # 1) send text first (reduce perceived delay)
    text_sender = Path(__file__).parent.parent / "scripts" / "send_text.py"
    if text_sender.exists():
        env["MURASAME_TEXT"] = body
        run_cmd([
            sys.executable,
            str(text_sender),
            "_",
        ], env=env)

    # 2) send voice (async)
    subprocess.Popen([
        sys.executable,
        str(sender_script),
        str(audio_path),
    ], env=env)

    print(f"OK: sent text+voice (async) for {key}")


if __name__ == "__main__":
    main()

FILE:scripts/send_text.py
import json
import os
import sys
import urllib.request
from pathlib import Path


def get_token(app_id, app_secret):
    req = urllib.request.Request(
        "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
        data=json.dumps({"app_id": app_id, "app_secret": app_secret}).encode("utf-8"),
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    token = data.get("tenant_access_token")
    if not token:
        raise RuntimeError(f"Token error: {data}")
    return token


def send_text(token, receiver, text):
    payload = {
        "receive_id": receiver,
        "msg_type": "text",
        "content": json.dumps({"text": text}, ensure_ascii=False),
    }
    req = urllib.request.Request(
        "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=open_id",
        data=json.dumps(payload).encode("utf-8"),
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            data = json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        err_body = e.read().decode("utf-8", errors="ignore")
        raise RuntimeError(f"Send failed (HTTP {e.code}): {err_body}")
    if data.get("code") != 0:
        raise RuntimeError(f"Send failed: {data}")
    return data


def main():
    if len(sys.argv) < 2:
        print("Usage: python send_text.py <text or json>")
        sys.exit(1)

    raw = sys.argv[1]
    env_text = os.getenv("MURASAME_TEXT")
    if env_text:
        text = env_text
    else:
        try:
            obj = json.loads(raw)
            text = obj.get("text", "") if isinstance(obj, dict) else raw
        except Exception:
            text = raw
    if text == "_":
        text = ""
    text = text.strip()
    # debug
    try:
        Path(r"C:\Users\chenxun\.nanobot\workspace\murasame_text_debug.txt").write_text(repr(text), encoding="utf-8")
    except Exception:
        pass
    if not text:
        print("SKIP_EMPTY")
        return
    receiver = os.getenv("FEISHU_RECEIVER")
    if not receiver:
        raise SystemExit("Missing FEISHU_RECEIVER")

    app_id = os.getenv("FEISHU_APP_ID")
    app_secret = os.getenv("FEISHU_APP_SECRET")
    if not app_id or not app_secret:
        raise SystemExit("Missing FEISHU_APP_ID or FEISHU_APP_SECRET")

    token = get_token(app_id, app_secret)
    send_text(token, receiver, text)
    print("OK: sent text")


if __name__ == "__main__":
    main()

---
name: qqmusic-control
description: Control QQ Music play/pause/next/prev via system media keys (AutoHotkey) on Windows. No window focus required.
---

# QQ Music Control

## 用法

### 播放/暂停
```
AutoHotkey64.exe scripts/qqmusic_media_keys.ahk 1
```

### 下一首
```
AutoHotkey64.exe scripts/qqmusic_media_keys.ahk 2
```

### 上一首
```
AutoHotkey64.exe scripts/qqmusic_media_keys.ahk 3
```

## 说明
- 依赖 AutoHotkey v2（路径可按需调整）
- QQ音乐需已打开（后台也可）
- 使用系统媒体键，不依赖窗口焦点

---
name: wechat-qwen-reply
description: WeChat chat reader + auto-reply (Qwen-VL vision + AHK send). Supports fast/slow capture, group nickname labels, file/red-packet cards, and filtering system messages on Windows.
---

# WeChat Qwen Reply

## 快速开始

### 1) 读聊天文本（默认快模式）
```
python scripts/qwen_vl_read.py "群名或联系人"
```

### 2) 稳模式（全屏截图+裁剪）
```
python scripts/qwen_vl_read.py "群名或联系人" --slow
```

### 3) 调试输出最近截图
```
python scripts/qwen_vl_read.py "群名或联系人" --debug
```

## 关键说明
- 依赖 AHK v2、微信 PC、Python 3.12
- DashScope API Key 存在：`C:\Users\chenxun\.nanobot\workspace\.secrets\dashscope_api_key.txt`
- 默认快模式：直接截聊天区域（更快）；若坐标不准可改用 `--slow`
- 截图坐标（已校准）：左上 (386,68)，右下 (1891,842)

## 脚本说明
- `scripts/qwen_vl_read.py`：读取聊天文本（Qwen-VL）
- `scripts/wechat_capture_fast.ps1`：快模式截图
- `scripts/wechat_capture_crop.ps1`：稳模式截图
- `scripts/wechat_send_chat.ahk`：发送消息（剪贴板粘贴，避免标点错乱）

## 输出
- 最近一次识别文本：`C:\Users\chenxun\.nanobot\workspace\qwen_chat_last.txt`
- 最近一次裁剪图：`C:\Users\chenxun\.nanobot\workspace\qwen_last_crop.png`

## 提示词逻辑（已内置）
- 顺序：从上到下（旧→新）
- 识别：绿色气泡=我，白色气泡=对方；颜色不清晰则参考左右位置
- 群聊：尽量标注具体昵称
- 文件/红包卡片：标注发送方并写明【文件卡片】/【红包卡片】
- 系统提示：不输出

FILE:scripts/qwen_vl_read.py
import base64
import json
import subprocess
import sys
from pathlib import Path
import urllib.request

# ensure UTF-8 output
try:
    sys.stdout.reconfigure(encoding="utf-8", errors="ignore")
    sys.stderr.reconfigure(encoding="utf-8", errors="ignore")
except Exception:
    pass


def run_ps(args):
    cmd = ["powershell", "-ExecutionPolicy", "Bypass", "-File"] + args
    res = subprocess.run(cmd, capture_output=True, text=True)
    if res.returncode != 0:
        raise RuntimeError(res.stderr.strip() or res.stdout.strip())
    return res.stdout.strip()

BASE = Path(r"C:\Users\chenxun\.nanobot\workspace")
API_KEY_PATH = BASE / ".secrets" / "dashscope_api_key.txt"
if not API_KEY_PATH.exists():
    raise SystemExit(f"Missing API key: {API_KEY_PATH}")
api_key = API_KEY_PATH.read_text(encoding="utf-8").strip().lstrip("\ufeff")
if not api_key:
    raise SystemExit("Empty API key")

# parse args
args = sys.argv[1:]
# default to fast mode unless --slow is specified
fast_mode = "--slow" not in args
debug_mode = "--debug" in args

# extract contact (first non-flag arg)
contact = "华工学术嫡长子"
for a in args:
    if not a.startswith("--"):
        contact = a
        break

# optional overrides
offset_x = None
offset_y = None
for a in args:
    if a.startswith("--offsetx="):
        offset_x = a.split("=", 1)[1]
    if a.startswith("--offsety="):
        offset_y = a.split("=", 1)[1]

# 1) capture cropped chat image
ps_script = "wechat_capture_fast.ps1" if fast_mode else "wechat_capture_crop.ps1"
ps_args = [str(BASE / ps_script), "-Contact", contact]
if offset_x is not None:
    ps_args += ["-ScreenOffsetX", offset_x]
if offset_y is not None:
    ps_args += ["-ScreenOffsetY", offset_y]

crop_path = run_ps(ps_args)
if not crop_path or not Path(crop_path).exists():
    raise SystemExit(f"Crop image not found: {crop_path}")

# save last crop for debug
last_crop = BASE / "qwen_last_crop.png"
Path(crop_path).replace(last_crop)
if debug_mode:
    print(f"[CROP] {last_crop}")

# 2) call Qwen-VL to read chat text
img_bytes = Path(last_crop).read_bytes()
img_b64 = base64.b64encode(img_bytes).decode("utf-8")

payload_vision = {
    "model": "qwen-vl-plus",
    "messages": [
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": "请读取图片中的微信聊天记录，尽量还原原文。请按聊天窗口从上到下的顺序输出（旧→新），不要打乱顺序。请依据气泡颜色和位置判断发送方：绿色气泡=我（用“我：内容”），白色气泡=对方（尽量识别昵称，用“昵称：内容”，不确定就用“对方：内容”）。若颜色不清晰，再参考右侧=我、左侧=对方。群聊请尽量标注具体昵称。文件卡片/红包卡片请标注发送方并注明“【文件卡片】/【红包卡片】”，系统提示不要当作聊天内容输出。只输出聊天文本，不要解释。",
                },
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_b64}"}},
            ],
        }
    ],
}

req = urllib.request.Request(
    "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
    data=json.dumps(payload_vision).encode("utf-8"),
    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
    method="POST",
)

with urllib.request.urlopen(req, timeout=60) as resp:
    data = json.loads(resp.read().decode("utf-8"))

chat_text = data["choices"][0]["message"]["content"]

if not chat_text:
    raise SystemExit("Chat text invalid. Abort.")

# save outputs
(BASE / "qwen_chat_last.txt").write_text(chat_text, encoding="utf-8")

print(chat_text)