implement identify

beancount_cde_importer/__init__.py

@@ -1,6 +1,46 @@
-from beancount.ingest import importer
+import csv
+from itertools import islice
+from typing import Optional
+
+from beancount.ingest import cache, importer  # type: ignore
+
+
+INDEX_DATE = 0
+INDEX_TRANSACTION_NUMBER = 1
+INDEX_LABEL = 2
+INDEX_DEBIT = 3
+INDEX_CREDIT = 4
+INDEX_DETAIL = 5
+
+
+def is_valid_header(header: list[str]) -> bool:
+    return (
+        header[INDEX_DATE] == "Date"
+        and header[INDEX_TRANSACTION_NUMBER] == "Numéro d'opération"
+        and header[INDEX_LABEL] == "Libellé"
+        and header[INDEX_DEBIT] == "Débit"
+        and header[INDEX_CREDIT] == "Crédit"
+        and header[INDEX_DETAIL] == "Détail"
+    )
+
 
 class CaisseDEpargneImporter(importer.ImporterProtocol):
+    def identify(self, file: cache._FileMemo) -> bool:
+        try:
+            # NOTE: beancount.ingest.cache._FileMemo handles automatic encoding
+            # detection
+            lines: list[str] = file.head().splitlines()
+            csv_reader = csv.reader(
+                lines, delimiter=";", strict=True, quoting=csv.QUOTE_NONE
+            )
 
-    def identify(self, file):
+            # header is actually on the 5th line, the previous ones contain
+            # miscellaneous information
+            header: Optional[list[str]] = next(islice(csv_reader, 4, None))
+
+            if header is None:
+                return False
+            return is_valid_header(header)
+
+        except:
             return False