Spaces:
Sleeping
Sleeping
File size: 6,271 Bytes
300df0f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | """
Validation & reporting for cross-reference extraction (T2.4).
"""
from __future__ import annotations
import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from neo4j import Driver
logger = logging.getLogger(__name__)
@dataclass
class ValidationReport:
"""Aggregate statistics produced by CrossReferenceValidator."""
# ── Counts ──────────────────────────────────────────────────────────
total_internal: int = 0
resolved_internal: int = 0
total_external: int = 0
resolved_external_exact: int = 0
resolved_external_fuzzy: int = 0
unresolved_external: int = 0
total_modification: int = 0
resolved_modification: int = 0
unresolved_modification: int = 0
# ── Unresolved details (for manual review) ──────────────────────────
unresolved_external_list: list[dict] = field(default_factory=list)
unresolved_modification_list: list[dict] = field(default_factory=list)
# ── Fuzzy confidence distribution ───────────────────────────────────
fuzzy_confidence_buckets: dict[str, int] = field(default_factory=lambda: {
"0.8-1.0": 0, "0.6-0.8": 0, "0.4-0.6": 0, "<0.4": 0,
})
# ── Derived metrics ─────────────────────────────────────────────────
@property
def external_resolution_rate(self) -> float:
if self.total_external == 0:
return 1.0
return (self.resolved_external_exact + self.resolved_external_fuzzy) / self.total_external
@property
def modification_resolution_rate(self) -> float:
if self.total_modification == 0:
return 1.0
return self.resolved_modification / self.total_modification
def to_dict(self) -> dict:
return {
"internal": {
"total": self.total_internal,
"resolved": self.resolved_internal,
},
"external": {
"total": self.total_external,
"resolved_exact": self.resolved_external_exact,
"resolved_fuzzy": self.resolved_external_fuzzy,
"unresolved": self.unresolved_external,
"resolution_rate": round(self.external_resolution_rate, 4),
},
"modification": {
"total": self.total_modification,
"resolved": self.resolved_modification,
"unresolved": self.unresolved_modification,
"resolution_rate": round(self.modification_resolution_rate, 4),
},
"fuzzy_confidence_distribution": self.fuzzy_confidence_buckets,
"unresolved_external_sample": self.unresolved_external_list[:50],
"unresolved_modification_sample": self.unresolved_modification_list[:50],
}
def save(self, path: str | Path) -> None:
Path(path).write_text(json.dumps(self.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8")
logger.info("Validation report saved to %s", path)
def print_summary(self) -> None:
print(
f"\n{'='*60}\n"
f"Cross-Reference Validation Report\n"
f"{'='*60}\n"
f"Internal refs : {self.resolved_internal}/{self.total_internal}\n"
f"External refs : resolved {self.resolved_external_exact} exact + "
f"{self.resolved_external_fuzzy} fuzzy / {self.total_external} total "
f"({self.external_resolution_rate:.1%})\n"
f"Modification : {self.resolved_modification}/{self.total_modification} "
f"({self.modification_resolution_rate:.1%})\n"
f"{'='*60}\n"
f"Targets: External ≥95%, Modification ≥95%\n"
)
class CrossReferenceValidator:
"""
Queries Neo4j to verify that extracted relationships point to valid nodes.
Usage
-----
validator = CrossReferenceValidator(driver)
report = validator.validate()
report.print_summary()
report.save("output/cross_ref_validation.json")
"""
def __init__(self, driver: "Driver") -> None:
self._driver = driver
def validate(self) -> ValidationReport:
"""
Run validation queries against Neo4j and return a populated ValidationReport.
"""
report = ValidationReport()
with self._driver.session() as session:
# 1. Internal References
res_int = session.run("MATCH ()-[r:REFERENCES_INTERNAL]->() RETURN count(r) as total")
report.total_internal = report.resolved_internal = res_int.single()["total"]
# 2. External References by match_method
res_ext = session.run("""
MATCH ()-[r:REFERENCES_EXTERNAL]->()
RETURN r.match_method as method, r.confidence as conf, count(r) as count
""")
for record in res_ext:
method = record["method"]
count = record["count"]
conf = record["conf"]
report.total_external += count
if method == "exact" or method == "short_title_map":
report.resolved_external_exact += count
else:
report.resolved_external_fuzzy += count
# Bucketing confidence
if conf >= 0.8: report.fuzzy_confidence_buckets["0.8-1.0"] += count
elif conf >= 0.6: report.fuzzy_confidence_buckets["0.6-0.8"] += count
elif conf >= 0.4: report.fuzzy_confidence_buckets["0.4-0.6"] += count
else: report.fuzzy_confidence_buckets["<0.4"] += count
# 3. Modification References
res_mod = session.run("MATCH ()-[r:MODIFIES]->() RETURN count(r) as total")
report.total_modification = report.resolved_modification = res_mod.single()["total"]
return report
|