调整skill位置

This commit is contained in:
马宝龙
2026-07-28 15:58:05 +08:00
parent 5c3b306aae
commit 92b718c096
7 changed files with 112 additions and 36 deletions

View File

@@ -0,0 +1,129 @@
---
name: database-connect
description: 连接 MySQL 数据库并完成安全的查询、数据写入、更新、删除、表结构探查、事务处理和数据库诊断。用户要求 Codex 连接 MySQL、查看或修改数据库数据、分析表结构、执行 SQL、排查数据库连接或查询问题时使用。
---
# MySQL 数据库连接
## 目标
按照“发现配置、验证连接、了解结构、最小化操作、核对结果”的流程连接项目 MySQL并完成用户授权范围内的数据库操作。优先复用项目已有的驱动、配置和命令不擅自安装依赖或改变数据库环境。
## 安全边界
- 将数据库凭据视为机密。读取 `.env``.env.example`配置,只提取连接所需值,绝不在回复、日志、补丁或命令输出中打印密码、完整连接串或令牌。
- 禁止把密码直接写在命令行参数、SQL、源码或新建文件中。优先使用项目现有的安全配置、环境变量、凭据助手或 MySQL 客户端配置文件。
- 默认使用只读方式探查:`SELECT``SHOW``DESCRIBE``EXPLAIN``information_schema` 查询可以先执行;不要把探查语句改成写入语句。
- 执行 `INSERT``UPDATE``DELETE``REPLACE``TRUNCATE``DROP``ALTER``CREATE`、批量导入、权限变更或存储过程时,先说明目标、范围、预计影响和事务策略。
- 对删除、清空、改表、批量更新等不可逆或高影响操作,先展示将执行的精确 SQL敏感值脱敏并请求用户确认用户已在当前请求中明确给出精确语句和范围时可视为已确认但仍先做影响评估。
- 不执行未限定范围的 `UPDATE``DELETE`,不执行生产库上的 `DROP DATABASE``TRUNCATE` 或全表更新,除非用户明确确认精确目标和范围。
- 不为了“修复连接”而修改防火墙、开放公网端口、重置密码、创建高权限账号或暴露数据库服务;这些需要单独授权。
## 标准工作流
### 1. 发现连接方式
按以下顺序检查,找到一种可用方式后停止扩散搜索:
1. 查看项目的 `README`、数据库迁移配置和应用配置,识别 host、port、database、user 的来源。密码只在本地使用,不回显。
2. 检查当前环境变量,如 `MYSQL_HOST``MYSQL_PORT``MYSQL_DATABASE``MYSQL_USER``MYSQL_PASSWORD` 或项目自定义变量。优先确认变量是否存在,不输出其值。
3. 检查项目依赖和现有数据库访问代码。Java 项目优先复用 JDBC/迁移工具Node.js 项目优先复用现有 MySQL 驱动Python 项目优先复用已安装的驱动。
4. 检查 `mysql` 客户端是否可用及版本。客户端不可用时,先报告缺少的工具,并优先使用项目已有驱动;不要未经授权下载或安装软件。
连接参数不完整时,只询问缺少的参数。接受用户提供的 `host``port`、数据库名和用户名,但不要要求用户把密码粘贴到聊天中;建议通过环境变量或本地凭据配置提供密码。
### 2. 验证连接
建立连接后先执行轻量验证,不直接操作业务数据:
```sql
SELECT DATABASE() AS current_database,
USER() AS connected_user,
VERSION() AS mysql_version;
```
验证以下事项:
- 实际连接的主机、端口、数据库和账号是否符合目标;不要仅凭配置文件推断已经连对。
- 当前账号是否有完成任务所需的最小权限。
- 是否存在只读副本、开发库和生产库混淆的风险。发现环境不明确时暂停写操作并询问。
- 字符集和时区是否可能影响结果,必要时读取 `SELECT @@character_set_connection, @@time_zone`
连接失败时按错误码和现象诊断:先区分 DNS/网络不可达、端口拒绝、认证失败、数据库不存在、权限不足、TLS 配置错误和客户端/驱动缺失;不要用反复重试掩盖根因。记录可公开的错误类型和下一步,不回显凭据。
### 3. 探查结构和数据
在写操作前确认表、列、键和约束。按需执行:
```sql
SHOW TABLES;
SHOW CREATE TABLE `table_name`;
SELECT TABLE_NAME, TABLE_ROWS
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = DATABASE();
SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_KEY, COLUMN_DEFAULT
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'table_name'
ORDER BY ORDINAL_POSITION;
```
- 从实际结构确认表名、列名、主键、唯一键、外键、默认值和软删除字段;不要根据名称猜测 schema。
- 样本数据只取必要列并加 `LIMIT`,避免 `SELECT *` 扫描大表。需要总量时优先 `COUNT(*)`,需要性能判断时使用 `EXPLAIN`
- 识别敏感列(密码、身份证、手机号、令牌、支付信息等),查询结果脱敏或只返回聚合结果。
- 表名和列名不能用普通 SQL 参数绑定。对动态标识符只允许来自已探查且白名单校验通过的名称,并使用 MySQL 反引号转义。
### 4. 执行查询或修改
- 复杂任务拆成可验证的小步骤,每一步说明目的和结果。
- 所有用户输入值使用驱动的参数绑定(如 `?``:name`),不要拼接值生成 SQL。只在无法参数化的标识符上使用白名单。
- 先用等价的 `SELECT` 验证筛选条件,再执行 `UPDATE`/`DELETE`。修改前记录主键集合或预估影响行数。
- 查询大表时必须有合理的过滤条件、索引或分页;避免在未了解数据量时执行全表排序、笛卡尔积和无条件扫描。
- 对写操作优先使用事务:开始事务,执行变更,检查受影响行数和关键结果,确认无误后提交;异常时回滚。不要在无法确认事务状态时重复提交。
- 批量操作分批执行并记录每批结果。遇到锁等待、死锁、超时或影响行数异常时停止后续批次,回滚当前事务并报告。
- DDL 可能隐式提交,执行前明确告知这一点;涉及表结构变更时先备份/导出或确认已有恢复方案。
通用写操作模板:
```sql
START TRANSACTION;
-- 先用同一 WHERE 条件确认目标和数量
SELECT `primary_key`
FROM `table_name`
WHERE ...
LIMIT 1000;
-- 使用参数绑定执行精确变更
UPDATE `table_name`
SET `column_name` = ?
WHERE `primary_key` IN (...);
-- 核对受影响行数和结果后再提交
COMMIT;
-- 任一步失败时执行 ROLLBACK;
```
### 5. 核对并汇报
操作完成后重新查询关键结果,核对实际影响行数、唯一性、关联完整性和业务约束。向用户报告:使用的连接环境(不含密码)、执行的操作、结果摘要、是否提交事务、异常或未验证事项;不要粘贴大批量原始数据。
## 工具选择
- 已有应用或迁移工具能安全提供连接时,优先使用它,避免引入第二套连接配置。
- 使用 `mysql` CLI 时,把密码交给安全凭据机制或交互式输入,不使用 `--password=明文`。长 SQL 用受控的临时文件或标准输入传递,并在使用后清理临时文件。
- 使用脚本时复用项目锁定的依赖和配置,设置连接超时、查询超时(若驱动支持)和结果上限;脚本结束时关闭连接。
- 不把结果写入仓库,不提交临时凭据,不把生产数据导出到工作区。确需导出时先确认文件位置、字段范围和保留期限。
## 常见问题处理
- `Access denied`: 核对实际用户名、认证插件、目标主机和账号来源;不要通过授予 `ALL PRIVILEGES` 规避权限问题。
- `Unknown database`: 确认数据库名和当前连接环境,先列出允许访问的数据库或检查 Compose/迁移配置。
- `Table doesn't exist`: 核对大小写、schema、迁移状态和连接库不要直接创建同名表覆盖问题。
- `Lock wait timeout` 或死锁: 回滚当前事务,识别长事务和索引缺失,缩小批次;不要盲目重试写入。
- 字符集乱码: 检查连接、表和列的字符集与排序规则,先读取并评估已有数据,再决定是否迁移。
- 结果数量异常: 重新检查 JOIN 条件、NULL 语义、时区、软删除条件和分页边界,并保留可复现的只读查询。
## 完成标准
只有在连接目标已验证、操作范围已确认、SQL 使用参数绑定、写操作结果已核对且事务状态明确时,才报告数据库任务完成。任何凭据、环境或影响范围无法确认时,保持只读并明确阻塞原因。

