init: 三端统一工作区目录结构规范 (W1-W6 catalog + templates + scripts + policies)
This commit is contained in:
Executable
+215
@@ -0,0 +1,215 @@
|
||||
#!/usr/bin/env python3
|
||||
"""classify_workspace.py — 工作区半自动分类判定（规范 catalog 驱动）
|
||||
|
||||
扫描目标目录，按 catalog.yaml 判定顺序 + 信号扫描，输出「建议类型 + 依据 + 置信度」。
|
||||
由人确认后写入 .workspace.yaml。不自动搬任何文件。
|
||||
|
||||
用法:
|
||||
python3 classify_workspace.py <TARGET_PATH> [--catalog PATH] [--write] [--quiet]
|
||||
|
||||
--catalog 规范 catalog.yaml 路径（默认自动找：环境变量 STANDARD_ROOT 或
|
||||
/srv/share/docs/standards/workspace-directory-standard/catalog.yaml）
|
||||
--write 确认后写入 .workspace.yaml（交互询问；--yes 跳过询问）
|
||||
--yes 配合 --write 自动确认
|
||||
--quiet 只输出一行结论
|
||||
|
||||
信号说明（catalog.yaml signals）：
|
||||
扫描顶层文件/目录名 + 少量文件内容特征，为每个类别给出命中信号列表。
|
||||
最终建议 = 判定顺序中首个「命中其强制信号或人工问答为是」的类别。
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
DEFAULT_CATALOG = "/srv/share/docs/standards/workspace-directory-standard/catalog.yaml"
|
||||
|
||||
# 强制信号：这些特征存在即强指向该类别（比 catalog signals 更强）
|
||||
STRONG_SIGNALS = {
|
||||
# W1 三层结构需全部齐全（防服务型误判：服务也有 projects/）
|
||||
"W1": ["ALL3"],
|
||||
"W2": ["systemd", "Dockerfile", "docker-compose.yml", "docker-compose.yaml", "supervisord.conf",
|
||||
"*.service", "*server*", "*_api", "*bridge*", "*.db", "*runtime*"],
|
||||
"W3": ["pyproject.toml", "package.json", "Cargo.toml", "go.mod", "setup.py", "requirements.txt", "pom.xml"],
|
||||
"W4": ["inbox", "domains", "indexes", "*.db", "search*", "semantic_search", "vector*", "*knowledge*", "*memory*", "_规范", "_模板", "*域*"], # 知识库/检索特征
|
||||
"W6": ["inventory", "shared", "共享", "migration"],
|
||||
}
|
||||
|
||||
|
||||
def load_catalog(path: Path) -> dict:
|
||||
try:
|
||||
import yaml
|
||||
except ImportError:
|
||||
print("ERROR: 需要 pyyaml（pip install pyyaml）", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
with open(path, encoding="utf-8") as f:
|
||||
return yaml.safe_load(f)
|
||||
|
||||
|
||||
def scan_signals(target: Path) -> dict:
|
||||
"""扫描目录，返回每个类别的命中信号列表"""
|
||||
hits = {k: [] for k in STRONG_SIGNALS}
|
||||
names = [p.name for p in target.iterdir()] if target.is_dir() else []
|
||||
lower_names = [n.lower() for n in names]
|
||||
|
||||
for cls, sigs in STRONG_SIGNALS.items():
|
||||
for s in sigs:
|
||||
if s == "ALL3":
|
||||
# W1 三层结构齐全判定：methods+projects+deliverables 全存在
|
||||
need = {"methods", "projects", "deliverables"}
|
||||
present = need & set(lower_names)
|
||||
if len(present) == 3:
|
||||
hits[cls].append("methods+projects+deliverables 齐全")
|
||||
continue
|
||||
if s.startswith("*") and s.endswith("*"):
|
||||
pat = s.strip("*").lower()
|
||||
if any(pat in n for n in lower_names):
|
||||
hits[cls].append(s)
|
||||
elif s.startswith("*"):
|
||||
pat = s[1:].lower()
|
||||
if any(n.endswith(pat) for n in lower_names):
|
||||
hits[cls].append(s)
|
||||
elif s.endswith("*"):
|
||||
pat = s[:-1].lower()
|
||||
if any(n.startswith(pat) for n in lower_names):
|
||||
hits[cls].append(s)
|
||||
else:
|
||||
if s.lower() in lower_names:
|
||||
hits[cls].append(s)
|
||||
|
||||
# 内容嗅探：文件内关键信号（只读前几层，不深扫）
|
||||
def sniff(path: Path, sigs: list, limit: int = 4000) -> list:
|
||||
found = []
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8", errors="ignore") as f:
|
||||
head = f.read(limit)
|
||||
for s in sigs:
|
||||
if s.lower() in head.lower():
|
||||
found.append(s)
|
||||
except Exception:
|
||||
pass
|
||||
return found
|
||||
|
||||
for p in target.iterdir():
|
||||
if p.is_file() and p.suffix.lower() in (".py", ".sh", ".yaml", ".yml", ".json", ".toml", ".ini"):
|
||||
if p.name == ".workspace.yaml":
|
||||
continue
|
||||
if "Dockerfile" in hits["W2"] or "docker-compose" in str(p).lower():
|
||||
pass
|
||||
for cls, sigs in {
|
||||
"W2": ["systemd", "uvicorn", "gunicorn", "0.0.0.0", "127.0.0.1", "FastAPI", "Flask", "scheduler"],
|
||||
"W3": ["pytest", "unittest", "if __name__", "typer", "click", "setuptools"],
|
||||
}.items():
|
||||
found = sniff(p, sigs)
|
||||
for s in found:
|
||||
if s not in hits[cls]:
|
||||
hits[cls].append(s)
|
||||
return hits
|
||||
|
||||
|
||||
def decide(target: Path, hits: dict, order: list, questions: dict, interactive: bool = True) -> tuple:
|
||||
"""判定顺序执行：首个命中强信号或人工确认为是 → 建议类别"""
|
||||
reasons = []
|
||||
for cls in order:
|
||||
strong = hits.get(cls, [])
|
||||
if strong:
|
||||
reasons.append(f"检测到 {cls} 特征: {', '.join(strong)}")
|
||||
return cls, reasons, 0.9
|
||||
# 无强信号 → 走 catalog 决策问题（交互）
|
||||
if not interactive:
|
||||
return "W5", ["非交互模式，无强特征，兜底 W5 单项目（请人工复核）"], 0.4
|
||||
print("未检测到强特征目录，请回答几个问题：")
|
||||
for cls in order:
|
||||
q = questions.get(cls)
|
||||
if not q:
|
||||
continue
|
||||
while True:
|
||||
ans = input(f" {q} [y/N] ").strip().lower()
|
||||
if ans in ("y", "yes"):
|
||||
reasons.append(f"问答确认: {q}")
|
||||
return cls, reasons, 0.7
|
||||
if ans in ("n", "no", ""):
|
||||
break
|
||||
print(" 请输入 y 或 n")
|
||||
return "W5", ["无强特征且未命中其他类，兜底 W5 单项目"], 0.6
|
||||
|
||||
|
||||
def write_workspace_yaml(target: Path, cls: str, catalog: dict) -> None:
|
||||
"""写入 .workspace.yaml（登记卡）"""
|
||||
cfg = catalog["classes"][cls]
|
||||
tpl = f"""schema_version: 1
|
||||
name: {target.name}
|
||||
class: {cls}
|
||||
profile: {cfg.get('en', '')}
|
||||
business_domain: 待定
|
||||
owner: user
|
||||
primary_location: 待定
|
||||
status: active
|
||||
created_at: {Path('_').stem or ''}
|
||||
last_reviewed_at: {Path('_').stem or ''}
|
||||
canonical_path: {target}
|
||||
related_services: []
|
||||
git_remote: ""
|
||||
"""
|
||||
# 用当前日期
|
||||
import datetime
|
||||
today = datetime.date.today().isoformat()
|
||||
tpl = tpl.replace(f"created_at: {Path('_').stem or ''}", f"created_at: {today}") \
|
||||
.replace(f"last_reviewed_at: {Path('_').stem or ''}", f"last_reviewed_at: {today}")
|
||||
(target / ".workspace.yaml").write_text(tpl, encoding="utf-8")
|
||||
print(f"已写入: {target / '.workspace.yaml'}")
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="工作区半自动分类判定")
|
||||
ap.add_argument("target", help="目标目录路径")
|
||||
ap.add_argument("--catalog", default=os.environ.get("STANDARD_ROOT", DEFAULT_CATALOG))
|
||||
ap.add_argument("--write", action="store_true", help="确认后写入 .workspace.yaml")
|
||||
ap.add_argument("--yes", action="store_true", help="跳过交互确认")
|
||||
ap.add_argument("--quiet", action="store_true")
|
||||
args = ap.parse_args()
|
||||
|
||||
target = Path(args.target)
|
||||
if not target.is_dir():
|
||||
print(f"ERROR: 不是目录: {target}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
catalog_path = Path(args.catalog)
|
||||
if not catalog_path.exists():
|
||||
print(f"ERROR: catalog 不存在: {catalog_path}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
catalog = load_catalog(catalog_path)
|
||||
order = catalog["decision_order"]
|
||||
questions = {
|
||||
cls: info["decisions"][0]["question"]
|
||||
for cls, info in catalog["classes"].items() if info.get("decisions")
|
||||
}
|
||||
|
||||
hits = scan_signals(target)
|
||||
cls, reasons, conf = decide(target, hits, order, questions, interactive=not args.quiet)
|
||||
|
||||
if args.quiet:
|
||||
print(f"{cls} (置信度 {conf:.0%})")
|
||||
return
|
||||
|
||||
print(f"\n建议分类: {cls} — {catalog['classes'][cls]['name']} (置信度 {conf:.0%})")
|
||||
for r in reasons:
|
||||
print(f" 依据: {r}")
|
||||
print(f" 模板: {catalog['classes'][cls]['template']}")
|
||||
print(f" 采用 v2 三层结构: {'是' if catalog['classes'][cls].get('uses_v2') else '否'}")
|
||||
|
||||
if args.write:
|
||||
if args.yes:
|
||||
write_workspace_yaml(target, cls, catalog)
|
||||
else:
|
||||
confirm = input("\n确认写入 .workspace.yaml? [y/N] ").strip().lower()
|
||||
if confirm in ("y", "yes"):
|
||||
write_workspace_yaml(target, cls, catalog)
|
||||
else:
|
||||
print("未写入。可手动编辑 .workspace.yaml 或调整分类后重跑。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user