init: edu-content 内容库骨架（L1-L4 四层数据模型 + B类迁移 + lint/CI 工具）

tools/oss_sync.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""oss_sync.py — 上传 L2 原始文件到 OSS 并回填 sha256/size/fetch_date。
+
+用法：
+  python tools/oss_sync.py --raw-root <DIR> --dry-run      # 仅报告将要上传的文件
+  EDU_OSS_KEY_ID=... EDU_OSS_KEY_SECRET=... \\
+  python tools/oss_sync.py --raw-root <DIR> --apply        # 实际上传 + 回填 YAML
+
+约定：
+- 本地 DIR 结构：<raw-root>/{source-id}/{filename...}，与 OSS key 的
+  edu-raw/{k9|exams|internal}/{学科或考试}/{类目}/{source-id}/{filename} 对应。
+- 上传后自动把 oss_key / sha256 / size / fetch_date 写回对应 source YAML 的 item。
+
+凭据：从环境变量 EDU_OSS_KEY_ID / EDU_OSS_KEY_SECRET / EDU_OSS_BUCKET / EDU_OSS_ENDPOINT 读取，
+      绝不写死、不进 git。缺省 bucket=edu-content-hy，endpoint=https://oss-cn-heyuan.aliyuncs.com。
+"""
+import os, sys, glob, json, hashlib, datetime, argparse, yaml
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+SHA_RE = __import__("re").compile(r"^[a-f0-9]{64}$")
+
+def sha256_file(p):
+    h = hashlib.sha256()
+    with open(p, "rb") as f:
+        for chunk in iter(lambda: f.read(1 << 20), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+def load_sources():
+    import yaml
+    docs = []
+    for path in glob.glob(os.path.join(REPO, "sources", "**", "*.yaml"), recursive=True):
+        try: doc = yaml.safe_load(open(path, encoding="utf-8"))
+        except Exception: continue
+        docs.append((path, doc))
+    return docs
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--raw-root", required=True)
+    ap.add_argument("--dry-run", action="store_true")
+    ap.add_argument("--apply", action="store_true")
+    args = ap.parse_args()
+    if not os.path.isdir(args.raw_root):
+        print("raw-root 不存在:", args.raw_root); sys.exit(2)
+
+    docs = load_sources()
+    pending = []
+    for path, doc in docs:
+        for s in (doc or {}).get("sources", []) or []:
+            sid = s.get("id")
+            for it in s.get("items", []) or []:
+                # 本地文件：raw-root/{sid}/{filename}
+                local = os.path.join(args.raw_root, sid, os.path.basename(it.get("oss_key") or it.get("title") or it.get("id")))
+                if os.path.isfile(local):
+                    sha = sha256_file(local)
+                    oss_key = it.get("oss_key") or f"edu-raw/exams/_staging/{sid}/{os.path.basename(local)}"
+                    pending.append((path, sid, it.get("id"), local, oss_key, sha, os.path.getsize(local)))
+
+    if not pending:
+        print("无可上传文件（raw-root 下未匹配到 source item 对应的本地文件）。")
+        sys.exit(0)
+
+    print(f"待上传 {len(pending)} 个文件：")
+    for p, sid, iid, local, ok, sha, size in pending:
+        print(f"  {sid}/{iid}: {local} -> {ok} ({size}B, sha256={sha[:12]}...)")
+
+    if args.dry_run or not args.apply:
+        print("\n[dry-run] 未实际上传。加 --apply 执行。")
+        sys.exit(0)
+
+    # 实际上传
+    try:
+        import oss2
+    except Exception as e:
+        print("oss2 不可用，无法上传:", e); sys.exit(3)
+    key_id = os.environ.get("EDU_OSS_KEY_ID")
+    key_secret = os.environ.get("EDU_OSS_KEY_SECRET")
+    bucket = os.environ.get("EDU_OSS_BUCKET", "edu-content-hy")
+    endpoint = os.environ.get("EDU_OSS_ENDPOINT", "https://oss-cn-heyuan.aliyuncs.com")
+    if not (key_id and key_secret):
+        print("缺少 EDU_OSS_KEY_ID / EDU_OSS_KEY_SECRET 环境变量"); sys.exit(3)
+    auth = oss2.Auth(key_id, key_secret)
+    b = oss2.Bucket(auth, endpoint, bucket)
+    today = datetime.date.today().isoformat()
+    # 上传 + 回填 YAML
+    for path, sid, iid, local, oss_key, sha, size in pending:
+        b.put_object_from_file(oss_key, local)
+        print(f"  uploaded {oss_key}")
+        # 回填
+        doc = yaml.safe_load(open(path, encoding="utf-8"))
+        for s in doc["sources"]:
+            if s.get("id") == sid:
+                for it in s.get("items", []):
+                    if it.get("id") == iid:
+                        it["oss_key"] = oss_key
+                        it["sha256"] = sha
+                        it["size"] = size
+                        it["fetch_date"] = today
+                        it["status"] = "fetched"
+        yaml.safe_dump(doc, open(path, "w", encoding="utf-8"), allow_unicode=True, sort_keys=False)
+    print("已回填 sha256/size 到 sources YAML。请 git commit。")
+
+if __name__ == "__main__":
+    main()