from __future__ import annotations
|
|
import hashlib
|
import json
|
import re
|
from pathlib import Path
|
from typing import Any
|
|
from .full_report import HEADINGS, render_report
|
from .snapshot_builder import CORE_VALUE_PATHS, protected_hash
|
|
|
FORBIDDEN_PLACEHOLDERS = ("TODO", "TBD", "待计算", "NaN", "Infinity")
|
FORBIDDEN_TRADING = ("买入", "卖出", "加仓", "减仓", "止损", "保证收益")
|
DIRECTIONAL_WITHOUT_JUDGMENT = (
|
"明显偏贵",
|
"明显低估",
|
"安全边际充足",
|
"建议持有",
|
"当前价格高于",
|
"当前价格低于",
|
)
|
|
|
def _tables_valid(report: str) -> bool:
|
active_width: int | None = None
|
for line in report.splitlines():
|
if line.startswith("|") and line.endswith("|"):
|
width = len(line.split("|"))
|
if active_width is None:
|
active_width = width
|
elif width != active_width:
|
return False
|
else:
|
active_width = None
|
return True
|
|
|
def _local_links_valid(report: str, artifact_root: Path | None) -> bool:
|
for target in re.findall(r"\[[^\]]+\]\(([^)]+)\)", report):
|
if re.match(r"https?://", target):
|
continue
|
if artifact_root is None:
|
return False
|
candidate = artifact_root / target
|
if not candidate.is_file():
|
return False
|
return True
|
|
|
def run_qa(
|
report: str,
|
snapshot: dict[str, Any],
|
build_report: dict[str, Any],
|
results: dict[str, Any] | None,
|
artifact_root: Path | None = None,
|
) -> dict[str, Any]:
|
errors: list[str] = []
|
headings = re.findall(r"^## \d+\. (.+)$", report, re.M)
|
if headings != HEADINGS or len(set(headings)) != 16:
|
errors.append("16 节标题不齐全、重复或顺序错误")
|
for token in FORBIDDEN_PLACEHOLDERS:
|
if token in report:
|
errors.append(f"存在占位词：{token}")
|
for token in FORBIDDEN_TRADING:
|
if token in report:
|
errors.append(f"存在交易指令/承诺：{token}")
|
if report.count("```") % 2:
|
errors.append("Markdown fence 未闭合")
|
if not _tables_valid(report):
|
errors.append("Markdown 表格列数不一致")
|
if not _local_links_valid(report, artifact_root):
|
errors.append("本地产物链接不存在")
|
if results is None:
|
if "GAP：需要人工判断覆盖层" not in report:
|
errors.append("无 judgment 未声明判断缺口")
|
for token in DIRECTIONAL_WITHOUT_JUDGMENT:
|
if token in report:
|
errors.append(f"无 judgment 出现方向结论：{token}")
|
if not build_report.get("as_of_safe"):
|
errors.append("as-of 检查未通过")
|
raw_hashes = set(build_report.get("raw_hashes", []))
|
if not raw_hashes:
|
errors.append("缺少原始响应哈希")
|
source_by_id = {
|
item.get("id"): item for item in snapshot.get("sources", []) if item.get("id")
|
}
|
source_ids = set(source_by_id)
|
as_of = snapshot.get("meta", {}).get("as_of_date")
|
lineage = build_report.get("field_lineage", {})
|
for field in CORE_VALUE_PATHS:
|
item = lineage.get(field)
|
if not item:
|
errors.append(f"核心字段缺少血缘：{field}")
|
continue
|
if item.get("source_id") not in source_ids:
|
errors.append(f"核心字段 source_id 不存在：{field}")
|
else:
|
structured = source_by_id[item["source_id"]]
|
if structured.get("period_end") != item.get("data_date"):
|
errors.append(f"核心字段 data_date/source period 不一致：{field}")
|
if structured.get("publish_date") != item.get("publish_date"):
|
errors.append(f"核心字段 publish_date/source 不一致：{field}")
|
if not any(
|
field == support or field.startswith(f"{support}.")
|
for support in structured.get("supports", [])
|
):
|
errors.append(f"核心字段结构化来源不支持字段：{field}")
|
if item.get("raw_hash") not in raw_hashes:
|
errors.append(f"核心字段 raw hash 不存在：{field}")
|
if not item.get("publish_date") or not item.get("data_date"):
|
errors.append(f"核心字段日期缺失：{field}")
|
elif item["publish_date"] > as_of or item["data_date"] > as_of:
|
errors.append(f"核心字段含未来日期：{field}")
|
if field.startswith(("financials.", "balance_sheet.")):
|
if item.get("a1_source_id") not in source_ids or item.get("a1_raw_hash") not in raw_hashes:
|
errors.append(f"核心财务缺少 A1/raw 回链：{field}")
|
if not item.get("a1_publish_date") or not item.get("a1_period_end"):
|
errors.append(f"核心财务 A1 日期缺失：{field}")
|
a1 = source_by_id.get(item.get("a1_source_id"))
|
support = item.get("a1_support")
|
if a1 is None or support not in a1.get("supports", []):
|
errors.append(f"核心财务 A1 不支持字段：{field}")
|
elif (
|
a1.get("publish_date") != item.get("a1_publish_date")
|
or a1.get("period_end") != item.get("a1_period_end")
|
):
|
errors.append(f"核心财务 A1 日期/期间不一致：{field}")
|
if field.startswith("financials.prior_year_same_period."):
|
relation = item.get("comparison_relation") or {}
|
if (
|
relation.get("type") != "same_response_comparative_row"
|
or relation.get("comparison_period_end") != item.get("data_date")
|
or relation.get("current_period_end") != item.get("a1_period_end")
|
or relation.get("shared_publish_date") != item.get("a1_publish_date")
|
or item.get("publish_date") != item.get("a1_publish_date")
|
):
|
errors.append(f"上年同期比较关系/A1 不一致：{field}")
|
elif item.get("a1_period_end") != item.get("data_date"):
|
errors.append(f"核心财务 data_date/A1 period 不一致：{field}")
|
if protected_hash(snapshot) != build_report.get("mechanical_protected_sha256"):
|
errors.append("judgment 改写了机械保护字段")
|
expected_report = render_report(snapshot, build_report, build_report.get("gaps", []), results)
|
if report != expected_report:
|
errors.append("报告不是 snapshot/V1 results 的确定渲染")
|
if results is not None:
|
for value in (
|
results["metrics"]["market_cap"],
|
results["metrics"]["ttm_revenue"],
|
results["metrics"]["ttm_attributable_profit"],
|
results["metrics"]["normalized_profit"],
|
results["metrics"]["pb"],
|
results["metrics"]["ps"],
|
):
|
if str(value) not in report:
|
errors.append(f"V1 关键数值未进入报告：{value}")
|
return {
|
"schema_version": 2,
|
"status": "PASS" if not errors else "FAIL",
|
"checks": {
|
"headings": headings == HEADINGS and len(set(headings)) == 16,
|
"markdown_fences": report.count("```") % 2 == 0,
|
"markdown_tables": _tables_valid(report),
|
"local_links": _local_links_valid(report, artifact_root),
|
"as_of": bool(build_report.get("as_of_safe")),
|
"field_lineage": all(field in lineage for field in CORE_VALUE_PATHS),
|
"raw_hashes": bool(raw_hashes),
|
"protected_fields": protected_hash(snapshot)
|
== build_report.get("mechanical_protected_sha256"),
|
"deterministic_render": report == expected_report,
|
"judgment_mode": results is not None,
|
},
|
"errors": errors,
|
}
|
|
|
def verify_manifest(root: Path) -> list[str]:
|
errors: list[str] = []
|
path = root / "manifest.json"
|
try:
|
manifest = json.loads(path.read_text(encoding="utf-8"))
|
except (OSError, json.JSONDecodeError) as exc:
|
return [f"manifest 不可读：{exc}"]
|
if not manifest.get("complete"):
|
errors.append("manifest complete 不是 true")
|
for rel, expected in manifest.get("artifacts", {}).items():
|
artifact = root / rel
|
try:
|
data = artifact.read_bytes()
|
except OSError:
|
errors.append(f"manifest 产物缺失：{rel}")
|
continue
|
if len(data) != expected.get("bytes") or hashlib.sha256(data).hexdigest() != expected.get("sha256"):
|
errors.append(f"manifest hash/bytes 不一致：{rel}")
|
return errors
|
