diff --git a/sanguo_factor/data_gap_check.py b/sanguo_factor/data_gap_check.py index 1d69e029..d4115087 100644 --- a/sanguo_factor/data_gap_check.py +++ b/sanguo_factor/data_gap_check.py @@ -16,12 +16,15 @@ 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 load_registry, open_gitea_issue +from sanguo_factor.version_registry import GITEA_BASE_URL, load_registry, \ + open_gitea_issue def load_manifest(path: str) -> dict[str, Any]: @@ -67,35 +70,120 @@ def render_gap_issue(gaps: dict[str, list[str]], manifest_path: str) -> tuple[st 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(仓库 issue 量小,首页 50 足够;失败返 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", - "http://192.168.2.154:3000") - url = f"{base_url}/api/v1/repos/{repo}/issues?state=open&type=issues&limit=50" - req = urllib.request.Request(url, headers={"Authorization": f"token {token}"}) - try: + 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: - for it in json.loads(resp.read().decode("utf-8")): - if it.get("title") == title: - return it["number"] - except Exception: - return None + 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 + 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) -> int | None: - """有 token 且同指纹无 open 单才开;一切失败静默降级返 None(告警不依赖网络).""" + manifest_path: str, + ledger_path: str = GAP_ISSUE_LEDGER) -> int | None: + """有 token 且同指纹无 open 单(台账+网络双层去重)才开;失败降级返 None. + + 去重三重(P2-13): ①本地台账命中→跳过不打网络;②open issue 双层查询 + (首页 LIST+search 兜底)命中→跳过;③查询网络异常→跳过开单本月 + (stderr 留痕,下月再试)——只有「确认不存在」才真正开单, + 开单成功后回写台账。 + """ token = os.environ.get("SANGUO_GITEA_TOKEN") if not token: return None - title, body = render_gap_issue(gaps, manifest_path) - if find_open_issue(title, token) is not 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 - return open_gitea_issue(title, body, token) + 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: diff --git a/sanguo_factor/version_registry.py b/sanguo_factor/version_registry.py index 4664da14..eec46095 100644 --- a/sanguo_factor/version_registry.py +++ b/sanguo_factor/version_registry.py @@ -21,6 +21,10 @@ import yaml VALID_STATUSES = frozenset( {"incubating", "assessable", "promoted", "decaying", "retired", "graveyard"}) +# 内网 Gitea 缺省地址(B' 双机: NAS=内网 IP, VPS=FRP 127.0.0.1:13000)—— +# 单源常量,data_gap_check 等同仓消费方 import 此处,禁再抄字面量(P3-20). +GITEA_BASE_URL = "http://192.168.2.154:3000" + # 合法迁移:普通链路逐级+跳过 assessable 的快车道不存在;graveyard 终态不可复活. _VALID_TRANSITIONS: dict[str, frozenset[str]] = { "incubating": frozenset({"assessable", "graveyard"}), @@ -216,8 +220,7 @@ def open_gitea_issue(title: str, body: str, token: str, """ import urllib.request if base_url is None: - base_url = os.environ.get("SANGUO_GITEA_BASE_URL", - "http://192.168.2.154:3000") + base_url = os.environ.get("SANGUO_GITEA_BASE_URL", GITEA_BASE_URL) url = f"{base_url}/api/v1/repos/{repo}/issues" payload = json.dumps({"title": title, "body": body}).encode("utf-8") req = urllib.request.Request( diff --git a/tests/factor/test_data_gap_check.py b/tests/factor/test_data_gap_check.py index ae5a6c72..52708ad5 100644 --- a/tests/factor/test_data_gap_check.py +++ b/tests/factor/test_data_gap_check.py @@ -69,25 +69,29 @@ def test_render_gap_issue_shape(): assert "data session" in body -def test_maybe_open_dedup_skips_existing(monkeypatch): +def test_maybe_open_dedup_skips_existing(monkeypatch, tmp_path): import sanguo_factor.data_gap_check as dgc monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok") monkeypatch.setattr(dgc, "find_open_issue", lambda *a, **k: 42) opened = [] monkeypatch.setattr(dgc, "open_gitea_issue", lambda *a, **k: opened.append(a) or 99) - assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml") is None + assert dgc.maybe_open_gap_issue( + {"fa_x": ["s_new"]}, "m.yaml", + ledger_path=str(tmp_path / "ledger.jsonl")) is None assert opened == [] # 同指纹在册→跳过 -def test_maybe_open_creates_when_absent(monkeypatch): +def test_maybe_open_creates_when_absent(monkeypatch, tmp_path): import sanguo_factor.data_gap_check as dgc monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok") monkeypatch.setattr(dgc, "find_open_issue", lambda *a, **k: None) opened = [] monkeypatch.setattr(dgc, "open_gitea_issue", lambda title, body, token, **k: opened.append(title) or 7) - assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml") == 7 + assert dgc.maybe_open_gap_issue( + {"fa_x": ["s_new"]}, "m.yaml", + ledger_path=str(tmp_path / "ledger.jsonl")) == 7 assert opened and opened[0].startswith("[数据缺口]") @@ -116,3 +120,173 @@ def test_find_open_issue_matches_exact_title(monkeypatch): monkeypatch.setattr(urllib.request, "urlopen", lambda req, timeout=None: FakeResp()) assert dgc.find_open_issue("[数据缺口] 1 因子待接源 [abcd1234]", "tok") == 11 assert dgc.find_open_issue("不在册标题", "tok") is None + + +# —— P2-13 三重加固: search 兜底/异常≠不存在/本地台账 + P3-20 地址单源 —— +import json as _json + + +class _JsonResp: + """按给定 items 回 JSON 的 urlopen 假响应.""" + + def __init__(self, items): + self._items = items + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return _json.dumps(self._items).encode() + + +def test_gitea_base_url_single_source(): + """P3-20: data_gap_check 的内网地址必须 import 自 version_registry(同一对象).""" + import sanguo_factor.data_gap_check as dgc + import sanguo_factor.version_registry as vr + assert dgc.GITEA_BASE_URL is vr.GITEA_BASE_URL + assert vr.GITEA_BASE_URL == "http://192.168.2.154:3000" + + +def test_find_open_issue_list_hit_no_search(monkeypatch): + import sanguo_factor.data_gap_check as dgc + urls = [] + + def fake(req, timeout=None): + urls.append(req.full_url) + return _JsonResp([{"number": 3, + "title": "[数据缺口] 1 因子待接源 [abcd1234]"}]) + + monkeypatch.setattr("urllib.request.urlopen", fake) + assert dgc.find_open_issue( + "[数据缺口] 1 因子待接源 [abcd1234]", "tok") == 3 + assert len(urls) == 1 and "limit=50" in urls[0] # 首页命中不触发 search + + +def test_find_open_issue_search_fallback_beyond_first_page(monkeypatch): + """首页 50 条塞满他人单,目标单在首页外——search API 按指纹兜底命中.""" + import sanguo_factor.data_gap_check as dgc + title = "[数据缺口] 1 因子待接源 [abcd1234]" + urls = [] + + def fake(req, timeout=None): + urls.append(req.full_url) + if "limit=50" in req.full_url: + return _JsonResp([{"number": i, "title": f"占位单{i}"} + for i in range(50)]) + return _JsonResp([{"number": 88, "title": title}]) + + monkeypatch.setattr("urllib.request.urlopen", fake) + assert dgc.find_open_issue(title, "tok") == 88 + assert len(urls) == 2 + assert "q=abcd1234" in urls[1] # search 按指纹兜底 + + +def test_find_open_issue_both_miss_confirmed_absent(monkeypatch): + import sanguo_factor.data_gap_check as dgc + + def fake(req, timeout=None): + return _JsonResp([]) + + monkeypatch.setattr("urllib.request.urlopen", fake) + assert dgc.find_open_issue("不在册标题", "tok") is None # 双 miss=确认不存在 + + +def test_find_open_issue_list_error_raises_not_none(monkeypatch): + """LIST 网络异常≠不存在——抛 GiteaLookupError 供调用方跳过开单.""" + import pytest + + import sanguo_factor.data_gap_check as dgc + + def boom(req, timeout=None): + raise OSError("connection refused") + + monkeypatch.setattr("urllib.request.urlopen", boom) + with pytest.raises(dgc.GiteaLookupError): + dgc.find_open_issue("[数据缺口] 1 因子待接源 [abcd1234]", "tok") + + +def test_find_open_issue_search_error_raises(monkeypatch): + """search 异常同义:异常≠空,归 GiteaLookupError(调用方跳过开单本月).""" + import pytest + + import sanguo_factor.data_gap_check as dgc + + def fake(req, timeout=None): + if "limit=50" in req.full_url: + return _JsonResp([]) + raise OSError("search timeout") + + monkeypatch.setattr("urllib.request.urlopen", fake) + with pytest.raises(dgc.GiteaLookupError): + dgc.find_open_issue("[数据缺口] 1 因子待接源 [abcd1234]", "tok") + + +def test_maybe_open_skips_on_lookup_error(monkeypatch, tmp_path, capsys): + """LIST 异常时跳过开单本月(stderr 留痕,下月再试),不再 fail-open.""" + import sanguo_factor.data_gap_check as dgc + + def boom(*a, **k): + raise dgc.GiteaLookupError("conn refused") + + monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok") + monkeypatch.setattr(dgc, "find_open_issue", boom) + opened = [] + monkeypatch.setattr(dgc, "open_gitea_issue", + lambda *a, **k: opened.append(a) or 99) + ledger = str(tmp_path / "ledger.jsonl") + assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml", + ledger_path=ledger) is None + assert opened == [] # 异常不开单 + assert "跳过开单" in capsys.readouterr().err # stderr 留痕 + + +def test_maybe_open_ledger_hit_skips_before_network(monkeypatch, tmp_path, + capsys): + """台账第二道去重:指纹命中即跳过,不再打网络.""" + import sanguo_factor.data_gap_check as dgc + fp = dgc.gap_fingerprint({"fa_x": ["s_new"]}) + ledger = tmp_path / "ledger.jsonl" + ledger.write_text(_json.dumps( + {"fingerprint": fp, "issue": 42, "opened_at": "2026-10-01"}) + "\n", + encoding="utf-8") + + def no_network(*a, **k): + raise AssertionError("台账命中后不应再触网") + + monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok") + monkeypatch.setattr(dgc, "find_open_issue", no_network) + monkeypatch.setattr(dgc, "open_gitea_issue", no_network) + assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml", + ledger_path=str(ledger)) is None + assert "台账命中" in capsys.readouterr().err + + +def test_maybe_open_appends_ledger_after_opening(monkeypatch, tmp_path): + """开单成功→台账 append 一行(指纹/issue号/日期三键).""" + import sanguo_factor.data_gap_check as dgc + monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok") + monkeypatch.setattr(dgc, "find_open_issue", lambda *a, **k: None) + monkeypatch.setattr(dgc, "open_gitea_issue", lambda *a, **k: 77) + ledger = tmp_path / "sub" / "ledger.jsonl" # 父目录不存在也建 + assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml", + ledger_path=str(ledger)) == 77 + rec = _json.loads(ledger.read_text(encoding="utf-8").strip()) + assert rec["fingerprint"] == dgc.gap_fingerprint({"fa_x": ["s_new"]}) + assert rec["issue"] == 77 and rec["opened_at"] + + +def test_append_ledger_failure_degrades_not_blocks(monkeypatch, tmp_path, + capsys): + """台账写失败(此处:路径是目录)不阻断开单主流程,stderr 降级留痕.""" + import sanguo_factor.data_gap_check as dgc + monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok") + monkeypatch.setattr(dgc, "find_open_issue", lambda *a, **k: None) + monkeypatch.setattr(dgc, "open_gitea_issue", lambda *a, **k: 77) + bad_ledger = tmp_path / "its_a_dir" # makedirs 会成功,open 会失败 + bad_ledger.mkdir() + assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml", + ledger_path=str(bad_ledger)) == 77 + assert "台账写入失败" in capsys.readouterr().err