Skills › Scientific Agent Skills › market-research-reports › validate_evidence_ledger.py
validate_evidence_ledger.py
291 lines · skills/market-research-reports/scripts/validate_evidence_ledger.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate a local market-research source/evidence ledger."""
4from __future__ import annotations
6import argparse
7from datetime import date
8from pathlib import Path, PurePosixPath
9from typing import Any
10from urllib.parse import urlsplit
12from _common import (
13 MAX_ROWS,
14 ValidationError,
15 error_exit,
16 parse_currency,
17 parse_iso_date,
18 parse_year,
19 read_csv_records,
20 read_json,
21 require_identifier,
22 require_list,
23 require_object,
24 require_text,
25 require_unique_identifiers,
26 write_json_report,
27)
29REQUIRED_FIELDS = (
30 "source_id",
31 "title",
32 "publisher",
33 "url",
34 "source_type",
35 "publication_date",
36 "retrieval_date",
37 "geography",
38 "currency",
39 "base_year",
40 "price_basis",
41 "measure_type",
42 "unit",
43 "taxonomy",
44 "taxonomy_version",
45 "revision_status",
46 "method",
47 "sample",
48 "limitations",
49 "license_or_terms",
50 "archive_path",
51)
53SOURCE_TYPES = {
54 "official_statistics",
55 "regulator_filing",
56 "company_filing",
57 "survey",
58 "interview",
59 "academic",
60 "industry_association",
61 "paid_secondary",
62 "news",
63 "methodological_guidance",
64 "other",
65}
66PRICE_BASES = {
67 "nominal",
68 "real",
69 "current",
70 "constant",
71 "chained",
72 "mixed",
73 "not-applicable",
74}
75MEASURE_TYPES = {
76 "stock",
77 "flow",
78 "index",
79 "share",
80 "count",
81 "rate",
82 "price",
83 "mixed",
84 "not-applicable",
85}
86REVISION_STATUSES = {
87 "preliminary",
88 "revised",
89 "final",
90 "current",
91 "vintage",
92 "unknown",
93 "not-applicable",
94}
97def _publication_date(value: Any, context: str) -> str:
98 text = require_text(value, context, maximum=10)
99 if text == "not-stated":
100 return text
101 if len(text) == 4:
102 parse_year(text, context)
103 return text
104 return parse_iso_date(text, context)
107def _url(value: Any, context: str) -> str:
108 text = require_text(value, context, maximum=2_048)
109 parsed = urlsplit(text)
110 if parsed.scheme not in {"http", "https"} or not parsed.netloc:
111 raise ValidationError(f"{context} must be an absolute HTTP(S) URL")
112 if parsed.username or parsed.password:
113 raise ValidationError(f"{context} must not contain embedded credentials")
114 return text
117def _archive_path(value: Any, context: str) -> str:
118 text = require_text(value, context, allow_empty=True, maximum=1_024)
119 if not text:
120 return ""
121 path = PurePosixPath(text)
122 if path.is_absolute() or ".." in path.parts:
123 raise ValidationError(
124 f"{context} must be a relative path without parent traversal"
125 )
126 return text
129def _load_records(path: str | Path) -> list[dict[str, Any]]:
130 suffix = Path(path).suffix.lower()
131 if suffix == ".csv":
132 return read_csv_records(path, required_fields=REQUIRED_FIELDS)
133 if suffix == ".json":
134 payload = read_json(path)
135 if isinstance(payload, dict):
136 records = require_list(
137 payload.get("sources"), "sources", minimum=1, maximum=MAX_ROWS
138 )
139 else:
140 records = require_list(payload, "source ledger", minimum=1, maximum=MAX_ROWS)
141 return [require_object(row, f"sources[{index}]") for index, row in enumerate(records)]
142 raise ValidationError("input must be .csv or .json")
145def validate_records(records: list[dict[str, Any]]) -> dict[str, Any]:
146 errors: list[str] = []
147 warnings: list[str] = []
148 valid_ids: list[str] = []
149 type_counts: dict[str, int] = {}
151 for index, record in enumerate(records, start=1):
152 context = f"record {index}"
153 missing = [field for field in REQUIRED_FIELDS if field not in record]
154 if missing:
155 errors.append(f"{context}: missing fields: {', '.join(missing)}")
156 continue
157 try:
158 source_id = require_identifier(record["source_id"], f"{context}.source_id")
159 valid_ids.append(source_id)
160 require_text(record["title"], f"{context}.title")
161 require_text(record["publisher"], f"{context}.publisher")
162 _url(record["url"], f"{context}.url")
164 source_type = require_text(
165 record["source_type"], f"{context}.source_type", maximum=40
166 )
167 if source_type not in SOURCE_TYPES:
168 raise ValidationError(
169 f"{context}.source_type must be one of "
170 f"{', '.join(sorted(SOURCE_TYPES))}"
171 )
172 type_counts[source_type] = type_counts.get(source_type, 0) + 1
174 publication = _publication_date(
175 record["publication_date"], f"{context}.publication_date"
176 )
177 retrieval = parse_iso_date(
178 record["retrieval_date"], f"{context}.retrieval_date"
179 )
180 if publication == "not-stated":
181 warnings.append(f"{source_id}: publication date is not stated")
182 elif len(publication) == 10 and date.fromisoformat(
183 retrieval
184 ) < date.fromisoformat(publication):
185 raise ValidationError(
186 f"{context}.retrieval_date precedes publication_date"
187 )
189 require_text(record["geography"], f"{context}.geography")
190 currency = parse_currency(
191 record["currency"], f"{context}.currency", allow_empty=True
192 )
193 base_year_raw = require_text(
194 record["base_year"], f"{context}.base_year", allow_empty=True, maximum=4
195 )
196 if base_year_raw:
197 parse_year(base_year_raw, f"{context}.base_year")
199 price_basis = require_text(
200 record["price_basis"], f"{context}.price_basis", maximum=20
201 )
202 if price_basis not in PRICE_BASES:
203 raise ValidationError(
204 f"{context}.price_basis must be one of "
205 f"{', '.join(sorted(PRICE_BASES))}"
206 )
207 if currency and (not base_year_raw or price_basis == "not-applicable"):
208 raise ValidationError(
209 f"{context}: monetary evidence requires base_year and price_basis"
210 )
212 measure_type = require_text(
213 record["measure_type"], f"{context}.measure_type", maximum=20
214 )
215 if measure_type not in MEASURE_TYPES:
216 raise ValidationError(
217 f"{context}.measure_type must be one of "
218 f"{', '.join(sorted(MEASURE_TYPES))}"
219 )
220 require_text(record["unit"], f"{context}.unit")
221 require_text(
222 record["taxonomy"], f"{context}.taxonomy", allow_empty=True
223 )
224 require_text(
225 record["taxonomy_version"],
226 f"{context}.taxonomy_version",
227 allow_empty=True,
228 )
230 revision_status = require_text(
231 record["revision_status"], f"{context}.revision_status", maximum=20
232 )
233 if revision_status not in REVISION_STATUSES:
234 raise ValidationError(
235 f"{context}.revision_status must be one of "
236 f"{', '.join(sorted(REVISION_STATUSES))}"
237 )
238 if revision_status == "unknown":
239 warnings.append(f"{source_id}: revision status is unknown")
241 require_text(record["method"], f"{context}.method")
242 require_text(record["sample"], f"{context}.sample", allow_empty=True)
243 require_text(record["limitations"], f"{context}.limitations")
244 require_text(
245 record["license_or_terms"], f"{context}.license_or_terms"
246 )
247 _archive_path(record["archive_path"], f"{context}.archive_path")
248 except ValidationError as exc:
249 errors.append(str(exc))
251 try:
252 require_unique_identifiers(valid_ids, "source ledger")
253 except ValidationError as exc:
254 errors.append(str(exc))
256 return {
257 "valid": not errors,
258 "source_count": len(records),
259 "source_type_counts": dict(sorted(type_counts.items())),
260 "warnings": warnings,
261 "errors": errors,
262 }
265def build_parser() -> argparse.ArgumentParser:
266 parser = argparse.ArgumentParser(
267 description=(
268 "Validate a strict local CSV/JSON evidence ledger. "
269 "No network requests are made."
270 )
271 )
272 parser.add_argument("ledger", help="Local .csv or .json source ledger")
273 parser.add_argument("--output", help="Optional local .json validation report")
274 parser.add_argument(
275 "--force", action="store_true", help="Replace an existing --output file"
276 )
277 return parser
280def main() -> int:
281 args = build_parser().parse_args()
282 try:
283 report = validate_records(_load_records(args.ledger))
284 write_json_report(report, args.output, force=args.force)
285 return 0 if report["valid"] else 1
286 except ValidationError as exc:
287 return error_exit(exc)
290if __name__ == "__main__":
291 raise SystemExit(main())