gen_ai_skill_sql.py 2.51 KB
"""把 skills/*.md 转成 ai_skill 注入 SQL(开发工作流:技能只维护 md 文件,DB 种子由本脚本生成)。

用法(uv,无需项目环境):
  uv run --no-project sql/gen_ai_skill_sql.py > sql/ai_skill_seed.sql
  docker exec -i mysql-saas mysql -uroot -plocal xlyweberp_saas < sql/ai_skill_seed.sql

语义 = 把 DB 里的**同名技能重置为文件版**(INSERT … ON DUPLICATE KEY UPDATE,按 sName 幂等)。
⚠️ 会覆盖用户在 DB 里对同名技能的自定义(bEnabled 保留不动);文件里没有的 DB 技能不受影响。

文件格式与生产 SkillService 同源:首行=技能名,次行=一句话用途,其余=正文。
超 600 token(中文≈1字/词元的保守口径)的技能拒绝生成(退出码 1)。
"""

import pathlib
import sys

ROOT = pathlib.Path(__file__).resolve().parent.parent
SKILLS_DIR = ROOT / "src/main/resources/skills"
TOKEN_LIMIT = 600


def estimate_tokens(s: str) -> int:
    cjk = sum(1 for c in s if ord(c) >= 0x2E80)
    other = len(s) - cjk
    return cjk + (other + 2) // 3


def q(s: str) -> str:
    return "'" + s.replace("\\", "\\\\").replace("'", "\\'").replace("\n", "\\n") + "'"


def main() -> int:
    files = sorted(SKILLS_DIR.glob("*.md"))
    if not files:
        print(f"-- 未找到技能文件:{SKILLS_DIR}", file=sys.stderr)
        return 1
    lines = [
        "-- 由 sql/gen_ai_skill_sql.py 生成,勿手改。语义 = 把 DB 同名技能重置为文件版(覆盖自定义,bEnabled 保留)。",
    ]
    for p in files:
        parts = p.read_text(encoding="utf-8").split("\n", 2)
        if len(parts) < 3 or not parts[0].strip() or not parts[2].strip():
            print(f"-- 跳过(格式不对,需 首行名/次行用途/正文):{p.name}", file=sys.stderr)
            continue
        name, brief, body = parts[0].strip(), parts[1].strip(), parts[2].strip()
        tokens = estimate_tokens(body)
        if tokens > TOKEN_LIMIT:
            print(f"技能「{name}」约 {tokens}t,超 {TOKEN_LIMIT}t 上限,拒绝生成", file=sys.stderr)
            return 1
        sid = "skill-" + p.stem
        lines.append(
            f"INSERT INTO ai_skill (sId,sName,sBrief,sBody,sUpdatePerson) "
            f"VALUES ({q(sid)},{q(name)},{q(brief)},{q(body)},'skills-md') "
            f"ON DUPLICATE KEY UPDATE sBrief=VALUES(sBrief), sBody=VALUES(sBody), "
            f"sUpdatePerson='skills-md', tUpdateDate=NOW();"
        )
    print("\n".join(lines))
    return 0


if __name__ == "__main__":
    sys.exit(main())