beancount_cde_importer/__init__.py

@@ -11,20 +11,44 @@ from beancount.core.number import Decimal  # type: ignore
 from beancount.ingest import cache, importer  # type: ignore
 
 
-COL_DATE = "Date de comptabilisation"
+INDEX_DATE = 0
-COL_LABEL = "Libelle operation"
+INDEX_TRANSACTION_NUMBER = 1
-COL_DEBIT = "Debit"
+INDEX_LABEL = 2
-COL_CREDIT = "Credit"
+INDEX_DEBIT = 3
-COL_DETAIL = "Informations complementaires"
+INDEX_CREDIT = 4
+INDEX_DETAIL = 5
 
 END_DATE_REGEX = "Date de fin de téléchargement : ([0-3][0-9]/[0-1][0-9]/[0-9]{4})"
 START_DATE_REGEX = "Date de début de téléchargement : ([0-3][0-9]/[0-1][0-9]/[0-9]{4})"
 
-EXPECTED_HEADER = "Date de comptabilisation;Libelle simplifie;Libelle operation;Reference;Informations complementaires;Type operation;Categorie;Sous categorie;Debit;Credit;Date operation;Date de valeur;Pointage operation"
+
+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"
+    )
 
 
-def is_valid_header(header: str) -> bool:
+def get_date(file: cache._FileMemo, regex: str) -> Optional[date]:
-    return header == EXPECTED_HEADER
+    match: Optional[re.Match] = re.search(regex, file.head())
+    if match is None:
+        return None
+    date_str: Optional[str] = match.group(1)
+    if date_str is None:
+        return None
+    return datetime.strptime(date_str, "%d/%m/%Y").date()
+
+
+def get_end_date(file: cache._FileMemo) -> Optional[date]:
+    return get_date(file, END_DATE_REGEX)
+
+
+def get_start_date(file: cache._FileMemo) -> Optional[date]:
+    return get_date(file, START_DATE_REGEX)
 
 
 class CDEImporter(importer.ImporterProtocol):
@@ -36,7 +60,16 @@ class CDEImporter(importer.ImporterProtocol):
             # NOTE: beancount.ingest.cache._FileMemo handles automatic encoding
             # detection
             lines: list[str] = file.head().splitlines()
-            header: str = lines[0]
+            csv_reader = csv.reader(
+                lines, delimiter=";", strict=True, quoting=csv.QUOTE_NONE
+            )
+
+            # 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:
@@ -49,38 +82,73 @@ class CDEImporter(importer.ImporterProtocol):
         return "CaisseEpargne_Statement.csv"
 
     def file_date(self, file: cache._FileMemo) -> Optional[date]:
-        lines: list[str] = file.contents().splitlines()
+        return get_end_date(file)
-        csv_reader: csv.DictReader = csv.DictReader(
-            lines, delimiter=";", strict=True, quoting=csv.QUOTE_NONE
-        )
-
-        row: Optional[dict[str, str]] = next(csv_reader)
-        if row is None:
-            return None
-
-        return datetime.strptime(
-            row[COL_DATE], "%d/%m/%Y"
-        ).date()
 
     def extract(self, file: cache._FileMemo, existing_entries=None) -> list[Any]:
         directives: list[Any] = []
 
+        end_date: Optional[date] = get_end_date(file)
+        start_date: Optional[date] = get_start_date(file)
+        if end_date is None or start_date is None:
+            return directives
+
         lines: list[str] = file.contents().splitlines()
