61 行
3.1 KiB
Python
61 行
3.1 KiB
Python
#!/usr/bin/env python3
|
||
"""lint_sources.py — 校验 sources/ 资源清单合规。
|
||
|
||
检查项：
|
||
- 每个 source 必填字段、id kebab-case 唯一、license/type/status 枚举合法
|
||
- status=fetched 的 item 必须有 oss_key + 64位 sha256
|
||
- license=unknown 且 fetched_at 距今 > 30 天 → warning
|
||
|
||
退出码非 0 表示有 error（CI 应标红）。
|
||
"""
|
||
import sys, os, glob, re, datetime, hashlib
|
||
|
||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
LICENSES = {"official-free","cc-by","cc-by-sa","cc-nc-education","restricted","internal-only","llm-generated","unknown"}
|
||
TYPES = {"dataset","website","book","internal","tool"}
|
||
STATUSES = {"planned","fetching","fetched","failed","archived"}
|
||
ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
|
||
SHA_RE = re.compile(r"^[a-f0-9]{64}$")
|
||
|
||
def err(msg): print("ERROR: " + msg)
|
||
def warn(msg): print("WARN: " + msg)
|
||
|
||
def main():
|
||
errors, warnings = 0, 0
|
||
seen_ids = {}
|
||
for path in sorted(glob.glob(os.path.join(REPO, "sources", "**", "*.yaml"), recursive=True)):
|
||
import yaml
|
||
try:
|
||
doc = yaml.safe_load(open(path, encoding="utf-8"))
|
||
except Exception as e:
|
||
err(f"{path}: YAML 解析失败: {e}"); errors += 1; continue
|
||
if not doc or "sources" not in doc:
|
||
err(f"{path}: 缺少 sources 列表"); errors += 1; continue
|
||
for s in doc["sources"]:
|
||
sid = s.get("id")
|
||
if not sid: err(f"{path}: source 缺 id"); errors += 1; continue
|
||
if not ID_RE.match(sid): err(f"{sid}: id 非 kebab-case"); errors += 1
|
||
if sid in seen_ids: err(f"{sid}: id 重复 ({seen_ids[sid]} 与 {path})"); errors += 1
|
||
else: seen_ids[sid] = path
|
||
for f in ("title","publisher","type","license","status"):
|
||
if not s.get(f): err(f"{sid}: 缺必填 {f}"); errors += 1
|
||
if s.get("type") not in TYPES: err(f"{sid}: type 非法 {s.get('type')}"); errors += 1
|
||
if s.get("license") not in LICENSES: err(f"{sid}: license 非法 {s.get('license')}"); errors += 1
|
||
if s.get("status") not in STATUSES: err(f"{sid}: status 非法 {s.get('status')}"); errors += 1
|
||
if s.get("license") == "unknown" and s.get("fetched_at"):
|
||
try:
|
||
d = datetime.date.fromisoformat(s["fetched_at"])
|
||
if (datetime.date.today() - d).days > 30: warn(f"{sid}: license=unknown 已超 30 天，请核查"); warnings += 1
|
||
except Exception: pass
|
||
for it in s.get("items", []) or []:
|
||
iid = it.get("id")
|
||
if not ID_RE.match(iid or ""): err(f"{sid}/{iid}: item id 非 kebab-case"); errors += 1
|
||
if s.get("status") == "fetched" and it.get("status") == "fetched":
|
||
if not it.get("oss_key"): err(f"{sid}/{iid}: fetched 但缺 oss_key"); errors += 1
|
||
if not SHA_RE.match(str(it.get("sha256") or "")): err(f"{sid}/{iid}: sha256 非 64 位 hex"); errors += 1
|
||
print(f"\n校验完成：{len(seen_ids)} 条 source | {errors} error | {warnings} warn")
|
||
sys.exit(1 if errors else 0)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|