View File

@@ -0,0 +1,4 @@
interface:
display_name: "MySQL 数据库连接"
short_description: "连接 MySQL 数据库并安全执行查询、写入与结构操作"
default_prompt: "使用 $database-connect 连接项目 MySQL 数据库,先检查连接和表结构,再按要求安全执行数据库操作。"

View File

@@ -0,0 +1,67 @@
---
name: fullstack-start-verify
description: 启动本地项目的前端和后端服务并通过进程状态、HTTP 健康检查、接口检查和可用时的浏览器冒烟测试验证整体功能。用于 Codex 需要运行全栈项目、验证前后端联调、排查启动失败,或确认用户流程是否端到端可用的场景。
---
# 全栈启动与验证
当用户要求启动前后端、验证整体功能、跑通本地联调或确认项目是否可用时使用本技能。目标是得到可复现的验证结论,而不是只确认某个进程曾经启动过。
## 工作流程
1. 在启动前检查项目。
- 阅读适用的 `AGENTS.md``README.md`、根目录环境变量示例,以及前后端的项目清单文件。
- 确认实际使用的包管理器和启动命令。优先使用项目文档和项目自带包装器中的命令,例如 `npm``pnpm``yarn``mvnw``gradlew``uv` 等。
- 找到后端健康检查地址、前端地址、API 基础地址、所需的数据库或缓存服务,以及文档中提供的测试账号。
- 不要为了让检查通过而虚构凭据、覆盖 `.env`、执行迁移或重置数据。
2. 执行启动前检查。
- 确认所需运行时和依赖目录可用。前端已有 `node_modules` 时直接启动,不要每次重复执行 `npm ci`;仅在依赖缺失或用户明确要求重装时安装。
- 检查计划使用的端口。只有在进程和健康响应明确属于当前项目时才复用服务,绝不要结束未知进程。
- 仅在项目文档说明了启动方式且用户已将其纳入范围时,启动 MySQL 或 Redis 等基础设施。
- 如果缺少必要的密钥、数据库或账号,应报告阻塞原因,不要降低验证标准。
3. 使用 `scripts/start_and_verify.py` 启动两个应用服务。脚本默认使用 0.25 秒快速轮询,并行等待两个服务就绪。
- 从项目根目录运行脚本。
- 传入明确的工作目录、启动命令和地址。对于健康检查之外的重要 API 路由,使用 `--check-url label=url` 添加只读检查;不要重复添加已经作为 `--backend-url` 的健康地址。
- 后端应使用项目专用的健康检查地址,而不是只检查 TCP 端口是否打开;前端应检查开发服务器根路径或已知路由。
- 脚本只管理和清理自己启动的进程。只有用户明确要求服务继续运行时,才使用 `--keep-running`
- 排查失败时使用 `--keep-logs` 保留日志。避免把密钥放在命令行参数中。
Example:
```powershell
python skills/fullstack-start-verify/scripts/start_and_verify.py `
--backend-cmd "mvn -f backend/pom.xml spring-boot:run" `
--backend-dir . `
--backend-url http://127.0.0.1:8080/api/v1/health `
--frontend-cmd "npm run dev -- --host 127.0.0.1" `
--frontend-dir frontend `
--frontend-url http://127.0.0.1:5173/ `
--check-url summary=http://127.0.0.1:8080/api/v1/users/summary
```
4. 按递进层级验证应用。
- 快速路径先确认两个健康检查地址和 1 至 2 个主要只读 API脚本会并行执行等待和额外检查。
- 只有代码变更、用户明确要求回归,或快速路径暴露问题时,才执行前端构建、后端测试等较慢命令。
- 只有涉及页面或交互变更且有浏览器工具时,才执行浏览器冒烟:加载页面、完成最小有意义的操作,同时验证页面可见结果和网络/API 结果。
- 除非用户明确要求写入流程且测试数据安全,否则不执行新增、修改、删除或数据库迁移。
- 如果没有浏览器工具,应明确说明未验证 UI 交互;仅凭 HTTP 检查不能声称完成了端到端验证。
5. 汇报并清理环境。
- 汇报验证任务 ID、具体命令、地址、执行的检查、通过或失败状态以及第一个可执行的失败原因。
- 分开汇报基础设施、后端、前端和浏览器结果,明确哪些部分已通过。
- 失败时给出日志位置或相关日志尾部,同时隐藏密码、令牌和连接字符串。
- 除非用户要求服务持续运行,否则在结束前确认脚本启动的进程已经停止。
## 失败处理
- 服务在地址就绪前退出,属于启动失败。重试前先检查捕获的日志。
- 超时不算通过。检查依赖是否可用、端口是否冲突、环境变量是否加载、代理配置和服务日志。
- 后端健康但前端 API 调用失败,属于集成失败。检查前端代理或基础地址,以及 CORS 配置。
- 前端页面能加载但浏览器操作失败,属于用户流程失败。记录准确路由、操作、响应状态和控制台错误。
- 验证期间不要修复无关代码、修改生产配置或删除数据,除非用户明确扩大任务范围。
## 内置脚本
`scripts/start_and_verify.py` 是一个不依赖第三方库的进程运行器,负责本流程中的启动和 HTTP 检查。脚本会为检查请求添加 `X-Request-Id`,输出验证任务 ID并行等待服务与执行额外检查。使用不常见选项前先阅读 `--help` 输出。脚本默认创建临时日志并在清理后删除;排查问题时传入 `--keep-logs` 保留日志。错误日志摘要会脱敏,但保留日志前仍需确认服务自身没有输出敏感信息。

View File

@@ -0,0 +1,4 @@
interface:
display_name: "全栈启动验证"
short_description: "自动启动前后端服务并完成健康检查与端到端功能验证"
default_prompt: "使用 $fullstack-start-verify 启动项目的前后端,执行健康检查和端到端验证并汇报结果。"

View File

@@ -0,0 +1,332 @@
#!/usr/bin/env python3
"""启动两个本地服务,等待 HTTP 就绪后执行检查并清理进程。"""
from __future__ import annotations
import argparse
import concurrent.futures
import os
import re
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
import uuid
from dataclasses import dataclass
from pathlib import Path
@dataclass
class Service:
name: str
command: str
cwd: Path
url: str
process: subprocess.Popen[bytes] | None = None
log_path: Path | None = None
log_file: object | None = None
SENSITIVE_VALUE_PATTERN = re.compile(
r"(?i)(password|token|secret|authorization|api[_-]?key)"
r"(\s*[\"']?\s*[:=]\s*[\"']?)([^\"'\s,;}\]]+)"
)
BEARER_PATTERN = re.compile(r"(?i)(Bearer\s+)[^\s,;]+")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="启动前后端命令,检查 HTTP 地址并清理进程。"
)
parser.add_argument("--backend-cmd", required=True, help="后端启动命令")
parser.add_argument("--backend-dir", default=".", help="后端工作目录")
parser.add_argument("--backend-url", required=True, help="后端健康检查地址")
parser.add_argument("--frontend-cmd", required=True, help="前端启动命令")
parser.add_argument("--frontend-dir", default=".", help="前端工作目录")
parser.add_argument("--frontend-url", required=True, help="前端就绪检查地址")
parser.add_argument(
"--check-url",
action="append",
default=[],
metavar="LABEL=URL",
help="额外的 GET 检查,可重复传入",
)
parser.add_argument(
"--timeout",
type=float,
default=60.0,
help="每个服务的就绪超时时间,单位为秒,默认 60",
)
parser.add_argument(
"--interval",
type=float,
default=0.25,
help="每次就绪检查的间隔秒数,默认 0.25",
)
parser.add_argument(
"--check-timeout",
type=float,
default=5.0,
help="额外接口检查的单请求超时时间,默认 5 秒",
)
parser.add_argument(
"--log-dir",
type=Path,
help="日志目录,运行结束后保留",
)
parser.add_argument(
"--keep-logs",
action="store_true",
help="运行结束后保留临时日志",
)
parser.add_argument(
"--keep-running",
action="store_true",
help="验证结束后保持已启动的服务运行",
)
return parser.parse_args()
def resolve_dir(value: str) -> Path:
path = Path(value).resolve()
if not path.is_dir():
raise RuntimeError(f"工作目录不存在:{path}")
return path
def start_service(service: Service, log_dir: Path) -> None:
service.log_path = log_dir / f"{service.name}.log"
service.log_file = service.log_path.open("w", encoding="utf-8", buffering=1)
creationflags = 0
start_new_session = False
if os.name == "nt":
creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
else:
start_new_session = True
try:
service.process = subprocess.Popen(
service.command,
cwd=service.cwd,
shell=True,
stdout=service.log_file,
stderr=subprocess.STDOUT,
creationflags=creationflags,
start_new_session=start_new_session,
)
except Exception:
service.log_file.close()
service.log_file = None
raise
print(f"启动 {service.name}:进程={service.process.pid},工作目录={service.cwd}")
def request(url: str, timeout: float, request_id: str) -> tuple[bool, str]:
request_obj = urllib.request.Request(
url,
headers={
"User-Agent": "fullstack-start-verify/1.1",
"X-Request-Id": request_id,
},
)
try:
with urllib.request.urlopen(request_obj, timeout=timeout) as response:
status = response.status
return 200 <= status < 400, str(status)
except urllib.error.HTTPError as error:
return 200 <= error.code < 400, str(error.code)
except (urllib.error.URLError, TimeoutError, OSError) as error:
reason = getattr(error, "reason", error)
return False, type(reason).__name__
def wait_for_service(
service: Service, timeout: float, interval: float, run_id: str
) -> None:
if service.process is None:
raise RuntimeError(f"{service.name} 未启动")
deadline = time.monotonic() + timeout
last_reason = "not checked"
while time.monotonic() < deadline:
exit_code = service.process.poll()
if exit_code is not None:
raise RuntimeError(f"{service.name} 在就绪检查前退出,退出码为 {exit_code}")
request_timeout = min(max(interval * 2, 0.5), 5.0)
ok, reason = request(service.url, request_timeout, f"{run_id}-{service.name}")
last_reason = reason
if ok:
print(f"通过 {service.name}{service.url} [{reason}]")
return
time.sleep(min(max(interval, 0.05), max(deadline - time.monotonic(), 0)))
raise RuntimeError(f"{service.name}{timeout:g} 秒内未就绪 [{last_reason}]")
def parse_extra_checks(values: list[str]) -> list[tuple[str, str]]:
checks: list[tuple[str, str]] = []
for value in values:
if "=" in value:
label, url = value.split("=", 1)
else:
label, url = value, value
label = label.strip() or url.strip()
url = url.strip()
if not url.startswith(("http://", "https://")):
raise RuntimeError(f"检查地址无效:{url}")
checks.append((label, url))
return checks
def wait_for_services(
services: list[Service], timeout: float, interval: float, run_id: str
) -> None:
"""并行等待服务,整体耗时取最慢服务的就绪时间。"""
executor = concurrent.futures.ThreadPoolExecutor(max_workers=len(services))
failed = True
try:
futures = [
executor.submit(wait_for_service, service, timeout, interval, run_id)
for service in services
]
done, _ = concurrent.futures.wait(
futures, return_when=concurrent.futures.FIRST_EXCEPTION
)
for future in done:
future.result()
failed = False
finally:
executor.shutdown(wait=not failed, cancel_futures=failed)
def run_extra_checks(
checks: list[tuple[str, str]], timeout: float, run_id: str
) -> None:
"""并行执行只读接口检查,避免多个接口检查串行等待。"""
if not checks:
return
def execute_check(
item: tuple[int, tuple[str, str]]
) -> tuple[str, str, tuple[bool, str]]:
index, (label, url) = item
return label, url, request(url, timeout, f"{run_id}-check-{index}")
with concurrent.futures.ThreadPoolExecutor(max_workers=len(checks)) as executor:
results = list(executor.map(execute_check, enumerate(checks, start=1)))
for label, url, (ok, reason) in results:
if not ok:
raise RuntimeError(f"检查失败:{label} {url} [{reason}]")
print(f"通过 {label}{url} [{reason}]")
def stop_service(service: Service) -> None:
process = service.process
if process is None or process.poll() is not None:
return
if os.name == "nt":
try:
subprocess.run(
["taskkill", "/PID", str(process.pid), "/T", "/F"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
timeout=2,
)
except subprocess.TimeoutExpired:
pass
try:
process.wait(timeout=1)
except subprocess.TimeoutExpired:
process.kill()
try:
process.wait(timeout=1)
except subprocess.TimeoutExpired:
process.returncode = -1
except OSError:
# taskkill 已经结束进程,但 Windows 可能同时使句柄失效。
process.returncode = -1
else:
try:
os.killpg(process.pid, signal.SIGTERM)
except ProcessLookupError:
return
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
os.killpg(process.pid, signal.SIGKILL)
def tail(path: Path | None, lines: int = 20) -> list[str]:
if path is None or not path.exists():
return []
return path.read_text(encoding="utf-8", errors="replace").splitlines()[-lines:]
def redact(line: str) -> str:
"""仅在终端输出日志时脱敏,避免错误摘要泄露凭据。"""
line = SENSITIVE_VALUE_PATTERN.sub(r"\1\2[REDACTED]", line)
return BEARER_PATTERN.sub(r"\1[REDACTED]", line)
def main() -> int:
args = parse_args()
if args.timeout <= 0 or args.interval <= 0 or args.check_timeout <= 0:
raise SystemExit("timeout、interval 和 check-timeout 必须大于 0")
run_id = uuid.uuid4().hex[:12]
print(f"验证任务 ID{run_id}")
checks = parse_extra_checks(args.check_url)
temp_log_dir: Path | None = None
log_dir = args.log_dir
if log_dir is None:
temp_log_dir = Path(tempfile.mkdtemp(prefix="fullstack-start-verify-"))
log_dir = temp_log_dir
log_dir = log_dir.resolve()
log_dir.mkdir(parents=True, exist_ok=True)
services = [
Service("backend", args.backend_cmd, resolve_dir(args.backend_dir), args.backend_url),
Service("frontend", args.frontend_cmd, resolve_dir(args.frontend_dir), args.frontend_url),
]
failure: Exception | None = None
try:
for service in services:
start_service(service, log_dir)
wait_for_services(services, args.timeout, args.interval, run_id)
run_extra_checks(checks, args.check_timeout, run_id)
print(f"全栈验证通过FULLSTACK_VERIFY=PASS任务 ID{run_id}")
return 0
except (OSError, RuntimeError, ValueError) as error:
failure = error
print(f"全栈验证失败:{error}", file=sys.stderr)
for service in services:
for line in tail(service.log_path):
print(f"日志 {service.name}{redact(line)}", file=sys.stderr)
return 1
finally:
if not args.keep_running:
for service in reversed(services):
stop_service(service)
for service in services:
if service.log_file is not None:
service.log_file.close()
if args.keep_running:
print(f"服务仍在运行,日志目录:{log_dir}")
elif temp_log_dir is not None and not args.keep_logs:
shutil.rmtree(temp_log_dir, ignore_errors=True)
elif temp_log_dir is not None:
print(f"日志已保留:{temp_log_dir}")
elif failure is not None:
print(f"日志已保留:{log_dir}")
if __name__ == "__main__":
raise SystemExit(main())