Skills › Scientific Agent Skills › market-research-reports › validate_competitor_matrix.py
validate_competitor_matrix.py
223 lines · skills/market-research-reports/scripts/validate_competitor_matrix.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate a complete, evidence-linked competitor-feature matrix CSV."""
4from __future__ import annotations
6import argparse
7from pathlib import Path
8from typing import Any
10from _common import (
11 MAX_ROWS,
12 ValidationError,
13 error_exit,
14 parse_iso_date,
15 read_csv_records,
16 read_json,
17 require_identifier,
18 require_list,
19 require_object,
20 require_text,
21 split_ids,
22 write_json_report,
23)
25REQUIRED_FIELDS = (
26 "competitor_id",
27 "competitor_name",
28 "feature_id",
29 "feature_name",
30 "status",
31 "evidence_source_ids",
32 "as_of_date",
33 "geography",
34 "product_scope",
35 "notes",
36)
37STATUSES = {"yes", "no", "partial", "unknown", "not-applicable"}
40def _source_ids(path: str | Path) -> set[str]:
41 if Path(path).suffix.lower() == ".csv":
42 records: list[dict[str, Any]] = read_csv_records(
43 path, required_fields=("source_id",)
44 )
45 elif Path(path).suffix.lower() == ".json":
46 payload = read_json(path)
47 if isinstance(payload, dict):
48 raw = require_list(
49 payload.get("sources"), "sources", minimum=1, maximum=MAX_ROWS
50 )
51 else:
52 raw = require_list(payload, "sources", minimum=1, maximum=MAX_ROWS)
53 records = [
54 require_object(record, f"sources[{index}]")
55 for index, record in enumerate(raw)
56 ]
57 else:
58 raise ValidationError("source ledger must be .csv or .json")
59 identifiers: set[str] = set()
60 for index, record in enumerate(records):
61 source_id = require_identifier(
62 record.get("source_id"), f"sources[{index}].source_id"
63 )
64 if source_id in identifiers:
65 raise ValidationError(f"duplicate source_id in ledger: {source_id}")
66 identifiers.add(source_id)
67 return identifiers
70def validate(
71 rows: list[dict[str, str]], known_source_ids: set[str] | None = None
72) -> dict[str, Any]:
73 errors: list[str] = []
74 warnings: list[str] = []
75 competitors: dict[str, str] = {}
76 features: dict[str, str] = {}
77 pairs: set[tuple[str, str]] = set()
78 scope_values: set[tuple[str, str, str]] = set()
79 cited_ids: set[str] = set()
80 status_counts: dict[str, int] = {}
82 for index, row in enumerate(rows, start=2):
83 context = f"row {index}"
84 try:
85 competitor_id = require_identifier(
86 row["competitor_id"], f"{context}.competitor_id"
87 )
88 competitor_name = require_text(
89 row["competitor_name"], f"{context}.competitor_name", maximum=200
90 )
91 existing_competitor = competitors.get(competitor_id)
92 if existing_competitor and existing_competitor != competitor_name:
93 raise ValidationError(
94 f"{context}: competitor_id {competitor_id} has inconsistent names"
95 )
96 competitors[competitor_id] = competitor_name
98 feature_id = require_identifier(row["feature_id"], f"{context}.feature_id")
99 feature_name = require_text(
100 row["feature_name"], f"{context}.feature_name", maximum=200
101 )
102 existing_feature = features.get(feature_id)
103 if existing_feature and existing_feature != feature_name:
104 raise ValidationError(
105 f"{context}: feature_id {feature_id} has inconsistent names"
106 )
107 features[feature_id] = feature_name
109 pair = (competitor_id, feature_id)
110 if pair in pairs:
111 raise ValidationError(
112 f"{context}: duplicate competitor-feature pair "
113 f"{competitor_id}/{feature_id}"
114 )
115 pairs.add(pair)
117 status = require_text(row["status"], f"{context}.status", maximum=20)
118 if status not in STATUSES:
119 raise ValidationError(
120 f"{context}.status must be one of {', '.join(sorted(STATUSES))}"
121 )
122 status_counts[status] = status_counts.get(status, 0) + 1
123 evidence_ids = split_ids(
124 row["evidence_source_ids"],
125 f"{context}.evidence_source_ids",
126 allow_empty=True,
127 )
128 if status not in {"unknown", "not-applicable"} and not evidence_ids:
129 raise ValidationError(
130 f"{context}: status {status!r} requires evidence_source_ids"
131 )
132 cited_ids.update(evidence_ids)
134 as_of_date = parse_iso_date(row["as_of_date"], f"{context}.as_of_date")
135 geography = require_text(
136 row["geography"], f"{context}.geography", maximum=200
137 )
138 product_scope = require_text(
139 row["product_scope"], f"{context}.product_scope", maximum=500
140 )
141 scope_values.add((as_of_date, geography, product_scope))
142 require_text(row["notes"], f"{context}.notes", allow_empty=True)
143 except ValidationError as exc:
144 errors.append(str(exc))
146 expected_pairs = {
147 (competitor_id, feature_id)
148 for competitor_id in competitors
149 for feature_id in features
150 }
151 missing_pairs = sorted(expected_pairs - pairs)
152 if missing_pairs:
153 preview = ", ".join(f"{a}/{b}" for a, b in missing_pairs[:20])
154 suffix = "" if len(missing_pairs) <= 20 else ", ..."
155 errors.append(
156 f"matrix is incomplete; missing {len(missing_pairs)} pairs: "
157 f"{preview}{suffix}"
158 )
159 if len(scope_values) > 1:
160 errors.append(
161 "matrix rows use inconsistent as_of_date/geography/product_scope values"
162 )
164 missing_sources: list[str] = []
165 if known_source_ids is None:
166 warnings.append(
167 "evidence IDs were syntax-checked but not cross-checked; pass "
168 "--source-ledger for referential integrity"
169 )
170 else:
171 missing_sources = sorted(cited_ids - known_source_ids)
172 if missing_sources:
173 errors.append(
174 "evidence IDs missing from source ledger: "
175 + ", ".join(missing_sources)
176 )
178 return {
179 "valid": not errors,
180 "row_count": len(rows),
181 "competitor_count": len(competitors),
182 "feature_count": len(features),
183 "expected_pair_count": len(expected_pairs),
184 "status_counts": dict(sorted(status_counts.items())),
185 "cited_source_count": len(cited_ids),
186 "warnings": warnings,
187 "errors": errors,
188 }
191def build_parser() -> argparse.ArgumentParser:
192 parser = argparse.ArgumentParser(
193 description=(
194 "Validate a strict local competitor-feature matrix for complete coverage, "
195 "consistent scope, and evidence links."
196 )
197 )
198 parser.add_argument("matrix", help="Local .csv competitor matrix")
199 parser.add_argument(
200 "--source-ledger",
201 help="Optional local .csv/.json source ledger for ID cross-checking",
202 )
203 parser.add_argument("--output", help="Optional local .json validation report")
204 parser.add_argument(
205 "--force", action="store_true", help="Replace an existing --output file"
206 )
207 return parser
210def main() -> int:
211 args = build_parser().parse_args()
212 try:
213 rows = read_csv_records(args.matrix, required_fields=REQUIRED_FIELDS)
214 known = _source_ids(args.source_ledger) if args.source_ledger else None
215 report = validate(rows, known)
216 write_json_report(report, args.output, force=args.force)
217 return 0 if report["valid"] else 1
218 except ValidationError as exc:
219 return error_exit(exc)
222if __name__ == "__main__":
223 raise SystemExit(main())