Check article expiry (#59)

* build artefacts in gitignore

* Add support for date_fin attribute and is_open_ended method

* add test case

* start check_expiry

* add cli command

* archive support

* add latest_version_id attribute and related logic

* move to datetime

* fixes

* fix

* assert_never requires typing_extensions before python 3.11

* typer needs an explicit Exit exception to return an error code

.gitignore

@@ -12,3 +12,5 @@ __pycache__/
 # Unit test / coverage reports
 .coverage
 
+.tox/
+dist/

pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
   "mdformat>=0.7.16",
   "more-itertools",
   "typer[all]",
+  "typing-extensions"
 ]
 dynamic = ["version"]
 

src/catleg/catleg.py

@@ -4,6 +4,7 @@
 
 import typer
 
+from catleg.check_expiry import check_expiry as expiry
 from catleg.cli_util import set_basic_loglevel
 from catleg.find_changes import find_changes
 from catleg.query import get_backend
@@ -26,6 +27,16 @@ def diff(file: Path):
         asyncio.run(find_changes(f, file_path=file))
 
 
+@app.command()
+def check_expiry(file: Path):
+    """
+    Check articles in a catala file for expiry.
+    """
+    with open(file) as f:
+        retcode = asyncio.run(expiry(f, file_path=file))
+        raise typer.Exit(retcode)
+
+
 @app.command()
 def query(article_id: str):
     """

src/catleg/check_expiry.py

@@ -0,0 +1,49 @@
+import logging
+import warnings
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import TextIO
+
+from catleg.parse_catala_markdown import parse_catala_file
+from catleg.query import get_backend
+
+
+logger = logging.getLogger(__name__)
+
+
+async def check_expiry(f: TextIO, *, file_path: Path | None = None):
+    # parse articles from file
+    articles = parse_catala_file(f, file_path=file_path)
+
+    back = get_backend("legifrance")
+    ref_articles = await back.articles([article.id.upper() for article in articles])
+    has_expired_articles = False
+    now = datetime.now(timezone.utc)
+
+    for article, ref_article in zip(articles, ref_articles):
+        if ref_article is None:
+            warnings.warn(f"Could not retrieve article '{article.id}'")
+            continue
+
+        if article.is_archive:
+            logger.info("article '%s' is achived, skipping expiry check", article.id)
+            continue
+
+        logger.info("checking article '%s'", article.id)
+
+        if not ref_article.is_open_ended:
+            if now > ref_article.end_date:
+                warnings.warn(
+                    f"Article '{article.id}' has expired "
+                    f"(on {ref_article.end_date.date()}). "
+                    f"It has been replaced by '{ref_article.latest_version_id}'."
+                )
+                has_expired_articles = True
+            else:
+                warnings.warn(
+                    f"Article '{article.id}' will expire "
+                    f"on {ref_article.end_date.date()}. "
+                    f"It will be replaced by '{ref_article.latest_version_id}'"
+                )
+
+    return 0 if not has_expired_articles else 1

src/catleg/law_text_fr.py

@@ -54,6 +54,7 @@ class CatalaFileArticle:
     text: str
     file_path: Path | None
     start_line: int
+    is_archive: bool
 
 
 def parse_article_id(article_id: str) -> tuple[ArticleType, str]:

src/catleg/parse_catala_markdown.py

@@ -45,6 +45,7 @@ def _parse_catala_doc(
                     typ, id = type_id
                     text = []
                     curr_elem = window[0]
+                    is_archive = "[archive]" in window[-1].content
 
                     # start line of the first text block
                     inline_map = window[1].map
@@ -72,6 +73,7 @@ def _parse_catala_doc(
                             text=" ".join(text),
                             start_line=start_line,
                             file_path=file_path,
+                            is_archive=is_archive,
                         )
                     )
 

src/catleg/query.py

@@ -5,21 +5,30 @@
   - legistix database (auto-api via datasette)
 """
 
-import datetime
 import functools
 import logging
 from collections.abc import Iterable
+from datetime import date, datetime, timedelta, timezone
 from typing import Protocol
 
 import aiometer
 import httpx
 from markdownify import markdownify as md  # type: ignore
