216 lines
8.5 KiB
Python
Executable File
216 lines
8.5 KiB
Python
Executable File
#!/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()
|