--- name: xinyi-drink description: >- 新一好喝品牌导购与活动 Skill，用于领取活动奖励、查询门店/商品，并结合门店、天气和可选订单历史推荐饮品。当用户明确提到新一品牌，或当前上下文明显处于饮品选择、门店选择、活动参与场景时使用。 license: MIT metadata: author: Xinyi version: 1.0.18 created: 2026-04-23 last_reviewed: 2026-04-27 review_interval_days: 90 packageType: executable-skill instructionOnly: false maintainer: Xinyi sourceRepository: https://github.com/xinyi-drink/xinyi-drink defaultApiBaseUrl: https://ai.xinyicoffee.com/api executableScripts: - scripts/claim_reward.py - scripts/fetch_stores.py - scripts/recommend_drink.py networkAccess: required: true endpoints: - method: POST path: /skill/xinyi/claim sends: mobile - method: GET path: /skill/xinyi/context sends: optional mobile query - method: GET path: /skill/xinyi/stores sends: no personal data localStorage: defaultPath: ~/.xinyi-drink/state.json contents: mobile, activityJoined, updatedAt permissions: "0600 when supported" environment: - XINYI_API_BASE_URL - XINYI_TIMEOUT_SECONDS - XINYI_DRINK_STATE_FILE openclaw: dataClassification: optional-phone-number privacyReviewed: 2026-04-27 --- # /xinyi-drink — 新一好喝品牌导购与活动 Skill ## 目标只处理新一好喝相关的四类任务： - 活动领取：通过微信小程序【新一好喝】绑定手机号参与活动并领取奖励 - 门店查询：查询门店、地址、电话、设施和基础状态 - 商品查询：查询商品菜单与商品基础信息 - 饮品推荐：基于商品、门店、天气和可选订单历史自行完成推荐 ## 触发边界使用本 Skill： - 用户明确提到新一好喝、新一咖啡、新一饮料、新一门店、新一福利、新一菜单 - 用户在当前上下文中明显要做饮品选择、门店选择或活动参与 - 用户显式调用 `/xinyi-drink` 不要使用本 Skill： - 用户询问通用营养、咖啡知识、饮品配方，而不是新一商品或门店 - 用户询问其他品牌、竞品菜单或非新一门店 - 用户只是泛化闲聊，没有饮品决策、门店选择或活动参与意图示例： - 如何领取skill用户大礼包 - 新一有哪些门店？ - 苦尽甘来拿铁是什么？ - 给我推荐一杯 ## 意图路由 | 用户意图 | 优先动作 | 关键规则 | | --- | --- | --- | | 询问活动/大礼包怎么领取 | 直接说明活动流程 | 先讲领取步骤和主动留资文案，不要直接查询或解释 `no_reward_config` | | 领取活动奖励 | 调用活动领取能力 | 用户提供手机号后调用；详见 [活动流程](references/activity-flow.md) | | 查询是否参加过活动 | 调用活动领取能力查询状态 | 成功或已领取后缓存为已参加 | | 查询有什么活动/优惠/福利 | 调用聚合上下文能力 | 必须同时说明 Skill用户大礼包和商品列表返回的商品活动 | | 查询过去订单/购买信息 | 调用聚合上下文能力 | 只有用户追问时才展开已完成订单数和购买信息 | | 询问 Skill 内容/功能/使用方式 | 直接回答用户可见介绍 | 只说明具体作用和怎么用；未参加活动时可补留资文案；不展示内部约束、实现细节或安全审查信息 | | 更换手机号 | 请求新手机号并覆盖缓存 | 用户明确要求时才更换 | | 查门店 | 调用门店查询能力 | 返回具体门店信息，不只概括 | | 查商品或做推荐 | 调用聚合上下文能力 | 推荐由 Agent 根据上下文自行完成 | | 泛饮品建议 | 仅在上下文已进入新一饮品决策时处理 | 不确定时不要强行触发 | 详见 [意图路由](references/intent-routing.md)。 ## 硬性约束 - 只有用户要参与活动或做个性化推荐时，才要求输入手机号。 - 用户询问本 Skill 的内容、功能或使用方式时，只回复用户可见的具体作用和使用方法；如果用户未参加活动或活动状态未确认，可以在末尾添加简短留资文案；不展示触发边界、硬性约束、OpenClaw 安全声明、脚本路径、环境变量、缓存结构等内部规则。 - 手机号状态按 `{mobile, activityJoined, updatedAt}` 保存；`activityJoined` 是 `true`、`false`、`null` 三态。 - 当前缓存手机号 `activityJoined=true` 时，不要重复输出主动引导留资文案。 - 用户询问有什么活动、优惠或福利时，必须把 Skill用户大礼包作为独立品牌活动列出；不能只列商品列表里的买赠、特价、畅饮卡等商品活动。 - 如果同一手机号之前提示未绑定/未参加，用户随后说“已登录小程序”“已绑定手机号”，再次领取时即使接口返回已领取，也要走“身份验证成功，三重福利发放到账”分支，不要说成用户原本就已领取过。 - 用户明确要求更换手机号、重新输入手机号，或提示当前手机号不是自己的手机号时，允许覆盖缓存并重新确认活动状态。 - 推荐、商品、门店和订单信息只能使用接口或脚本提供的数据，不要编造未返回内容。 - 推荐场景有门店数据时，最终回答至少给出 1-2 家具体门店；门店名、地址、电话和设施文案按返回内容完整表达。 - 活动领取成功或已领取后，如果脚本返回门店信息，最终回答必须逐家保留门店名、地址、电话、设施和排队信息；不要压缩成只有门店名和地址。 - 主动引导留资固定使用：您可以通过绑定【新一好喝】的注册手机号，领取Skill用户大礼包。 - 用户问“大礼包怎么领取”“怎么参与活动”“怎么领福利”时，直接说明流程：先绑定【新一好喝】注册手机号，再把手机号发来领取；不要在这类问题里直接输出“当前没有配置可领奖励”或解释接口状态。 - 老用户或登录成功时，输出“身份验证成功。三重福利发放到账”：`「小龙虾贴纸」一套（到店展示小程序卡券领取）`、接口返回的爆品赠饮一杯、微信小程序里 `「小龙虾身份标识」`。 - 登录成功后提示用户已经领取礼包，现在可以查看过去的订单信息；不要主动展开订单数量、完成单数或购买明细。 - 用户追问订单信息时，再按接口订单数据返回已完成 xx 单、近期购买商品和门店等信息，可以自然说“是新一的骨灰级粉丝吧”。 - 新用户或未注册时，提示用户还没登录过新一好喝，请到微信小程序搜索【新一咖啡】登录后获取全部福利和功能。 ## OpenClaw 安全声明 - 本 Skill 不是 instruction-only；它包含 Python 可执行脚本和 `install.sh`，用于查询门店、查询推荐上下文和提交活动领取请求。 - 默认后端为 `https://ai.xinyicoffee.com/api`，由 `config/defaults.json` 声明；可用 `XINYI_API_BASE_URL` 覆盖到可信后端。 - 网络请求只访问新一好喝业务接口：`/skill/xinyi/stores`、`/skill/xinyi/context`、`/skill/xinyi/claim`。 - 手机号只在参与活动、查询活动状态或个性化推荐时使用；`/skill/xinyi/claim` 会以 JSON 提交手机号，`/skill/xinyi/context` 会在有手机号时用 query 传递手机号。 - 本地只保存手机号活动状态，默认路径为 `~/.xinyi-drink/state.json`，内容为 `{mobile, activityJoined, updatedAt}`；可用 `XINYI_DRINK_STATE_FILE` 改到其他路径。 - 不读取 shell history、浏览器 Cookie、系统凭据、SSH 密钥或无关文件；不请求 API key、token 或密码。 ## 回答原则 - 推荐回答要像熟悉的店员在认真帮用户挑一杯，语气真诚、松弛、有温度。 - 推荐内容分成 3-4 个短块：主推饮品、适合原因、附近门店、活动留资提示。 - 主推饮品名和门店名必须加粗；备选饮品可适当加粗突出。 - 可以少量使用合适 emoji 做层次锚点或增加温度，但不要每行都加，也不要连续堆 emoji。 - 避免机械标题和报告腔，不要使用“根据你的历史订单偏好”“推荐理由”“历史偏好匹配”“天气适配”“推荐门店”。详见 [回答规范](references/response-guidelines.md)。 ## 当前限制 - 不支持实时库存 - 不支持最优用券测算 - 当前订单与历史订单只提供精简字段 ## 参考细则 - [意图路由](references/intent-routing.md) - [活动流程](references/activity-flow.md) - [回答规范](references/response-guidelines.md) - [能力地图](references/capability-map.md) - [注意事项](references/gotchas.md) - [隐私边界](references/privacy-boundaries.md) - [平台安装说明](references/platform-install.md) FILE:install.sh #!/bin/sh set -eu SKILL_NAME="xinyi-drink" SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" PLATFORM="" DRY_RUN=false while [ $# -gt 0 ]; do case "$1" in --platform) if [ $# -lt 2 ]; then echo "--platform 需要指定平台名称" >&2 exit 1 fi PLATFORM="$2" shift 2 ;; --dry-run) DRY_RUN=true shift ;; -h|--help) cat < ~/.claude/skills/ cursor -> .cursor/rules/ codex -> ~/.agents/skills/ openclaw -> ~/.openclaw/skills/ hermes -> ~/.hermes/skills/ qclaw -> ~/.agents/skills/ lobsterai -> ~/.agents/skills/ workbuddy -> ~/.agents/skills/ universal -> ~/.agents/skills/ 说明： - openclaw 安装到 ~/.openclaw/skills/ - hermes 安装到 ~/.hermes/skills/ - qclaw / lobsterai / workbuddy / codex / universal 安装到 ~/.agents/skills/ - claude-code 安装到 ~/.claude/skills/ - cursor 安装到 .cursor/rules/ EOF exit 0 ;; *) echo "未知参数: $1" >&2 exit 1 ;; esac done if [ -z "$PLATFORM" ]; then if [ -d "$HOME/.openclaw" ]; then PLATFORM="openclaw" elif [ -d "$HOME/.hermes" ]; then PLATFORM="hermes" elif [ -d "$HOME/.claude" ]; then PLATFORM="claude-code" elif [ -d "$HOME/.agents" ]; then PLATFORM="universal" else PLATFORM="universal" fi fi case "$PLATFORM" in claude-code) DEST="$HOME/.claude/skills/$SKILL_NAME" ;; openclaw) DEST="$HOME/.openclaw/skills/$SKILL_NAME" ;; hermes) DEST="$HOME/.hermes/skills/$SKILL_NAME" ;; universal|codex|qclaw|lobsterai|lobsterAI|workbuddy) DEST="$HOME/.agents/skills/$SKILL_NAME" ;; cursor) DEST=".cursor/rules/$SKILL_NAME" ;; *) echo "不支持的平台: $PLATFORM" >&2 exit 2 ;; esac case "$DEST" in ""|"/"|"$HOME"|"$HOME/") echo "安装目标路径异常，已拒绝覆盖: $DEST" >&2 exit 3 ;; esac if $DRY_RUN; then echo "[DRY-RUN] 平台: $PLATFORM" echo "[DRY-RUN] 将安装 $SKILL_NAME 到: $DEST" echo "[DRY-RUN] 不会发起网络请求，不会读取或写入手机号状态" exit 0 fi mkdir -p "$(dirname "$DEST")" if [ -e "$DEST" ]; then BACKUP="$DEST.backup.$(date +%Y%m%d%H%M%S)" if [ -e "$BACKUP" ]; then BACKUP_INDEX=1 while [ -e "$BACKUP.$BACKUP_INDEX" ]; do BACKUP_INDEX=$((BACKUP_INDEX + 1)) done BACKUP="$BACKUP.$BACKUP_INDEX" fi mv "$DEST" "$BACKUP" echo "已备份旧版本到 $BACKUP" fi cp -R "$SCRIPT_DIR" "$DEST" find "$DEST" -type d -name '__pycache__' -prune -exec rm -rf {} + find "$DEST" -name '.DS_Store' -delete cat < Path: configured = os.getenv("XINYI_DRINK_STATE_FILE") if configured: return Path(configured).expanduser() return Path.home() / ".xinyi-drink" / "state.json" def load_state() -> dict: state_file = resolve_state_file() if not state_file.exists(): return {} try: return json.loads(state_file.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} def write_state(payload: dict) -> None: state_file = resolve_state_file() state_file.parent.mkdir(parents=True, exist_ok=True) state_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") try: os.chmod(state_file, STATE_FILE_MODE) except OSError: pass def save_mobile(mobile: str) -> None: payload = { "mobile": mobile, "activityJoined": None, "updatedAt": datetime.now(timezone.utc).isoformat(), } write_state(payload) def save_activity_state(mobile: str, activity_joined: bool) -> None: payload = { "mobile": mobile, "activityJoined": activity_joined, "updatedAt": datetime.now(timezone.utc).isoformat(), } write_state(payload) def mark_activity_joined(mobile: str) -> None: save_activity_state(mobile, True) def mark_activity_not_joined(mobile: str) -> None: save_activity_state(mobile, False) def load_activity_joined(mobile: str | None) -> bool | None: if not mobile: return None payload = load_state() if payload.get("mobile") != mobile: return None value = payload.get("activityJoined") return value if isinstance(value, bool) else None def has_activity_joined(mobile: str | None) -> bool: return load_activity_joined(mobile) is True def clear_mobile() -> None: payload = { "mobile": None, "activityJoined": None, "updatedAt": datetime.now(timezone.utc).isoformat(), } write_state(payload) def load_mobile() -> str | None: value = load_state().get("mobile") return value if isinstance(value, str) and value.strip() else None FILE:scripts/skill_http.py #!/usr/bin/env python3 from __future__ import annotations import json import sys import urllib.error import urllib.parse import urllib.request from typing import Any, Callable class SkillHttpError(RuntimeError): """Readable HTTP/JSON error for command-line skill scripts.""" def make_debug_logger(component: str) -> Callable[[bool, str], None]: def debug_log(enabled: bool, message: str) -> None: if enabled: print(f"DEBUG {component}: {message}", file=sys.stderr) return debug_log def build_url(base_url: str, path: str, query: dict[str, Any] | None = None) -> str: url = f"{base_url.rstrip('/')}/{path.lstrip('/')}" if not query: return url filtered_query = { key: value for key, value in query.items() if value is not None and str(value).strip() } if not filtered_query: return url return f"{url}?{urllib.parse.urlencode(filtered_query)}" def fetch_json(url: str, timeout: int) -> dict: request = urllib.request.Request( url, headers={"Accept": "application/json"}, method="GET", ) return request_json(request, timeout) def post_json(url: str, timeout: int, payload: dict) -> dict: request = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), headers={"Accept": "application/json", "Content-Type": "application/json"}, method="POST", ) return request_json(request, timeout) def request_json(request: urllib.request.Request, timeout: int) -> dict: try: with urllib.request.urlopen(request, timeout=timeout) as response: payload = json.loads(response.read().decode("utf-8")) if not isinstance(payload, dict): raise SkillHttpError("接口返回格式不是 JSON 对象") return payload except urllib.error.HTTPError as exc: raise SkillHttpError(f"接口返回 HTTP {exc.code}") from exc except urllib.error.URLError as exc: raise SkillHttpError(f"网络请求失败：{exc.reason}") from exc except TimeoutError as exc: raise SkillHttpError("网络请求超时") from exc except OSError as exc: raise SkillHttpError(f"网络请求失败：{exc}") from exc except json.JSONDecodeError as exc: raise SkillHttpError("接口返回了无法解析的 JSON") from exc except Exception as exc: if isinstance(exc, SkillHttpError): raise raise SkillHttpError(f"请求处理失败：{exc}") from exc FILE:scripts/build_response.py #!/usr/bin/env python3 from __future__ import annotations from typing import Any from recommendation_logic import ( build_recommendation_fallback_copy, build_recommendation_material_lines, ) from response_rendering import ( escape_table_cell, render_markdown_table, render_primary_response, render_text_section, ) WEATHER_LABELS = { "sunny": "晴", "cloudy": "多云", "rainy": "雨", "snowy": "雪", "unknown": "未知", } ORDER_STATE_LABELS = { 0: "未知", 1: "待支付", 2: "制作中", 3: "待取餐", 4: "已取消", 5: "退款中", 6: "已完成", } LEAD_CAPTURE_COPY = "您可以通过绑定【新一好喝】的注册手机号，领取Skill用户大礼包。" UNREGISTERED_LOGIN_COPY = "目前还没登录过新一好喝，请到微信小程序搜索【新一咖啡】登录后获取全部福利和功能。" ACTIVITY_GIFT_SUMMARY = "三重福利包含：「小龙虾贴纸」一套、根据接口返回的爆品赠饮一杯、微信小程序里「小龙虾身份标识」。" ACTIVITY_QUERY_KEYWORDS = ("活动", "福利", "优惠", "券", "见面礼", "龙虾", "领取") ORDER_QUERY_KEYWORDS = ("订单", "过去", "历史", "完成", "买过", "购买", "消费", "几单", "多少单", "下过单") def build_activity_flow_lines() -> list[str]: return [ LEAD_CAPTURE_COPY, "领取流程：先绑定【新一好喝】注册手机号，再把手机号发来，我会帮您领取。", "如果还没登录过新一好喝，请到微信小程序搜索【新一咖啡】登录后获取全部福利和功能。", ] def pick_store_contact(store: dict[str, Any]) -> Any: for key in ("storeMobile", "contactPhone", "phone", "telephone", "tel"): value = store.get(key) if value: return value return None def split_store_facilities(store: dict[str, Any]) -> list[str]: facilities = store.get("facilities") if not facilities: return [] text = str(facilities).replace("，", ",").replace("。", ",") return [item.strip() for item in text.split(",") if item.strip()] def render_store_facilities_text(store: dict[str, Any]) -> Any: return store.get("facilities") or None def format_coupon_label(coupon_names: list[str]) -> str: if not coupon_names: return "" if len(coupon_names) == 1: return f"「{coupon_names[0]}」" return "、".join(f"「{name}」" for name in coupon_names) def format_coupon_reward(coupon_names: list[str]) -> str: if not coupon_names: return "爆品赠饮一杯（具体饮品以小程序卡券为准）" if len(coupon_names) == 1: return f"{format_coupon_label(coupon_names)}一杯" return f"{format_coupon_label(coupon_names)}各一杯" def render_context_section(context: dict[str, Any]) -> str: mobile_source = "-" mobile = context.get("mobile") if mobile: mobile_source = "本地缓存" if context.get("mobileFromStore") else "本次输入" activity_joined = context.get("activityJoined") if activity_joined is True: activity_joined_label = "是" elif activity_joined is False: activity_joined_label = "否" else: activity_joined_label = "未确认" rows = [ ["手机号", mobile], ["手机号来源", mobile_source], ["是否已参加活动", activity_joined_label], ["用户问题", context.get("query")], ["推荐场景", context.get("scene")], ["用户偏好", context.get("preference")], ] return render_markdown_table( title="用户上下文", headers=["字段", "内容"], rows=rows, empty_text="暂无上下文信息。", ) def is_activity_query(context: dict[str, Any]) -> bool: text = " ".join( str(context.get(key) or "") for key in ("query", "scene", "preference") ) return any(keyword in text for keyword in ACTIVITY_QUERY_KEYWORDS) def is_order_query(context: dict[str, Any]) -> bool: text = " ".join( str(context.get(key) or "") for key in ("query", "scene", "preference") ) return any(keyword in text for keyword in ORDER_QUERY_KEYWORDS) def render_brand_activity_section(context: dict[str, Any]) -> str: activity_joined = context.get("activityJoined") if activity_joined is True: lines = [ f"**Skill用户大礼包**：用户身份已验证成功，{ACTIVITY_GIFT_SUMMARY}", "回答时可以说明福利已经到账，不要再要求用户重新登录小程序或再次告知手机号。", ] else: lines = [ f"**Skill用户大礼包**：{LEAD_CAPTURE_COPY}", "用户可以提供【新一好喝】注册手机号领取；如果仍未注册，提醒去微信小程序搜索【新一咖啡】登录后获取全部福利和功能。", ] lines.append("这是品牌活动，必须和商品列表里的买一赠一、特价、畅饮卡等商品活动区分开。") return render_text_section("品牌活动", lines) def render_weather_section(weather: dict[str, Any] | None) -> str: rows: list[list[Any]] = [] if weather: rows.append( [ weather.get("city"), WEATHER_LABELS.get(weather.get("condition"), weather.get("condition")), weather.get("temperatureC"), ] ) return render_markdown_table( title="天气", headers=["城市", "天气", "温度(°C)"], rows=rows, empty_text="暂无天气数据。", ) def format_order_goods(goods: list[dict[str, Any]]) -> str: entries: list[str] = [] for item in goods: name = item.get("name") or "-" detail_parts = [part for part in [item.get("spec"), item.get("attr")] if part] if detail_parts: entries.append(f"{name}（{' / '.join(detail_parts)}）") else: entries.append(name) return "；".join(entries) if entries else "-" def format_order_state(state: Any) -> str: if isinstance(state, int): return ORDER_STATE_LABELS.get(state, f"状态{state}") return escape_table_cell(state) def build_store_pickup_lines(stores: list[dict[str, Any]]) -> list[str]: if not stores: return [] highlighted_stores = stores[:2] lines = ["贴纸领取门店信息我给你列全（地址、电话、设施、排队）："] for store in highlighted_stores: wait_parts: list[str] = [] if store.get("makingCupCount") is not None: wait_parts.append(f"制作中{store.get('makingCupCount')}杯") if store.get("makingCupMinutes") is not None: wait_parts.append(f"预计{store.get('makingCupMinutes')}分钟") lines.append( "；".join( [ f"**{store.get('name') or '-'}**：地址：{store.get('address') or '-'}", f"电话：{pick_store_contact(store) or '未提供联系电话'}", f"设施：{render_store_facilities_text(store) or '未提供设施文案'}", f"排队：{'，'.join(wait_parts) if wait_parts else '暂无排队信息'}", ] ) ) return lines def build_activity_completion_lines(kind: str, coupon_names: list[str]) -> list[str]: if kind in {"granted", "already_claimed", "obtained_after_registration"}: return [ "身份验证成功。三重福利发放到账：", f"「小龙虾贴纸」一套（到店展示小程序卡券领取）；{format_coupon_reward(coupon_names)}；微信小程序里「小龙虾身份标识」。", "你已经领取礼包，现在可以查看你过去的订单信息。", ] return [] def render_orders_section(orders: dict[str, Any] | None) -> str: if orders is None: return "## 订单历史\n未提供手机号，未查询订单历史。" rows = [ [ item.get("createdAt"), item.get("orderSn"), format_order_state(item.get("state")), item.get("pickNo"), item.get("serverTime"), item.get("store", {}).get("name") if item.get("store") else None, format_order_goods(item.get("goods", [])), item.get("goodsNum"), ] for item in orders.get("orders", []) ] return render_markdown_table( title="订单历史", headers=[ "下单时间", "订单号", "状态", "排号", "预计取餐时间", "门店", "商品", "杯数", ], rows=rows, empty_text="暂无订单记录。", ) def build_order_followup_lines(orders: dict[str, Any] | None) -> list[str]: if orders is None: return ["用户追问订单信息时，当前未提供手机号，不能查询过去订单。"] order_list = orders.get("orders", []) completed_orders = [order for order in order_list if order.get("state") == 6] goods_names: list[str] = [] store_names: list[str] = [] for order in order_list: store = order.get("store") or {} if store.get("name"): store_names.append(str(store.get("name"))) for good in order.get("goods", []): if good.get("name"): goods_names.append(str(good.get("name"))) lines = [ f"用户追问订单信息时再展开：你已完成{len(completed_orders)}单，是新一的骨灰级粉丝吧。", f"当前可见订单数：{len(order_list)}单。", ] if goods_names: lines.append(f"买过的商品可以提这些：{'、'.join(goods_names[:5])}。") if store_names: unique_store_names = list(dict.fromkeys(store_names)) lines.append(f"到过的门店可以提这些：{'、'.join(unique_store_names[:3])}。") return lines def render_order_followup_section( context: dict[str, Any], orders: dict[str, Any] | None, ) -> str: if not is_order_query(context): return "" return render_text_section("订单追问素材", build_order_followup_lines(orders)) def render_goods_section(goods: list[dict[str, Any]]) -> str: rows = [ [ item.get("name"), item.get("categories", []), item.get("price"), item.get("cupSizes", []), item.get("temperatures", []), item.get("sugarLevels", []), item.get("calories"), item.get("ingredients", []), ] for item in goods ] return render_markdown_table( title="商品列表", headers=["商品名称", "分类", "价格", "杯型", "温度", "糖度", "卡路里", "配料"], rows=rows, empty_text="暂无商品数据。", ) def format_store_status(store: dict[str, Any]) -> str: parts: list[str] = [] if store.get("businessStatus") is not None: parts.append(f"营业状态={store.get('businessStatus')}") if store.get("operatingStatus") is not None: parts.append(f"运营状态={store.get('operatingStatus')}") if store.get("realtimeState") is not None: parts.append(f"实时状态={store.get('realtimeState')}") return " / ".join(parts) if parts else "-" def build_store_feature_tags(store: dict[str, Any]) -> list[str]: feature_tags: list[str] = [] def append_tag(tag: Any) -> None: if not tag: return text = str(tag).strip() if text and text not in feature_tags: feature_tags.append(text) for facility in split_store_facilities(store): append_tag(facility) for label in store.get("labels", []): if isinstance(label, dict): append_tag(label.get("name")) if store.get("supportUnattendedMode") == 1: append_tag("支持无人模式") if store.get("storeType") == 2: append_tag("Box 门店") return feature_tags def build_store_summary_lines(stores: list[dict[str, Any]]) -> list[str]: lines: list[str] = [] for store in stores[:2]: feature_tags = build_store_feature_tags(store) feature_text = "、".join(feature_tags) if feature_tags else "暂无特色信息" wait_parts: list[str] = [] if store.get("makingCupCount") is not None: wait_parts.append(f"制作中{store.get('makingCupCount')}杯") if store.get("makingCupMinutes") is not None: wait_parts.append(f"预计{store.get('makingCupMinutes')}分钟") wait_text = "，".join(wait_parts) if wait_parts else "暂无排队信息" contact = pick_store_contact(store) or "未提供联系电话" lines.append( "；".join( [ f"门店名：{store.get('name') or '-'}", f"地址：{store.get('address') or '-'}", f"电话：{contact}", f"设施：{render_store_facilities_text(store) or '未提供设施文案'}", f"特色：{feature_text}", f"排队：{wait_text}", ] ) ) return lines def render_stores_section(stores: list[dict[str, Any]]) -> str: rows = [ [ item.get("name"), item.get("address"), pick_store_contact(item), render_store_facilities_text(item), format_store_status(item), build_store_feature_tags(item), item.get("makingCupCount"), item.get("makingCupMinutes"), ] for item in stores ] return render_markdown_table( title="门店列表", headers=["门店名称", "地址", "联系电话", "设施文案", "状态", "特色标签", "制作中杯数", "制作时长(分钟)"], rows=rows, empty_text="暂无门店数据。", ) def render_answer_requirements_section( stores: list[dict[str, Any]], context: dict[str, Any], ) -> str: lines = [ "先给出一段有温度的主推荐文案，像熟悉的店员在认真帮用户挑一杯，而不是像报告摘要。", "回答需要有层次和重点：推荐饮品、适合它的几个原因、附近可去的门店、活动留资提示分成 3-4 个短块，不要堆成一整段。", "主推饮品名和门店名必须加粗；如果提到备选饮品，也可以加粗突出。", "可以少量使用合适 emoji 做层次锚点或增强温度，比如饮品、天气、门店、活动各 0-1 个；不要每行都加，也不要连续堆 emoji。", "把推荐依据自然融进表达里，可以用轻量分点，但不要使用“根据你的历史订单偏好”“推荐理由”“历史偏好匹配”“天气适配”“推荐门店”这类机械标题。", "语气要真诚、松弛、有人情味；少用营销腔、感叹号和模板化开场，不要复述固定模板。", "登录成功后只提示用户已经领取礼包、现在可以查看过去的订单信息；不要主动展开已完成多少单或购买明细。", "只有用户追问订单、完成多少单、买过什么时，才根据订单数据返回已完成订单数和购买信息；可以自然说“你已完成xx单，是新一的骨灰级粉丝吧”。", ] if context.get("activityJoined"): lines.append("用户已参加过活动，不要再输出主动留资文案；可说明身份验证成功，三重福利已经到账。") else: lines.append( f"用户未参加过活动或当前手机号状态未确认，回答末尾可以用分割线 `---` 单独隔开主动留资文案：{LEAD_CAPTURE_COPY} 用户提交手机号后如果仍未注册，再提示：{UNREGISTERED_LOGIN_COPY}" ) if is_activity_query(context): lines.append( "用户正在问活动/福利/优惠，最终回答必须把“**Skill用户大礼包**”作为独立品牌活动，和商品列表里的买一赠一、特价、畅饮卡等商品活动并列展示，不能只列商品活动。" ) if stores: lines.extend( [ "若返回了门店数据，最终回答里至少给出 1-2 家具体门店，带上门店名和详细地址。", "若有门店电话也一并给出；如果没有电话字段，再明确说明未提供联系电话。", "若门店返回了 facilities，必须明确返回这段设施文案，不要省略。", "门店部分用“如果你在附近，可以去……”这类自然说法承接；门店名加粗，地址、电话、设施和排队信息用短行呈现。", ] ) else: lines.append("如果没有门店数据，明确说明当前未拿到可用门店信息。") return render_text_section("回答要求", lines) def render_store_summary_section(stores: list[dict[str, Any]]) -> str: return render_text_section( "门店摘要建议", build_store_summary_lines(stores) if stores else ["当前未拿到可直接复述的门店摘要。"], ) def render_recommendation_context( *, context: dict[str, Any], goods: list[dict[str, Any]], stores: list[dict[str, Any]], weather: dict[str, Any] | None, orders: dict[str, Any] | None, ) -> str: sections = [ render_context_section(context), render_brand_activity_section(context), render_weather_section(weather), render_orders_section(orders), render_order_followup_section(context, orders), render_goods_section(goods), render_stores_section(stores), render_store_summary_section(stores), render_answer_requirements_section(stores, context), ] recommendation_material_lines = build_recommendation_material_lines(goods, orders, weather) if recommendation_material_lines: sections.append( render_text_section( "推荐素材", recommendation_material_lines, ) ) return "\n\n".join(section for section in sections if section.strip()) def render_claim_result( data: dict[str, Any], context_data: dict[str, Any] | None = None, ) -> str: kind = data.get("kind") items = data.get("items", []) user = data.get("user") if kind == "unregistered": return render_primary_response( "领取结果：请先登录小程序", [ UNREGISTERED_LOGIN_COPY, ], ) coupon_names = [ item.get("coupon", {}).get("name") for item in items if item.get("state") == 1 and item.get("coupon") ] coupon_names = [name for name in coupon_names if name] context_data = context_data or {} goods = context_data.get("goods", []) stores = context_data.get("stores", []) weather = context_data.get("weather") orders = context_data.get("orders") recommendation_copy = build_recommendation_fallback_copy(goods, orders, weather) title = "领取结果：处理完成" lines: list[str] = [] if kind == "granted": title = "领取结果：身份验证成功" lines.extend(build_activity_completion_lines(kind, coupon_names)) elif kind == "obtained_after_registration": title = "领取结果：身份验证成功" lines.extend(build_activity_completion_lines(kind, coupon_names)) elif kind == "already_claimed": title = "领取结果：身份验证成功" lines.extend(build_activity_completion_lines(kind, coupon_names)) elif kind == "no_reward_config": title = "领取方式" lines.extend(build_activity_flow_lines()) else: lines.append("活动处理已完成。") if user and user.get("nickname"): lines.append(f"当前识别用户：{user.get('nickname')}") lines.extend(build_store_pickup_lines(stores)) if recommendation_copy: lines.append(recommendation_copy) failed_messages = [item.get("message") for item in items if item.get("state") != 1 and item.get("message")] if failed_messages and kind == "no_reward_config": lines.append(f"失败原因：{'；'.join(str(message) for message in failed_messages)}") return render_primary_response(title, lines) FILE:scripts/skill_config.py #!/usr/bin/env python3 from __future__ import annotations import json import os from pathlib import Path from typing import Any def load_config() -> dict[str, Any]: config_path = Path(__file__).resolve().parents[1] / "config" / "defaults.json" config = json.loads(config_path.read_text(encoding="utf-8")) api_base_url = os.getenv("XINYI_API_BASE_URL") if api_base_url: config["apiBaseUrl"] = api_base_url timeout_seconds = os.getenv("XINYI_TIMEOUT_SECONDS") if timeout_seconds: try: config["timeoutSeconds"] = int(timeout_seconds) except ValueError: pass return config FILE:scripts/fetch_stores.py #!/usr/bin/env python3 from __future__ import annotations import sys from build_response import render_stores_section from skill_config import load_config from skill_http import SkillHttpError, build_url, fetch_json, make_debug_logger debug_log = make_debug_logger("fetch_stores") def main() -> int: debug_enabled = "--debug" in sys.argv[1:] config = load_config() url = build_url(config["apiBaseUrl"], "/skill/xinyi/stores") debug_log(debug_enabled, f"fetching stores from {url}") try: payload = fetch_json(url, config["timeoutSeconds"]) except SkillHttpError as exc: sys.stdout.write(f"门店查询失败：{exc}") return 1 sys.stdout.write(render_stores_section(payload.get("data", {}).get("stores", []))) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/claim_reward.py #!/usr/bin/env python3 from __future__ import annotations import argparse import sys from build_response import render_claim_result from skill_config import load_config from skill_http import SkillHttpError, build_url, fetch_json, make_debug_logger, post_json from user_state import load_activity_joined, mark_activity_joined, mark_activity_not_joined, save_mobile debug_log = make_debug_logger("claim_reward") def main() -> int: parser = argparse.ArgumentParser(description="按手机号参与新一好喝活动并领取奖励") parser.add_argument("--mobile", required=True, help="用户手机号") parser.add_argument("--debug", action="store_true", help="输出调试信息到 stderr") args = parser.parse_args() config = load_config() base_url = config["apiBaseUrl"].rstrip("/") url = build_url(base_url, "/skill/xinyi/claim") debug_log(args.debug, f"posting claim request to {url}") previous_activity_joined = load_activity_joined(args.mobile) save_mobile(args.mobile) try: parsed = post_json(url, config["timeoutSeconds"], {"mobile": args.mobile}) except SkillHttpError as exc: sys.stdout.write(f"领取活动失败：{exc}") return 1 if isinstance(parsed.get("data"), dict): parsed["data"].setdefault("requestedMobile", args.mobile) if parsed["data"].get("kind") == "already_claimed" and previous_activity_joined is False: parsed["data"]["kind"] = "obtained_after_registration" context_data = None if parsed.get("code") == 200: claim_data = parsed.get("data", {}) if claim_data.get("user"): debug_log(args.debug, "user matched; saving mobile") if claim_data.get("kind") in {"granted", "already_claimed", "obtained_after_registration"}: mark_activity_joined(args.mobile) else: mark_activity_not_joined(args.mobile) context_url = build_url( base_url, "/skill/xinyi/context", {"mobile": args.mobile}, ) try: debug_log(args.debug, f"fetching context from {context_url}") context_response = fetch_json( context_url, config["timeoutSeconds"], ) context_data = context_response.get("data", {}) debug_log( args.debug, "context includes weather data" if context_data.get("weather") is not None else "context missing weather; using generic recommendation copy", ) except SkillHttpError: debug_log(args.debug, "context enrichment failed; keep primary success message only") context_data = None else: mark_activity_not_joined(args.mobile) debug_log(args.debug, "user not found; marking activity as not joined") sys.stdout.write(render_claim_result(parsed.get("data", {}), context_data)) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/response_rendering.py #!/usr/bin/env python3 from __future__ import annotations from typing import Any def render_primary_response(title: str, lines: list[str]) -> str: parts = [title.strip()] parts.extend(line.strip() for line in lines if line and line.strip()) return "\n".join(parts) def escape_table_cell(value: Any) -> str: if value is None: return "-" if isinstance(value, bool): text = "是" if value else "否" elif isinstance(value, (list, tuple, set)): items = [escape_table_cell(item) for item in value] text = "、".join(item for item in items if item != "-") else: text = str(value).strip() if not text: return "-" return text.replace("|", "\\|").replace("\n", "
") def render_markdown_table( title: str, headers: list[str], rows: list[list[Any]], empty_text: str, ) -> str: parts = [f"## {title}"] if not rows: parts.append(empty_text) return "\n".join(parts) parts.append(f"| {' | '.join(headers)} |") parts.append(f"| {' | '.join('---' for _ in headers)} |") for row in rows: parts.append(f"| {' | '.join(escape_table_cell(cell) for cell in row)} |") return "\n".join(parts) def render_text_section(title: str, lines: list[str]) -> str: return render_primary_response( f"## {title}", lines, ) FILE:scripts/recommendation_logic.py #!/usr/bin/env python3 from __future__ import annotations from typing import Any def describe_weather_feel(weather: dict[str, Any] | None) -> str | None: temperature = weather.get("temperatureC") if weather else None if not isinstance(temperature, (int, float)): return None if temperature <= 10: return "有点冷" if temperature <= 18: return "偏凉" if temperature <= 26: return "挺舒服" return "有点热" def choose_recommendation_good( goods: list[dict[str, Any]], orders: dict[str, Any] | None, weather: dict[str, Any] | None, ) -> dict[str, Any] | None: if not goods: return None order_goods_names = [ order_good.get("name") for order in (orders or {}).get("orders", []) for order_good in order.get("goods", []) if order_good.get("name") ] for ordered_name in order_goods_names: for good in goods: good_name = good.get("name") if not good_name: continue if good_name == ordered_name or good_name in ordered_name or ordered_name in good_name: return good feel = describe_weather_feel(weather) if feel in {"有点冷", "偏凉"}: for good in goods: temperatures = good.get("temperatures") or [] if any("热" in str(item) for item in temperatures): return good if feel == "有点热": for good in goods: temperatures = good.get("temperatures") or [] if any("冰" in str(item) for item in temperatures): return good return goods[0] def build_recommendation_reason_signals( recommended_good: dict[str, Any], orders: dict[str, Any] | None, weather: dict[str, Any] | None, ) -> list[str]: signals: list[str] = [] recommended_name = recommended_good.get("name") order_goods_names = [ order_good.get("name") for order in (orders or {}).get("orders", []) for order_good in order.get("goods", []) if order_good.get("name") ] if recommended_name and order_goods_names: matched_history = [ name for name in order_goods_names if name == recommended_name or recommended_name in name or name in recommended_name ] if matched_history: signals.append(f"历史订单里出现过：{'、'.join(matched_history[:3])}") else: signals.append(f"用户有历史订单，可参考近期喝过：{'、'.join(order_goods_names[:3])}") weather_feel = describe_weather_feel(weather) if weather_feel: signals.append(f"当前天气体感：{weather_feel}") temperatures = recommended_good.get("temperatures") or [] if temperatures: signals.append(f"可选温度：{'、'.join(str(item) for item in temperatures)}") sugar_levels = recommended_good.get("sugarLevels") or [] if sugar_levels: signals.append(f"可选糖度：{'、'.join(str(item) for item in sugar_levels)}") calories = recommended_good.get("calories") if calories: signals.append(f"热量信息：{calories}") ingredients = recommended_good.get("ingredients") or [] if ingredients: signals.append(f"主要配料：{'、'.join(str(item) for item in ingredients)}") categories = recommended_good.get("categories") or [] if categories: signals.append(f"商品分类：{'、'.join(str(item) for item in categories)}") return signals def build_recommendation_fallback_copy( goods: list[dict[str, Any]], orders: dict[str, Any] | None, weather: dict[str, Any] | None, ) -> str | None: recommended_good = choose_recommendation_good(goods, orders, weather) if not recommended_good: return None recommended_name = recommended_good.get("name") if not recommended_name: return None weather_feel = describe_weather_feel(weather) has_history = bool((orders or {}).get("orders")) if weather_feel: if has_history: return f"哇我们的老朋友，今天天气{weather_feel}，建议您喝{recommended_name}。" return f"今天天气{weather_feel}，建议您喝{recommended_name}。" if has_history: return f"哇我们的老朋友，今天建议您喝{recommended_name}。" return f"今天建议您喝{recommended_name}。" def build_recommendation_material_lines( goods: list[dict[str, Any]], orders: dict[str, Any] | None, weather: dict[str, Any] | None, ) -> list[str]: recommended_good = choose_recommendation_good(goods, orders, weather) if not recommended_good: return [] recommended_name = recommended_good.get("name") if not recommended_name: return [] lines = [ f"推荐候选饮品：{recommended_name}", f"推荐素材字段：candidateName={recommended_name}", "主推荐文案由大模型根据下方素材自行生成，像熟悉的店员给朋友建议一样自然、温和、有温度，不要照搬固定模板。", "回答要有清晰层次：主推饮品、适合原因、附近门店、活动留资提示分成短块；主推饮品名和门店名要加粗。", "可以少量使用合适 emoji 做层次锚点或增加温度，但不要每行都加，也不要连续堆 emoji。", "推荐依据要融进自然短句里，可以写 2-4 个轻量要点，但不要使用“推荐理由”“历史偏好匹配”“天气适配”这类生硬标题。", "活动留资提示以“回答要求”区块为准；已参加活动不要输出主动留资文案。", "订单统计和购买明细只在用户追问订单信息时展开，不要在礼包领取成功后主动列出。", "只使用已提供的历史订单、天气、商品属性和门店信息，不要编造未返回的数据。", ] reason_signals = build_recommendation_reason_signals( recommended_good, orders, weather, ) if reason_signals: lines.append(f"推荐素材字段：reasonSignals={' | '.join(reason_signals)}") lines.append(f"可用推荐依据：{'；'.join(reason_signals)}。") return lines FILE:scripts/recommend_drink.py #!/usr/bin/env python3 from __future__ import annotations import argparse import sys from build_response import render_recommendation_context from skill_config import load_config from skill_http import SkillHttpError, build_url, fetch_json, make_debug_logger, post_json from user_state import ( clear_mobile, load_activity_joined, load_mobile, mark_activity_joined, mark_activity_not_joined, save_mobile, ) debug_log = make_debug_logger("recommend_drink") def main() -> int: parser = argparse.ArgumentParser( description="获取聚合推荐上下文，并整理成可直接提供给大模型的文本/表格" ) parser.add_argument("--mobile", help="可选手机号，用于个性化推荐") parser.add_argument( "--clear-mobile", action="store_true", help="清空已保存的手机号，下次不再自动带出", ) parser.add_argument("--query", help="用户问题或饮品名称") parser.add_argument("--scene", help="场景，如提神、下午茶、轻负担") parser.add_argument("--preference", help="偏好，如咖啡、茶、低卡") parser.add_argument("--debug", action="store_true", help="输出调试信息到 stderr") args = parser.parse_args() if args.clear_mobile: clear_mobile() debug_log(args.debug, "cleared saved mobile") resolved_mobile = args.mobile or load_mobile() if args.mobile: debug_log(args.debug, "resolved mobile from cli argument") elif resolved_mobile: debug_log(args.debug, "resolved mobile from local state") else: debug_log(args.debug, "no mobile resolved") if args.mobile: save_mobile(args.mobile) debug_log(args.debug, "saved mobile from cli argument before claim lookup") config = load_config() base_url = config["apiBaseUrl"].rstrip("/") timeout = config["timeoutSeconds"] mobile_for_context = resolved_mobile if resolved_mobile: claim_url = build_url(base_url, "/skill/xinyi/claim") debug_log(args.debug, f"posting claim request to {claim_url}") try: claim_response = post_json( claim_url, timeout, {"mobile": resolved_mobile}, ) except Exception as exc: debug_log( args.debug, f"claim lookup failed; continue with context lookup and cached activity state: {exc}", ) else: claim_data = claim_response.get("data", {}) if claim_data.get("user"): debug_log(args.debug, "claim matched user") if claim_data.get("kind") in {"granted", "already_claimed"}: mark_activity_joined(resolved_mobile) else: mark_activity_not_joined(resolved_mobile) else: mark_activity_not_joined(resolved_mobile) debug_log( args.debug, "claim did not match user; keep saved mobile and continue with context lookup", ) context_url = build_url( base_url, "/skill/xinyi/context", {"mobile": mobile_for_context}, ) debug_log(args.debug, f"fetching context from {context_url}") try: context_response = fetch_json(context_url, timeout) except SkillHttpError as exc: sys.stdout.write(f"获取推荐上下文失败：{exc}") return 1 context_data = context_response.get("data", {}) weather_data = context_data.get("weather") debug_log( args.debug, "context includes weather data" if weather_data is not None else "context missing weather; using generic recommendation copy", ) rendered_context = render_recommendation_context( context={ "mobile": mobile_for_context, "mobileFromStore": bool(mobile_for_context and not args.mobile), "preference": args.preference, "query": args.query, "scene": args.scene, "activityJoined": load_activity_joined(mobile_for_context), }, goods=context_data.get("goods", []), stores=context_data.get("stores", []), weather=weather_data, orders=context_data.get("orders"), ) sys.stdout.write(rendered_context) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:spec/fetch_stores_test.py from __future__ import annotations import io import sys import unittest from pathlib import Path from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import fetch_stores class FetchStoresScriptTest(unittest.TestCase): @patch.object( fetch_stores, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_fetch_stores_outputs_table_text( self, urlopen_mock, _load_config_mock, ) -> None: response = urlopen_mock.return_value.__enter__.return_value response.read.return_value = ( '{"data":{"stores":[{"name":"幂茶幂咖望京店","address":"北京市朝阳区望京街9号",' '"facilities":"外摆区，休息区，宠物友好。",' '"storeMobile":"010-12345678",' '"businessStatus":1,"operatingStatus":1,"realtimeState":1,' '"labels":[{"name":"休息区"}],"makingCupCount":4,"makingCupMinutes":18,' '"storeType":2,"supportUnattendedMode":1}]}}' ).encode("utf-8") stdout = io.StringIO() with patch("sys.stdout", stdout): exit_code = fetch_stores.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("## 门店列表", output) self.assertIn("幂茶幂咖望京店", output) self.assertIn("010-12345678", output) self.assertIn("宠物友好", output) self.assertIn("外摆区，休息区，宠物友好。", output) self.assertIn("休息区", output) self.assertIn("Box 门店", output) self.assertIn("支持无人模式", output) self.assertNotIn('"stores"', output) @patch.object( fetch_stores, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_fetch_stores_outputs_debug_logs_to_stderr( self, urlopen_mock, _load_config_mock, ) -> None: response = urlopen_mock.return_value.__enter__.return_value response.read.return_value = '{"data":{"stores":[]}}'.encode("utf-8") stdout = io.StringIO() stderr = io.StringIO() with patch.object( sys, "argv", ["fetch_stores.py", "--debug"], ), patch("sys.stdout", stdout), patch("sys.stderr", stderr): exit_code = fetch_stores.main() self.assertEqual(exit_code, 0) self.assertIn("DEBUG fetch_stores: fetching stores from http://127.0.0.1:8020/skill/xinyi/stores", stderr.getvalue()) self.assertIn("## 门店列表", stdout.getvalue()) @patch.object( fetch_stores, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen", side_effect=OSError("connection refused")) def test_fetch_stores_outputs_readable_error_when_request_fails( self, _urlopen_mock, _load_config_mock, ) -> None: stdout = io.StringIO() with patch("sys.stdout", stdout): exit_code = fetch_stores.main() self.assertEqual(exit_code, 1) self.assertIn("门店查询失败：网络请求失败", stdout.getvalue()) if __name__ == "__main__": unittest.main() FILE:spec/skill_config_test.py from __future__ import annotations import os import sys import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import skill_config class SkillConfigTest(unittest.TestCase): def test_load_config_allows_environment_overrides(self) -> None: previous_base_url = os.environ.get("XINYI_API_BASE_URL") previous_timeout = os.environ.get("XINYI_TIMEOUT_SECONDS") os.environ["XINYI_API_BASE_URL"] = "http://127.0.0.1:8020" os.environ["XINYI_TIMEOUT_SECONDS"] = "3" try: config = skill_config.load_config() finally: if previous_base_url is None: os.environ.pop("XINYI_API_BASE_URL", None) else: os.environ["XINYI_API_BASE_URL"] = previous_base_url if previous_timeout is None: os.environ.pop("XINYI_TIMEOUT_SECONDS", None) else: os.environ["XINYI_TIMEOUT_SECONDS"] = previous_timeout self.assertEqual(config["apiBaseUrl"], "http://127.0.0.1:8020") self.assertEqual(config["timeoutSeconds"], 3) if __name__ == "__main__": unittest.main() FILE:spec/recommend_drink_test.py from __future__ import annotations import io import sys import unittest from pathlib import Path from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import recommend_drink from skill_http import SkillHttpError class RecommendDrinkScriptTest(unittest.TestCase): def setUp(self) -> None: self.mark_activity_joined_patcher = patch.object(recommend_drink, "mark_activity_joined") self.mark_activity_not_joined_patcher = patch.object(recommend_drink, "mark_activity_not_joined") self.load_activity_joined_patcher = patch.object( recommend_drink, "load_activity_joined", return_value=None, ) self.mark_activity_joined_mock = self.mark_activity_joined_patcher.start() self.mark_activity_not_joined_mock = self.mark_activity_not_joined_patcher.start() self.load_activity_joined_mock = self.load_activity_joined_patcher.start() self.addCleanup(self.mark_activity_joined_patcher.stop) self.addCleanup(self.mark_activity_not_joined_patcher.stop) self.addCleanup(self.load_activity_joined_patcher.stop) @patch.object(recommend_drink, "post_json") @patch.object(recommend_drink, "save_mobile") @patch.object(recommend_drink, "load_mobile", return_value=None) @patch.object( recommend_drink, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch.object(recommend_drink, "fetch_json") def test_main_outputs_llm_friendly_text_instead_of_json( self, fetch_json_mock, _load_config_mock, _load_mobile_mock, save_mobile_mock, post_json_mock, ) -> None: events: list[tuple[str, str]] = [] save_mobile_mock.side_effect = lambda mobile: events.append(("save", mobile)) post_json_mock.side_effect = lambda url, timeout, payload: ( events.append(("claim", payload["mobile"])), { "data": { "kind": "already_claimed", "user": {"mobile": "15712459595", "nickname": "双龙"}, } }, )[1] self.load_activity_joined_mock.return_value = True fetch_json_mock.return_value = { "data": { "goods": [ { "name": "杨枝甘露|轻乳版", "categories": ["果咖", "轻负担"], "price": "16.80", "cupSizes": ["大杯"], "temperatures": ["少冰", "去冰"], "sugarLevels": ["3分糖", "5分糖"], "calories": "120\nkcal", "ingredients": ["芒果", "西柚"], } ], "stores": [ { "name": "幂茶幂咖|望京店", "address": "北京市朝阳区望京街9号\n商业楼1层", "facilities": "外摆区，休息区，宠物友好。", "storeMobile": "010-12345678", "businessStatus": 1, "labels": [{"name": "休息区"}], "lat": "39.990326", "lng": "116.483659", "operatingStatus": 1, "realtimeState": 1, "makingCupCount": 4, "makingCupMinutes": 18, "storeType": 2, "supportUnattendedMode": 1, } ], "weather": { "city": "Beijing", "condition": "sunny", "temperatureC": 26, }, "orders": { "orders": [ { "createdAt": "2025-08-08 14:16:25", "orderSn": "20250808141625274275", "state": 2, "pickNo": "Z108", "serverTime": "2025-08-08 14:36:25", "goodsNum": 1, "goods": [ { "name": "葡萄毛尖轻咖", "spec": "中杯", "attr": "正常冰|7分糖", } ], "store": { "name": "幂茶幂咖望京小街店", }, } ] }, } } stdout = io.StringIO() with patch.object( sys, "argv", [ "recommend_drink.py", "--mobile", "15712459595", "--query", "想喝不苦的", "--scene", "下午茶", "--preference", "低卡", ], ), patch("sys.stdout", stdout): exit_code = recommend_drink.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("## 用户上下文", output) self.assertIn("## 商品列表", output) self.assertIn("## 门店列表", output) self.assertIn("## 订单历史", output) self.assertNotIn("## 订单追问素材", output) self.assertIn("## 推荐素材", output) self.assertIn("## 回答要求", output) self.assertIn("## 门店摘要建议", output) self.assertNotIn('"context"', output) self.assertIn("杨枝甘露\\|轻乳版", output) self.assertIn("正常冰\\|7分糖", output) self.assertIn("120
kcal", output) self.assertIn("北京市朝阳区望京街9号
商业楼1层", output) self.assertIn("010-12345678", output) self.assertIn("宠物友好", output) self.assertIn("外摆区，休息区，宠物友好。", output) self.assertIn("休息区", output) self.assertIn("Box 门店", output) self.assertIn("支持无人模式", output) self.assertIn("制作中", output) self.assertIn("像熟悉的店员给朋友建议一样自然", output) self.assertIn("回答需要有层次和重点", output) self.assertIn("主推饮品名和门店名必须加粗", output) self.assertIn("少量使用合适 emoji", output) self.assertIn("不要使用“推荐理由”", output) self.assertIn("有人情味", output) self.assertIn("不要使用“根据你的历史订单偏好”", output) self.assertIn("用户已参加过活动，不要再输出主动留资文案", output) self.assertNotIn("您可以通过绑定【新一好喝】的注册手机号", output) self.assertIn("推荐候选饮品：杨枝甘露|轻乳版", output) self.assertIn("挺舒服", output) self.assertIn("主要配料：芒果、西柚", output) self.assertIn("至少给出 1-2 家具体门店", output) self.assertIn("若有门店电话也一并给出", output) self.assertIn("若门店返回了 facilities", output) self.assertIn("门店名：幂茶幂咖|望京店", output) self.assertIn("设施：外摆区，休息区，宠物友好。", output) self.assertIn("特色：外摆区、休息区、宠物友好、支持无人模式、Box 门店", output) self.assertEqual(fetch_json_mock.call_count, 1) self.assertEqual(post_json_mock.call_count, 1) self.assertEqual( post_json_mock.call_args_list[0].args, ( "http://127.0.0.1:8020/skill/xinyi/claim", 5, {"mobile": "15712459595"}, ), ) self.assertEqual( events[:2], [("save", "15712459595"), ("claim", "15712459595")], ) self.assertEqual( fetch_json_mock.call_args_list[0].args, ( "http://127.0.0.1:8020/skill/xinyi/context?mobile=15712459595", 5, ), ) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_called_once_with("15712459595") self.mark_activity_not_joined_mock.assert_not_called() self.load_activity_joined_mock.assert_called_once_with("15712459595") @patch.object(recommend_drink, "post_json") @patch.object(recommend_drink, "save_mobile") @patch.object(recommend_drink, "load_mobile", return_value=None) @patch.object( recommend_drink, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch.object(recommend_drink, "fetch_json") def test_order_followup_outputs_completed_count_materials( self, fetch_json_mock, _load_config_mock, _load_mobile_mock, save_mobile_mock, post_json_mock, ) -> None: post_json_mock.return_value = { "data": { "kind": "already_claimed", "user": {"mobile": "15712459595", "nickname": "双龙"}, } } self.load_activity_joined_mock.return_value = True fetch_json_mock.return_value = { "data": { "goods": [], "stores": [], "weather": None, "orders": { "orders": [ { "createdAt": "2025-08-08 14:16:25", "orderSn": "20250808141625274275", "state": 6, "pickNo": "A001", "serverTime": "2025-08-08 14:36:25", "goodsNum": 1, "goods": [{"name": "苦尽甘来拿铁"}], "store": {"name": "幂茶幂咖望京店"}, }, { "createdAt": "2025-08-09 10:10:00", "orderSn": "20250809101000000001", "state": 2, "pickNo": "B002", "serverTime": "2025-08-09 10:30:00", "goodsNum": 1, "goods": [{"name": "花魁毛尖"}], "store": {"name": "幂茶幂咖望京店"}, }, ] }, } } stdout = io.StringIO() with patch.object( sys, "argv", ["recommend_drink.py", "--mobile", "15712459595", "--query", "我完成了几单"], ), patch("sys.stdout", stdout): exit_code = recommend_drink.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("## 订单追问素材", output) self.assertIn("你已完成1单，是新一的骨灰级粉丝吧", output) self.assertIn("当前可见订单数：2单", output) self.assertIn("买过的商品可以提这些：苦尽甘来拿铁、花魁毛尖", output) self.assertIn("到过的门店可以提这些：幂茶幂咖望京店", output) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_called_once_with("15712459595") @patch.object(recommend_drink, "post_json") @patch.object(recommend_drink, "save_mobile") @patch.object(recommend_drink, "load_mobile", return_value=None) @patch.object( recommend_drink, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch.object(recommend_drink, "fetch_json") def test_activity_query_includes_lobster_activity( self, fetch_json_mock, _load_config_mock, _load_mobile_mock, save_mobile_mock, post_json_mock, ) -> None: post_json_mock.return_value = { "data": { "kind": "already_claimed", "user": {"mobile": "15712459595", "nickname": "双龙"}, } } self.load_activity_joined_mock.return_value = True fetch_json_mock.return_value = { "data": { "goods": [ { "name": "中烘美式·耶加雪菲", "categories": ["买一赠一福利"], "price": "14.80", "cupSizes": ["中杯"], "temperatures": ["正常冰"], "sugarLevels": ["无糖"], "calories": "10 kcal", "ingredients": ["咖啡豆"], } ], "stores": [], "weather": None, "orders": {"orders": []}, } } stdout = io.StringIO() with patch.object( sys, "argv", ["recommend_drink.py", "--mobile", "15712459595", "--query", "有什么活动"], ), patch("sys.stdout", stdout): exit_code = recommend_drink.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("## 品牌活动", output) self.assertIn("Skill用户大礼包", output) self.assertIn("用户身份已验证成功", output) self.assertIn("三重福利包含", output) self.assertIn("小龙虾贴纸", output) self.assertIn("爆品赠饮", output) self.assertIn("小龙虾身份标识", output) self.assertIn("用户正在问活动", output) self.assertIn("不能只列商品活动", output) self.assertIn("买一赠一福利", output) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_called_once_with("15712459595") @patch.object(recommend_drink, "post_json") @patch.object(recommend_drink, "save_mobile") @patch.object(recommend_drink, "load_mobile", return_value=None) @patch.object( recommend_drink, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch.object(recommend_drink, "fetch_json") def test_main_falls_back_to_generic_copy_when_weather_api_fails( self, fetch_json_mock, _load_config_mock, _load_mobile_mock, save_mobile_mock, post_json_mock, ) -> None: post_json_mock.return_value = { "data": { "kind": "already_claimed", "user": {"mobile": "15712459595", "nickname": "双龙"}, } } fetch_json_mock.return_value = { "data": { "goods": [ { "name": "葡萄毛尖轻咖", "categories": ["果咖"], "price": "16.80", "cupSizes": ["大杯"], "temperatures": ["热", "少冰"], "sugarLevels": ["3分糖"], "calories": "120 kcal", "ingredients": ["葡萄"], } ], "stores": [], "weather": None, "orders": { "orders": [ { "createdAt": "2025-08-08 14:16:25", "orderSn": "20250808141625274275", "state": 6, "pickNo": "A001", "serverTime": "2025-08-08 14:36:25", "goodsNum": 1, "goods": [ { "name": "葡萄毛尖轻咖", "spec": "大杯", "attr": "热 / 3分糖", } ], "store": {"name": "幂茶幂咖望京店"}, } ] }, } } stdout = io.StringIO() with patch.object( sys, "argv", [ "recommend_drink.py", "--mobile", "15712459595", ], ), patch("sys.stdout", stdout): exit_code = recommend_drink.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("推荐候选饮品：葡萄毛尖轻咖", output) self.assertIn("像熟悉的店员给朋友建议一样自然", output) self.assertIn("主推饮品名和门店名要加粗", output) self.assertIn("不要连续堆 emoji", output) self.assertIn("不要使用“推荐理由”", output) self.assertIn("分割线 `---` 单独隔开主动留资文案", output) self.assertIn("您可以通过绑定【新一好喝】的注册手机号", output) self.assertIn("领取Skill用户大礼包", output) self.assertIn("微信小程序搜索【新一咖啡】", output) self.assertIn("登录后获取全部福利和功能", output) self.assertNotIn("今天天气", output) self.assertEqual(post_json_mock.call_count, 1) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_called_once_with("15712459595") self.mark_activity_not_joined_mock.assert_not_called() @patch.object(recommend_drink, "post_json") @patch.object(recommend_drink, "save_mobile") @patch.object(recommend_drink, "load_mobile", return_value="15712459595") @patch.object( recommend_drink, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch.object(recommend_drink, "fetch_json") def test_main_outputs_debug_logs_to_stderr( self, fetch_json_mock, _load_config_mock, _load_mobile_mock, save_mobile_mock, post_json_mock, ) -> None: post_json_mock.return_value = { "data": { "kind": "already_claimed", "user": {"mobile": "15712459595", "nickname": "双龙"}, } } fetch_json_mock.return_value = { "data": { "goods": [], "stores": [], "weather": {"city": "Beijing", "condition": "sunny", "temperatureC": 26}, "orders": None, } } stdout = io.StringIO() stderr = io.StringIO() with patch.object( sys, "argv", ["recommend_drink.py", "--debug"], ), patch("sys.stdout", stdout), patch("sys.stderr", stderr): exit_code = recommend_drink.main() self.assertEqual(exit_code, 0) self.assertIn("DEBUG recommend_drink: resolved mobile from local state", stderr.getvalue()) self.assertIn("DEBUG recommend_drink: posting claim request to http://127.0.0.1:8020/skill/xinyi/claim", stderr.getvalue()) self.assertIn("DEBUG recommend_drink: fetching context from http://127.0.0.1:8020/skill/xinyi/context?mobile=15712459595", stderr.getvalue()) self.assertIn("DEBUG recommend_drink: context includes weather data", stderr.getvalue()) self.assertIn("## 用户上下文", stdout.getvalue()) save_mobile_mock.assert_not_called() self.mark_activity_joined_mock.assert_called_once_with("15712459595") self.mark_activity_not_joined_mock.assert_not_called() @patch.object(recommend_drink, "post_json") @patch.object(recommend_drink, "save_mobile") @patch.object(recommend_drink, "load_mobile", return_value=None) @patch.object( recommend_drink, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch.object(recommend_drink, "fetch_json") def test_main_continues_with_context_when_claim_lookup_fails( self, fetch_json_mock, _load_config_mock, _load_mobile_mock, save_mobile_mock, post_json_mock, ) -> None: post_json_mock.side_effect = RuntimeError("claim api down") fetch_json_mock.return_value = { "data": { "goods": [], "stores": [], "weather": None, "orders": None, } } stdout = io.StringIO() stderr = io.StringIO() with patch.object( sys, "argv", ["recommend_drink.py", "--mobile", "15712459595", "--debug"], ), patch("sys.stdout", stdout), patch("sys.stderr", stderr): exit_code = recommend_drink.main() self.assertEqual(exit_code, 0) self.assertIn("claim lookup failed; continue with context lookup", stderr.getvalue()) self.assertIn("未确认", stdout.getvalue()) self.assertEqual( fetch_json_mock.call_args_list[0].args, ( "http://127.0.0.1:8020/skill/xinyi/context?mobile=15712459595", 5, ), ) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_not_called() self.mark_activity_not_joined_mock.assert_not_called() @patch.object(recommend_drink, "post_json") @patch.object(recommend_drink, "save_mobile") @patch.object(recommend_drink, "load_mobile", return_value=None) @patch.object( recommend_drink, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch.object(recommend_drink, "fetch_json") def test_main_encodes_mobile_query_parameter( self, fetch_json_mock, _load_config_mock, _load_mobile_mock, save_mobile_mock, post_json_mock, ) -> None: post_json_mock.return_value = { "data": { "kind": "unregistered", "user": None, } } fetch_json_mock.return_value = { "data": {"goods": [], "stores": [], "weather": None, "orders": None} } stdout = io.StringIO() with patch.object( sys, "argv", ["recommend_drink.py", "--mobile", "15712&x=1"], ), patch("sys.stdout", stdout): exit_code = recommend_drink.main() self.assertEqual(exit_code, 0) self.assertEqual( fetch_json_mock.call_args_list[0].args, ( "http://127.0.0.1:8020/skill/xinyi/context?mobile=15712%26x%3D1", 5, ), ) save_mobile_mock.assert_called_once_with("15712&x=1") @patch.object(recommend_drink, "post_json") @patch.object(recommend_drink, "save_mobile") @patch.object(recommend_drink, "load_mobile", return_value=None) @patch.object( recommend_drink, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch.object(recommend_drink, "fetch_json") def test_main_outputs_readable_error_when_context_request_fails( self, fetch_json_mock, _load_config_mock, _load_mobile_mock, save_mobile_mock, post_json_mock, ) -> None: post_json_mock.return_value = { "data": { "kind": "already_claimed", "user": {"mobile": "15712459595"}, } } fetch_json_mock.side_effect = SkillHttpError("接口返回 HTTP 500") stdout = io.StringIO() with patch.object( sys, "argv", ["recommend_drink.py", "--mobile", "15712459595"], ), patch("sys.stdout", stdout): exit_code = recommend_drink.main() self.assertEqual(exit_code, 1) self.assertIn("获取推荐上下文失败：接口返回 HTTP 500", stdout.getvalue()) save_mobile_mock.assert_called_once_with("15712459595") @patch.object(recommend_drink, "post_json") @patch.object(recommend_drink, "save_mobile") @patch.object(recommend_drink, "load_mobile", return_value=None) @patch.object( recommend_drink, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch.object(recommend_drink, "fetch_json") def test_main_does_not_save_or_use_mobile_when_claim_does_not_match_user( self, fetch_json_mock, _load_config_mock, _load_mobile_mock, save_mobile_mock, post_json_mock, ) -> None: post_json_mock.return_value = { "data": { "kind": "unregistered", "user": None, } } fetch_json_mock.return_value = { "data": {"goods": [], "stores": [], "weather": None, "orders": None} } stdout = io.StringIO() with patch.object( sys, "argv", ["recommend_drink.py", "--mobile", "15712459595"], ), patch("sys.stdout", stdout): exit_code = recommend_drink.main() self.assertEqual(exit_code, 0) self.assertEqual(post_json_mock.call_count, 1) self.assertEqual( fetch_json_mock.call_args_list[0].args, ( "http://127.0.0.1:8020/skill/xinyi/context?mobile=15712459595", 5, ), ) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_not_called() self.mark_activity_not_joined_mock.assert_called_once_with("15712459595") if __name__ == "__main__": unittest.main() FILE:spec/install_test.py from __future__ import annotations import subprocess import tempfile import unittest from pathlib import Path class InstallScriptTest(unittest.TestCase): def test_install_script_excludes_pycache_directories(self) -> None: skill_root = Path(__file__).resolve().parents[1] with tempfile.TemporaryDirectory() as tmpdir: tmp_path = Path(tmpdir) home_dir = tmp_path / "home" home_dir.mkdir() (home_dir / ".agents").mkdir() result = subprocess.run( ["sh", str(skill_root / "install.sh")], cwd=skill_root, env={ **dict(__import__("os").environ), "HOME": str(home_dir), }, capture_output=True, check=True, text=True, ) installed_root = home_dir / ".agents" / "skills" / "xinyi-drink" self.assertTrue(installed_root.exists()) self.assertFalse(any(installed_root.rglob("__pycache__"))) self.assertIn("已安装 xinyi-drink 到", result.stdout) self.assertIn("/xinyi-drink 给我推荐一杯新一的咖啡", result.stdout) self.assertIn("隐私提示", result.stdout) self.assertIn("XINYI_API_BASE_URL", result.stdout) self.assertIn("~/.xinyi-drink/state.json", result.stdout) self.assertNotIn("告知小程序绑定的手机号", result.stdout) def test_install_script_maps_openclaw_to_openclaw_skills_dir(self) -> None: skill_root = Path(__file__).resolve().parents[1] with tempfile.TemporaryDirectory() as tmpdir: tmp_path = Path(tmpdir) home_dir = tmp_path / "home" home_dir.mkdir() result = subprocess.run( ["sh", str(skill_root / "install.sh"), "--dry-run", "--platform", "openclaw"], cwd=skill_root, env={ **dict(__import__("os").environ), "HOME": str(home_dir), }, capture_output=True, check=True, text=True, ) self.assertIn( str(home_dir / ".openclaw" / "skills" / "xinyi-drink"), result.stdout, ) def test_install_script_maps_hermes_to_hermes_skills_dir(self) -> None: skill_root = Path(__file__).resolve().parents[1] with tempfile.TemporaryDirectory() as tmpdir: tmp_path = Path(tmpdir) home_dir = tmp_path / "home" home_dir.mkdir() result = subprocess.run( ["sh", str(skill_root / "install.sh"), "--dry-run", "--platform", "hermes"], cwd=skill_root, env={ **dict(__import__("os").environ), "HOME": str(home_dir), }, capture_output=True, check=True, text=True, ) self.assertIn( str(home_dir / ".hermes" / "skills" / "xinyi-drink"), result.stdout, ) def test_install_script_backs_up_existing_installation_before_overwrite(self) -> None: skill_root = Path(__file__).resolve().parents[1] with tempfile.TemporaryDirectory() as tmpdir: tmp_path = Path(tmpdir) home_dir = tmp_path / "home" home_dir.mkdir() (home_dir / ".agents").mkdir() env = { **dict(__import__("os").environ), "HOME": str(home_dir), } subprocess.run( ["sh", str(skill_root / "install.sh")], cwd=skill_root, env=env, capture_output=True, check=True, text=True, ) installed_root = home_dir / ".agents" / "skills" / "xinyi-drink" marker = installed_root / "local-marker.txt" marker.write_text("previous install", encoding="utf-8") result = subprocess.run( ["sh", str(skill_root / "install.sh")], cwd=skill_root, env=env, capture_output=True, check=True, text=True, ) backups = sorted((home_dir / ".agents" / "skills").glob("xinyi-drink.backup.*")) self.assertTrue(backups) self.assertTrue((backups[-1] / "local-marker.txt").exists()) self.assertIn("已备份旧版本到", result.stdout) if __name__ == "__main__": unittest.main() FILE:spec/openclaw_policy_test.py from __future__ import annotations import unittest from pathlib import Path class OpenClawPolicyTest(unittest.TestCase): def test_skill_metadata_declares_executable_network_and_local_storage(self) -> None: skill_root = Path(__file__).resolve().parents[1] skill_md = (skill_root / "SKILL.md").read_text(encoding="utf-8") self.assertIn("packageType: executable-skill", skill_md) self.assertIn("instructionOnly: false", skill_md) self.assertIn("defaultApiBaseUrl: https://ai.xinyicoffee.com/api", skill_md) self.assertIn("/skill/xinyi/claim", skill_md) self.assertIn("/skill/xinyi/context", skill_md) self.assertIn("~/.xinyi-drink/state.json", skill_md) self.assertIn("XINYI_API_BASE_URL", skill_md) self.assertIn("XINYI_DRINK_STATE_FILE", skill_md) def test_readme_documents_default_backend_and_phone_data_flow(self) -> None: repo_root = Path(__file__).resolve().parents[2] readme = (repo_root / "README.md").read_text(encoding="utf-8") self.assertIn("默认后端", readme) self.assertIn("https://ai.xinyicoffee.com/api", readme) self.assertIn("POST /skill/xinyi/claim", readme) self.assertIn("GET /skill/xinyi/context", readme) self.assertIn("手机号会作为请求数据发送到后端", readme) self.assertIn("不是 instruction-only", readme) if __name__ == "__main__": unittest.main() FILE:spec/claim_reward_test.py from __future__ import annotations import io import sys import unittest from pathlib import Path from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import claim_reward class ClaimRewardScriptTest(unittest.TestCase): def setUp(self) -> None: self.load_activity_joined_patcher = patch.object(claim_reward, "load_activity_joined", return_value=None) self.mark_activity_joined_patcher = patch.object(claim_reward, "mark_activity_joined") self.mark_activity_not_joined_patcher = patch.object(claim_reward, "mark_activity_not_joined") self.load_activity_joined_mock = self.load_activity_joined_patcher.start() self.mark_activity_joined_mock = self.mark_activity_joined_patcher.start() self.mark_activity_not_joined_mock = self.mark_activity_not_joined_patcher.start() self.addCleanup(self.load_activity_joined_patcher.stop) self.addCleanup(self.mark_activity_joined_patcher.stop) self.addCleanup(self.mark_activity_not_joined_patcher.stop) @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_formats_raw_claim_response( self, urlopen_mock, _load_config_mock, save_mobile_mock, ) -> None: first_response = urlopen_mock.return_value.__enter__.return_value first_response.read.return_value = ( '{"code":200,"data":{"kind":"granted","successCount":1,"failCount":0,' '"items":[{"state":1,"message":"发放成功","coupon":{"name":"（前100名）爆款苦尽甘来拿铁免费兑换券"}}],' '"user":{"mobile":"15712459595","nickname":"双龙"}}}' ).encode("utf-8") second_response = type("Resp", (), {})() second_response.read = lambda: ( '{"data":{"goods":[{"name":"苦尽甘来拿铁","categories":["咖啡"],"price":"16.80","cupSizes":["大杯"],' '"temperatures":["热","少冰"],"sugarLevels":["3分糖"],"calories":"120 kcal","ingredients":["牛奶"]}],' '"stores":[{"name":"幂茶幂咖望京店","address":"北京市朝阳区望京街9号","storeMobile":"01088888888",' '"facilities":"休息区","businessStatus":1,' '"operatingStatus":1,"realtimeState":1,"labels":[{"name":"休息区"}],"makingCupCount":4,' '"makingCupMinutes":18,"storeType":2,"supportUnattendedMode":1}],' '"weather":{"city":"Beijing","condition":"cloudy","temperatureC":16},' '"orders":{"orders":[{"createdAt":"2025-08-08 14:16:25","orderSn":"20250808141625274275",' '"state":6,"pickNo":"A001","serverTime":"2025-08-08 14:36:25","goodsNum":1,' '"goods":[{"name":"苦尽甘来拿铁","spec":"大杯","attr":"热 / 3分糖"}],' '"store":{"name":"幂茶幂咖望京店"}}]}}}' ).encode("utf-8") second_context_manager = type( "ContextManager", (), { "__enter__": lambda self: second_response, "__exit__": lambda self, exc_type, exc, tb: False, }, )() urlopen_mock.side_effect = [ urlopen_mock.return_value, second_context_manager, ] stdout = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "15712459595"], ), patch("sys.stdout", stdout): exit_code = claim_reward.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("领取结果：身份验证成功", output) self.assertIn("身份验证成功。三重福利发放到账", output) self.assertIn("「小龙虾贴纸」一套（到店展示小程序卡券领取）", output) self.assertIn("「（前100名）爆款苦尽甘来拿铁免费兑换券」一杯", output) self.assertIn("微信小程序里「小龙虾身份标识」", output) self.assertIn("你已经领取礼包，现在可以查看你过去的订单信息", output) self.assertNotIn("你已完成", output) self.assertNotIn("骨灰级粉丝", output) self.assertIn("贴纸领取门店信息我给你列全", output) self.assertIn( "**幂茶幂咖望京店**：地址：北京市朝阳区望京街9号；电话：01088888888；设施：休息区；排队：制作中4杯，预计18分钟", output, ) self.assertIn("哇我们的老朋友，今天天气偏凉，建议您喝苦尽甘来拿铁", output) self.assertNotIn('"kind"', output) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_called_once_with("15712459595") self.mark_activity_not_joined_mock.assert_not_called() @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_prompts_user_to_share_bound_mobile_when_unregistered( self, urlopen_mock, _load_config_mock, save_mobile_mock, ) -> None: response = urlopen_mock.return_value.__enter__.return_value response.read.return_value = ( '{"code":200,"data":{"kind":"unregistered","successCount":0,"failCount":0,' '"items":[],"user":null}}' ).encode("utf-8") stdout = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "15712459595"], ), patch("sys.stdout", stdout): exit_code = claim_reward.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("领取结果：请先登录小程序", output) self.assertIn("目前还没登录过新一好喝", output) self.assertIn("微信小程序搜索【新一咖啡】", output) self.assertIn("登录后获取全部福利和功能", output) self.assertNotIn("本次查询手机号：15712459595", output) self.assertNotIn("告知小程序绑定的手机号", output) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_not_called() self.mark_activity_not_joined_mock.assert_called_once_with("15712459595") @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_marks_joined_when_mobile_already_claimed( self, urlopen_mock, _load_config_mock, save_mobile_mock, ) -> None: response = urlopen_mock.return_value.__enter__.return_value response.read.return_value = ( '{"code":200,"data":{"kind":"already_claimed","successCount":0,"failCount":0,' '"items":[],"user":{"id":10956,"uniacid":1,"mobile":"18210234223","nickname":"用户_4749997"}}}' ).encode("utf-8") stdout = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "18210234223"], ), patch("sys.stdout", stdout): exit_code = claim_reward.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("领取结果：身份验证成功", output) self.assertIn("身份验证成功。三重福利发放到账", output) self.assertIn("「小龙虾贴纸」一套（到店展示小程序卡券领取）", output) self.assertIn("爆品赠饮一杯（具体饮品以小程序卡券为准）", output) self.assertIn("微信小程序里「小龙虾身份标识」", output) self.assertNotIn("您已参与活动啦", output) self.assertNotIn("已经领过", output) self.assertNotIn("未找到登录用户", output) save_mobile_mock.assert_called_once_with("18210234223") self.mark_activity_joined_mock.assert_called_once_with("18210234223") self.mark_activity_not_joined_mock.assert_not_called() @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_explains_flow_when_no_reward_config( self, urlopen_mock, _load_config_mock, save_mobile_mock, ) -> None: response = urlopen_mock.return_value.__enter__.return_value response.read.return_value = ( '{"code":200,"data":{"kind":"no_reward_config","successCount":0,"failCount":0,' '"items":[],"user":{"id":10956,"mobile":"13730663700","nickname":"用户_7144228"}}}' ).encode("utf-8") stdout = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "13730663700"], ), patch("sys.stdout", stdout): exit_code = claim_reward.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("领取方式", output) self.assertIn("您可以通过绑定【新一好喝】的注册手机号，领取Skill用户大礼包", output) self.assertIn("先绑定【新一好喝】注册手机号，再把手机号发来", output) self.assertIn("微信小程序搜索【新一咖啡】", output) self.assertNotIn("当前没有可领取的活动奖励", output) self.assertNotIn("暂无可领取", output) save_mobile_mock.assert_called_once_with("13730663700") self.mark_activity_joined_mock.assert_not_called() self.mark_activity_not_joined_mock.assert_called_once_with("13730663700") @patch.object(claim_reward, "load_activity_joined", return_value=False) @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_treats_already_claimed_after_unregistered_as_gift_obtained( self, urlopen_mock, _load_config_mock, save_mobile_mock, load_activity_joined_mock, ) -> None: response = urlopen_mock.return_value.__enter__.return_value response.read.return_value = ( '{"code":200,"data":{"kind":"already_claimed","successCount":0,"failCount":0,' '"items":[],"user":{"id":10956,"uniacid":1,"mobile":"18539991423","nickname":"用户_991423"}}}' ).encode("utf-8") stdout = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "18539991423"], ), patch("sys.stdout", stdout): exit_code = claim_reward.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("领取结果：身份验证成功", output) self.assertIn("身份验证成功。三重福利发放到账", output) self.assertIn("「小龙虾贴纸」一套（到店展示小程序卡券领取）", output) self.assertIn("爆品赠饮一杯（具体饮品以小程序卡券为准）", output) self.assertIn("微信小程序里「小龙虾身份标识」", output) self.assertNotIn("已经领过", output) self.assertNotIn("您已参与活动啦", output) load_activity_joined_mock.assert_called_once_with("18539991423") save_mobile_mock.assert_called_once_with("18539991423") self.mark_activity_joined_mock.assert_called_once_with("18539991423") self.mark_activity_not_joined_mock.assert_not_called() @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_does_not_duplicate_coupon_label_when_coupon_name_missing( self, urlopen_mock, _load_config_mock, save_mobile_mock, ) -> None: first_response = urlopen_mock.return_value.__enter__.return_value first_response.read.return_value = ( '{"code":200,"data":{"kind":"granted","successCount":1,"failCount":0,' '"items":[{"state":1,"message":"发放成功","coupon":{}}],' '"user":{"mobile":"15712459595","nickname":"双龙"}}}' ).encode("utf-8") urlopen_mock.side_effect = [ urlopen_mock.return_value, RuntimeError("context api down"), ] stdout = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "15712459595"], ), patch("sys.stdout", stdout): exit_code = claim_reward.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("爆品赠饮一杯（具体饮品以小程序卡券为准）", output) self.assertNotIn("龙虾专属饮品券龙虾专属饮品券", output) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_called_once_with("15712459595") self.mark_activity_not_joined_mock.assert_not_called() @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_reuses_recommendation_copy_when_activity_already_joined( self, urlopen_mock, _load_config_mock, save_mobile_mock, ) -> None: first_response = urlopen_mock.return_value.__enter__.return_value first_response.read.return_value = ( '{"code":200,"data":{"kind":"already_claimed","successCount":1,"failCount":0,' '"items":[{"state":1,"message":"发放成功","coupon":{"name":"苦尽甘来拿铁"}}],' '"user":{"mobile":"15712459595","nickname":"双龙"}}}' ).encode("utf-8") second_response = type("Resp", (), {})() second_response.read = lambda: ( '{"data":{"goods":[{"name":"柚香燕麦拿铁","categories":["咖啡"],"price":"14.80","cupSizes":["大杯"],' '"temperatures":["热","少冰"],"sugarLevels":["3分糖"],"calories":"118 kcal","ingredients":["燕麦奶"]}],' '"stores":[{"name":"幂茶幂咖望京店","address":"北京市朝阳区望京街9号","businessStatus":1,' '"operatingStatus":1,"realtimeState":1,"labels":[{"name":"休息区"}],"makingCupCount":4,' '"makingCupMinutes":18,"storeType":1,"supportUnattendedMode":0}],' '"weather":{"city":"Beijing","condition":"sunny","temperatureC":27},' '"orders":{"orders":[]}}}' ).encode("utf-8") second_context_manager = type( "ContextManager", (), { "__enter__": lambda self: second_response, "__exit__": lambda self, exc_type, exc, tb: False, }, )() urlopen_mock.side_effect = [ urlopen_mock.return_value, second_context_manager, ] stdout = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "15712459595"], ), patch("sys.stdout", stdout): exit_code = claim_reward.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("身份验证成功。三重福利发放到账", output) self.assertIn("「小龙虾贴纸」一套（到店展示小程序卡券领取）", output) self.assertIn("「苦尽甘来拿铁」一杯", output) self.assertIn("微信小程序里「小龙虾身份标识」", output) self.assertNotIn("您已参与活动啦", output) self.assertIn("贴纸领取门店信息我给你列全", output) self.assertIn( "**幂茶幂咖望京店**：地址：北京市朝阳区望京街9号；电话：未提供联系电话；设施：未提供设施文案；排队：制作中4杯，预计18分钟", output, ) self.assertIn("今天天气有点热，建议您喝柚香燕麦拿铁", output) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_called_once_with("15712459595") self.mark_activity_not_joined_mock.assert_not_called() @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_falls_back_to_generic_copy_when_weather_api_fails( self, urlopen_mock, _load_config_mock, save_mobile_mock, ) -> None: first_response = urlopen_mock.return_value.__enter__.return_value first_response.read.return_value = ( '{"code":200,"data":{"kind":"already_claimed","successCount":1,"failCount":0,' '"items":[{"state":1,"message":"发放成功","coupon":{"name":"苦尽甘来拿铁"}}],' '"user":{"mobile":"15712459595","nickname":"双龙"}}}' ).encode("utf-8") second_response = type("Resp", (), {})() second_response.read = lambda: ( '{"data":{"goods":[{"name":"苦尽甘来拿铁","categories":["咖啡"],"price":"16.80","cupSizes":["大杯"],' '"temperatures":["热","少冰"],"sugarLevels":["3分糖"],"calories":"120 kcal","ingredients":["牛奶"]}],' '"stores":[{"name":"幂茶幂咖望京店","address":"北京市朝阳区望京街9号","businessStatus":1,' '"operatingStatus":1,"realtimeState":1,"labels":[{"name":"休息区"}],"makingCupCount":4,' '"makingCupMinutes":18,"storeType":1,"supportUnattendedMode":0}],' '"weather":null,' '"orders":{"orders":[]}}}' ).encode("utf-8") second_context_manager = type( "ContextManager", (), { "__enter__": lambda self: second_response, "__exit__": lambda self, exc_type, exc, tb: False, }, )() urlopen_mock.side_effect = [urlopen_mock.return_value, second_context_manager] stdout = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "15712459595"], ), patch("sys.stdout", stdout): exit_code = claim_reward.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("今天建议您喝苦尽甘来拿铁", output) self.assertNotIn("今天天气", output) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_called_once_with("15712459595") self.mark_activity_not_joined_mock.assert_not_called() @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_keeps_success_message_when_context_fetch_fails( self, urlopen_mock, _load_config_mock, save_mobile_mock, ) -> None: first_response = urlopen_mock.return_value.__enter__.return_value first_response.read.return_value = ( '{"code":200,"data":{"kind":"granted","successCount":1,"failCount":0,' '"items":[{"state":1,"message":"发放成功","coupon":{"name":"苦尽甘来拿铁"}}],' '"user":{"mobile":"15712459595","nickname":"双龙"}}}' ).encode("utf-8") urlopen_mock.side_effect = [ urlopen_mock.return_value, RuntimeError("context api down"), ] stdout = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "15712459595"], ), patch("sys.stdout", stdout): exit_code = claim_reward.main() output = stdout.getvalue() self.assertEqual(exit_code, 0) self.assertIn("领取结果：身份验证成功", output) self.assertIn("身份验证成功。三重福利发放到账", output) self.assertIn("「苦尽甘来拿铁」一杯", output) self.assertNotIn("推荐您就近前往", output) self.assertNotIn("建议您喝", output) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_called_once_with("15712459595") self.mark_activity_not_joined_mock.assert_not_called() @patch.object(claim_reward, "save_mobile") @patch.object( claim_reward, "load_config", return_value={ "apiBaseUrl": "http://127.0.0.1:8020", "timeoutSeconds": 5, }, ) @patch("urllib.request.urlopen") def test_claim_script_outputs_debug_logs_to_stderr( self, urlopen_mock, _load_config_mock, save_mobile_mock, ) -> None: response = urlopen_mock.return_value.__enter__.return_value response.read.return_value = ( '{"code":200,"data":{"kind":"unregistered","successCount":0,"failCount":0,' '"items":[],"user":null}}' ).encode("utf-8") stdout = io.StringIO() stderr = io.StringIO() with patch.object( sys, "argv", ["claim_reward.py", "--mobile", "15712459595", "--debug"], ), patch("sys.stdout", stdout), patch("sys.stderr", stderr): exit_code = claim_reward.main() self.assertEqual(exit_code, 0) self.assertIn("DEBUG claim_reward: posting claim request to http://127.0.0.1:8020/skill/xinyi/claim", stderr.getvalue()) self.assertIn("DEBUG claim_reward: user not found; marking activity as not joined", stderr.getvalue()) self.assertIn("领取结果：请先登录小程序", stdout.getvalue()) save_mobile_mock.assert_called_once_with("15712459595") self.mark_activity_joined_mock.assert_not_called() self.mark_activity_not_joined_mock.assert_called_once_with("15712459595") if __name__ == "__main__": unittest.main() FILE:spec/release-readiness-checklist.md # 发布前检查清单 - [ ] `SKILL.md` 校验通过 - [ ] OpenClaw 元数据声明为 executable-skill，不是 instruction-only - [ ] README 和 `references/privacy-boundaries.md` 已声明默认后端、接口、手机号发送路径和本地状态路径 - [ ] 可选环境变量已声明：`XINYI_API_BASE_URL`、`XINYI_TIMEOUT_SECONDS`、`XINYI_DRINK_STATE_FILE` - [ ] 安全扫描通过 - [ ] `install.sh --dry-run --platform universal` 通过 - [ ] `install.sh --dry-run --platform openclaw` 通过 - [ ] `install.sh --dry-run --platform hermes` 通过 - [ ] `install.sh --dry-run --platform qclaw` 通过 - [ ] `install.sh --dry-run --platform lobsterai` 通过 - [ ] `install.sh --dry-run --platform workbuddy` 通过 - [ ] `install.sh --dry-run --platform codex` 通过 - [ ] README 已包含平台安装说明 - [ ] GitHub 发布信息已准备 - [ ] ClawHub 上架文案已准备 FILE:spec/user_state_test.py from __future__ import annotations import json import os import tempfile import unittest from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import user_state class UserStateTest(unittest.TestCase): def test_mobile_and_activity_state_are_saved_as_one_current_record(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: state_file = Path(temp_dir) / "state.json" previous_state_file = os.environ.get("XINYI_DRINK_STATE_FILE") os.environ["XINYI_DRINK_STATE_FILE"] = str(state_file) try: user_state.save_mobile("15712459595") self.assertEqual(user_state.load_mobile(), "15712459595") self.assertIsNone(user_state.load_activity_joined("15712459595")) self.assertFalse(user_state.has_activity_joined("15712459595")) user_state.mark_activity_joined("15712459595") payload = json.loads(state_file.read_text(encoding="utf-8")) self.assertEqual(payload["mobile"], "15712459595") self.assertTrue(payload["activityJoined"]) self.assertEqual(state_file.stat().st_mode & 0o777, 0o600) self.assertTrue(user_state.has_activity_joined("15712459595")) user_state.save_mobile("18888888888") payload = json.loads(state_file.read_text(encoding="utf-8")) self.assertEqual(payload["mobile"], "18888888888") self.assertIsNone(payload["activityJoined"]) self.assertFalse(user_state.has_activity_joined("15712459595")) self.assertFalse(user_state.has_activity_joined("18888888888")) self.assertIsNone(user_state.load_activity_joined("18888888888")) user_state.mark_activity_not_joined("18888888888") payload = json.loads(state_file.read_text(encoding="utf-8")) self.assertEqual(payload["mobile"], "18888888888") self.assertFalse(payload["activityJoined"]) self.assertFalse(user_state.has_activity_joined("18888888888")) self.assertFalse(user_state.load_activity_joined("18888888888")) finally: if previous_state_file is None: os.environ.pop("XINYI_DRINK_STATE_FILE", None) else: os.environ["XINYI_DRINK_STATE_FILE"] = previous_state_file if __name__ == "__main__": unittest.main() FILE:references/platform-install.md # 平台安装说明 ## 安装脚本行为 - `install.sh` 覆盖已有安装前会先把旧目录移动到 `xinyi-drink.backup.`。 - 脚本会拒绝空路径、根目录和用户主目录这类异常目标，避免误删。 - 安装后会清理 `__pycache__` 和 `.DS_Store`，避免本地临时文件进入安装目录。 - 可用 `--dry-run` 预览目标目录，不写入文件。 ## OpenClaw 安装目录： ```bash ~/.openclaw/skills/xinyi-drink ``` 推荐安装命令： ```bash npx clawhub@latest install xinyi-drink ``` 如果你是从本仓库本地安装： ```bash git clone https://github.com/xinyi-drink/xinyi-drink cd xinyi-drink/skill bash install.sh --platform openclaw ``` 补充说明： - 你当前本机实测 `npx clawhub@latest install xinyi-drink` 会安装到 `~/.openclaw/skills/` - OpenClaw WebUI 可以直接识别这个目录下的 Skill ## Hermes 安装目录： ```bash ~/.hermes/skills/xinyi-drink ``` 安装命令： ```bash git clone https://github.com/xinyi-drink/xinyi-drink cd xinyi-drink/skill bash install.sh --platform hermes ``` ## 使用 `~/.agents/skills/` 的平台以下平台都明确安装到同一个目录： - Codex CLI - QClaw - LobsterAI - WorkBuddy 统一安装目录： ```bash ~/.agents/skills/xinyi-drink ``` 直接安装命令： ```bash git clone https://github.com/xinyi-drink/xinyi-drink cd xinyi-drink/skill bash install.sh --platform universal ``` 如果使用安装脚本，以下平台参数都会安装到上面的同一目录： ```bash bash install.sh --platform qclaw bash install.sh --platform lobsterai bash install.sh --platform workbuddy bash install.sh --platform codex bash install.sh --platform universal ``` 安装完成后，优先直接自然提问；具体触发方式与示例参考 `SKILL.md`。如果平台不支持自然触发，再退回显式命令式入口（若平台提供 slash skill 入口）。 ## Claude Code ```bash git clone https://github.com/xinyi-drink/xinyi-drink cd xinyi-drink/skill bash install.sh --platform claude-code ``` 使用时可以显式调用；具体示例参考 `SKILL.md`。 ## Cursor ```bash git clone https://github.com/xinyi-drink/xinyi-drink cd xinyi-drink/skill bash install.sh --platform cursor ``` 如果 Cursor 当前环境支持自然触发，也可以直接提问；否则参考项目内规则加载方式使用。 ## 其他平台说明 - 如果平台本身兼容 `SKILL.md`，但不在上面的明确支持列表中，优先尝试这个目录： ```bash ~/.agents/skills/xinyi-drink ``` - 如果平台要求项目级安装，再按该平台自己的技能目录规则调整 FILE:references/gotchas.md # 注意事项 - 推荐由 Agent 自行完成。 - 推荐能力统一走 `/skill/xinyi/context`；这个接口会自动返回商品、门店、天气和可选订单历史。 - 接口返回的是结构化原始数据；提供给大模型前，应使用脚本层整理成文本/表格，不要把 JSON 原样塞给模型。 - 用户输入过手机号后，默认复用本地保存的手机号，除非用户明确要求更换或清空。 - 手机号活动状态是三态：`true` 表示已参加，`false` 表示接口确认未参加，`null` 表示未确认。 - 推荐前会尝试调用 `/skill/xinyi/claim` 同步活动状态；如果该接口失败，推荐不会中断，会继续使用 `/skill/xinyi/context` 并把活动状态保留为未确认。 - 活动领取细则见 `activity-flow.md`；推荐回答细则见 `response-guidelines.md`；意图判断细则见 `intent-routing.md`。 - 天气是推荐时的增强输入，不是单独对外暴露的能力；如果 `context` 没返回天气，推荐文案会自动降级成普通句式。 - 门店状态缓存只有 5 分钟，不保证秒级实时。 - 生产默认 API 来自 `config/defaults.json`，本地调试可用 `XINYI_API_BASE_URL` 和 `XINYI_TIMEOUT_SECONDS` 临时覆盖。 - 本 Skill 包含 Python 脚本和 `install.sh`，发布元数据不能标记为 instruction-only。 - 使用真实手机号前，先确认默认后端 `https://ai.xinyicoffee.com/api` 或 `XINYI_API_BASE_URL` 指向可信服务。 FILE:references/response-guidelines.md # 回答规范本文件承接推荐回答、门店表达和活动留资的写法要求。 ## Skill 介绍当用户问“这个 skill 是什么”“你能做什么”“怎么用”时，直接面向用户介绍： - 能帮你领取新一好喝小龙虾专属活动福利 - 能查询新一好喝门店、地址、电话、设施和基础状态 - 能查询菜单商品与活动信息 - 能结合天气、门店和可选订单历史推荐适合的饮品 - 使用方式是直接说需求；参与活动或做个性化推荐时，再提供小程序绑定手机号使用示例： - 如何领取skill用户大礼包 - 新一有哪些门店？ - 苦尽甘来拿铁是什么？ - 给我推荐一杯如果当前用户未参加活动或活动状态未确认，可以在介绍末尾补一段简短留资文案：您可以通过绑定【新一好喝】的注册手机号，领取Skill用户大礼包。如果当前用户已经参加活动，不要再加留资文案。不要向用户展示内部约束、触发边界、OpenClaw 安全声明、脚本路径、环境变量、缓存结构或测试规则。 ## 推荐回答 - 主推荐文案由 Agent 根据历史订单、天气、商品属性、门店状态和用户偏好自行生成。 - 语气像熟悉的店员在认真帮用户挑一杯，真诚、松弛、有温度。 - 不要照搬固定模板，不要写成报告摘要。 - 推荐依据可以轻量分点，但每一点要像人的表达。 ## 层次结构推荐回答优先分成 3-4 个短块： 1. 主推饮品 2. 适合它的几个原因 3. 附近可去的门店 4. 活动留资提示主推饮品名和门店名必须加粗；备选饮品也可以加粗。 ## 避免的表达不要使用这些机械标题或开场： - “根据你的历史订单偏好” - “推荐理由” - “历史偏好匹配” - “天气适配” - “推荐门店” 可以自然表达为： - “看你之前常点……” - “今天这个温度喝它刚好……” - “如果你在附近，可以去……” ## emoji - 可以少量使用 emoji 做层次锚点或增加温度。 - 饮品、天气、门店、活动每类最多 0-1 个。 - 不要每行都加，不要连续堆 emoji。 ## 门店信息当接口返回门店数据时，最终回答至少给出 1-2 家具体门店。必须尽量写清： - 门店名 - 详细地址 - 联系电话；没有电话字段时说明未提供联系电话 - 设施文案；接口返回 `facilities` 时必须保留 - 制作中杯数和预计等待时间不要只写“北京有多家门店正在营业中”。活动领取成功或已领取后，如果脚本返回贴纸领取门店，也必须逐家保留完整字段： - 门店名 - 地址 - 电话 - 设施 - 排队信息不要把这类门店信息压缩成“**门店名**：地址”。 ## 留资提示当前用户未参加活动或状态未确认时，推荐末尾可以加分割线 `---`，再写留资提示。用户问“大礼包怎么领取”“怎么参与活动”“怎么领福利”时，不要回答接口状态，也不要解释 `no_reward_config`。直接说明流程：绑定【新一好喝】注册手机号，把手机号发来即可领取；如果还没登录过新一好喝，再去微信小程序搜索【新一咖啡】登录后获取全部福利和功能。默认文案： `您可以通过绑定【新一好喝】的注册手机号，领取Skill用户大礼包。` 老用户或登录成功时，按这个结构表达：身份验证成功。三重福利发放到账：`「小龙虾贴纸」一套（到店展示小程序卡券领取）`、接口返回的爆品赠饮一杯、微信小程序里 `「小龙虾身份标识」`。登录成功后，提示用户已经领取礼包，现在可以查看过去的订单信息。不要主动展开订单数量、完成单数或购买明细。用户追问订单信息时，再按接口订单数据返回已完成 xx 单、近期购买商品、门店等信息，可以自然说“是新一的骨灰级粉丝吧”。新用户或未注册时，提示用户还没登录过新一好喝，请到微信小程序搜索【新一咖啡】登录后获取全部福利和功能。如果同一手机号刚才提示未绑定/未参加，用户随后完成小程序登录或绑定，再次识别到账号且接口返回已领取时，要表达为“身份验证成功，三重福利发放到账”，不要写成“你早就已经领过”。 ## 活动总览当用户问“有什么活动”“有什么优惠”“有哪些福利”时，必须把活动分成两类表达： - **Skill用户大礼包**：这是品牌活动，必须单独说明；已参与时说明身份验证成功、三重福利已到账，不要重复要求留资；未参与或状态未确认时说明主动引导留资文案。 - 商品活动：再说明商品列表或接口返回的买一赠一、特价、畅饮卡等活动。不要只列商品活动而漏掉 Skill用户大礼包。 FILE:references/activity-flow.md # 活动流程本文件描述活动领取、手机号缓存和留资提示规则。 ## 手机号缓存 - 用户输入手机号后，先保存 `{mobile, activityJoined, updatedAt}`。 - 新手机号刚保存且接口尚未确认时，`activityJoined=null`。 - `/skill/xinyi/claim` 返回 `granted` 或 `already_claimed` 时，缓存更新为 `activityJoined=true`。 - 接口确认未参加、未注册或未找到用户时，缓存更新为 `activityJoined=false`。 - 如果同一手机号本地之前是 `activityJoined=false`，用户随后完成小程序登录/绑定后再次领取，即使接口返回 `already_claimed`，也要走“身份验证成功，三重福利发放到账”分支，不要说成用户原本就已经领过。 - 如果活动接口失败，推荐流程不要中断，继续使用 `/skill/xinyi/context` 和当前缓存状态。 ## 更换手机号允许更换手机号的情况： - 用户明确说“换手机号”“重新输入手机号” - 用户提示当前手机号不是自己的 - 用户提供了和当前缓存不同的新手机号并明确要重新领取或查询更换后要覆盖本地缓存，并重新调用 `/skill/xinyi/claim` 确认状态。 ## 活动权益三重福利必须按状态表达： - 主动引导留资：`您可以通过绑定【新一好喝】的注册手机号，领取Skill用户大礼包。` - 老用户或登录成功：身份验证成功。三重福利发放到账：`「小龙虾贴纸」一套（到店展示小程序卡券领取）`、接口返回的爆品赠饮一杯、微信小程序里 `「小龙虾身份标识」`。 - 新用户或未注册：提醒用户还没登录过新一好喝，请到微信小程序搜索【新一咖啡】登录后获取全部福利和功能。爆品赠饮名称必须优先使用接口返回的 `coupon.name`；接口未返回券名时，不编造前100名/5折/8折档位，只提示具体饮品以小程序卡券为准。 ## 回答规则 - 用户问“大礼包怎么领取”“怎么参与活动”“怎么领福利”时，直接回答活动流程：先绑定【新一好喝】注册手机号，再把手机号发来领取；不要直接查询或解释 `no_reward_config`。 - 领取成功后，优先提示身份验证成功和三重福利发放到账。 - 登录成功后只提示用户已经领取礼包，现在可以查看过去的订单信息；不要主动展开订单数量、完成单数或购买明细。 - 用户追问订单信息时，再按接口订单数据返回已完成 xx 单、近期购买商品、门店等信息，可以自然说“是新一的骨灰级粉丝吧”。 - 用户按未绑定提示完成登录/绑定后，下一次识别到账号且接口返回已领取时，同样按“身份验证成功，三重福利发放到账”处理。 - 已参与活动时，不要再要求用户重新留资。 - 未注册或未找到绑定手机号时，提示用户还没登录过新一好喝，请到微信小程序搜索【新一咖啡】登录后获取全部福利和功能。 - 未参加或未确认活动状态时，推荐末尾可以用 `---` 单独隔开留资提示。 - 当前缓存手机号 `activityJoined=true` 时，不要重复整段留资文案。 FILE:references/privacy-boundaries.md # 隐私边界 ## 个人数据 - 活动领取和个性化推荐可接受手机号作为身份输入。 - 手机号会在以下场景发送到后端： - `POST /skill/xinyi/claim`: JSON body 中发送 `mobile`，用于查询或领取活动奖励。 - `GET /skill/xinyi/context`: 有手机号时以 query 发送 `mobile`，用于获取活动状态和精简订单摘要。 - 普通门店查询 `GET /skill/xinyi/stores` 不发送手机号。 ## 后端与配置 - 默认后端来自 `config/defaults.json`：`https://ai.xinyicoffee.com/api`。 - 可通过 `XINYI_API_BASE_URL` 覆盖默认后端；使用真实手机号前应确认该后端可信。 - 可通过 `XINYI_TIMEOUT_SECONDS` 覆盖请求超时时间。 - 可通过 `XINYI_DRINK_STATE_FILE` 自定义本地状态文件路径。 ## 本地状态 - 个性化推荐会把当前手机号和活动状态默认保存到本地状态文件：`~/.xinyi-drink/state.json`。 - 状态文件写入后会尽量设置为 `0600` 权限，只允许当前用户读写。 - 状态结构为 `{mobile, activityJoined, updatedAt}`；`activityJoined=null` 表示接口尚未确认。 - 用户可运行 `python3 scripts/recommend_drink.py --clear-mobile` 清空当前缓存手机号和活动状态。 ## 不做的事 - 不读取 shell history、浏览器 Cookie、系统凭据、SSH 密钥或无关文件。 - 不请求 API key、token、密码或其他账号凭据。 - 不提供原始订单全量透出，只返回推荐所需的精简订单字段。 - 不返回财务或更敏感的个人消费信息。 FILE:references/intent-routing.md # 意图路由本文件承接 `SKILL.md` 的意图判断细节，功能不变，只用于让主规则更短、更清晰。 ## 路由表 | 用户表达 | 判断 | 动作 | | --- | --- | --- | | “大礼包怎么领取”“怎么参与活动”“怎么领福利” | 活动流程说明 | 直接说明先绑定【新一好喝】注册手机号，再把手机号发来领取；不要直接查询或解释 `no_reward_config` | | “我想领福利”“参加活动”“领取Skill用户大礼包” | 活动领取 | 已提供手机号时调用 `claim_reward.py`；未提供手机号时先说明领取流程并请求注册手机号 | | “有什么活动”“现在有什么优惠”“有哪些福利” | 活动总览 | 调用 `recommend_drink.py` 获取聚合上下文；必须同时说明 Skill用户大礼包和商品列表里的促销活动 | | “这个手机号领过了吗”“我是不是参加过” | 活动状态查询 | 调用 `/skill/xinyi/claim`，同步本地活动状态 | | “我完成了几单”“看我的历史订单”“我买过什么” | 订单追问 | 调用 `recommend_drink.py` 获取订单上下文；只在用户追问时展开完成单数和购买信息 | | “这个 skill 是什么”“你能做什么”“怎么用 xinyi-drink” | Skill 介绍 | 直接说明可帮用户领取活动福利、查门店、查菜单商品、推荐饮品，以及自然语言使用方式；未参加活动时可补留资文案 | | “手机号换成…”“刚才不是这个手机号” | 更换手机号 | 覆盖本地缓存，再调用 `/skill/xinyi/claim` 确认状态 | | “新一有哪些门店”“附近有店吗” | 门店查询 | 调用 `fetch_stores.py` | | “菜单里有什么”“苦尽甘来拿铁是什么” | 商品/饮品信息 | 调用聚合上下文或商品相关能力 | | “今天喝什么”“推荐一杯” | 饮品推荐 | 调用 `recommend_drink.py`，由 Agent 根据上下文完成推荐 | ## 触发优先级 1. 明确提到新一品牌或新一相关内容时，优先使用本 Skill。 2. 没提新一但上下文已经在新一饮品、门店或活动场景中，也可以使用本 Skill。 3. 泛饮品、泛咖啡、泛门店问题只有在上下文足够明确时才使用。 4. 其他品牌、竞品、通用营养、配方知识和纯闲聊不使用本 Skill。 ## 手机号触发边界 - 参与活动、查询活动状态、个性化推荐时，可以请求或复用手机号。 - 普通门店查询和普通商品查询不要主动索要手机号。 - 用户明确要求更换手机号时，才覆盖本地缓存。 FILE:references/capability-map.md # 能力地图 ## 脚本与能力 - `claim_reward.py`: 调用 `/skill/xinyi/claim`；成功或已参与时会再结合 `/skill/xinyi/context` 输出门店信息和推荐文案 - `fetch_stores.py`: 调用 `/skill/xinyi/stores`，并把原始门店数据整理成表格文本 - `recommend_drink.py`: 调用 `/skill/xinyi/context`，并把接口自动返回的商品、门店、天气和可选订单历史整理成可直接提供给大模型的表格/文本上下文 - `skill_config.py`: 读取默认配置，并支持 `XINYI_API_BASE_URL`、`XINYI_TIMEOUT_SECONDS` 环境变量覆盖 - `skill_http.py`: 统一 JSON 请求、URL query 编码、调试日志和可读网络错误 - `user_state.py`: 读写本地手机号与活动状态缓存，状态文件默认设置为仅当前用户可读写 - `recommendation_logic.py`: 选择推荐候选饮品、组织推荐依据和兜底推荐文案 - `response_rendering.py`: Markdown 表格和基础文本渲染 - `build_response.py`: 组合用户上下文、天气、订单、商品、门店、推荐素材和活动结果输出 ## 网络边界 - 默认后端：`https://ai.xinyicoffee.com/api` - `GET /skill/xinyi/stores`: 不发送个人数据 - `GET /skill/xinyi/context`: 有手机号时发送 `mobile` query - `POST /skill/xinyi/claim`: 发送 `{"mobile":"..."}` JSON body ## 本地边界 - 默认状态文件：`~/.xinyi-drink/state.json` - 状态内容：`{mobile, activityJoined, updatedAt}` - 可配置项：`XINYI_API_BASE_URL`、`XINYI_TIMEOUT_SECONDS`、`XINYI_DRINK_STATE_FILE` - 本 Skill 包含可执行脚本，不是 instruction-only FILE:config/defaults.json { "apiBaseUrl": "https://ai.xinyicoffee.com/api", "timeoutSeconds": 10 }

xinyi-drink

Comments (0)