1 files changed, 136 insertions, 0 deletions

diff --git a/beancount_extras_kris7t/importers/hetzner_pdf.py b/beancount_extras_kris7t/importers/hetzner_pdf.py
new file mode 100644
index 0000000..7eb89b6
--- /dev/null
+++ b/beancount_extras_kris7t/importers/hetzner_pdf.py
@@ -0,0 +1,136 @@
+'''
+Importer for Hetzner PDF invoices.
+'''
+__copyright__ = 'Copyright (c) 2020  Kristóf Marussy <kristof@marussy.com>'
+__license__ = 'GNU GPLv2'
+
+import datetime as dt
+import logging
+import re
+from typing import cast, Iterable, Optional
+
+from beancount.core import amount as am, data
+from beancount.core.amount import Amount
+from beancount.core.flags import FLAG_OKAY, FLAG_WARNING
+from beancount.core.number import D
+from beancount.ingest.cache import _FileMemo as FileMemo
+from beancount.ingest.importer import ImporterProtocol
+
+from pdfminer.high_level import extract_pages
+from pdfminer.layout import LTPage, LTTextContainer
+
+from beancount_extras_kris7t.importers.utils import MISSING_AMOUNT
+
+INVOICE_REGEX = re.compile(
+    r'.*Hetzner_(?P<date>\d{4}-\d{2}-\d{2})_(?P<number>R\d+)\.pdf$', re.IGNORECASE)
+AMOUNT_REGEX = re.compile(r'Amount due: € (?P<amount>\d+(\.\d+)?)', re.IGNORECASE)
+BALANCE_REGEX = re.compile(
+    'The amount has been charged to the credit balance on your client credit account.',
+    re.IGNORECASE)
+CARD_REGEX = re.compile(
+    'The invoice amount will soon be debited from your credit card.',
+    re.IGNORECASE)
+MIXED_REGEX = re.compile(
+    r'The amount of € (?P<balance>\d+(\.\d+)?) has been charged to the credit balance ' +
+    r'on your client credit account. The remaining amount of € (?P<card>\d(\.\d+)?) ' +
+    'will be debited by credit card in the next few days.',
+    re.IGNORECASE)
+
+
+def _extract_match(pages: Iterable[LTPage], pattern: re.Pattern) -> Optional[re.Match]:
+    for page in pages:
+        for element in page:
+            if isinstance(element, LTTextContainer):
+                text = element.get_text().strip().replace('\n', ' ')
+                if match := pattern.match(text):
+                    return match
+    return None
+
+
+class Importer(ImporterProtocol):
+    '''
+    Importer for Hetzner PDF invoices.
+    '''
+
+    _log: logging.Logger
+    _liability: str
+    _credit_balance: str
+    _expense: str
+
+    def __init__(self, liability: str, credit_balance: str, expense: str):
+        self._log = logging.getLogger(type(self).__qualname__)
+        self._liability = liability
+        self._credit_balance = credit_balance
+        self._expense = expense
+
+    def identify(self, file: FileMemo) -> bool:
+        return INVOICE_REGEX.match(file.name) is not None
+
+    def file_name(self, file: FileMemo) -> str:
+        if match := INVOICE_REGEX.match(file.name):
+            number = match.group('number')
+            return f'Hetzner_{number}.pdf'
+        else:
+            raise RuntimeError(f'Not an invoice: {file.name}')
+
+    def file_account(self, file: FileMemo) -> str:
+        if INVOICE_REGEX.match(file.name) is None:
+            raise RuntimeError(f'Not an invoice: {file.name}')
+        else:
+            return self._liability
+
+    def file_date(self, file: FileMemo) -> dt.date:
+        if match := INVOICE_REGEX.match(file.name):
+            date_str = match.group('date')
+            return dt.datetime.strptime(date_str, '%Y-%m-%d').date()
+        else:
+            raise RuntimeError(f'Not an invoice: {file.name}')
+
+    def extract(self, file: FileMemo) -> data.Entries:
+        if match := INVOICE_REGEX.match(file.name):
+            invoice_number = match.group('number')
+        else:
+            self._log.warn('Not an invoice: %s', file.name)
+            return []
+        date = self.file_date(file)
+        pages = [page for page in cast(Iterable[LTPage], extract_pages(file.name))]
+        if match := _extract_match(pages, AMOUNT_REGEX):
+            number = D(match.group('amount'))
+            assert number is not None
+            amount = Amount(number, 'EUR')
+        else:
+            self._log.warn('Not amount found in %s', file.name)
+            return []
+        postings = []
+        flag = FLAG_OKAY
+        if _extract_match(pages, BALANCE_REGEX) is not None:
+            postings.append(
+                data.Posting(self._credit_balance, -amount, None, None, None, None))
+        elif _extract_match(pages, CARD_REGEX) is not None:
+            postings.append(
+                data.Posting(self._liability, -amount, None, None, None, None))
+        elif match := _extract_match(pages, MIXED_REGEX):
+            balance_number = D(match.group('balance'))
+            assert balance_number is not None
+            balance_amount = Amount(balance_number, 'EUR')
+            postings.append(
+                data.Posting(self._credit_balance, -balance_amount, None, None, None, None))
+            card_number = D(match.group('card'))
+            assert card_number is not None
+            card_amount = Amount(card_number, 'EUR')
+            postings.append(
+                data.Posting(self._liability, -card_amount, None, None, None, None))
+            if am.add(balance_amount, card_amount) != amount:
+                self._log.warn('Payments do not cover total amount in %s', file.name)
+                flag = FLAG_WARNING
+        else:
+            self._log.warn('Unknown payment method in %s', file.name)
+            flag = FLAG_WARNING
+        if flag == FLAG_OKAY:
+            amount = MISSING_AMOUNT
+        postings.append(
+            data.Posting(self._expense, amount, None, None, None, None))
+        return [
+            data.Transaction(data.new_metadata(file.name, 0), date, flag, 'Hetzner', 'Invoice',
+                             set(), {f'hetzner_{invoice_number}'}, postings)
+        ]