main
py 730 lines 24.9 KB
Raw
1 #!/usr/bin/env python3
2 """Map/reduce vs. single-pass comparison framework.
3
4 Generates a structured comparison report that tracks quality deltas between
5 the map/reduce candidate and the current single-pass analyzer output. This
6 report feeds the promotion criteria defined in docs/map-reduce-promotion-path.md.
7
8 Never publishes content. Produces comparison-report.json alongside the
9 candidate artifacts for QA analysis.
10 """
11
12 from __future__ import annotations
13
14 import argparse
15 import hashlib
16 import json
17 import os
18 import re
19 import sys
20 from dataclasses import dataclass
21 from datetime import UTC, datetime
22 from pathlib import Path
23 from typing import Any
24
25 try:
26 from scripts.analysis_gate import validate_analysis, validate_publish_quality
27 except ModuleNotFoundError: # pragma: no cover
28 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
29 from scripts.analysis_gate import validate_analysis, validate_publish_quality
30
31 COMPARISON_SCHEMA = "comparison_report_v1"
32 PROMOTION_MIN_QUALITY = int(os.environ.get("MAPREDUCE_MIN_QUALITY_SCORE", "60"))
33 PROMOTION_MIN_COVERAGE = float(os.environ.get("MAPREDUCE_MIN_COVERAGE", "0.85"))
34 PROMOTION_HARD_FLOOR_QUALITY = 55
35 PROMOTION_HARD_FLOOR_COVERAGE = 0.70
36 TIME_BUDGET_SECONDS = int(os.environ.get("MAPREDUCE_TIME_BUDGET_SECONDS", "300"))
37 MARKDOWN_LINK_PATTERN = re.compile(r"\[[^\]]+\]\((https?://[^)]+)\)")
38 WEEK_PATTERN = re.compile(r"^(?P<year>\d{4})-W(?P<week>\d{2})$")
39 VALID_REJECTION_REASONS = {
40 "duplicate",
41 "malformed_finding",
42 "unresolved_contradiction",
43 "weak_citation",
44 }
45
46
47 @dataclass(frozen=True)
48 class ArtifactInfo:
49 path: str
50 sha256: str
51 quality_score: int
52 gate_passed: bool
53 evidence_coverage: float
54 citation_count: int
55 word_count: int
56
57
58 def sha256_file(path: Path) -> str:
59 return hashlib.sha256(path.read_bytes()).hexdigest()
60
61
62 def count_citations(text: str) -> int:
63 """Count markdown repo/article links as citations."""
64 import re
65
66 return len(re.findall(r"\[[^\]]+\]\(https?://[^)]+\)", text))
67
68
69 def extract_quality_score(text: str) -> int:
70 """Extract quality_score from frontmatter."""
71 import re
72
73 match = re.search(r"^quality_score:\s*(\d+)", text, re.MULTILINE)
74 return int(match.group(1)) if match else 0
75
76
77 def compute_evidence_coverage(qa_report: dict[str, Any]) -> float:
78 """Compute evidence coverage from QA report checks."""
79 checks = qa_report.get("checks", {})
80 # Use mapper contracts to estimate input count
81 mapper_contracts = checks.get("mapper_contracts", {})
82 errors_by_mapper = mapper_contracts.get("errors_by_mapper", {})
83 # If no errors, assume good coverage
84 if not any(errors_by_mapper.values()):
85 return 0.90
86 # Degrade based on mapper failures
87 failed_mappers = sum(1 for errs in errors_by_mapper.values() if errs)
88 total_mappers = max(len(errors_by_mapper), 1)
89 return max(0.0, 1.0 - (failed_mappers / total_mappers) * 0.5)
90
91
92 def compute_evidence_coverage_from_ledgers(
93 candidate_dir: Path,
94 ) -> float:
95 """Compute evidence coverage from mapper ledger files."""
96 total_input = 0
97 total_mapped = 0
98 for ledger_path in sorted((candidate_dir / "maps").glob("*.json")):
99 try:
100 ledger = json.loads(ledger_path.read_text(encoding="utf-8"))
101 except (json.JSONDecodeError, OSError):
102 continue
103 coverage = ledger.get("coverage", {})
104 total_input += int(coverage.get("repo_count_input", 0)) + int(
105 coverage.get("article_count_input", 0)
106 )
107 total_mapped += int(coverage.get("repo_count_mapped", 0)) + int(
108 coverage.get("article_count_mapped", 0)
109 )
110 if total_input == 0:
111 return 0.0
112 return total_mapped / total_input
113
114
115 def analyze_single_pass(
116 summary_path: Path, raw_payload: dict[str, Any], current_datetime: str
117 ) -> ArtifactInfo:
118 """Analyze the single-pass baseline artifact."""
119 text = summary_path.read_text(encoding="utf-8")
120 structural_errors, word_count = validate_analysis(text, raw_payload, current_datetime)
121 publish_errors, _gates = validate_publish_quality(
122 text, raw_payload, source="copilot-cli", model="claude-sonnet-4.6"
123 )
124 non_provenance = [e for e in publish_errors if not e.startswith("AI provenance")]
125 gate_passed = not structural_errors and not non_provenance
126 return ArtifactInfo(
127 path=summary_path.as_posix(),
128 sha256=sha256_file(summary_path),
129 quality_score=extract_quality_score(text),
130 gate_passed=gate_passed,
131 evidence_coverage=0.92, # Single-pass baseline assumed high coverage
132 citation_count=count_citations(text),
133 word_count=word_count,
134 )
135
136
137 def analyze_map_reduce(
138 candidate_dir: Path, raw_payload: dict[str, Any], current_datetime: str
139 ) -> tuple[ArtifactInfo, dict[str, Any]]:
140 """Analyze the map/reduce candidate artifact and QA report."""
141 week = str(raw_payload.get("week") or "").strip()
142 candidate_path = resolve_candidate_path(candidate_dir, week)
143 qa_path = resolve_sidecar_path(
144 candidate_dir,
145 preferred_name="qa-comparison-report.json",
146 legacy_name="qa-report.json",
147 )
148 plan_path = candidate_dir / "editorial-plan.json"
149 contradictions_path = candidate_dir / "sidecars" / "contradictions.json"
150 rejected_claims_path = candidate_dir / "sidecars" / "rejected-claims.json"
151
152 if not candidate_path.exists():
153 raise FileNotFoundError(f"Candidate summary not found: {candidate_path}")
154
155 text = candidate_path.read_text(encoding="utf-8")
156 structural_errors, word_count = validate_analysis(text, raw_payload, current_datetime)
157 publish_errors, _gates = validate_publish_quality(
158 text, raw_payload, source="map-reduce-dry-run", model="local-deterministic"
159 )
160 non_provenance = [e for e in publish_errors if not e.startswith("AI provenance")]
161 gate_passed = not structural_errors and not non_provenance
162
163 artifact_errors: list[str] = []
164 qa_report = load_optional_json(
165 qa_path,
166 artifact_name="QA comparison report",
167 errors=artifact_errors,
168 )
169 editorial_plan = load_optional_json(
170 plan_path,
171 artifact_name="editorial plan",
172 errors=artifact_errors,
173 )
174 contradictions_sidecar = load_optional_json(
175 contradictions_path,
176 artifact_name="contradictions sidecar",
177 errors=artifact_errors,
178 )
179 rejected_claims_sidecar = load_optional_json(
180 rejected_claims_path,
181 artifact_name="rejected claims sidecar",
182 errors=artifact_errors,
183 )
184
185 evidence_coverage = compute_evidence_coverage_from_ledgers(candidate_dir)
186 orphaned_citations = compute_orphaned_citations(
187 text=text,
188 editorial_plan=editorial_plan,
189 )
190 unresolved_contradictions = count_unresolved_contradictions(contradictions_sidecar)
191 invalid_rejected_claims = count_invalid_rejected_claims(
192 rejected_claims_sidecar,
193 contradictions_sidecar=contradictions_sidecar,
194 )
195 checks = qa_report.get("checks", {}) if isinstance(qa_report, dict) else {}
196 sidecars_present = checks.get("sidecars_present", {}) if isinstance(checks, dict) else {}
197
198 extra = {
199 "mapper_errors": checks.get("mapper_contracts", {}).get("errors_by_mapper", {}),
200 "claims_rejected": sidecars_present.get(
201 "rejected_count",
202 len(rejected_claims_sidecar.get("rejected_claims", [])),
203 ),
204 "unresolved_contradictions": sidecars_present.get(
205 "contradiction_count",
206 unresolved_contradictions,
207 ),
208 "orphaned_citations": orphaned_citations,
209 "invalid_rejected_claims": invalid_rejected_claims,
210 "artifact_errors": artifact_errors,
211 }
212
213 info = ArtifactInfo(
214 path=candidate_path.as_posix(),
215 sha256=sha256_file(candidate_path),
216 quality_score=extract_quality_score(text),
217 gate_passed=gate_passed,
218 evidence_coverage=evidence_coverage,
219 citation_count=count_citations(text),
220 word_count=word_count,
221 )
222 return info, extra
223
224
225 def compute_verdict(
226 single_pass: ArtifactInfo,
227 map_reduce: ArtifactInfo,
228 deltas: dict[str, Any],
229 ) -> tuple[str, list[str]]:
230 """Determine pass/fail verdict and list blockers."""
231 blockers: list[str] = []
232
233 # Gate regression check
234 if deltas.get("gate_regression"):
235 blockers.append("Gate regression: single-pass passes but map/reduce fails.")
236
237 # Quality floor
238 if map_reduce.quality_score < PROMOTION_MIN_QUALITY:
239 blockers.append(
240 f"Quality score {map_reduce.quality_score} below minimum {PROMOTION_MIN_QUALITY}."
241 )
242
243 # Hard quality floor (rollback trigger)
244 if map_reduce.quality_score < PROMOTION_HARD_FLOOR_QUALITY:
245 blockers.append(
246 f"Quality score {map_reduce.quality_score} below hard floor "
247 f"{PROMOTION_HARD_FLOOR_QUALITY} (rollback trigger)."
248 )
249
250 # Evidence coverage
251 if map_reduce.evidence_coverage < PROMOTION_MIN_COVERAGE:
252 blockers.append(
253 f"Evidence coverage {map_reduce.evidence_coverage:.2f} below minimum "
254 f"{PROMOTION_MIN_COVERAGE}."
255 )
256
257 # Hard coverage floor (rollback trigger)
258 if map_reduce.evidence_coverage < PROMOTION_HARD_FLOOR_COVERAGE:
259 blockers.append(
260 f"Evidence coverage {map_reduce.evidence_coverage:.2f} below hard floor "
261 f"{PROMOTION_HARD_FLOOR_COVERAGE} (rollback trigger)."
262 )
263
264 if deltas.get("orphaned_citations", 0) > 0:
265 blockers.append(
266 "Citation integrity failed: "
267 f"{deltas['orphaned_citations']} orphaned rendered citation(s) were not "
268 "bound in the editorial plan."
269 )
270
271 if deltas.get("unresolved_contradictions", 0) > 0:
272 blockers.append(
273 "Contradiction handling failed: "
274 f"{deltas['unresolved_contradictions']} unresolved contradiction(s) remain "
275 "in the map/reduce sidecars."
276 )
277
278 if deltas.get("invalid_rejected_claims", 0) > 0:
279 blockers.append(
280 "Claim rejection audit failed: "
281 f"{deltas['invalid_rejected_claims']} rejected claim(s) were missing a "
282 "supported audit reason."
283 )
284
285 for artifact_error in deltas.get("artifact_errors", []):
286 blockers.append(f"Comparison artifact unavailable: {artifact_error}")
287
288 verdict = "pass" if not blockers else "fail"
289 return verdict, blockers
290
291
292 def generate_comparison_report(
293 *,
294 week: str,
295 single_pass: ArtifactInfo,
296 map_reduce: ArtifactInfo,
297 map_reduce_extra: dict[str, Any],
298 run_datetime: str,
299 ) -> dict[str, Any]:
300 """Generate the full comparison report."""
301 deltas = {
302 "quality_score": map_reduce.quality_score - single_pass.quality_score,
303 "evidence_coverage": round(map_reduce.evidence_coverage - single_pass.evidence_coverage, 4),
304 "citation_count": map_reduce.citation_count - single_pass.citation_count,
305 "word_count": map_reduce.word_count - single_pass.word_count,
306 "gate_regression": single_pass.gate_passed and not map_reduce.gate_passed,
307 "orphaned_citations": int(map_reduce_extra.get("orphaned_citations", 0)),
308 "unresolved_contradictions": int(map_reduce_extra.get("unresolved_contradictions", 0)),
309 "invalid_rejected_claims": int(map_reduce_extra.get("invalid_rejected_claims", 0)),
310 "artifact_errors": list(map_reduce_extra.get("artifact_errors", [])),
311 }
312
313 verdict, blockers = compute_verdict(single_pass, map_reduce, deltas)
314
315 return {
316 "schema_version": COMPARISON_SCHEMA,
317 "week": week,
318 "run_datetime": run_datetime,
319 "single_pass": {
320 "artifact_path": single_pass.path,
321 "sha256": single_pass.sha256,
322 "quality_score": single_pass.quality_score,
323 "gate_passed": single_pass.gate_passed,
324 "evidence_coverage": single_pass.evidence_coverage,
325 "citation_count": single_pass.citation_count,
326 "word_count": single_pass.word_count,
327 },
328 "map_reduce": {
329 "artifact_path": map_reduce.path,
330 "sha256": map_reduce.sha256,
331 "quality_score": map_reduce.quality_score,
332 "gate_passed": map_reduce.gate_passed,
333 "evidence_coverage": map_reduce.evidence_coverage,
334 "citation_count": map_reduce.citation_count,
335 "word_count": map_reduce.word_count,
336 **map_reduce_extra,
337 },
338 "deltas": deltas,
339 "verdict": verdict,
340 "blockers": blockers,
341 }
342
343
344 def check_promotion_eligibility(reports: list[dict[str, Any]]) -> dict[str, Any]:
345 """Check whether a set of comparison reports meets promotion criteria.
346
347 Returns a promotion status object indicating readiness and any blockers.
348 """
349 if len(reports) < 3:
350 return {
351 "eligible": False,
352 "reason": f"Only {len(reports)} comparison runs available; need ≥ 3.",
353 "runs_passing": len([r for r in reports if r.get("verdict") == "pass"]),
354 "runs_total": len(reports),
355 }
356
357 ordered_reports = sorted(reports, key=promotion_report_sort_key)
358 recent_reports = ordered_reports[-3:]
359 passing = [r for r in recent_reports if r.get("verdict") == "pass"]
360 if len(passing) < 3:
361 return {
362 "eligible": False,
363 "reason": (
364 f"Only {len(passing)}/{len(recent_reports)} most recent runs passed; "
365 "need ≥ 3 consecutive."
366 ),
367 "runs_passing": len(passing),
368 "runs_total": len(reports),
369 }
370
371 # Check average quality across the required consecutive window.
372 avg_quality = sum(r["map_reduce"]["quality_score"] for r in recent_reports) / len(
373 recent_reports
374 )
375 if avg_quality < 65:
376 return {
377 "eligible": False,
378 "reason": f"Average quality score {avg_quality:.1f} below 65 threshold.",
379 "runs_passing": len(passing),
380 "runs_total": len(reports),
381 }
382
383 # Check staleness (28 days)
384 now = datetime.now(UTC)
385 for report in recent_reports:
386 run_dt = report.get("run_datetime", "")
387 try:
388 report_time = datetime.fromisoformat(run_dt.replace("Z", "+00:00"))
389 if (now - report_time).days > 28:
390 return {
391 "eligible": False,
392 "reason": f"Comparison run from {run_dt} is older than 28 days.",
393 "runs_passing": len(passing),
394 "runs_total": len(reports),
395 }
396 except (ValueError, TypeError):
397 pass
398
399 return {
400 "eligible": True,
401 "reason": "All promotion criteria met. Awaiting operator opt-in and team sign-off.",
402 "runs_passing": len(passing),
403 "runs_total": len(reports),
404 "average_quality": round(avg_quality, 1),
405 }
406
407
408 def should_rollback(report: dict[str, Any]) -> tuple[bool, str]:
409 """Determine if automatic rollback should trigger based on a comparison report.
410
411 Returns (should_rollback, reason).
412 """
413 mr = report.get("map_reduce", {})
414
415 # Gate failure when single-pass passes
416 if report.get("deltas", {}).get("gate_regression"):
417 return True, "Gate regression: map/reduce fails gates that single-pass passes."
418
419 # Hard quality floor
420 quality = mr.get("quality_score", 0)
421 if quality < PROMOTION_HARD_FLOOR_QUALITY:
422 return True, f"Quality score {quality} below hard floor {PROMOTION_HARD_FLOOR_QUALITY}."
423
424 # Hard coverage floor
425 coverage = mr.get("evidence_coverage", 0.0)
426 if coverage < PROMOTION_HARD_FLOOR_COVERAGE:
427 return (
428 True,
429 f"Evidence coverage {coverage:.2f} below hard floor {PROMOTION_HARD_FLOOR_COVERAGE}.",
430 )
431
432 # Mapper failure
433 mapper_errors = mr.get("mapper_errors", {})
434 if any(errs for errs in mapper_errors.values() if isinstance(errs, list) and errs):
435 failed = [k for k, v in mapper_errors.items() if v]
436 return True, f"Mapper failures in: {', '.join(failed)}."
437
438 return False, ""
439
440
441 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
442 parser = argparse.ArgumentParser(
443 description="Compare map/reduce candidate against single-pass baseline."
444 )
445 parser.add_argument(
446 "--raw-json",
447 required=True,
448 type=Path,
449 help="Weekly raw crawl payload.",
450 )
451 parser.add_argument(
452 "--single-pass-summary",
453 required=True,
454 type=Path,
455 help="Current single-pass analyzed summary.",
456 )
457 parser.add_argument(
458 "--candidate-dir",
459 required=True,
460 type=Path,
461 help="Map/reduce candidate output directory.",
462 )
463 parser.add_argument(
464 "--current-datetime",
465 default=datetime.now(UTC).isoformat(),
466 help="ISO-8601 timestamp for the comparison run.",
467 )
468 parser.add_argument(
469 "--output",
470 type=Path,
471 help="Output path for comparison report (default: candidate-dir/comparison-report.json).",
472 )
473 parser.add_argument(
474 "--check-promotion",
475 action="store_true",
476 help="Also check promotion eligibility across all available comparison reports.",
477 )
478 return parser.parse_args(argv)
479
480
481 def run(args: argparse.Namespace) -> dict[str, Any]:
482 """Execute comparison and return the report."""
483 raw_payload = json.loads(args.raw_json.read_text(encoding="utf-8"))
484 week = raw_payload.get("week", "unknown")
485
486 single_pass = analyze_single_pass(args.single_pass_summary, raw_payload, args.current_datetime)
487 map_reduce, mr_extra = analyze_map_reduce(
488 args.candidate_dir, raw_payload, args.current_datetime
489 )
490
491 report = generate_comparison_report(
492 week=week,
493 single_pass=single_pass,
494 map_reduce=map_reduce,
495 map_reduce_extra=mr_extra,
496 run_datetime=args.current_datetime,
497 )
498
499 # Check rollback
500 rollback, rollback_reason = should_rollback(report)
501 if rollback:
502 report["rollback"] = True
503 report["rollback_reason"] = rollback_reason
504
505 # Write report
506 output_path = args.output or (args.candidate_dir / "comparison-report.json")
507 output_path.parent.mkdir(parents=True, exist_ok=True)
508 output_path.write_text(
509 json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
510 encoding="utf-8",
511 )
512
513 # Optional promotion check
514 if args.check_promotion:
515 reports_dir = args.candidate_dir.parent
516 all_reports = []
517 for rdir in sorted(reports_dir.iterdir()):
518 rpath = rdir / "comparison-report.json"
519 if rpath.exists():
520 try:
521 all_reports.append(json.loads(rpath.read_text(encoding="utf-8")))
522 except (json.JSONDecodeError, OSError):
523 pass
524 promotion = check_promotion_eligibility(all_reports)
525 report["promotion_status"] = promotion
526
527 # Re-write with promotion status
528 output_path.write_text(
529 json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
530 encoding="utf-8",
531 )
532
533 return report
534
535
536 def main(argv: list[str] | None = None) -> int:
537 args = parse_args(argv)
538 try:
539 report = run(args)
540 except FileNotFoundError as exc:
541 print(f"ERROR: {exc}", file=sys.stderr)
542 return 1
543
544 verdict = report.get("verdict", "unknown")
545 rollback = report.get("rollback", False)
546 print(f"Comparison complete: week={report.get('week')} verdict={verdict}", flush=True)
547 if rollback:
548 print(f"⚠️ ROLLBACK TRIGGERED: {report.get('rollback_reason')}", flush=True)
549 if report.get("promotion_status"):
550 status = report["promotion_status"]
551 if status.get("eligible"):
552 print("✅ Promotion eligible — awaiting operator opt-in.", flush=True)
553 else:
554 print(f"⏳ Not yet eligible: {status.get('reason')}", flush=True)
555
556 return 1 if rollback or verdict != "pass" else 0
557
558
559 def resolve_candidate_path(candidate_dir: Path, week: str) -> Path:
560 preferred = candidate_dir / f"{week}-map-reduce-candidate.md"
561 if preferred.exists():
562 return preferred
563 legacy = candidate_dir / "candidate-summary.md"
564 if legacy.exists():
565 return legacy
566 return preferred
567
568
569 def resolve_sidecar_path(
570 candidate_dir: Path,
571 *,
572 preferred_name: str,
573 legacy_name: str,
574 ) -> Path:
575 preferred = candidate_dir / preferred_name
576 if preferred.exists():
577 return preferred
578 legacy = candidate_dir / legacy_name
579 if legacy.exists():
580 return legacy
581 return preferred
582
583
584 def load_optional_json(
585 path: Path,
586 *,
587 artifact_name: str,
588 errors: list[str],
589 ) -> dict[str, Any]:
590 if not path.exists():
591 errors.append(f"{artifact_name} missing: {path}")
592 return {}
593 try:
594 payload = json.loads(path.read_text(encoding="utf-8"))
595 except (json.JSONDecodeError, OSError) as exc:
596 errors.append(f"{artifact_name} unreadable: {path} ({exc})")
597 return {}
598 if not isinstance(payload, dict):
599 errors.append(f"{artifact_name} must be a JSON object: {path}")
600 return {}
601 return payload
602
603
604 def normalize_repo_url(url: str) -> str | None:
605 prefix = "https://github.com/"
606 if not url.startswith(prefix):
607 return None
608 repo_path = url.removeprefix(prefix).split("?", 1)[0].split("#", 1)[0]
609 parts = [part for part in repo_path.split("/") if part]
610 if len(parts) < 2:
611 return None
612 return f"{parts[0]}/{parts[1]}"
613
614
615 def compute_orphaned_citations(
616 *,
617 text: str,
618 editorial_plan: dict[str, Any],
619 ) -> int:
620 if not editorial_plan:
621 return len(MARKDOWN_LINK_PATTERN.findall(text))
622
623 selected_claims = (
624 editorial_plan.get("selected_claims", [])
625 if isinstance(editorial_plan.get("selected_claims"), list)
626 else []
627 )
628 key_references = (
629 editorial_plan.get("key_references", {})
630 if isinstance(editorial_plan.get("key_references"), dict)
631 else {}
632 )
633
634 allowed_repos = {
635 str(editorial_plan.get("top_repo", "")).strip(),
636 *(str(repo).strip() for repo in key_references.get("notable_projects", [])),
637 }
638 allowed_articles = {str(url).strip() for url in key_references.get("press_articles", [])}
639 for claim in selected_claims:
640 if not isinstance(claim, dict):
641 continue
642 bindings = (
643 claim.get("citation_bindings", {})
644 if isinstance(claim.get("citation_bindings"), dict)
645 else {}
646 )
647 allowed_repos.update(str(repo).strip() for repo in bindings.get("repos", []))
648 allowed_articles.update(str(url).strip() for url in bindings.get("articles", []))
649
650 allowed_repos.discard("")
651 allowed_articles.discard("")
652
653 orphaned = 0
654 for url in MARKDOWN_LINK_PATTERN.findall(text):
655 repo_name = normalize_repo_url(url)
656 if repo_name is not None:
657 if repo_name not in allowed_repos:
658 orphaned += 1
659 continue
660 if url not in allowed_articles:
661 orphaned += 1
662 return orphaned
663
664
665 def count_unresolved_contradictions(contradictions_sidecar: dict[str, Any]) -> int:
666 contradictions = (
667 contradictions_sidecar.get("contradictions", [])
668 if isinstance(contradictions_sidecar.get("contradictions"), list)
669 else []
670 )
671 return len(contradictions)
672
673
674 def count_invalid_rejected_claims(
675 rejected_claims_sidecar: dict[str, Any],
676 *,
677 contradictions_sidecar: dict[str, Any],
678 ) -> int:
679 rejected_claims = (
680 rejected_claims_sidecar.get("rejected_claims", [])
681 if isinstance(rejected_claims_sidecar.get("rejected_claims"), list)
682 else []
683 )
684 contradiction_claim_ids = {
685 str(entry.get("claim_id"))
686 for entry in contradictions_sidecar.get("contradictions", [])
687 if isinstance(entry, dict) and entry.get("claim_id") is not None
688 }
689 invalid = 0
690 for entry in rejected_claims:
691 if not isinstance(entry, dict):
692 invalid += 1
693 continue
694 reason = str(entry.get("reason") or "").strip()
695 if reason not in VALID_REJECTION_REASONS:
696 invalid += 1
697 continue
698 if reason == "unresolved_contradiction":
699 claim_id = entry.get("claim_id")
700 if claim_id is None or str(claim_id) not in contradiction_claim_ids:
701 invalid += 1
702 return invalid
703
704
705 def promotion_report_sort_key(report: dict[str, Any]) -> float:
706 run_datetime = report.get("run_datetime")
707 if isinstance(run_datetime, str):
708 try:
709 parsed = datetime.fromisoformat(run_datetime.replace("Z", "+00:00"))
710 except ValueError:
711 pass
712 else:
713 return parsed.astimezone(UTC).timestamp()
714
715 week = report.get("week")
716 if isinstance(week, str):
717 match = WEEK_PATTERN.fullmatch(week)
718 if match:
719 report_week = datetime.fromisocalendar(
720 int(match.group("year")),
721 int(match.group("week")),
722 1,
723 )
724 return report_week.replace(tzinfo=UTC).timestamp()
725
726 return float("-inf")
727
728
729 if __name__ == "__main__":
730 raise SystemExit(main())