Checks EUR-Lex for harmonised standards amendments via CELLAR SPARQL API. Three-layer detection: 1. SPARQL query: Get all consolidated versions of CID 2021/1182 from CELLAR, compare latest consolidated date against last known. 2. Standards count comparison: Compare current h
| 2565 | # --------------------------------------------------------------------------- |
| 2566 | |
| 2567 | class EURLexChecker: |
| 2568 | """ |
| 2569 | Checks EUR-Lex for harmonised standards amendments via CELLAR SPARQL API. |
| 2570 | |
| 2571 | Three-layer detection: |
| 2572 | 1. SPARQL query: Get all consolidated versions of CID 2021/1182 from CELLAR, |
| 2573 | compare latest consolidated date against last known. |
| 2574 | 2. Standards count comparison: Compare current harmonised standards count against |
| 2575 | local _index.json to detect additions/removals between amendment cycles. |
| 2576 | 3. Fallback HEAD redirect (unreliable, kept as belt-and-suspenders). |
| 2577 | |
| 2578 | Note: The old HEAD-redirect approach to eur-lex.europa.eu returns HTTP 202 with |
| 2579 | empty body (anti-bot), so Layer 1 SPARQL is the primary detection mechanism. |
| 2580 | """ |
| 2581 | SPARQL_URL = "https://publications.europa.eu/webapi/rdf/sparql" |
| 2582 | SPARQL_QUERY = """ |
| 2583 | PREFIX cdm: <http://publications.europa.eu/ontology/cdm#> |
| 2584 | SELECT DISTINCT ?celex ?date ?consoDate WHERE { |
| 2585 | ?work cdm:resource_legal_id_celex ?celex . |
| 2586 | FILTER(STRSTARTS(?celex, "02021D1182")) |
| 2587 | OPTIONAL { ?work cdm:work_date_document ?date . } |
| 2588 | OPTIONAL { ?work cdm:work_date_creation ?consoDate . } |
| 2589 | } ORDER BY DESC(?celex) LIMIT 5 |
| 2590 | """ |
| 2591 | |
| 2592 | def __init__(self, session: requests.Session, state: dict): |
| 2593 | self.session = session |
| 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 "" |