Drjunchenfeng
a006c8c504
一些检查失败了
ci / lint (push) Failing after 52s
登记 chinese-poetry / ck12 / phet，并让 source schema、lint 与 INDEX 能按学科和抓取进度消费。 Co-authored-by: Cursor <cursoragent@cursor.com>
153 行
4.9 KiB
Python
153 行
4.9 KiB
Python
#!/usr/bin/env python3
|
||
"""yaml_patch.py — 对 sources/*.yaml 做**行级**字段回填，保留注释与原有格式。
|
||
|
||
为什么需要它：
|
||
`yaml.safe_load` + `yaml.safe_dump` 会把整个文件重写，**注释全部丢失**。
|
||
sources/*.yaml 里的注释承载了关键信息（下载模板、许可解释、未纳入原因、运维提示），
|
||
必须保住。因此所有自动回填（fetch_source.py / oss_sync.py）统一走本模块的行级替换。
|
||
|
||
约定的缩进层级（与 sources/*.yaml 一致）：
|
||
sources:
|
||
- id: <source_id> # 2 空格 + "- "
|
||
status: planned # 4 空格
|
||
items:
|
||
- id: <item_id> # 6 空格 + "- "
|
||
oss_key: ... # 8 空格
|
||
"""
|
||
import re
|
||
|
||
|
||
def _yaml_scalar(v):
|
||
"""把 Python 值渲染成 YAML 标量。"""
|
||
if v is None:
|
||
return "null"
|
||
if isinstance(v, bool):
|
||
return "true" if v else "false"
|
||
if isinstance(v, (int, float)):
|
||
return str(v)
|
||
s = str(v)
|
||
# 需要引号的情况：空串、含特殊起始字符、含冒号空格、纯数字样式
|
||
if s == "" or s[0] in "&*?|-<>=!%@`{[\"'" or ": " in s or s.endswith(":"):
|
||
return '"' + s.replace('\\', '\\\\').replace('"', '\\"') + '"'
|
||
return s
|
||
|
||
|
||
def patch_item_fields(path, source_id, item_id, fields):
|
||
"""回填某 source 下某 item 的字段。返回 True 表示有改动。
|
||
|
||
fields: dict，如 {"sha256": "ab..", "size": 123, "status": "fetched"}
|
||
已存在的键 → 原地替换；不存在的键 → 追加到该 item 块末尾。
|
||
"""
|
||
with open(path, encoding="utf-8") as f:
|
||
lines = f.read().split("\n")
|
||
|
||
src_re = re.compile(r"^ - id: (\S+)\s*$")
|
||
item_re = re.compile(r"^ - id: (\S+)\s*$")
|
||
|
||
# 1) 定位 source 块范围
|
||
s_start = s_end = None
|
||
for i, ln in enumerate(lines):
|
||
m = src_re.match(ln)
|
||
if m:
|
||
if m.group(1) == source_id and s_start is None:
|
||
s_start = i
|
||
elif s_start is not None and s_end is None:
|
||
s_end = i
|
||
break
|
||
if s_start is None:
|
||
return False
|
||
if s_end is None:
|
||
s_end = len(lines)
|
||
|
||
# 2) 在 source 块内定位 item 块范围
|
||
i_start = i_end = None
|
||
for i in range(s_start, s_end):
|
||
m = item_re.match(lines[i])
|
||
if m:
|
||
if m.group(1) == item_id and i_start is None:
|
||
i_start = i
|
||
elif i_start is not None and i_end is None:
|
||
i_end = i
|
||
break
|
||
if i_start is None:
|
||
return False
|
||
if i_end is None:
|
||
# item 块延伸到 source 块末尾；剔除尾部空行与注释行
|
||
j = s_end - 1
|
||
while j > i_start and (lines[j].strip() == "" or lines[j].lstrip().startswith("#")):
|
||
j -= 1
|
||
i_end = j + 1
|
||
|
||
# 3) 逐字段替换或追加
|
||
changed = False
|
||
for key, val in fields.items():
|
||
new_line = f" {key}: {_yaml_scalar(val)}"
|
||
hit = None
|
||
for i in range(i_start, i_end):
|
||
if re.match(rf"^ {re.escape(key)}:", lines[i]):
|
||
hit = i
|
||
break
|
||
if hit is not None:
|
||
if lines[hit] != new_line:
|
||
lines[hit] = new_line
|
||
changed = True
|
||
else:
|
||
lines.insert(i_end, new_line)
|
||
i_end += 1
|
||
changed = True
|
||
|
||
if changed:
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(lines))
|
||
return changed
|
||
|
||
|
||
def patch_source_fields(path, source_id, fields):
|
||
"""回填 source 级字段（如 status / fetched_at）。返回 True 表示有改动。"""
|
||
with open(path, encoding="utf-8") as f:
|
||
lines = f.read().split("\n")
|
||
|
||
src_re = re.compile(r"^ - id: (\S+)\s*$")
|
||
s_start = s_end = None
|
||
for i, ln in enumerate(lines):
|
||
m = src_re.match(ln)
|
||
if m:
|
||
if m.group(1) == source_id and s_start is None:
|
||
s_start = i
|
||
elif s_start is not None and s_end is None:
|
||
s_end = i
|
||
break
|
||
if s_start is None:
|
||
return False
|
||
if s_end is None:
|
||
s_end = len(lines)
|
||
|
||
# items: 之前才是 source 级字段区
|
||
limit = s_end
|
||
for i in range(s_start, s_end):
|
||
if re.match(r"^ items:", lines[i]):
|
||
limit = i
|
||
break
|
||
|
||
changed = False
|
||
for key, val in fields.items():
|
||
new_line = f" {key}: {_yaml_scalar(val)}"
|
||
hit = None
|
||
for i in range(s_start, limit):
|
||
if re.match(rf"^ {re.escape(key)}:", lines[i]):
|
||
hit = i
|
||
break
|
||
if hit is not None:
|
||
if lines[hit] != new_line:
|
||
lines[hit] = new_line
|
||
changed = True
|
||
else:
|
||
lines.insert(limit, new_line)
|
||
limit += 1
|
||
changed = True
|
||
|
||
if changed:
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(lines))
|
||
return changed
|