Run version checks and return report.
(
source_filter: str | None = None,
category_filter: str | None = None,
report_path: str | None = None,
)
| 249 | |
| 250 | |
| 251 | def run_check( |
| 252 | source_filter: str | None = None, |
| 253 | category_filter: str | None = None, |
| 254 | report_path: str | None = None, |
| 255 | ) -> dict: |
| 256 | """Run version checks and return report.""" |
| 257 | standards = load_all_standards() |
| 258 | |
| 259 | if source_filter: |
| 260 | standards = [s for s in standards if s["source_type"] == source_filter] |
| 261 | if category_filter: |
| 262 | standards = [s for s in standards if s["category"] == category_filter] |
| 263 | |
| 264 | # Skip sources without checkers (eurlex, aami, bsi, other) |
| 265 | checkable = [s for s in standards if s["source_type"] in CHECKERS] |
| 266 | skipped = [s for s in standards if s["source_type"] not in CHECKERS] |
| 267 | |
| 268 | print(f"Standards to check: {len(checkable)} (skipping {len(skipped)} without checker)") |
| 269 | print(f" ISO: {sum(1 for s in checkable if s['source_type'] == 'iso')}") |
| 270 | print(f" IEC: {sum(1 for s in checkable if s['source_type'] == 'iec')}") |
| 271 | print(f" ASTM: {sum(1 for s in checkable if s['source_type'] == 'astm')}") |
| 272 | print(f" iTeh: {sum(1 for s in checkable if s['source_type'] == 'iteh')}") |
| 273 | print() |
| 274 | |
| 275 | updates = [] |
| 276 | errors = [] |
| 277 | checked = 0 |
| 278 | |
| 279 | for i, std in enumerate(checkable): |
| 280 | checker = CHECKERS[std["source_type"]] |
| 281 | result = checker(std) |
| 282 | checked += 1 |
| 283 | |
| 284 | if result: |
| 285 | if result["type"] == "error": |
| 286 | errors.append({**std, "error": result["detail"]}) |
| 287 | print(f" [{i+1}/{len(checkable)}] ERROR {std['number']}: {result['detail']}") |
| 288 | else: |
| 289 | updates.append({ |
| 290 | "category": std["category"], |
| 291 | "number": std["number"], |
| 292 | "current_url": std["url"], |
| 293 | "signal": result.get("signal", ""), |
| 294 | "new_version": result.get("new_version", ""), |
| 295 | "new_url": result.get("new_url", ""), |
| 296 | "action_required": result["type"], |
| 297 | }) |
| 298 | print(f" [{i+1}/{len(checkable)}] UPDATE {std['number']}: {result['signal']}") |
| 299 | else: |
| 300 | if (i + 1) % 20 == 0: |
| 301 | print(f" [{i+1}/{len(checkable)}] ... {std['number']} OK") |
| 302 | |
| 303 | time.sleep(FETCH_DELAY) |
| 304 | |
| 305 | report = { |
| 306 | "checked_at": datetime.now(timezone.utc).isoformat(), |
| 307 | "total_standards": len(standards), |
| 308 | "total_checked": checked, |
no test coverage detected