+from typing_extensions import assert_never
 
 from catleg.config import settings
 
 from catleg.law_text_fr import Article, ArticleType, parse_article_id
 
 
+def _lf_timestamp_to_datetime(ts):
+    return datetime.fromtimestamp(ts / 1000, timezone.utc)
+
+
+# Legifrance uses 2999-01-01 to mark a non-expired or non-expiring text
+END_OF_TIME = _lf_timestamp_to_datetime(32472144000000)
+
+
 class Backend(Protocol):
     async def article(self, id: str) -> Article | None:
         """
@@ -87,7 +96,7 @@ async def _list_codes(self, page_size=20):
         return results, nb_results
 
     async def code_toc(self, id: str):
-        params = {"textId": id, "date": str(datetime.date.today())}
+        params = {"textId": id, "date": str(date.today())}
         reply = await self.client.post(
             f"{self.API_BASE_URL}/consult/legi/tableMatieres", json=params
         )
@@ -124,12 +133,35 @@ def get_backend(spec: str):
 
 
 class LegifranceArticle(Article):
-    def __init__(self, id: str, text: str, text_html: str, nota: str, nota_html: str):
+    def __init__(
+        self,
+        id: str,
+        text: str,
+        text_html: str,
+        nota: str,
+        nota_html: str,
+        end_date: int | str | None,
+        latest_version_id: str,
+    ):
         self._id: str = id
         self._text: str = text
         self._text_html: str = text_html
         self._nota: str = nota
         self._nota_html: str = nota_html
+        self._end_date: datetime = (
+            _lf_timestamp_to_datetime(int(end_date))
+            if end_date is not None
+            else END_OF_TIME
+        )
+        self._latest_version_id = latest_version_id
+
+    @property
+    def end_date(self) -> datetime:
+        return self._end_date
+
+    @property
+    def is_open_ended(self) -> bool:
+        return self._end_date == END_OF_TIME
 
     @property
     def id(self) -> str:
@@ -151,6 +183,10 @@ def nota(self) -> str:
     def nota_html(self) -> str:
         return self._nota_html
 
+    @property
+    def latest_version_id(self) -> str:
+        return self._latest_version_id
+
     def text_and_nota(self) -> str:
         if len(self.nota):
             return f"{self._text}\n\nNOTA :\n\n{self._nota}"
@@ -192,12 +228,26 @@ def _article_from_legifrance_reply(reply) -> Article | None:
     else:
         raise ValueError("Could not parse Legifrance reply")
 
+    article_type, article_id = parse_article_id(article["id"])
+    match article_type:
+        case ArticleType.CETATEXT:
+            latest_version_id = article_id
+        case ArticleType.LEGIARTI | ArticleType.JORFARTI:
+            article_versions = sorted(
+                article["articleVersions"], key=lambda d: int(d["dateDebut"])
+            )
+            latest_version_id = article_versions[-1]["id"]
+        case _:
+            assert_never()
+
     return LegifranceArticle(
         id=article["id"],
         text=article["texte"],
         text_html=article["texteHtml"],
         nota=article["nota"] or "",
         nota_html=article["notaHtml"] or "",
+        end_date=article["dateFin"],
+        latest_version_id=latest_version_id,
     )
 
 
@@ -217,10 +267,10 @@ def __init__(self, client_id: str, client_secret: str):
         self.client_id = client_id
         self.client_secret = client_secret
         self.token = None
-        self.token_expires_at: datetime.datetime | None = None
+        self.token_expires_at: datetime | None = None
 
     def auth_flow(self, request: httpx.Request):
-        if self.token is None or self.token_expires_at <= datetime.datetime.now():
+        if self.token is None or self.token_expires_at <= datetime.now():
             logging.info("Requesting auth token")
             data = {
                 "grant_type": "client_credentials",
@@ -234,9 +284,7 @@ def auth_flow(self, request: httpx.Request):
             resp_json = resp.json()
             self.token = resp_json["access_token"]
             expires_in = int(resp_json["expires_in"])
-            self.token_expires_at = datetime.datetime.now() + datetime.timedelta(
-                seconds=expires_in
-            )
+            self.token_expires_at = datetime.now() + timedelta(seconds=expires_in)
         else:
             logging.info("Using existing auth token")