mirror of
https://github.com/tradecatlabs/vibe-coding-cn.git
synced 2026-07-27 18:57:50 +00:00
docs: add research domains and resource registry
- add local external resources registry - add research domain contract and per-repository research domains - add raw fact snapshot governance and validation scripts Validation: - make test - git diff --check HEAD~1..HEAD Note: - welcome workflow failed because its action input names are stale; core content CI passed.
This commit is contained in:
tradecatlabs
+3
-1
@@ -1,11 +1,13 @@
|
||||
# scripts/ Agent 指南
|
||||
|
||||
本目录维护仓库级自动化脚本，主要用于 Markdown、链接、锚点、metadata 和 AI 引用资产校验。
|
||||
本目录维护仓库级自动化脚本，主要用于 Markdown、链接、锚点、metadata、AI 引用资产、外部资源注册表校验、研究域 raw 原始事实层与 Git 工作树检查和拉取。
|
||||
|
||||
## 约束
|
||||
|
||||
- 脚本默认从仓库根目录运行，路径解析必须稳定。
|
||||
- 新增检查脚本时，同步更新 `scripts/README.md`、`Makefile` 和根目录 `AGENTS.md` 的命令清单；只有 CI 环境稳定具备所需输入时才纳入 CI。
|
||||
- `fetch-research-raw.py` 访问 GitHub 网络 API，只作为手动刷新命令，不纳入 `make test` 或 CI 硬门禁。
|
||||
- `check-research-raw.py` 只检查本地文件结构和 JSON 形态，可以纳入 `make test`。
|
||||
- `check-directory-docs.py` 对根 `.github/` 只要求 `AGENTS.md`，不要重新补 `.github/README.md`。
|
||||
- 修改 docs README 或主题正文的主章节、锚点、索引后，优先运行 `python3 scripts/sync-doc-toc.py`，再运行 `make test`。
|
||||
- 修改 GitHub Wiki 独立仓库后，运行 `make check-wiki WIKI_DIR=/path/to/wiki`；不要把 Wiki checkout 提交进主仓。
|
||||
|
||||
@@ -10,5 +10,8 @@
|
||||
- `check-directory-docs.py`：仓库自有目录 `README.md` / `AGENTS.md` 覆盖检查脚本；根 `.github/` 仅要求 `AGENTS.md`，避免 GitHub 首页误展示平台配置说明。
|
||||
- `check-metadata.py`：`metadata/taxonomy.yml` 与 `metadata/redirects.yml` 路径和锚点检查脚本。
|
||||
- `check-ai-citation.py`：`llms.txt`、`assets/ai-citation/llms-full.txt` 与 AI 引用语料路径和锚点检查脚本。
|
||||
- `check-external-resources.py`：本地外部资源注册表字段、分类统计、ID 与链接形态检查脚本。
|
||||
- `check-research-raw.py`：研究域 raw 原始事实层、Git 工作树、来源清单和核心材料文件检查脚本。
|
||||
- `fetch-research-raw.py`：按 `docs/research/*/domain.yml` 拉取 GitHub 研究对象的 raw 原始事实层和 `repository/` 工作树。
|
||||
- `check-wiki.py`：GitHub Wiki 独立仓库本地 checkout 的页面覆盖、内链和旧口径检查脚本。
|
||||
- `sync-doc-toc.py`：兼容旧线性 README 的细粒度目录生成脚本；当前拆分结构下通常无变更。
|
||||
|
||||
@@ -65,6 +65,8 @@ def should_skip(directory: Path) -> bool:
|
||||
rel = directory.relative_to(ROOT)
|
||||
if any(part in SKIP_PARTS for part in rel.parts):
|
||||
return True
|
||||
if is_research_repository_snapshot(rel):
|
||||
return True
|
||||
if directory.is_symlink():
|
||||
return True
|
||||
if any(rel == prefix or prefix in rel.parents for prefix in SKIP_PREFIXES):
|
||||
@@ -72,6 +74,17 @@ def should_skip(directory: Path) -> bool:
|
||||
return any(vendor in rel.parents for vendor in VENDOR_SUBTREES)
|
||||
|
||||
|
||||
def is_research_repository_snapshot(rel: Path) -> bool:
|
||||
parts = rel.parts
|
||||
return (
|
||||
len(parts) >= 5
|
||||
and parts[0] == "docs"
|
||||
and parts[1] == "research"
|
||||
and parts[3] == "raw"
|
||||
and parts[4] == "repository"
|
||||
)
|
||||
|
||||
|
||||
def repository_owned_dirs() -> list[Path]:
|
||||
dirs = set(REQUIRED_DIRS)
|
||||
for directory in ROOT.rglob("*"):
|
||||
|
||||
@@ -54,9 +54,22 @@ def should_skip(path: Path) -> bool:
|
||||
rel = path.relative_to(ROOT)
|
||||
if any(part in SKIP_PARTS for part in rel.parts):
|
||||
return True
|
||||
if is_research_repository_snapshot(rel):
|
||||
return True
|
||||
return any(rel == prefix or prefix in rel.parents for prefix in SKIP_PREFIXES)
|
||||
|
||||
|
||||
def is_research_repository_snapshot(rel: Path) -> bool:
|
||||
parts = rel.parts
|
||||
return (
|
||||
len(parts) >= 5
|
||||
and parts[0] == "docs"
|
||||
and parts[1] == "research"
|
||||
and parts[3] == "raw"
|
||||
and parts[4] == "repository"
|
||||
)
|
||||
|
||||
|
||||
def duplicate_manual_anchors(path: Path) -> list[str]:
|
||||
text = strip_fenced_code(path.read_text(encoding="utf-8", errors="ignore"))
|
||||
seen: dict[str, int] = {}
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Validate the local external resources registry."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import sys
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
REGISTRY_DIR = ROOT / "assets" / "external-resources"
|
||||
MANIFEST = REGISTRY_DIR / "resources.yml"
|
||||
SCHEMA = REGISTRY_DIR / "schema.yml"
|
||||
ID_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
||||
|
||||
|
||||
def load_yaml(path: Path) -> object:
|
||||
with path.open(encoding="utf-8") as fh:
|
||||
return yaml.safe_load(fh)
|
||||
|
||||
|
||||
def as_dict(value: object, label: str, errors: list[str]) -> dict:
|
||||
if isinstance(value, dict):
|
||||
return value
|
||||
errors.append(f"{label}: expected mapping")
|
||||
return {}
|
||||
|
||||
|
||||
def as_list(value: object, label: str, errors: list[str]) -> list:
|
||||
if isinstance(value, list):
|
||||
return value
|
||||
errors.append(f"{label}: expected list")
|
||||
return []
|
||||
|
||||
|
||||
def main() -> int:
|
||||
errors: list[str] = []
|
||||
|
||||
if not MANIFEST.is_file():
|
||||
print(f"EXTERNAL_RESOURCE_ERRORS\n{MANIFEST.relative_to(ROOT)}: missing manifest")
|
||||
return 1
|
||||
if not SCHEMA.is_file():
|
||||
print(f"EXTERNAL_RESOURCE_ERRORS\n{SCHEMA.relative_to(ROOT)}: missing schema")
|
||||
return 1
|
||||
|
||||
manifest = as_dict(load_yaml(MANIFEST), str(MANIFEST.relative_to(ROOT)), errors)
|
||||
schema = as_dict(load_yaml(SCHEMA), str(SCHEMA.relative_to(ROOT)), errors)
|
||||
|
||||
resource_schema = as_dict(schema.get("resource"), "schema.yml resource", errors)
|
||||
required_resource_fields = set(as_list(resource_schema.get("required_fields"), "schema.yml required_fields", errors))
|
||||
enums = as_dict(schema.get("enums"), "schema.yml enums", errors)
|
||||
allowed_status = set(as_list(enums.get("status"), "schema.yml enums.status", errors))
|
||||
allowed_verification = set(as_list(enums.get("verification_status"), "schema.yml enums.verification_status", errors))
|
||||
allowed_risk_flags = set(as_list(enums.get("risk_flags"), "schema.yml enums.risk_flags", errors))
|
||||
allowed_resource_types = set(as_list(enums.get("resource_type"), "schema.yml enums.resource_type", errors))
|
||||
allowed_link_kinds = set(as_dict(as_dict(schema.get("link"), "schema.yml link", errors).get("kinds"), "schema.yml link.kinds", errors))
|
||||
|
||||
ids: list[str] = []
|
||||
total_resources = 0
|
||||
risk_counter: Counter[str] = Counter()
|
||||
|
||||
categories = as_list(manifest.get("categories"), "resources.yml categories", errors)
|
||||
for index, category_ref in enumerate(categories, start=1):
|
||||
category_ref = as_dict(category_ref, f"resources.yml categories[{index}]", errors)
|
||||
category_id = category_ref.get("id")
|
||||
category_file = category_ref.get("file")
|
||||
expected_count = category_ref.get("count")
|
||||
if not category_id or not isinstance(category_id, str):
|
||||
errors.append(f"resources.yml categories[{index}].id: missing string")
|
||||
continue
|
||||
if not category_file or not isinstance(category_file, str):
|
||||
errors.append(f"resources.yml categories[{index}].file: missing string")
|
||||
continue
|
||||
|
||||
path = REGISTRY_DIR / category_file
|
||||
if not path.is_file():
|
||||
errors.append(f"{path.relative_to(ROOT)}: missing category file")
|
||||
continue
|
||||
|
||||
payload = as_dict(load_yaml(path), str(path.relative_to(ROOT)), errors)
|
||||
category = as_dict(payload.get("category"), f"{path.relative_to(ROOT)} category", errors)
|
||||
if category.get("id") != category_id:
|
||||
errors.append(f"{path.relative_to(ROOT)}: category.id does not match resources.yml")
|
||||
|
||||
resources = as_list(payload.get("resources"), f"{path.relative_to(ROOT)} resources", errors)
|
||||
if expected_count != len(resources):
|
||||
errors.append(
|
||||
f"{path.relative_to(ROOT)}: expected {expected_count} resources from manifest, found {len(resources)}"
|
||||
)
|
||||
|
||||
for row_index, resource in enumerate(resources, start=1):
|
||||
label = f"{path.relative_to(ROOT)} resources[{row_index}]"
|
||||
resource = as_dict(resource, label, errors)
|
||||
missing_fields = sorted(field for field in required_resource_fields if field not in resource)
|
||||
for field in missing_fields:
|
||||
errors.append(f"{label}.{field}: missing required field")
|
||||
|
||||
resource_id = resource.get("id")
|
||||
if not isinstance(resource_id, str) or not ID_PATTERN.match(resource_id):
|
||||
errors.append(f"{label}.id: must be lowercase kebab-case")
|
||||
elif not resource_id.startswith(f"{category_id}-"):
|
||||
errors.append(f"{label}.id: must start with category prefix {category_id}-")
|
||||
else:
|
||||
ids.append(resource_id)
|
||||
|
||||
resource_type = resource.get("resource_type")
|
||||
if resource_type not in allowed_resource_types:
|
||||
errors.append(f"{label}.resource_type: unknown value {resource_type!r}")
|
||||
|
||||
status = resource.get("status")
|
||||
if status not in allowed_status:
|
||||
errors.append(f"{label}.status: unknown value {status!r}")
|
||||
|
||||
verification_status = resource.get("verification_status")
|
||||
if verification_status not in allowed_verification:
|
||||
errors.append(f"{label}.verification_status: unknown value {verification_status!r}")
|
||||
|
||||
link = as_dict(resource.get("link"), f"{label}.link", errors)
|
||||
link_kind = link.get("kind")
|
||||
if link_kind not in allowed_link_kinds:
|
||||
errors.append(f"{label}.link.kind: unknown value {link_kind!r}")
|
||||
elif link_kind == "external_url":
|
||||
url = link.get("url")
|
||||
domain = link.get("domain")
|
||||
parsed = urlparse(str(url))
|
||||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||||
errors.append(f"{label}.link.url: must be http(s) URL")
|
||||
if not isinstance(domain, str) or not domain:
|
||||
errors.append(f"{label}.link.domain: missing domain")
|
||||
elif parsed.netloc and domain != parsed.netloc.lower():
|
||||
errors.append(f"{label}.link.domain: does not match URL netloc")
|
||||
elif link_kind in {"internal_reference", "search_instruction"}:
|
||||
if not link.get("locator"):
|
||||
errors.append(f"{label}.link.locator: required for {link_kind}")
|
||||
if status != "needs-review":
|
||||
errors.append(f"{label}.status: non-URL locator must remain needs-review")
|
||||
elif link_kind == "missing":
|
||||
if status != "needs-review":
|
||||
errors.append(f"{label}.status: missing link must remain needs-review")
|
||||
|
||||
risk_flags = as_list(resource.get("risk_flags"), f"{label}.risk_flags", errors)
|
||||
unknown_flags = sorted(flag for flag in risk_flags if flag not in allowed_risk_flags)
|
||||
for flag in unknown_flags:
|
||||
errors.append(f"{label}.risk_flags: unknown value {flag!r}")
|
||||
risk_counter.update(risk_flags)
|
||||
total_resources += 1
|
||||
|
||||
duplicate_ids = sorted(resource_id for resource_id, count in Counter(ids).items() if count > 1)
|
||||
for resource_id in duplicate_ids:
|
||||
errors.append(f"duplicate resource id: {resource_id}")
|
||||
|
||||
stats = as_dict(manifest.get("stats"), "resources.yml stats", errors)
|
||||
if stats.get("total_resources") != total_resources:
|
||||
errors.append(f"resources.yml stats.total_resources: expected {total_resources}")
|
||||
manifest_risks = as_dict(stats.get("risk_flags"), "resources.yml stats.risk_flags", errors)
|
||||
if dict(sorted(risk_counter.items())) != manifest_risks:
|
||||
errors.append("resources.yml stats.risk_flags: does not match category files")
|
||||
|
||||
if errors:
|
||||
print("EXTERNAL_RESOURCE_ERRORS")
|
||||
for error in errors:
|
||||
print(error)
|
||||
print(f"TOTAL={len(errors)}")
|
||||
return 1
|
||||
|
||||
print(f"OK external resources checked: {total_resources} resources, {len(categories)} categories")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -101,9 +101,22 @@ def should_skip(path: Path) -> bool:
|
||||
rel = path.relative_to(ROOT)
|
||||
if any(part in SKIP_PARTS for part in rel.parts):
|
||||
return True
|
||||
if is_research_repository_snapshot(rel):
|
||||
return True
|
||||
return any(rel == prefix or prefix in rel.parents for prefix in SKIP_PREFIXES)
|
||||
|
||||
|
||||
def is_research_repository_snapshot(rel: Path) -> bool:
|
||||
parts = rel.parts
|
||||
return (
|
||||
len(parts) >= 5
|
||||
and parts[0] == "docs"
|
||||
and parts[1] == "research"
|
||||
and parts[3] == "raw"
|
||||
and parts[4] == "repository"
|
||||
)
|
||||
|
||||
|
||||
def normalize_target(raw: str) -> tuple[str, str]:
|
||||
raw = raw.strip()
|
||||
if not raw or raw.startswith(EXTERNAL_PREFIXES):
|
||||
|
||||
@@ -47,9 +47,22 @@ def should_skip(path: Path) -> bool:
|
||||
rel = path.relative_to(ROOT)
|
||||
if any(part in SKIP_PARTS for part in rel.parts):
|
||||
return True
|
||||
if is_research_repository_snapshot(rel):
|
||||
return True
|
||||
return any(rel == prefix or prefix in rel.parents for prefix in SKIP_PREFIXES)
|
||||
|
||||
|
||||
def is_research_repository_snapshot(rel: Path) -> bool:
|
||||
parts = rel.parts
|
||||
return (
|
||||
len(parts) >= 5
|
||||
and parts[0] == "docs"
|
||||
and parts[1] == "research"
|
||||
and parts[3] == "raw"
|
||||
and parts[4] == "repository"
|
||||
)
|
||||
|
||||
|
||||
def line_number(text: str, offset: int) -> int:
|
||||
return text.count("\n", 0, offset) + 1
|
||||
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Check research domains have local raw fact snapshots."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
RESEARCH_DIR = ROOT / "docs" / "research"
|
||||
REQUIRED_RAW_FILES = [
|
||||
"README.md",
|
||||
"AGENTS.md",
|
||||
"sources.yml",
|
||||
"repository",
|
||||
"github-repo.raw.json",
|
||||
"github-readme.raw.md.txt",
|
||||
"github-root-contents.raw.json",
|
||||
"github-languages.raw.json",
|
||||
]
|
||||
REQUIRED_SOURCE_FIELDS = ["version: 1", "pulled_at:", "puller: scripts/fetch-research-raw.py", "files:"]
|
||||
JSON_RAW_FILES = [
|
||||
"github-repo.raw.json",
|
||||
"github-root-contents.raw.json",
|
||||
"github-languages.raw.json",
|
||||
]
|
||||
|
||||
|
||||
def check_json(path: Path) -> str | None:
|
||||
try:
|
||||
json.loads(path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError as error:
|
||||
rel = path.relative_to(ROOT)
|
||||
return f"{rel}:{error.lineno}:{error.colno}: invalid JSON raw file"
|
||||
return None
|
||||
|
||||
|
||||
def check_git_worktree(path: Path) -> str | None:
|
||||
result = subprocess.run(
|
||||
["git", "-C", str(path), "rev-parse", "--is-inside-work-tree"],
|
||||
cwd=ROOT,
|
||||
check=False,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
if result.returncode != 0 or result.stdout.strip() != "true":
|
||||
return f"{path.relative_to(ROOT)}: raw repository is not a valid Git working tree"
|
||||
return None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
errors: list[str] = []
|
||||
domain_files = sorted(RESEARCH_DIR.glob("*/domain.yml"))
|
||||
|
||||
for domain_file in domain_files:
|
||||
domain_dir = domain_file.parent
|
||||
raw_dir = domain_dir / "raw"
|
||||
rel_raw = raw_dir.relative_to(ROOT)
|
||||
|
||||
if not raw_dir.is_dir():
|
||||
errors.append(f"{rel_raw}: missing raw fact directory")
|
||||
continue
|
||||
|
||||
for filename in REQUIRED_RAW_FILES:
|
||||
path = raw_dir / filename
|
||||
if filename == "repository":
|
||||
if not path.is_dir():
|
||||
errors.append(f"{path.relative_to(ROOT)}: missing required raw repository directory")
|
||||
continue
|
||||
if not path.is_file():
|
||||
errors.append(f"{path.relative_to(ROOT)}: missing required raw fact file")
|
||||
|
||||
sources = raw_dir / "sources.yml"
|
||||
if sources.is_file():
|
||||
text = sources.read_text(encoding="utf-8", errors="ignore")
|
||||
for field in REQUIRED_SOURCE_FIELDS:
|
||||
if field not in text:
|
||||
errors.append(f"{sources.relative_to(ROOT)}: missing sources field '{field}'")
|
||||
if "path: repository/" not in text:
|
||||
errors.append(f"{sources.relative_to(ROOT)}: missing repository source record")
|
||||
|
||||
raw_markdown = [
|
||||
path
|
||||
for path in raw_dir.glob("*.raw.md")
|
||||
if path.name not in {"README.md", "AGENTS.md"}
|
||||
]
|
||||
for path in raw_markdown:
|
||||
errors.append(
|
||||
f"{path.relative_to(ROOT)}: raw external markdown must use '.raw.md.txt' to avoid local link checks"
|
||||
)
|
||||
|
||||
for filename in JSON_RAW_FILES:
|
||||
path = raw_dir / filename
|
||||
if path.is_file():
|
||||
error = check_json(path)
|
||||
if error:
|
||||
errors.append(error)
|
||||
|
||||
repository = raw_dir / "repository"
|
||||
if repository.is_dir():
|
||||
error = check_git_worktree(repository)
|
||||
if error:
|
||||
errors.append(error)
|
||||
|
||||
if errors:
|
||||
print("RESEARCH_RAW_ERRORS")
|
||||
for error in errors:
|
||||
print(error)
|
||||
print(f"TOTAL={len(errors)}")
|
||||
return 1
|
||||
|
||||
print(f"OK research raw facts checked: {len(domain_files)} domains")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,365 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fetch raw GitHub facts for docs/research repository domains."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
RESEARCH_DIR = ROOT / "docs" / "research"
|
||||
|
||||
REPO_VIEW_FIELDS = [
|
||||
"nameWithOwner",
|
||||
"url",
|
||||
"description",
|
||||
"homepageUrl",
|
||||
"isArchived",
|
||||
"isFork",
|
||||
"isMirror",
|
||||
"defaultBranchRef",
|
||||
"licenseInfo",
|
||||
"primaryLanguage",
|
||||
"repositoryTopics",
|
||||
"stargazerCount",
|
||||
"forkCount",
|
||||
"watchers",
|
||||
"createdAt",
|
||||
"updatedAt",
|
||||
"pushedAt",
|
||||
"latestRelease",
|
||||
]
|
||||
|
||||
|
||||
def run_gh(args: list[str]) -> tuple[bool, str]:
|
||||
result = subprocess.run(
|
||||
["gh", *args],
|
||||
cwd=ROOT,
|
||||
check=False,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
if result.returncode == 0:
|
||||
return True, result.stdout
|
||||
return False, result.stderr.strip() or result.stdout.strip()
|
||||
|
||||
|
||||
def run_git(args: list[str], cwd: Path = ROOT) -> tuple[bool, str]:
|
||||
result = subprocess.run(
|
||||
["git", *args],
|
||||
cwd=cwd,
|
||||
check=False,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
if result.returncode == 0:
|
||||
return True, result.stdout.strip()
|
||||
return False, result.stderr.strip() or result.stdout.strip()
|
||||
|
||||
|
||||
def read_repo_slug(domain_file: Path) -> str | None:
|
||||
text = domain_file.read_text(encoding="utf-8")
|
||||
match = re.search(r"^\s+url:\s+https://github\.com/([^\s#]+)\s*$", text, re.MULTILINE)
|
||||
if not match:
|
||||
return None
|
||||
return match.group(1).strip().rstrip("/")
|
||||
|
||||
|
||||
def read_domain_id(domain_file: Path) -> str:
|
||||
text = domain_file.read_text(encoding="utf-8")
|
||||
match = re.search(r"^\s+domain_id:\s+(.+?)\s*$", text, re.MULTILINE)
|
||||
if match:
|
||||
return unquote_yaml_value(match.group(1))
|
||||
return domain_file.parent.name
|
||||
|
||||
|
||||
def unquote_yaml_value(value: str) -> str:
|
||||
value = value.strip()
|
||||
if (value.startswith("'") and value.endswith("'")) or (value.startswith('"') and value.endswith('"')):
|
||||
return value[1:-1]
|
||||
return value
|
||||
|
||||
|
||||
def write_text(path: Path, content: str) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if not content.endswith("\n"):
|
||||
content += "\n"
|
||||
path.write_text(content, encoding="utf-8")
|
||||
|
||||
|
||||
def write_json(path: Path, content: str) -> bool:
|
||||
try:
|
||||
parsed = json.loads(content)
|
||||
except json.JSONDecodeError:
|
||||
write_text(path, content)
|
||||
return False
|
||||
write_text(path, json.dumps(parsed, ensure_ascii=False, indent=2))
|
||||
return True
|
||||
|
||||
|
||||
def raw_readme(repo_slug: str) -> str:
|
||||
return f"""# raw 事实层
|
||||
|
||||
本目录保存 `{repo_slug}` 的本地原始材料快照。
|
||||
|
||||
这里不写分析结论，只保存可复查的一手资料：
|
||||
|
||||
- `sources.yml`：来源清单、拉取时间、命令和文件状态。
|
||||
- `repository/`：Git 仓库工作树；本目录由 `.gitignore` 忽略，只作为本地研究材料。
|
||||
- `github-repo.raw.json`：GitHub 仓库元数据。
|
||||
- `github-readme.raw.md.txt`：GitHub README 原文快照。
|
||||
- `github-license.raw.txt`：GitHub license 原文快照；仓库无 license 时可能不存在。
|
||||
- `github-root-contents.raw.json`：默认分支根目录内容快照。
|
||||
- `github-languages.raw.json`：GitHub language 统计快照。
|
||||
- `github-latest-release.raw.json`：最新 release 快照；无 release 时可能不存在。
|
||||
|
||||
`repository/` 是外部源码快照，不参与本仓库 Markdown、链接、README/AGENTS 覆盖检查。原始 README 使用 `.txt` 后缀保存，避免其中的外部相对链接被本仓库 Markdown 链接检查误判。
|
||||
"""
|
||||
|
||||
|
||||
def raw_agents(repo_slug: str) -> str:
|
||||
return f"""# raw/ Agent 指南
|
||||
|
||||
本目录是 `{repo_slug}` 的原始事实层。
|
||||
|
||||
## 维护规则
|
||||
|
||||
- 只保存从研究对象拉取的一手材料，不写分析判断。
|
||||
- `repository/` 是本地 Git 工作树，刷新时由脚本 clone 或 fast-forward pull。
|
||||
- 不手工改写 `*.raw.*` 文件内容；需要刷新时运行 `python3 scripts/fetch-research-raw.py`。
|
||||
- `sources.yml` 必须记录拉取时间、来源命令和每个文件的状态。
|
||||
- 外部 README 原文必须保存为 `.txt`，避免本仓库 Markdown 链接检查误判。
|
||||
- 分析、判断、采用建议和沉淀路径写回上一级 `README.md`、`analysis.md` 或 `decisions.md`。
|
||||
"""
|
||||
|
||||
|
||||
def fetch_domain(domain_file: Path, pulled_at: str, force: bool) -> tuple[str, bool, list[str]]:
|
||||
repo_slug = read_repo_slug(domain_file)
|
||||
domain_id = read_domain_id(domain_file)
|
||||
if not repo_slug:
|
||||
return domain_id, False, [f"{domain_file}: missing github repo url"]
|
||||
|
||||
raw_dir = domain_file.parent / "raw"
|
||||
raw_dir.mkdir(parents=True, exist_ok=True)
|
||||
write_text(raw_dir / "README.md", raw_readme(repo_slug))
|
||||
write_text(raw_dir / "AGENTS.md", raw_agents(repo_slug))
|
||||
|
||||
records: list[dict[str, str]] = []
|
||||
errors: list[str] = []
|
||||
|
||||
def record(path: str, kind: str, status: str, source: str, message: str | None = None) -> None:
|
||||
item = {
|
||||
"path": path,
|
||||
"kind": kind,
|
||||
"status": status,
|
||||
"source": source,
|
||||
}
|
||||
if message:
|
||||
item["message"] = message.replace("\n", " ")[:240]
|
||||
records.append(item)
|
||||
|
||||
def fetch_file(path: str, command: list[str], *, kind: str, json_file: bool = False) -> None:
|
||||
target = raw_dir / path
|
||||
ok, output = run_gh(command)
|
||||
command_text = "gh " + " ".join(command)
|
||||
if ok:
|
||||
if json_file:
|
||||
write_json(target, output)
|
||||
else:
|
||||
write_text(target, output)
|
||||
record(path, kind, "ok", command_text)
|
||||
return
|
||||
|
||||
if force and target.exists():
|
||||
target.unlink()
|
||||
record(path, kind, "missing_or_error", command_text, output)
|
||||
if kind not in {"latest-release", "license"}:
|
||||
errors.append(f"{domain_id}: {path}: {output}")
|
||||
|
||||
def fetch_repository() -> None:
|
||||
repository_dir = raw_dir / "repository"
|
||||
clone_url = f"https://github.com/{repo_slug}.git"
|
||||
if (repository_dir / ".git").exists():
|
||||
ok, output = run_git(["-C", str(repository_dir), "pull", "--ff-only"])
|
||||
command_text = f"git -C {repository_dir.relative_to(ROOT)} pull --ff-only"
|
||||
if ok:
|
||||
record("repository/", "git-working-tree", "ok", command_text, output)
|
||||
else:
|
||||
record("repository/", "git-working-tree", "missing_or_error", command_text, output)
|
||||
errors.append(f"{domain_id}: repository pull failed: {output}")
|
||||
return
|
||||
|
||||
if repository_dir.exists():
|
||||
message = "repository path exists but is not a Git working tree"
|
||||
record("repository/", "git-working-tree", "missing_or_error", f"git clone {clone_url}", message)
|
||||
errors.append(f"{domain_id}: {message}")
|
||||
return
|
||||
|
||||
ok, output = run_git(
|
||||
[
|
||||
"clone",
|
||||
"--depth=1",
|
||||
"--single-branch",
|
||||
"--no-tags",
|
||||
clone_url,
|
||||
str(repository_dir),
|
||||
]
|
||||
)
|
||||
command_text = (
|
||||
f"git clone --depth=1 --single-branch --no-tags {clone_url} "
|
||||
f"{repository_dir.relative_to(ROOT)}"
|
||||
)
|
||||
if ok:
|
||||
record("repository/", "git-working-tree", "ok", command_text, output)
|
||||
else:
|
||||
record("repository/", "git-working-tree", "missing_or_error", command_text, output)
|
||||
errors.append(f"{domain_id}: repository clone failed: {output}")
|
||||
|
||||
fetch_repository()
|
||||
fetch_file(
|
||||
"github-repo.raw.json",
|
||||
[
|
||||
"repo",
|
||||
"view",
|
||||
repo_slug,
|
||||
"--json",
|
||||
",".join(REPO_VIEW_FIELDS),
|
||||
"--jq",
|
||||
".",
|
||||
],
|
||||
kind="repository-metadata",
|
||||
json_file=True,
|
||||
)
|
||||
fetch_file(
|
||||
"github-readme.raw.md.txt",
|
||||
[
|
||||
"api",
|
||||
f"repos/{repo_slug}/readme",
|
||||
"-H",
|
||||
"Accept: application/vnd.github.raw",
|
||||
],
|
||||
kind="readme",
|
||||
)
|
||||
fetch_file(
|
||||
"github-license.raw.txt",
|
||||
[
|
||||
"api",
|
||||
f"repos/{repo_slug}/license",
|
||||
"-H",
|
||||
"Accept: application/vnd.github.raw",
|
||||
],
|
||||
kind="license",
|
||||
)
|
||||
fetch_file(
|
||||
"github-root-contents.raw.json",
|
||||
[
|
||||
"api",
|
||||
f"repos/{repo_slug}/contents",
|
||||
],
|
||||
kind="root-contents",
|
||||
json_file=True,
|
||||
)
|
||||
fetch_file(
|
||||
"github-languages.raw.json",
|
||||
[
|
||||
"api",
|
||||
f"repos/{repo_slug}/languages",
|
||||
],
|
||||
kind="languages",
|
||||
json_file=True,
|
||||
)
|
||||
fetch_file(
|
||||
"github-latest-release.raw.json",
|
||||
[
|
||||
"api",
|
||||
f"repos/{repo_slug}/releases/latest",
|
||||
],
|
||||
kind="latest-release",
|
||||
json_file=True,
|
||||
)
|
||||
|
||||
write_text(raw_dir / "sources.yml", sources_yml(domain_id, repo_slug, pulled_at, records))
|
||||
return domain_id, not errors, errors
|
||||
|
||||
|
||||
def yaml_quote(value: str) -> str:
|
||||
escaped = value.replace("'", "''")
|
||||
return f"'{escaped}'"
|
||||
|
||||
|
||||
def sources_yml(domain_id: str, repo_slug: str, pulled_at: str, records: list[dict[str, str]]) -> str:
|
||||
lines = [
|
||||
"version: 1",
|
||||
f"domain_id: {domain_id}",
|
||||
"object:",
|
||||
" type: github_repository",
|
||||
f" name: {repo_slug}",
|
||||
f" url: https://github.com/{repo_slug}",
|
||||
f"pulled_at: {yaml_quote(pulled_at)}",
|
||||
"puller: scripts/fetch-research-raw.py",
|
||||
"files:",
|
||||
]
|
||||
for record in records:
|
||||
lines.extend(
|
||||
[
|
||||
f" - path: {record['path']}",
|
||||
f" kind: {record['kind']}",
|
||||
f" status: {record['status']}",
|
||||
f" source: {yaml_quote(record['source'])}",
|
||||
]
|
||||
)
|
||||
if "message" in record:
|
||||
lines.append(f" message: {yaml_quote(record['message'])}")
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def domain_files(selected: set[str] | None) -> list[Path]:
|
||||
files = sorted(RESEARCH_DIR.glob("*/domain.yml"))
|
||||
if not selected:
|
||||
return files
|
||||
return [path for path in files if path.parent.name in selected or read_repo_slug(path) in selected]
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("domains", nargs="*", help="Optional domain id or owner/repo slug to fetch.")
|
||||
parser.add_argument("--force", action="store_true", help="Remove stale optional files when fetch fails.")
|
||||
args = parser.parse_args()
|
||||
|
||||
selected = set(args.domains) if args.domains else None
|
||||
files = domain_files(selected)
|
||||
if not files:
|
||||
print("No research domain.yml files found.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
pulled_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||||
all_errors: list[str] = []
|
||||
ok_count = 0
|
||||
for path in files:
|
||||
domain_id, ok, errors = fetch_domain(path, pulled_at, args.force)
|
||||
if ok:
|
||||
ok_count += 1
|
||||
all_errors.extend(errors)
|
||||
print(f"{domain_id}: {'OK' if ok else 'WARN'}")
|
||||
|
||||
if all_errors:
|
||||
print("FETCH_RESEARCH_RAW_WARNINGS")
|
||||
for error in all_errors:
|
||||
print(error)
|
||||
print(f"OK={ok_count} TOTAL={len(files)}")
|
||||
return 1
|
||||
|
||||
print(f"OK fetched raw research materials: {ok_count} domains")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user