-        csv_reader = csv.DictReader(
+        csv_reader = csv.reader(
             lines, delimiter=";", strict=True, quoting=csv.QUOTE_NONE
         )
 
+        # first 3 lines are useless
+        for _ in range(3):
+            next(csv_reader)
+
+        # 4th line is usually the final balance
+        row: Optional[list[str]] = next(csv_reader)
+        if row is None:
+            return directives
+
+        if row[0] == "Solde en fin de période":
+            meta = data.new_metadata(file.name, 4)
+            balance = Decimal(row[4].replace(",", "."))
+            directives.append(
+                data.Balance(
+                    meta=meta,
+                    date=end_date,
+                    account=self.account,
+                    amount=Amount(balance, "EUR"),
+                    tolerance=None,
+                    diff_amount=None,
+                )
+            )
+
+        # skip headings
+        next(csv_reader)
+
         for index, row in enumerate(csv_reader):
-            lineno: int = index + 2  # entries start at line 2
+            lineno: int = index + 6  # entries start at line 6
             meta = data.new_metadata(file.name, lineno)
 
+            if row[0] == "Solde en début de période":
+                balance = Decimal(row[4].replace(",", "."))
+                directives.append(
+                    data.Balance(
+                        meta=meta,
+                        date=start_date,
+                        account=self.account,
+                        amount=Amount(balance, "EUR"),
+                        tolerance=None,
+                        diff_amount=None,
+                    )
+                )
+
+                # should be the last line anyway
+                continue
+
             transaction_date: date = datetime.strptime(
-                row[COL_DATE], "%d/%m/%Y"
+                row[INDEX_DATE], "%d/%m/%y"
             ).date()
-            label: str = row[COL_LABEL]
+            label: str = row[INDEX_LABEL]
-            debit: str = row[COL_DEBIT]
+            debit: str = row[INDEX_DEBIT]
-            credit: str = row[COL_CREDIT]
+            credit: str = row[INDEX_CREDIT]
-            detail: str = row[COL_DETAIL] if row[COL_DETAIL] else ""
 
             postings: list[data.Posting] = []
 
@@ -106,7 +174,7 @@ class CDEImporter(importer.ImporterProtocol):
                     transaction_date,
                     self.FLAG,
                     label,
-                    detail,
+                    "",
                     data.EMPTY_SET,
                     data.EMPTY_SET,
                     postings,

flake.lock

@@ -2,11 +2,11 @@
   "nodes": {
     "flake-utils": {
       "locked": {
-        "lastModified": 1659877975,
+        "lastModified": 1642700792,
-        "narHash": "sha256-zllb8aq3YO3h8B/U0/J1WBgAL8EX5yWf5pMj3G0NAmc=",
+        "narHash": "sha256-XqHrk7hFb+zBvRg6Ghl+AZDq03ov6OshJLiSWOoX5es=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "c0e246b9b83f637f4681389ecabcb2681b4f3af0",
+        "rev": "846b2ae0fc4cc943637d3d1def4454213e203cba",
         "type": "github"
       },
       "original": {
@@ -17,11 +17,11 @@
     },
     "flake-utils_2": {
       "locked": {
-        "lastModified": 1659877975,
+        "lastModified": 1610051610,
-        "narHash": "sha256-zllb8aq3YO3h8B/U0/J1WBgAL8EX5yWf5pMj3G0NAmc=",
+        "narHash": "sha256-U9rPz/usA1/Aohhk7Cmc2gBrEEKRzcW4nwPWMPwja4Y=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "c0e246b9b83f637f4681389ecabcb2681b4f3af0",
+        "rev": "3982c9903e93927c2164caa727cd3f6a0e6d14cc",
         "type": "github"
       },
       "original": {
@@ -32,27 +32,27 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1666867875,
+        "lastModified": 1643247693,
-        "narHash": "sha256-3nD7iQXd/J6KjkT8IjozTuA5p8qjiLKTxvOUmH+AzNM=",
+        "narHash": "sha256-rmShxIuNjYBz4l83J0J++sug+MURUY1koPCzX4F8hfo=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "c132d0837dfb9035701dcd8fc91786c605c855c3",
+        "rev": "6c4b9f1a2fd761e2d384ef86cff0d208ca27fdca",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "nixos-22.05",
+        "ref": "nixos-21.11",
         "repo": "nixpkgs",
         "type": "github"
       }
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1666963493,
+        "lastModified": 1610729867,
-        "narHash": "sha256-RbTJWOQmAAge/7HqD5qDvvTV9devzs/lXwwcDKruOcM=",
+        "narHash": "sha256-bk/SBaBLqZX/PEqal27DMQwAHHl0dcZMp8NNksQr80s=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "38164d1660dcc24b41a5a22f1e9ef075a8e26714",
+        "rev": "04af07c659c6723a2259bb6bc00a47ec53330f20",
         "type": "github"
       },
       "original": {
@@ -67,11 +67,11 @@
         "nixpkgs": "nixpkgs_2"
       },
       "locked": {
-        "lastModified": 1666918719,
+        "lastModified": 1643339875,
-        "narHash": "sha256-BkK42fjAku+2WgCOv2/1NrPa754eQPV7gPBmoKQBWlc=",
+        "narHash": "sha256-9mIB+pCvo1xgLfZZzs9R6j0/9pO2dFCsVLiyuyLyAU0=",
         "owner": "nix-community",
         "repo": "poetry2nix",
-        "rev": "289efb187123656a116b915206e66852f038720e",
+        "rev": "9b601b8cd0f5545ddb0e6a36a01983aca17d9f2a",
         "type": "github"
       },
       "original": {

flake.nix

@@ -1,6 +1,6 @@
 {
   inputs = {
-    nixpkgs.url = "github:NixOS/nixpkgs/nixos-22.05";
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-21.11";
     flake-utils.url = "github:numtide/flake-utils";
     poetry2nix.url = "github:nix-community/poetry2nix";
   };