(self, source_id: str, source: dict)
| 2594 | self.state = state |
| 2595 | |
| 2596 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 2597 | prev = self.state.get(source_id, {}) |
| 2598 | last_known_amendment = prev.get("last_known_amendment", "") |
| 2599 | last_known_count = prev.get("standards_count", 0) |
| 2600 | |
| 2601 | # Ignore legacy placeholder "99990101" from old HEAD-redirect approach |
| 2602 | if last_known_amendment == "99990101": |
| 2603 | last_known_amendment = "" |
| 2604 | |
| 2605 | new_consol_date = "" |
| 2606 | sparql_ok = False |
| 2607 | |
| 2608 | try: |
| 2609 | # Layer 1: CELLAR SPARQL API -- authoritative consolidated version dates |
| 2610 | new_consol_date = self._sparql_latest_consolidated() |
| 2611 | if new_consol_date: |
| 2612 | sparql_ok = True |
| 2613 | print(f" INFO: SPARQL latest consolidated: {new_consol_date}") |
| 2614 | except Exception as e: |
| 2615 | print(f" WARNING: SPARQL failed ({e}), falling back to HEAD redirect") |
| 2616 | |
| 2617 | if not sparql_ok: |
| 2618 | # Layer 3 fallback: HEAD redirect (unreliable but harmless) |
| 2619 | try: |
| 2620 | consol_url = "https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX:02021D1182-99990101" |
| 2621 | resp = self.session.head(consol_url, timeout=15, allow_redirects=True) |
| 2622 | final_url = str(resp.url) if resp.url else "" |
| 2623 | date_match = re.search(r"02021D1182-(\d{8})", final_url) |
| 2624 | fallback_date = date_match.group(1) if date_match else "" |
| 2625 | if fallback_date and fallback_date != "99990101": |
| 2626 | new_consol_date = fallback_date |
| 2627 | except Exception: |
| 2628 | pass |
| 2629 | |
| 2630 | # Layer 2: Local standards count check |
| 2631 | current_count = self._get_local_standards_count() |
| 2632 | |
| 2633 | self.state[source_id] = { |
| 2634 | "url": source["url"], |
| 2635 | "last_checked": datetime.now().isoformat(), |
| 2636 | "last_known_amendment": new_consol_date or last_known_amendment, |
| 2637 | "standards_count": current_count, |
| 2638 | "detection_method": "sparql" if sparql_ok else "head_fallback", |
| 2639 | } |
| 2640 | |
| 2641 | if not last_known_amendment: |
| 2642 | print(f" INFO: EUR-Lex baseline established (consolidated date: {new_consol_date}, count: {current_count})") |
| 2643 | return None |
| 2644 | |
| 2645 | changes = [] |
| 2646 | if new_consol_date and new_consol_date > last_known_amendment: |
| 2647 | changes.append(f"consolidated date: {last_known_amendment} -> {new_consol_date}") |
| 2648 | |
| 2649 | if last_known_count and current_count != last_known_count: |
| 2650 | changes.append(f"standards count: {last_known_count} -> {current_count}") |
| 2651 | |
| 2652 | if changes: |
| 2653 | return _make_update( |
nothing calls this directly
no test coverage detected