235 lines
9.9 KiB
Python
235 lines
9.9 KiB
Python
# sanguo_factor/data_gap_check.py
|
|
"""数据缺口自动比对(决议 J/§11.3-②): 因子注册表数据需求 vs 底座 manifest.
|
|
|
|
需求=注册表最新版本 params.source(必填约定,J2 血统闸的数据侧);
|
|
供给=config/data_manifest.yaml(首版手工,data session 可接管自动生成).
|
|
缺口→data_gaps.json+exit 2(调度层可开 issue 转 data session 接源);
|
|
本件只比对不接源(接源归 data session+人).
|
|
契约对齐(P2-11): 本 CLI 缺 manifest=fail-fast exit 1 且不落文件(独立调度
|
|
场景,无批可跑);月度批(monthly_batch)同情形 fail-soft——落
|
|
{"manifest_missing": true} 标记件+stderr 告警照跑批。两处 data_gaps.json
|
|
同位(本件 out_dir / 月度批 report_dir 下)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import date
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from sanguo_factor.version_registry import GITEA_BASE_URL, load_registry, \
|
|
open_gitea_issue
|
|
|
|
|
|
def load_manifest(path: str) -> dict[str, Any]:
|
|
with open(path, encoding="utf-8") as f:
|
|
return yaml.safe_load(f) or {}
|
|
|
|
|
|
def collect_requirements(registry: dict) -> dict[str, list[str]]:
|
|
"""每因子数据需求=最新版本 params.source(无版本/无 source→空=不比对)."""
|
|
out: dict[str, list[str]] = {}
|
|
for name, entry in registry["factors"].items():
|
|
versions = entry.get("versions") or [{}]
|
|
src = (versions[-1].get("params") or {}).get("source")
|
|
out[name] = [src] if src else []
|
|
return out
|
|
|
|
|
|
def check_gaps(requirements: dict[str, list[str]],
|
|
manifest: dict[str, Any]) -> dict[str, list[str]]:
|
|
available = set(manifest.get("domains") or {})
|
|
return {n: [s for s in srcs if s not in available]
|
|
for n, srcs in requirements.items()
|
|
if any(s not in available for s in srcs)}
|
|
|
|
|
|
# —— 决议 M 半环: 缺口自动开单(指纹去重;复用 J1 open_gitea_issue,零新依赖) ——
|
|
def gap_fingerprint(gaps: dict[str, list[str]]) -> str:
|
|
"""缺口集合指纹: '因子:源' 排序行 sha1 前 8 位(集合变化=新指纹=新单)."""
|
|
lines = sorted(f"{f}:{s}" for f, srcs in gaps.items() for s in srcs)
|
|
return hashlib.sha1("\n".join(lines).encode("utf-8")).hexdigest()[:8]
|
|
|
|
|
|
def render_gap_issue(gaps: dict[str, list[str]], manifest_path: str) -> tuple[str, str]:
|
|
fp = gap_fingerprint(gaps)
|
|
title = f"[数据缺口] {len(gaps)} 因子待接源 [{fp}]"
|
|
body = [f"注册表数据需求 vs `{manifest_path}` 供给比对缺口"
|
|
"(决议 M:缺口自动开单进 data 待办):", ""]
|
|
for f in sorted(gaps):
|
|
body.append(f"- **{f}**: 待接 {', '.join(gaps[f])}")
|
|
body += ["", "处置: data session 接源后在 data_manifest.yaml 补该域,"
|
|
"缺口自愈(下月批自动重试,无须人工关单——新指纹才开新单)。",
|
|
f"指纹 `{fp}`。"]
|
|
return title, "\n".join(body)
|
|
|
|
|
|
class GiteaLookupError(RuntimeError):
|
|
"""open issue 查询网络异常(≠「确认不存在」)——调用方须跳过开单本月."""
|
|
|
|
|
|
# 指纹本地台账(append-only,每行一 JSON):第二道去重,网络盲区/失明时兜底.
|
|
GAP_ISSUE_LEDGER = "data/gap_issue_ledger.jsonl"
|
|
|
|
|
|
def find_open_issue(title: str, token: str, base_url: str | None = None,
|
|
repo: str = "sanguo/sanguo_vnpy_v2") -> int | None:
|
|
"""精确标题匹配 open issue,双层查询(P2-13①).
|
|
|
|
首页 LIST 50 条精确匹配→未中再 search API 兜底(按标题指纹 sha1 前 8 位
|
|
搜,破「目标单跌出首页→静默失明→同指纹月月重复开单」)。
|
|
返回 None=双层皆空=确认不存在;LIST/search 任一网络异常抛
|
|
GiteaLookupError(≠不存在,调用方异常时跳过开单本月,下月再试)。
|
|
"""
|
|
import urllib.parse
|
|
import urllib.request
|
|
if base_url is None:
|
|
base_url = os.environ.get("SANGUO_GITEA_BASE_URL", GITEA_BASE_URL)
|
|
|
|
def _get(url: str) -> list:
|
|
req = urllib.request.Request(
|
|
url, headers={"Authorization": f"token {token}"})
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
try:
|
|
list_url = (f"{base_url}/api/v1/repos/{repo}/issues"
|
|
"?state=open&type=issues&limit=50")
|
|
for it in _get(list_url):
|
|
if it.get("title") == title:
|
|
return it["number"]
|
|
m = re.search(r"\[([0-9a-f]{8})\]", title)
|
|
q = m.group(1) if m else title
|
|
search_url = (f"{base_url}/api/v1/repos/{repo}/issues"
|
|
f"?state=open&type=issues&q={urllib.parse.quote(q)}")
|
|
for it in _get(search_url):
|
|
if it.get("title") == title:
|
|
return it["number"]
|
|
except GiteaLookupError:
|
|
raise
|
|
except Exception as e:
|
|
raise GiteaLookupError(str(e)) from e
|
|
return None
|
|
|
|
|
|
def load_ledger_fps(path: str) -> set[str]:
|
|
"""台账已开指纹集合(坏行/空行跳过;文件缺=空集;读失败降级空集+stderr)."""
|
|
if not os.path.exists(path):
|
|
return set()
|
|
fps: set[str] = set()
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
fp = json.loads(line).get("fingerprint")
|
|
except json.JSONDecodeError:
|
|
continue
|
|
except AttributeError:
|
|
print(f"[data_gap_check] 台账坏行(合法 JSON 非 dict)跳过: "
|
|
f"{line}", file=sys.stderr) # T6-N1 降级留痕
|
|
continue
|
|
if fp:
|
|
fps.add(fp)
|
|
except OSError as e:
|
|
print(f"[data_gap_check] 台账读取失败(降级视为空集): {e}", file=sys.stderr)
|
|
return fps
|
|
|
|
|
|
def append_ledger(path: str, fingerprint: str, issue_number: int,
|
|
opened_at: str) -> None:
|
|
"""台账 append 一行(指纹/issue号/日期);写失败 stderr 降级不阻断开单."""
|
|
try:
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
with open(path, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps({"fingerprint": fingerprint,
|
|
"issue": issue_number,
|
|
"opened_at": opened_at},
|
|
ensure_ascii=False) + "\n")
|
|
except OSError as e:
|
|
print(f"[data_gap_check] 台账写入失败(降级不阻断): {e}", file=sys.stderr)
|
|
|
|
|
|
def maybe_open_gap_issue(gaps: dict[str, list[str]],
|
|
manifest_path: str,
|
|
ledger_path: str = GAP_ISSUE_LEDGER,
|
|
host: str | None = None) -> int | None:
|
|
"""有 token 且同指纹无 open 单(台账+网络双层去重)才开;失败降级返 None.
|
|
|
|
去重三重(P2-13): ①本地台账命中→跳过不打网络;②open issue 双层查询
|
|
(首页 LIST+search 兜底)命中→跳过;③查询网络异常→跳过开单本月
|
|
(stderr 留痕,下月再试)——只有「确认不存在」才真正开单,
|
|
开单成功后回写台账。
|
|
P3-14 开单权威单机制: host="vps" 结构性拒开(开单仅 NAS 单侧权威——
|
|
台账本地不跨机,双机同查同开窗口就此关死);host=None(人工运行)不限.
|
|
"""
|
|
if host == "vps":
|
|
print("[data_gap_check] host=vps 非开单权威侧(P3-14 权威单机制:"
|
|
"NAS 单侧开单),跳过开单", file=sys.stderr)
|
|
return None
|
|
token = os.environ.get("SANGUO_GITEA_TOKEN")
|
|
if not token:
|
|
return None
|
|
fp = gap_fingerprint(gaps)
|
|
if fp in load_ledger_fps(ledger_path):
|
|
print(f"[data_gap_check] 台账命中指纹 {fp}(issue 已开过),跳过开单",
|
|
file=sys.stderr)
|
|
return None
|
|
title, body = render_gap_issue(gaps, manifest_path)
|
|
try:
|
|
if find_open_issue(title, token) is not None:
|
|
return None
|
|
except GiteaLookupError as e:
|
|
print(f"[data_gap_check] Gitea 查询异常,本月跳过开单(下月再试): {e}",
|
|
file=sys.stderr)
|
|
return None
|
|
num = open_gitea_issue(title, body, token)
|
|
if num:
|
|
append_ledger(ledger_path, fp, num, date.today().isoformat())
|
|
return num
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
ap = argparse.ArgumentParser(description="数据缺口比对(需求=注册表,供给=manifest)")
|
|
ap.add_argument("--registry", default="config/factor_registry.yaml")
|
|
ap.add_argument("--manifest", default="config/data_manifest.yaml")
|
|
ap.add_argument("--out-dir", default="reports/factor_monthly")
|
|
ap.add_argument("--gitea-open", action="store_true",
|
|
help="有缺口时尝试开 Gitea issue(需 env SANGUO_GITEA_TOKEN;"
|
|
"同指纹 open 单存在则跳过)")
|
|
ap.add_argument("--host", choices=("nas", "vps"), default=None,
|
|
help="运行侧别声明(P3-14 开单权威单机制: vps 侧结构性"
|
|
"拒开缺口单,NAS=权威侧;缺省=人工运行不限)")
|
|
args = ap.parse_args(argv)
|
|
if not os.path.exists(args.manifest):
|
|
print(f"[data_gap_check] manifest 不存在: {args.manifest}", file=sys.stderr)
|
|
return 1
|
|
reqs = collect_requirements(load_registry(args.registry))
|
|
gaps = check_gaps(reqs, load_manifest(args.manifest))
|
|
os.makedirs(args.out_dir, exist_ok=True)
|
|
with open(os.path.join(args.out_dir, "data_gaps.json"), "w",
|
|
encoding="utf-8") as f:
|
|
json.dump(gaps, f, ensure_ascii=False, indent=1)
|
|
if gaps:
|
|
print(f"[data_gap_check] {len(gaps)} 因子有数据缺口(详见 data_gaps.json): "
|
|
+ ", ".join(sorted(gaps)))
|
|
if args.gitea_open:
|
|
num = maybe_open_gap_issue(gaps, args.manifest, host=args.host)
|
|
if num:
|
|
print(f"[data_gap_check] 已开数据缺口 issue #{num}")
|
|
return 2
|
|
print("[data_gap_check] 无缺口")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|