wip: snapshot das alterações não commitadas (2026-10-20)
Origem: D:/Documents/Github/dev-workflows Arquivos omitidos (segredos, bancos de dados, logs, >50 MB): 0
This commit is contained in:
1 parent
c17192c9fe
commit
2ef6f6be10
61 files changed
+7892
No files matched your search
@@ -0,0 +1,48 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PreToolUse guard for shell commands: denies gate bypasses, asks before destructive git/db commands.
|
||||
|
||||
Claude Code: reads the hook JSON on stdin and prints a permissionDecision.
|
||||
opencode: `bash_guard.py --check "<command>"` prints the reason and exits 2 when the command must not run.
|
||||
"""
|
||||
import json, re, sys
|
||||
|
||||
DENY = [
|
||||
(r"(^|\s)--no-verify(\s|$)", "--no-verify pula os hooks do git. Corrija o que o hook reclamou em vez de pular."),
|
||||
(r"\bgit\s+push\b(?=.*\s(--force|-f)(\s|$))(?!.*--force-with-lease)", "push --force reescreve histórico remoto. Use --force-with-lease e só com pedido explícito do usuário."),
|
||||
(r"\brm\s+(-[a-zA-Z]*r[a-zA-Z]*f|-[a-zA-Z]*f[a-zA-Z]*r)[a-zA-Z]*\s+(/|~|\$HOME|/\*|\.\.)(\s|/?$)", "rm -rf na raiz, no home ou no diretório pai não roda."),
|
||||
]
|
||||
ASK = [
|
||||
(r"\bgit\s+reset\s+--hard\b", "git reset --hard descarta trabalho não commitado."),
|
||||
(r"\bgit\s+clean\s+-[a-zA-Z]*f", "git clean -f apaga arquivos não rastreados."),
|
||||
(r"\bgit\s+(checkout|restore)\s+(--\s+)?\.(\s|$)", "descarta todas as mudanças locais."),
|
||||
(r"\b(DROP\s+(DATABASE|TABLE|SCHEMA)|TRUNCATE\s+TABLE)\b", "comando destrutivo de banco de dados."),
|
||||
]
|
||||
|
||||
|
||||
def decide(command):
|
||||
for pattern, why in DENY:
|
||||
if re.search(pattern, command, re.IGNORECASE):
|
||||
return "deny", why
|
||||
for pattern, why in ASK:
|
||||
if re.search(pattern, command, re.IGNORECASE):
|
||||
return "ask", why
|
||||
return None, None
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) > 2 and sys.argv[1] == "--check":
|
||||
decision, why = decide(sys.argv[2])
|
||||
if decision: # opencode has no "ask": both block, the agent must ask the user
|
||||
print(f"[dev-workflows] {why} Peça confirmação ao usuário.")
|
||||
return 2
|
||||
return 0
|
||||
event = json.load(sys.stdin)
|
||||
decision, why = decide(event.get("tool_input", {}).get("command", ""))
|
||||
if decision:
|
||||
print(json.dumps({"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": decision,
|
||||
"permissionDecisionReason": f"[dev-workflows] {why}"}}, ensure_ascii=False))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,58 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PostToolUse: formats the edited file with the formatter the project already uses. Never fails the edit.
|
||||
|
||||
Claude Code: reads tool_input.file_path from the hook JSON on stdin.
|
||||
opencode: `post_edit.py --file <path>`.
|
||||
"""
|
||||
import json, os, shutil, subprocess, sys
|
||||
|
||||
PRETTIER_EXT = {".js", ".jsx", ".ts", ".tsx", ".mjs", ".cjs", ".json", ".css", ".scss", ".html", ".vue", ".svelte", ".md", ".yaml", ".yml"}
|
||||
|
||||
|
||||
def project_root(path):
|
||||
d = os.path.dirname(os.path.abspath(path))
|
||||
while d != os.path.dirname(d):
|
||||
if any(os.path.exists(os.path.join(d, m)) for m in ("package.json", "pyproject.toml", "go.mod", "Cargo.toml", ".git")):
|
||||
return d
|
||||
d = os.path.dirname(d)
|
||||
return None
|
||||
|
||||
|
||||
def uses_ruff(root):
|
||||
if os.path.exists(os.path.join(root, "ruff.toml")) or os.path.exists(os.path.join(root, ".ruff.toml")):
|
||||
return True
|
||||
pyproject = os.path.join(root, "pyproject.toml")
|
||||
return os.path.exists(pyproject) and "[tool.ruff" in open(pyproject).read()
|
||||
|
||||
|
||||
def formatter(path, root):
|
||||
ext = os.path.splitext(path)[1].lower()
|
||||
prettier = os.path.join(root, "node_modules", ".bin", "prettier")
|
||||
if ext in PRETTIER_EXT and os.path.exists(prettier):
|
||||
return [prettier, "--write", "--log-level", "warn", path]
|
||||
if ext == ".py" and shutil.which("ruff") and uses_ruff(root):
|
||||
return ["ruff", "format", "--quiet", path]
|
||||
if ext == ".go" and shutil.which("gofmt"):
|
||||
return ["gofmt", "-w", path]
|
||||
if ext == ".rs" and shutil.which("rustfmt"):
|
||||
return ["rustfmt", path]
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) > 2 and sys.argv[1] == "--file":
|
||||
path = sys.argv[2]
|
||||
else:
|
||||
path = json.load(sys.stdin).get("tool_input", {}).get("file_path", "")
|
||||
root = project_root(path) if path and os.path.isfile(path) else None
|
||||
cmd = formatter(path, root) if root else None
|
||||
if cmd:
|
||||
try:
|
||||
subprocess.run(cmd, cwd=root, capture_output=True, timeout=30)
|
||||
except (subprocess.TimeoutExpired, OSError):
|
||||
pass
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,70 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Stop hook: blocks "done" while uncommitted code changes lack a fresh, non-failing quality-gate report.
|
||||
|
||||
Claude Code: reads the hook JSON on stdin, prints {"decision": "block", ...} to keep the agent working.
|
||||
opencode: `stop_gate.py --check <dir>` prints the reason and exits 2 when it would block.
|
||||
Disable with DW_GATE=off.
|
||||
"""
|
||||
import json, os, subprocess, sys
|
||||
|
||||
DOC_EXT = {".md", ".mdx", ".txt", ".rst", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"}
|
||||
PLUGIN_ROOT = os.environ.get("CLAUDE_PLUGIN_ROOT") or os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
GATE = os.path.join(PLUGIN_ROOT, "scripts", "quality-gate")
|
||||
|
||||
|
||||
def pending_code_files(cwd):
|
||||
p = subprocess.run(["git", "status", "--porcelain", "--untracked-files=all"], cwd=cwd, capture_output=True, text=True)
|
||||
if p.returncode != 0:
|
||||
return None
|
||||
files = []
|
||||
for line in p.stdout.splitlines():
|
||||
path = line[3:].split(" -> ")[-1].strip('"')
|
||||
if line[:2].strip() == "D" or path.startswith(".dev-workflows/"):
|
||||
continue
|
||||
if os.path.splitext(path)[1].lower() not in DOC_EXT and os.path.isfile(os.path.join(cwd, path)):
|
||||
files.append(path)
|
||||
return files
|
||||
|
||||
|
||||
def reason_to_block(cwd):
|
||||
if os.environ.get("DW_GATE") == "off":
|
||||
return None
|
||||
top = subprocess.run(["git", "rev-parse", "--show-toplevel"], cwd=cwd, capture_output=True, text=True)
|
||||
if top.returncode != 0:
|
||||
return None
|
||||
root = top.stdout.strip()
|
||||
files = pending_code_files(root)
|
||||
if not files:
|
||||
return None
|
||||
run = f'"{GATE}" --tier <nível do dev-router>'
|
||||
report = os.path.join(root, ".dev-workflows", "gate.json")
|
||||
if not os.path.isfile(report):
|
||||
return f"Há {len(files)} arquivo(s) de código alterados sem gate de qualidade. Rode {run} e trate o resultado antes de encerrar."
|
||||
newest = max(os.path.getmtime(os.path.join(root, f)) for f in files)
|
||||
if os.path.getmtime(report) < newest:
|
||||
return f"O código mudou depois do último gate. Rode de novo {run} antes de encerrar."
|
||||
data = json.load(open(report))
|
||||
if data.get("verdict") == "fail":
|
||||
failed = ", ".join(c["name"] for c in data.get("checks", []) if c.get("status") == "fail")
|
||||
return f"O gate falhou em: {failed}. Corrija (veja .dev-workflows/gate.json) ou explique ao usuário por que não dá para corrigir."
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) > 2 and sys.argv[1] == "--check":
|
||||
why = reason_to_block(sys.argv[2])
|
||||
if why:
|
||||
print(why)
|
||||
return 2
|
||||
return 0
|
||||
event = json.load(sys.stdin)
|
||||
if event.get("stop_hook_active"):
|
||||
return 0
|
||||
why = reason_to_block(event.get("cwd") or os.getcwd())
|
||||
if why:
|
||||
print(json.dumps({"decision": "block", "reason": why}, ensure_ascii=False))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,99 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Install dev-workflows into Claude Code, opencode or Hermes Agent.
|
||||
|
||||
python3 scripts/install.py claude # plugin via local marketplace (skills, agents, hooks)
|
||||
python3 scripts/install.py opencode # skills + agents + plugin under ~/.config/opencode
|
||||
python3 scripts/install.py hermes # skills under ~/.hermes/skills (no agents/hooks there)
|
||||
|
||||
opencode and Hermes do not substitute ${CLAUDE_PLUGIN_ROOT}, so the copies point at this checkout:
|
||||
keep it where it is, or re-run the install after moving it.
|
||||
"""
|
||||
import argparse, os, re, shutil, subprocess, sys
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
MARKER = ".dev-workflows-install"
|
||||
TEXT_EXT = {".md", ".txt", ".json", ".toml", ".yaml", ".yml"}
|
||||
|
||||
|
||||
def copy_tree(src, dst, force):
|
||||
if os.path.exists(dst) and not os.path.exists(os.path.join(dst, MARKER)) and not force:
|
||||
sys.exit(f"{dst} já existe e não foi instalado por este script. Use --force para sobrescrever.")
|
||||
shutil.rmtree(dst, ignore_errors=True)
|
||||
shutil.copytree(src, dst)
|
||||
for dirpath, _, names in os.walk(dst):
|
||||
for n in names:
|
||||
if os.path.splitext(n)[1] in TEXT_EXT:
|
||||
p = os.path.join(dirpath, n)
|
||||
text = open(p, encoding="utf-8", errors="ignore").read()
|
||||
if "${CLAUDE_PLUGIN_ROOT}" in text:
|
||||
open(p, "w", encoding="utf-8").write(text.replace("${CLAUDE_PLUGIN_ROOT}", ROOT))
|
||||
open(os.path.join(dst, MARKER), "w").write(ROOT + "\n")
|
||||
|
||||
|
||||
def install_skills(target_dir, force):
|
||||
os.makedirs(target_dir, exist_ok=True)
|
||||
names = sorted(d for d in os.listdir(os.path.join(ROOT, "skills"))
|
||||
if os.path.isfile(os.path.join(ROOT, "skills", d, "SKILL.md")))
|
||||
for name in names:
|
||||
copy_tree(os.path.join(ROOT, "skills", name), os.path.join(target_dir, name), force)
|
||||
print(f"{len(names)} skills em {target_dir}")
|
||||
|
||||
|
||||
def opencode_agent(path):
|
||||
"""Claude Code agent frontmatter -> opencode subagent frontmatter."""
|
||||
text = open(path, encoding="utf-8").read()
|
||||
m = re.match(r"---\n(.*?)\n---\n(.*)", text, re.S)
|
||||
meta = dict(re.findall(r"^(\w+):\s*(.*)$", m.group(1), re.M))
|
||||
tools = {t.strip() for t in meta.get("tools", "").split(",")}
|
||||
skills = re.findall(r"[\w-]+", meta.get("skills", ""))
|
||||
perm = {"edit": "allow" if tools & {"Edit", "Write"} else "deny",
|
||||
"bash": "allow" if "Bash" in tools else "deny",
|
||||
"webfetch": "allow" if "WebFetch" in tools else "deny"}
|
||||
body = m.group(2).replace("${CLAUDE_PLUGIN_ROOT}", ROOT)
|
||||
if skills:
|
||||
body = f"Antes de começar, carregue a(s) skill(s): {', '.join(skills)}.\n\n" + body
|
||||
head = ["# dev-workflows", f"description: {meta['description']}", "mode: subagent", "permission:"]
|
||||
head += [f" {k}: {v}" for k, v in perm.items()]
|
||||
return "---\n" + "\n".join(head) + "\n---\n\n" + body
|
||||
|
||||
|
||||
def install_opencode(force):
|
||||
base = os.path.expanduser("~/.config/opencode")
|
||||
install_skills(os.path.join(base, "skills"), force)
|
||||
agents_dir = os.path.join(base, "agents")
|
||||
os.makedirs(agents_dir, exist_ok=True)
|
||||
for f in sorted(os.listdir(os.path.join(ROOT, "agents"))):
|
||||
dst = os.path.join(agents_dir, f)
|
||||
if os.path.exists(dst) and "dev-workflows" not in open(dst, encoding="utf-8").read() and not force:
|
||||
sys.exit(f"{dst} já existe e não é deste pacote. Use --force para sobrescrever.")
|
||||
open(dst, "w", encoding="utf-8").write(opencode_agent(os.path.join(ROOT, "agents", f)))
|
||||
plugins_dir = os.path.join(base, "plugins")
|
||||
os.makedirs(plugins_dir, exist_ok=True)
|
||||
plugin = open(os.path.join(ROOT, "adapters", "opencode", "dev-workflows.ts"), encoding="utf-8").read()
|
||||
open(os.path.join(plugins_dir, "dev-workflows.ts"), "w", encoding="utf-8").write(plugin.replace("__DW_ROOT__", ROOT))
|
||||
print(f"agentes em {agents_dir}, plugin em {plugins_dir}/dev-workflows.ts")
|
||||
|
||||
|
||||
def install_claude():
|
||||
for cmd in (["claude", "plugin", "marketplace", "add", ROOT], ["claude", "plugin", "install", "dev-workflows@dev-workflows"]):
|
||||
print("$ " + " ".join(cmd))
|
||||
if subprocess.run(cmd).returncode != 0:
|
||||
sys.exit("falhou; rode o comando acima manualmente para ver o erro")
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
ap.add_argument("target", choices=["claude", "opencode", "hermes"])
|
||||
ap.add_argument("--force", action="store_true", help="sobrescreve pastas com o mesmo nome que não vieram deste script")
|
||||
a = ap.parse_args()
|
||||
if a.target == "claude":
|
||||
install_claude()
|
||||
elif a.target == "opencode":
|
||||
install_opencode(a.force)
|
||||
else:
|
||||
install_skills(os.path.expanduser("~/.hermes/skills"), a.force)
|
||||
print("Hermes: agentes e hooks não se aplicam; rode o gate manualmente.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,304 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Deterministic quality gate: runs only the checks of the requested tier, scoped to the diff.
|
||||
|
||||
Usage: quality-gate [--tier 0-3] [--base REF] [--dir PATH] [--list]
|
||||
Writes .dev-workflows/gate.json in the project root. Exit 1 when any check fails.
|
||||
A tool that is not installed is reported as "unverified", never as passed.
|
||||
"""
|
||||
import argparse, json, os, re, shutil, subprocess, sys, tempfile, time, tomllib
|
||||
|
||||
CODE_EXT = {".py", ".js", ".jsx", ".ts", ".tsx", ".mjs", ".cjs", ".vue", ".svelte", ".go", ".rs",
|
||||
".java", ".kt", ".rb", ".php", ".cs", ".sql", ".sh", ".css", ".scss", ".html"}
|
||||
JS_EXT = {".js", ".jsx", ".ts", ".tsx", ".mjs", ".cjs", ".vue", ".svelte"}
|
||||
|
||||
|
||||
def sh(cmd, cwd, timeout=600):
|
||||
t = time.time()
|
||||
try:
|
||||
p = subprocess.run(cmd, cwd=cwd, shell=isinstance(cmd, str), capture_output=True, text=True, timeout=timeout)
|
||||
out, code = (p.stdout + p.stderr), p.returncode
|
||||
except subprocess.TimeoutExpired:
|
||||
out, code = f"timeout after {timeout}s", 124
|
||||
return code, out, round(time.time() - t, 1)
|
||||
|
||||
|
||||
def git(args, cwd):
|
||||
code, out, _ = sh(["git", *args], cwd, 60)
|
||||
return out.strip() if code == 0 else None
|
||||
|
||||
|
||||
def find_base(root, explicit):
|
||||
if explicit:
|
||||
return explicit
|
||||
for ref in ("origin/HEAD", "origin/main", "origin/master", "main", "master"):
|
||||
mb = git(["merge-base", "HEAD", ref], root)
|
||||
if mb:
|
||||
return mb
|
||||
return "HEAD" if git(["rev-parse", "HEAD"], root) else None
|
||||
|
||||
|
||||
def changed_files(root, base):
|
||||
if base is None:
|
||||
return []
|
||||
names = (git(["diff", "--name-only", "--diff-filter=d", base], root) or "").splitlines()
|
||||
names += (git(["ls-files", "--others", "--exclude-standard"], root) or "").splitlines()
|
||||
return sorted({n for n in names if n and not n.startswith(".dev-workflows/") and os.path.isfile(os.path.join(root, n))})
|
||||
|
||||
|
||||
def diff_stats(root, base, files):
|
||||
added = removed = 0
|
||||
for line in (git(["diff", "--numstat", base], root) or "").splitlines() if base else []:
|
||||
a, r, _ = (line.split("\t") + ["", ""])[:3]
|
||||
added += int(a) if a.isdigit() else 0
|
||||
removed += int(r) if r.isdigit() else 0
|
||||
for f in (git(["ls-files", "--others", "--exclude-standard"], root) or "").splitlines():
|
||||
if f in files:
|
||||
try:
|
||||
added += sum(1 for _ in open(os.path.join(root, f), errors="ignore"))
|
||||
except OSError:
|
||||
pass
|
||||
return {"files": len(files), "added": added, "removed": removed, "net": added - removed}
|
||||
|
||||
|
||||
def deps_at(root, rev):
|
||||
"""Dependency names declared in package.json / requirements*.txt / pyproject.toml at a revision (None = working tree)."""
|
||||
def read(path):
|
||||
if rev is None:
|
||||
p = os.path.join(root, path)
|
||||
return open(p).read() if os.path.isfile(p) else None
|
||||
return git(["show", f"{rev}:{path}"], root)
|
||||
|
||||
names = set()
|
||||
pkg = read("package.json")
|
||||
if pkg:
|
||||
try:
|
||||
d = json.loads(pkg)
|
||||
for k in ("dependencies", "devDependencies", "optionalDependencies"):
|
||||
names |= {f"npm:{n}" for n in d.get(k, {})}
|
||||
except ValueError:
|
||||
pass
|
||||
for req in [f for f in os.listdir(root) if re.fullmatch(r"requirements.*\.txt", f)]:
|
||||
for line in (read(req) or "").splitlines():
|
||||
m = re.match(r"\s*([A-Za-z0-9_.\-]+)", line)
|
||||
if m and not line.strip().startswith(("#", "-")):
|
||||
names.add(f"py:{m.group(1).lower()}")
|
||||
py = read("pyproject.toml")
|
||||
if py:
|
||||
try:
|
||||
proj = tomllib.loads(py).get("project", {})
|
||||
for spec in proj.get("dependencies", []) + sum(proj.get("optional-dependencies", {}).values(), []):
|
||||
m = re.match(r"\s*([A-Za-z0-9_.\-]+)", spec)
|
||||
if m:
|
||||
names.add(f"py:{m.group(1).lower()}")
|
||||
except tomllib.TOMLDecodeError:
|
||||
pass
|
||||
return names
|
||||
|
||||
|
||||
def load_config(root):
|
||||
p = os.path.join(root, ".dev-workflows.toml")
|
||||
if not os.path.isfile(p):
|
||||
return {}
|
||||
with open(p, "rb") as f:
|
||||
return tomllib.load(f)
|
||||
|
||||
|
||||
def local_bin(root, name):
|
||||
p = os.path.join(root, "node_modules", ".bin", name)
|
||||
return p if os.path.exists(p) else shutil.which(name)
|
||||
|
||||
|
||||
def node_pm(root):
|
||||
for lock, pm in (("pnpm-lock.yaml", "pnpm"), ("yarn.lock", "yarn"), ("bun.lockb", "bun"), ("bun.lock", "bun")):
|
||||
if os.path.exists(os.path.join(root, lock)):
|
||||
return pm
|
||||
return "npm"
|
||||
|
||||
|
||||
def plan_checks(root, cfg, files, base):
|
||||
"""Returns [(name, tier, cmd|None, kind)] ; kind: gate (fails the gate) | info (report only)."""
|
||||
cmds = cfg.get("commands", {})
|
||||
exts = {os.path.splitext(f)[1] for f in files}
|
||||
js_files = [f for f in files if os.path.splitext(f)[1] in JS_EXT]
|
||||
py_files = [f for f in files if f.endswith(".py")]
|
||||
checks = []
|
||||
|
||||
pkg = {}
|
||||
if os.path.isfile(os.path.join(root, "package.json")):
|
||||
try:
|
||||
pkg = json.load(open(os.path.join(root, "package.json"))).get("scripts", {})
|
||||
except ValueError:
|
||||
pkg = {}
|
||||
pm = node_pm(root)
|
||||
is_py = any(os.path.exists(os.path.join(root, f)) for f in ("pyproject.toml", "setup.py", "requirements.txt"))
|
||||
pyproject = {}
|
||||
if os.path.isfile(os.path.join(root, "pyproject.toml")):
|
||||
try:
|
||||
pyproject = tomllib.load(open(os.path.join(root, "pyproject.toml"), "rb"))
|
||||
except tomllib.TOMLDecodeError:
|
||||
pass
|
||||
tool = pyproject.get("tool", {})
|
||||
|
||||
def pick(key, *candidates):
|
||||
if key in cmds:
|
||||
return cmds[key]
|
||||
return next((c for c in candidates if c), None)
|
||||
|
||||
lint = pick("lint",
|
||||
f"{pm} run lint" if "lint" in pkg else None,
|
||||
(local_bin(root, "eslint") and js_files and [local_bin(root, "eslint"), *js_files]),
|
||||
(shutil.which("ruff") and py_files and ["ruff", "check", *py_files]),
|
||||
("go vet ./..." if os.path.exists(os.path.join(root, "go.mod")) else None),
|
||||
("cargo clippy -- -D warnings" if os.path.exists(os.path.join(root, "Cargo.toml")) else None))
|
||||
typecheck = pick("typecheck",
|
||||
next((f"{pm} run {s}" for s in ("typecheck", "type-check", "tsc") if s in pkg), None),
|
||||
(os.path.exists(os.path.join(root, "tsconfig.json")) and local_bin(root, "tsc") and [local_bin(root, "tsc"), "--noEmit"]),
|
||||
(("pyright" in tool or os.path.exists(os.path.join(root, "pyrightconfig.json"))) and shutil.which("pyright") and ["pyright"]),
|
||||
("mypy" in tool and shutil.which("mypy") and ["mypy", "."]))
|
||||
venv_py = next((os.path.join(root, v, "bin", "python") for v in (".venv", "venv")
|
||||
if os.path.exists(os.path.join(root, v, "bin", "python"))), "python3")
|
||||
test = pick("test",
|
||||
f"{pm} run test" if "test" in pkg else None,
|
||||
(is_py and [venv_py, "-m", "pytest", "-q"]),
|
||||
("go test ./..." if os.path.exists(os.path.join(root, "go.mod")) else None),
|
||||
("cargo test" if os.path.exists(os.path.join(root, "Cargo.toml")) else None))
|
||||
build = pick("build", f"{pm} run build" if "build" in pkg else None)
|
||||
|
||||
checks += [("lint", 0, lint, "gate"), ("typecheck", 0, typecheck, "gate"), ("test", 0, test, "gate")]
|
||||
checks.append(("secrets", 0, "@gitleaks" if shutil.which("gitleaks") else None, "gate"))
|
||||
|
||||
cov = os.path.join(root, cfg.get("coverage", {}).get("report", "coverage.xml"))
|
||||
fail_under = cfg.get("thresholds", {}).get("diff_coverage", 80)
|
||||
checks.append(("diff-coverage", 1,
|
||||
shutil.which("diff-cover") and os.path.isfile(cov) and base
|
||||
and ["diff-cover", cov, f"--compare-branch={base}", f"--fail-under={fail_under}"], "gate"))
|
||||
checks.append(("semgrep", 1, shutil.which("semgrep") and base and
|
||||
["semgrep", "scan", "--config", "p/default", "--error", "--metrics=off", "--quiet",
|
||||
"--baseline-commit", base], "gate"))
|
||||
dead = (local_bin(root, "knip") and pkg and [local_bin(root, "knip"), "--no-exit-code"]) or \
|
||||
(shutil.which("vulture") and py_files and ["vulture", *py_files, "--min-confidence", "80"])
|
||||
checks.append(("dead-code", 1, dead, "info"))
|
||||
|
||||
checks.append(("build", 2, build, "gate") if build else ("build", 2, "@skip", "gate"))
|
||||
stryker_cfg = any(os.path.exists(os.path.join(root, f)) for f in
|
||||
("stryker.config.mjs", "stryker.config.js", "stryker.config.json", "stryker.conf.js", ".stryker.conf.json"))
|
||||
mut = None
|
||||
if stryker_cfg and local_bin(root, "stryker") and js_files:
|
||||
mut = [local_bin(root, "stryker"), "run", "--mutate", ",".join(js_files)]
|
||||
elif "mutmut" in tool and shutil.which("mutmut"):
|
||||
mut = ["mutmut", "run"] # ponytail: mutmut 3 runs its configured paths, not only the diff
|
||||
checks.append(("mutation", 2, cmds.get("mutation", mut), "gate"))
|
||||
checks.append(("osv-scanner", 2, shutil.which("osv-scanner") and ["osv-scanner", "scan", "source", "--recursive", "."], "gate"))
|
||||
cruiser = next((f for f in (".dependency-cruiser.js", ".dependency-cruiser.cjs", ".dependency-cruiser.json")
|
||||
if os.path.exists(os.path.join(root, f))), None)
|
||||
arch = (cruiser and local_bin(root, "depcruise") and js_files and [local_bin(root, "depcruise"), "--config", cruiser, *js_files]) or \
|
||||
(("importlinter" in tool or os.path.exists(os.path.join(root, ".importlinter"))) and shutil.which("lint-imports") and ["lint-imports"])
|
||||
checks.append(("architecture", 2, cmds.get("architecture", arch) or "@skip", "gate"))
|
||||
checks.append(("duplication", 2, local_bin(root, "jscpd") and js_files and [local_bin(root, "jscpd"), *js_files], "info"))
|
||||
|
||||
checks.append(("trivy", 3, shutil.which("trivy") and
|
||||
["trivy", "fs", "--exit-code", "1", "--severity", "HIGH,CRITICAL", "--scanners", "vuln,misconfig,secret", "."], "gate"))
|
||||
if cfg.get("spec", {}).get("file"):
|
||||
checks.append(("traceability", 3, "@trace", "gate"))
|
||||
|
||||
for extra in cfg.get("check", []):
|
||||
checks.append((extra["name"], int(extra.get("tier", 2)), extra["cmd"], extra.get("kind", "gate")))
|
||||
|
||||
if not exts & CODE_EXT and files:
|
||||
checks = [c for c in checks if c[0] == "secrets"]
|
||||
return checks
|
||||
|
||||
|
||||
def run_gitleaks(root, files):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
for f in files:
|
||||
dst = os.path.join(tmp, f)
|
||||
os.makedirs(os.path.dirname(dst), exist_ok=True)
|
||||
shutil.copy2(os.path.join(root, f), dst)
|
||||
return sh(["gitleaks", "dir", "--no-banner", "--redact", tmp], root)
|
||||
|
||||
|
||||
def run_trace(root, cfg):
|
||||
spec = cfg["spec"]
|
||||
text = open(os.path.join(root, spec["file"]), errors="ignore").read()
|
||||
ids = sorted(set(re.findall(r"\bREQ-\d+\b", text)))
|
||||
corpus = ""
|
||||
for d in spec.get("tests", ["tests", "test", "src"]):
|
||||
for dirpath, _, names in os.walk(os.path.join(root, d)):
|
||||
for n in names:
|
||||
corpus += open(os.path.join(dirpath, n), errors="ignore").read()
|
||||
missing = [i for i in ids if i not in corpus]
|
||||
msg = f"{len(ids)} requisitos, sem teste: {', '.join(missing) or 'nenhum'}"
|
||||
return (1 if missing or not ids else 0), msg, 0.0
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
ap.add_argument("--tier", type=int, default=int(os.environ.get("DW_TIER", 1)), choices=range(4))
|
||||
ap.add_argument("--base")
|
||||
ap.add_argument("--dir", default=".")
|
||||
ap.add_argument("--list", action="store_true", help="show the planned checks without running them")
|
||||
a = ap.parse_args()
|
||||
|
||||
root = git(["rev-parse", "--show-toplevel"], a.dir) or os.path.abspath(a.dir)
|
||||
cfg = load_config(root)
|
||||
base = find_base(root, a.base)
|
||||
files = changed_files(root, base)
|
||||
planned = [c for c in plan_checks(root, cfg, files, base) if c[1] <= a.tier]
|
||||
|
||||
if a.list:
|
||||
for name, tier, cmd, kind in planned:
|
||||
state = "skip" if cmd == "@skip" else ("unverified (tool/config missing)" if not cmd else "run")
|
||||
print(f"T{tier} {name:14} {kind:4} {state}")
|
||||
return 0
|
||||
|
||||
results = []
|
||||
for name, tier, cmd, kind in planned:
|
||||
if cmd == "@skip":
|
||||
continue
|
||||
if not cmd:
|
||||
results.append({"name": name, "tier": tier, "kind": kind, "status": "unverified"})
|
||||
continue
|
||||
if cmd == "@gitleaks":
|
||||
code, out, secs = run_gitleaks(root, files) if files else (0, "no changed files", 0.0)
|
||||
elif cmd == "@trace":
|
||||
code, out, secs = run_trace(root, cfg)
|
||||
else:
|
||||
code, out, secs = sh(cmd, root, int(cfg.get("timeout", 600)))
|
||||
missing_tool = code == 127 or re.search(r"No module named (pytest|mypy|pyright|ruff)\b", out)
|
||||
status = "unverified" if missing_tool else "info" if kind == "info" else ("pass" if code == 0 else "fail")
|
||||
results.append({"name": name, "tier": tier, "kind": kind, "status": status, "exit": code, "seconds": secs,
|
||||
"cmd": cmd if isinstance(cmd, str) else " ".join(cmd),
|
||||
"output_tail": "\n".join(out.strip().splitlines()[-30:])})
|
||||
|
||||
new_deps = sorted(deps_at(root, None) - deps_at(root, base)) if base else []
|
||||
verdict = "fail" if any(r["status"] == "fail" for r in results) else \
|
||||
"incomplete" if any(r["status"] == "unverified" for r in results) else "pass"
|
||||
report = {"verdict": verdict, "tier": a.tier, "base": base, "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
"diff": diff_stats(root, base, files), "changed_files": files, "new_dependencies": new_deps, "checks": results}
|
||||
|
||||
out_dir = os.path.join(root, ".dev-workflows")
|
||||
os.makedirs(out_dir, exist_ok=True)
|
||||
json.dump(report, open(os.path.join(out_dir, "gate.json"), "w"), indent=2, ensure_ascii=False)
|
||||
with open(os.path.join(out_dir, "log.jsonl"), "a") as log:
|
||||
log.write(json.dumps({k: report[k] for k in ("generated_at", "verdict", "tier", "diff")}) + "\n")
|
||||
exclude = os.path.join(root, ".git", "info", "exclude")
|
||||
if os.path.isdir(os.path.dirname(exclude)):
|
||||
current = open(exclude).read() if os.path.exists(exclude) else ""
|
||||
if ".dev-workflows/" not in current:
|
||||
with open(exclude, "a") as f:
|
||||
f.write("\n.dev-workflows/\n")
|
||||
|
||||
icon = {"pass": "ok", "fail": "FALHOU", "unverified": "não verificado", "info": "info"}
|
||||
for r in results:
|
||||
print(f" [{icon[r['status']]:>14}] T{r['tier']} {r['name']}")
|
||||
d = report["diff"]
|
||||
print(f"diff: {d['files']} arquivos, +{d['added']} -{d['removed']} (líquido {d['net']})")
|
||||
if new_deps:
|
||||
print("dependências novas (justifique cada uma): " + ", ".join(new_deps))
|
||||
print(f"veredito: {verdict} -> {os.path.relpath(os.path.join(out_dir, 'gate.json'), os.getcwd())}")
|
||||
return 1 if verdict == "fail" else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user