Files
daily-robots/bot/skills_service.py
yumao f533e31adb refactor: Phase 1 抽出 shared/skills_data 解耦 daily 与 bot
daily 不再通过 sys.path 导入 bot/skills_service,skills feed 数据层上移到 shared/。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-03 14:45:39 +08:00

227 lines
7.1 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""skills.sh 快查命令解析与企微回复格式化。"""
from __future__ import annotations
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal
_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
sys.path.insert(0, str(_REPO_ROOT))
from shared.skills_data import Board, board_items, format_installs, load_feed, warm_feed_cache
__all__ = ["Command", "handle_command", "parse_command", "warm_feed_cache"]
@dataclass
class Command:
kind: Literal["help", "list", "search", "detail"]
board: Board = "trending"
limit: int = 10
query: str = ""
def _normalize_text(text: str) -> str:
text = re.sub(r"@\S+\s*", "", text)
return text.strip().lower()
def _parse_limit(raw: str | None, default: int = 10) -> int:
if not raw:
return default
try:
n = int(raw)
except ValueError:
return default
return max(1, min(n, 30))
def _match_list(raw: str, board: Board, aliases: str) -> Command | None:
m = re.match(rf"^({aliases})(?:\s+top)?\s*(\d+)?$", raw)
if m:
return Command(kind="list", board=board, limit=_parse_limit(m.group(2)))
m = re.match(rf"^(查|查询)\s+({aliases})(?:\s+top)?\s*(\d+)?$", raw)
if m:
return Command(kind="list", board=board, limit=_parse_limit(m.group(3)))
return None
def parse_command(text: str) -> Command:
raw = _normalize_text(text)
if not raw or raw in {"help", "帮助", "?", "h"}:
return Command(kind="help")
for board, aliases in (
("trending", "trending|趋势|top"),
("hot", "hot|实时|热门"),
("all", "all|总榜|alltime|all-time"),
):
cmd = _match_list(raw, board, aliases)
if cmd:
return cmd
m = re.match(r"^(search|搜索|find|查)\s+(.+)$", raw)
if m:
return Command(kind="search", query=m.group(2).strip(), limit=5)
m = re.match(r"^(detail|详情|skill|info)\s+(.+)$", raw)
if m:
return Command(kind="detail", query=m.group(2).strip())
if raw.startswith("trending") or raw.startswith("趋势"):
parts = raw.split(maxsplit=1)
return Command(kind="list", board="trending", limit=_parse_limit(parts[1] if len(parts) > 1 else None))
return Command(kind="search", query=raw, limit=5)
def _board_title(board: Board) -> str:
return {
"trending": "Trending近期增长",
"hot": "Hot实时热度",
"all": "All Time总安装榜",
}[board]
def format_list(board: Board, limit: int) -> str:
feed = load_feed()
items = board_items(feed, board)[:limit]
updated = feed.get("updatedAt", "未知")[:10]
lines = [
f"**skills.sh {_board_title(board)} Top {limit}**",
f"> 数据更新:{updated}",
"",
]
for i, item in enumerate(items, 1):
title = item.get("title", "?")
source = item.get("source", "?")
installs = format_installs(item.get("installs", 0))
desc = item.get("description", "")
if len(desc) > 80:
desc = desc[:77] + "..."
link = item.get("link", "")
lines.append(f"{i}. **{title}** · {installs}")
lines.append(f" `{source}`")
if desc:
lines.append(f" {desc}")
if link:
lines.append(f" [查看]({link})")
lines.append("")
return "\n".join(lines).strip()
def format_search(query: str, limit: int) -> str:
feed = load_feed()
q = query.lower()
seen: set[str] = set()
matches: list[dict[str, Any]] = []
for board in ("topTrending", "topHot", "topAllTime"):
for item in feed.get(board, []):
item_id = item.get("id") or item.get("title", "")
if item_id in seen:
continue
haystack = " ".join(
[
item.get("title", ""),
item.get("source", ""),
item.get("description", ""),
]
).lower()
if q in haystack:
seen.add(item_id)
matches.append(item)
if len(matches) >= limit:
break
if len(matches) >= limit:
break
if not matches:
return f"未找到与 **{query}** 相关的 skill。\n\n试试:`trending 10` / `hot 10` / `搜索 react`"
lines = [f"**搜索「{query}」** 共 {len(matches)}", ""]
for i, item in enumerate(matches, 1):
title = item.get("title", "?")
source = item.get("source", "?")
installs = format_installs(item.get("installs", 0))
link = item.get("link", "")
lines.append(f"{i}. **{title}** · {installs} · `{source}`")
if link:
lines.append(f" [查看]({link})")
return "\n".join(lines)
def format_detail(name: str) -> str:
feed = load_feed()
q = name.lower().strip()
best: dict[str, Any] | None = None
for board in ("topTrending", "topHot", "topAllTime"):
for item in feed.get(board, []):
title = (item.get("title") or "").lower()
item_id = (item.get("id") or "").lower()
if title == q or q in title or q in item_id:
if best is None or item.get("installs", 0) > best.get("installs", 0):
best = item
if not best:
return f"未找到 skill**{name}**\n\n试试:`搜索 {name}`"
desc = best.get("description", "无描述")
return "\n".join(
[
f"**{best.get('title', '?')}**",
f"`{best.get('source', '?')}`",
f"安装量:**{format_installs(best.get('installs', 0))}**",
"",
desc,
"",
f"[skills.sh 详情]({best.get('link', 'https://skills.sh')})",
"",
f"安装:`npx skills add {best.get('source', '')}/{best.get('title', '')}`",
]
)
def format_help() -> str:
return "\n".join(
[
"**Skills 助手 · 命令帮助**",
"",
"`trending 10` / `趋势 10` — 近期增长榜",
"`hot 10` / `实时 10` — 实时热度榜",
"`all 10` / `总榜 10` — 历史总安装榜",
"`搜索 react` / `search tdd` — 关键词搜索",
"`详情 find-skills` — 查看单个 skill",
"`preview` / `截图` / `预览` — 单页截图",
"`browser 场景名` — 执行 YAML 场景(见 bot/scenarios/",
"自然语言 — 如:访问登录页,输入账号密码,点击登录,点击智能体管理,截图",
"`preview /about 5173` — 指定路径和端口",
"",
"示例:",
"• trending top10",
"• 查 grill",
"• 详情 remotion-render",
]
)
def handle_command(text: str) -> str:
cmd = parse_command(text)
if cmd.kind == "help":
return format_help()
if cmd.kind == "list":
return format_list(cmd.board, cmd.limit)
if cmd.kind == "search":
return format_search(cmd.query, cmd.limit)
if cmd.kind == "detail":
return format_detail(cmd.query)
return format_help()