Invoices#

I use this script to download, archive and print invoice files which are originally generated and handled by Sogei SpA and the Agenzia delle Entrate.

Invoice files are downloaded from PEC accounts (certified mail) as attachments. An HTML file corresponding to the decoded XML invoice file is archived and printed. Finally optional notifications about the operation are sent.

During this process cryptographical signatures and integrity checks are performed.

See also

  • A collection of scripts I have written and/or adapted that I currently use on my systems as automated tasks 1

  • scripts/getmail.py at master · markuz/scripts · GitHub 2

Important

CUPS must be up and running with a default printer set. If not, do this before proceeding.

  1. install the dependencies

    apt-get install python3-pip libcups2-dev ttf-dejavu
    
  2. create the jobs directories. See reference

    mkdir -p /home/jobs/{scripts,services}/by-user/myuser
    chown -R myuser:myuser /home/jobs/{scripts,services}/by-user/myuser
    chmod 700 -R /home/jobs/{scripts,services}/by-user/myuser
    
  3. create the script

    /home/jobs/scripts/by-user/myuser/archive_invoice_files.py#
      1#!/usr/bin/env python3
      2#
      3# archive_invoice_files.py
      4#
      5# Copyright (C) 2019-2022 Franco Masotti (franco \D\o\T masotti {-A-T-} tutanota \D\o\T com)
      6#
      7# This program is free software: you can redistribute it and/or modify
      8# it under the terms of the GNU General Public License as published by
      9# the Free Software Foundation, either version 3 of the License, or
     10# (at your option) any later version.
     11#
     12# This program is distributed in the hope that it will be useful,
     13# but WITHOUT ANY WARRANTY; without even the implied warranty of
     14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     15# GNU General Public License for more details.
     16#
     17# You should have received a copy of the GNU General Public License
     18# along with this program.  If not, see <https://www.gnu.org/licenses/>.
     19#
     20# See more copyrights and licenses below.
     21r"""Download, verify, archive and print invoice files."""
     22
     23import email
     24import imaplib
     25import pathlib
     26import shlex
     27import shutil
     28import subprocess
     29import sys
     30import tempfile
     31import traceback
     32from itertools import permutations
     33
     34import cups
     35import dateutil.parser
     36import fattura_elettronica_reader
     37import fpyutils
     38import lxml.etree
     39import yaml
     40from weasyprint import CSS, HTML
     41
     42
     43class EmailError(Exception):
     44    r"""Error."""
     45
     46
     47def get_attachments(config: dict):
     48    r"""Download and save the attachments."""
     49    validate_config_struct(config)
     50
     51    # Most of this function comes from
     52    # https://github.com/markuz/scripts/blob/master/getmail.py
     53    #
     54    # This file is part of my scripts project
     55    #
     56    # Copyright (c) 2011 Marco Antonio Islas Cruz
     57    #
     58    # This script is free software; you can redistribute it and/or modify
     59    # it under the terms of the GNU General Public License as published by
     60    # the Free Software Foundation; either version 2 of the License, or
     61    # (at your option) any later version.
     62    #
     63    # This script is distributed in the hope that it will be useful,
     64    # but WITHOUT ANY WARRANTY; without even the implied warranty of
     65    # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     66    # GNU General Public License for more details.
     67    #
     68    # You should have received a copy of the GNU General Public License
     69    # along with this program; if not, write to the Free Software
     70    # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
     71    #
     72    # @author    Marco Antonio Islas Cruz <markuz@islascruz.org>
     73    # @copyright 2011 Marco Antonio Islas Cruz
     74    # @license   http://www.gnu.org/licenses/gpl.txt
     75    conn = imaplib.IMAP4_SSL(host=config['certified_email']['host'], port=config['certified_email']['port'])
     76    conn.login(user=config['certified_email']['username'], password=config['certified_email']['password'])
     77    conn.select(mailbox=config['certified_email']['mailbox'])
     78
     79    # message_ids is 1-element list of message ids in BytesIO form.
     80    # Filter by subject and unread emails.
     81    # See:
     82    # https://tools.ietf.org/html/rfc2060.html
     83    # for all the commands and parameters.
     84    typ, message_ids = conn.search(None, '(SUBJECT "' + config['certified_email']['subject_filter'] + '")',
     85                                   '(UNSEEN)')
     86    if typ != 'OK':
     87        raise EmailError
     88
     89    # Email id.
     90    i = 0
     91    # Group attachments by email so that they can be processed easily.
     92    saved_files = dict()
     93    for m_id in message_ids[0].split():
     94        # Once the message is processed it will be set as SEEN (read).
     95
     96        # Attachment group.
     97        saved_files[i] = list()
     98
     99        # Returned data are tuples of message part envelope and data.
    100        # data is 1-element list.
    101        # data [0][0] corresponds to the header,
    102        # while data[0][1] corresponds to the text.
    103        # See:
    104        # https://tools.ietf.org/html/rfc2060.html#page-41
    105        # in particular the RFC822 and BODY parameters.
    106        typ, data = conn.fetch(m_id, '(RFC822)')
    107        if typ != 'OK':
    108            raise EmailError
    109
    110        # Load payload in the email data structure.
    111        text = data[0][1]
    112        msg = email.message_from_bytes(text)
    113
    114        # Get the receiving date of the email.
    115        date = msg['Date']
    116
    117        # Iterate through all the attachments of the email.
    118        for part in msg.walk():
    119            # Skip current element if necessary.
    120            if part.get_content_maintype() == 'multipart':
    121                print('iterating down the tree, skipping...')
    122                continue
    123            if part.get('Content-Disposition') is None:
    124                print('unkown content disposition, skipping...')
    125                continue
    126
    127            # Get the filename and the content.
    128            filename = part.get_filename()
    129            data = part.get_payload(decode=True)
    130
    131            # Get the year and month in terms of local time when the email
    132            # was received.
    133            dt = dateutil.parser.parse(date)
    134            # Define a subpath of 'year/month'.
    135            date_part_path = dt.astimezone(
    136                dateutil.tz.tzlocal()).strftime('%Y/%m')
    137
    138            if (filename is not None and data and
    139                    filename not in config['files']['ignore_attachments']):
    140                dst_directory = str(pathlib.Path(config['files']['destination_base_directory'], date_part_path))
    141                # Create the final directory.
    142                pathlib.Path(dst_directory).mkdir(mode=0o700,
    143                                                  parents=True,
    144                                                  exist_ok=True)
    145                # Compute the filename path based on the final directory.
    146                filename = str(pathlib.Path(dst_directory, filename))
    147
    148                # Write the attachment content to its file.
    149                with open(filename, 'wb') as f:
    150                    f.write(data)
    151                saved_files[i].append(filename)
    152            else:
    153                print(
    154                    'undefined filename or no attachments, marking as read anyway'
    155                )
    156        i += 1
    157
    158    conn.close()
    159    conn.logout()
    160
    161    return saved_files
    162
    163
    164def decode_invoice_file(metadata_file: str, invoice_file: str, extract_attachments: bool = False) -> dict:
    165    r"""Try to decode the invoice file."""
    166    source = 'invoice'
    167    file_type = 'p7m'
    168    data = {
    169        'patched': True,
    170        'configuration_file': str(),
    171        'write_default_configuration_file': False,
    172        'extract_attachments': extract_attachments,
    173        'invoice_xslt_type': 'ordinaria',
    174        'no_invoice_xml_validation': False,
    175        'force_invoice_schema_file_download': False,
    176        'generate_html_output': True,
    177        'invoice_filename': invoice_file,
    178        'no_checksum_check': False,
    179        'force_invoice_xml_stylesheet_file_download': False,
    180        'ignore_attachment_extension_whitelist': False,
    181        'ignore_attachment_filetype_whitelist': False,
    182        'metadata_file': metadata_file,
    183        'ignore_signature_check': False,
    184        'ignore_signers_certificate_check': False,
    185        'force_trusted_list_file_download': False,
    186        'keep_original_file': True,
    187        'ignore_assets_checksum': False,
    188        'destination_directory': str(pathlib.Path(invoice_file).parents[0])
    189    }
    190
    191    status = {
    192        'invoice_file': invoice_file,
    193        'valid_checksum': True,
    194        'valid_signature_and_signers_certificate': True,
    195        'valid_assets_checksum': True,
    196        'file_type': file_type,
    197    }
    198
    199    # Most probably a metadata file or a non-signed invoice file.
    200    # Metadata file must have .xml as extension
    201    # Avoid case sensitivity problems.
    202    if str(pathlib.PurePath(metadata_file).suffix).lower() == '.xml':
    203        done = False
    204    else:
    205        done = True
    206        # Unprocessed.
    207        status['invoice_file'] = str()
    208
    209    while not done:
    210        try:
    211            fattura_elettronica_reader.pipeline(source=source,
    212                                                file_type=file_type,
    213                                                data=data)
    214            done = True
    215        except fattura_elettronica_reader.exceptions.InvoiceFileChecksumFailed:
    216            if status['valid_checksum']:
    217                status['valid_checksum'] = False
    218                # Ignore checksum at the next iteration but mark the checksum
    219                # as invalid.
    220                data['no_checksum_check'] = True
    221        except fattura_elettronica_reader.exceptions.P7MFileNotAuthentic:
    222            if status['valid_signature_and_signers_certificate']:
    223                status['valid_signature_and_signers_certificate'] = False
    224                data['ignore_signature_check'] = True
    225                data['ignore_signers_certificate_check'] = True
    226        except fattura_elettronica_reader.exceptions.P7MFileDoesNotHaveACoherentCryptographicalSignature:
    227            if status['file_type'] == 'p7m':
    228                status['file_type'] = 'plain'
    229                file_type = 'plain'
    230        except lxml.etree.LxmlError:
    231            # The selected metadata file is the real invoice file.
    232            # Retry with the next loop from the caller function.
    233            done = True
    234            traceback.print_exc()
    235        except fattura_elettronica_reader.exceptions.AssetsChecksumDoesNotMatch:
    236            if status['valid_assets_checksum']:
    237                status['valid_assets_checksum'] = False
    238                data['ignore_assets_checksum'] = True
    239        except fattura_elettronica_reader.exceptions.CannotExtractOriginalP7MFile:
    240            # Fatal error.
    241            done = True
    242            traceback.print_exc()
    243            sys.exit(1)
    244
    245    return status
    246
    247
    248def validate_decoded_invoice_files_struct(struct: list):
    249    r"""Check if the data structure corresponds to the specifications."""
    250    for e in struct:
    251        if not isinstance(e, dict):
    252            raise TypeError
    253        if 'invoice_file' not in e:
    254            raise ValueError
    255        if 'valid_checksum' not in e:
    256            raise ValueError
    257        if 'valid_signature_and_signers_certificate' not in e:
    258            raise ValueError
    259        if 'valid_assets_checksum' not in e:
    260            raise ValueError
    261        if 'file_type' not in e:
    262            raise ValueError
    263        if not isinstance(e['invoice_file'], str):
    264            raise TypeError
    265        if not isinstance(e['valid_checksum'], bool):
    266            raise TypeError
    267        if not isinstance(e['valid_signature_and_signers_certificate'], bool):
    268            raise TypeError
    269        if not isinstance(e['valid_assets_checksum'], bool):
    270            raise TypeError
    271        if not isinstance(e['file_type'], str):
    272            raise TypeError
    273        if e['file_type'] not in ['p7m', 'plain']:
    274            raise ValueError
    275
    276
    277def validate_config_struct(data: dict):
    278    r"""Check if the data structure corresponds to the specifications."""
    279    if 'certified_email' not in data:
    280        raise ValueError
    281    if 'files' not in data:
    282        raise ValueError
    283    if 'print' not in data:
    284        raise ValueError
    285    if 'invoice' not in data:
    286        raise ValueError
    287    if 'status_page' not in data:
    288        raise ValueError
    289    if 'notify' not in data:
    290        raise ValueError
    291
    292    if 'host' not in data['certified_email']:
    293        raise ValueError
    294    if 'port' not in data['certified_email']:
    295        raise ValueError
    296    if 'username' not in data['certified_email']:
    297        raise ValueError
    298    if 'password' not in data['certified_email']:
    299        raise ValueError
    300    if 'mailbox' not in data['certified_email']:
    301        raise ValueError
    302    if 'subject_filter' not in data['certified_email']:
    303        raise ValueError
    304
    305    if 'destination_base_directory' not in data['files']:
    306        raise ValueError
    307    if 'ignore_attachments' not in data['files']:
    308        raise ValueError
    309
    310    if 'printer' not in data['print']:
    311        raise ValueError
    312    if 'css_string' not in data['print']:
    313        raise ValueError
    314
    315    if 'file' not in data['invoice']:
    316        raise ValueError
    317    if 'attachments' not in data['invoice']:
    318        raise ValueError
    319
    320    if 'file' not in data['status_page']:
    321        raise ValueError
    322    if 'show' not in data['status_page']:
    323        raise ValueError
    324    if 'status' not in data['status_page']:
    325        raise ValueError
    326
    327    if 'gotify' not in data['notify']:
    328        raise ValueError
    329
    330    if not isinstance(data['certified_email']['host'], str):
    331        raise TypeError
    332    if not isinstance(data['certified_email']['port'], int):
    333        raise TypeError
    334    if not isinstance(data['certified_email']['username'], str):
    335        raise TypeError
    336    if not isinstance(data['certified_email']['password'], str):
    337        raise TypeError
    338    if not isinstance(data['certified_email']['mailbox'], str):
    339        raise TypeError
    340    if not isinstance(data['certified_email']['subject_filter'], str):
    341        raise TypeError
    342
    343    if not isinstance(data['files']['destination_base_directory'], str):
    344        raise TypeError
    345    if not isinstance(data['files']['ignore_attachments'], list):
    346        raise TypeError
    347
    348    if not isinstance(data['print']['printer'], str):
    349        raise TypeError
    350    if not isinstance(data['print']['css_string'], str):
    351        raise TypeError
    352
    353    if 'print' not in data['invoice']['file']:
    354        raise ValueError
    355    if 'extract' not in data['invoice']['attachments']:
    356        raise ValueError
    357    if 'print' not in data['invoice']['attachments']:
    358        raise ValueError
    359
    360    if 'store' not in data['status_page']['file']:
    361        raise ValueError
    362    if 'print' not in data['status_page']['file']:
    363        raise ValueError
    364
    365    if 'info' not in data['status_page']['show']:
    366        raise ValueError
    367    if 'openssl_version' not in data['status_page']['show']:
    368        raise ValueError
    369
    370    if 'crypto' not in data['status_page']['status']:
    371        raise ValueError
    372    if 'checksum' not in data['status_page']['status']:
    373        raise ValueError
    374    if 'p7m' not in data['status_page']['status']:
    375        raise ValueError
    376    if 'assets' not in data['status_page']['status']:
    377        raise ValueError
    378
    379    if 'enabled' not in data['notify']['gotify']:
    380        raise ValueError
    381    if 'url' not in data['notify']['gotify']:
    382        raise ValueError
    383    if 'token' not in data['notify']['gotify']:
    384        raise ValueError
    385    if 'message' not in data['notify']['gotify']:
    386        raise ValueError
    387    if 'priority' not in data['notify']['gotify']:
    388        raise ValueError
    389
    390    for a in data['files']['ignore_attachments']:
    391        if not isinstance(a, str):
    392            raise TypeError
    393
    394    if not isinstance(data['invoice']['file']['print'], bool):
    395        raise TypeError
    396    if not isinstance(data['invoice']['attachments']['extract'], bool):
    397        raise TypeError
    398    if not isinstance(data['invoice']['attachments']['print'], bool):
    399        raise TypeError
    400
    401    if not isinstance(data['status_page']['file']['store'], bool):
    402        raise TypeError
    403    if not isinstance(data['status_page']['file']['print'], bool):
    404        raise TypeError
    405
    406    if 'enabled' not in data['status_page']['show']['info']:
    407        raise ValueError
    408    if 'url' not in data['status_page']['show']['info']:
    409        raise ValueError
    410
    411    if 'enabled' not in data['status_page']['show']['openssl_version']:
    412        raise ValueError
    413
    414    if 'enabled' not in data['status_page']['status']['crypto']:
    415        raise ValueError
    416    if 'message' not in data['status_page']['status']['crypto']:
    417        raise ValueError
    418    if 'valid_value' not in data['status_page']['status']['crypto']:
    419        raise ValueError
    420    if 'invalid_value' not in data['status_page']['status']['crypto']:
    421        raise ValueError
    422
    423    if 'enabled' not in data['status_page']['status']['checksum']:
    424        raise ValueError
    425    if 'message' not in data['status_page']['status']['checksum']:
    426        raise ValueError
    427    if 'valid_value' not in data['status_page']['status']['checksum']:
    428        raise ValueError
    429    if 'invalid_value' not in data['status_page']['status']['checksum']:
    430        raise ValueError
    431
    432    if 'enabled' not in data['status_page']['status']['p7m']:
    433        raise ValueError
    434    if 'message' not in data['status_page']['status']['p7m']:
    435        raise ValueError
    436    if 'valid_value' not in data['status_page']['status']['p7m']:
    437        raise ValueError
    438    if 'invalid_value' not in data['status_page']['status']['p7m']:
    439        raise ValueError
    440
    441    if 'enabled' not in data['status_page']['status']['assets']:
    442        raise ValueError
    443    if 'message' not in data['status_page']['status']['assets']:
    444        raise ValueError
    445    if 'valid_value' not in data['status_page']['status']['assets']:
    446        raise ValueError
    447    if 'invalid_value' not in data['status_page']['status']['assets']:
    448        raise ValueError
    449
    450    if not isinstance(data['status_page']['show']['info']['enabled'], bool):
    451        raise TypeError
    452    if not isinstance(data['status_page']['show']['info']['url'], str):
    453        raise TypeError
    454
    455    if not isinstance(data['status_page']['show']['openssl_version']['enabled'], bool):
    456        raise TypeError
    457
    458    if not isinstance(data['status_page']['status']['crypto']['enabled'], bool):
    459        raise TypeError
    460    if not isinstance(data['status_page']['status']['crypto']['message'], str):
    461        raise TypeError
    462    if not isinstance(data['status_page']['status']['crypto']['valid_value'], str):
    463        raise TypeError
    464    if not isinstance(data['status_page']['status']['crypto']['invalid_value'], str):
    465        raise TypeError
    466
    467    if not isinstance(data['status_page']['status']['checksum']['enabled'], bool):
    468        raise TypeError
    469    if not isinstance(data['status_page']['status']['checksum']['message'], str):
    470        raise TypeError
    471    if not isinstance(data['status_page']['status']['checksum']['valid_value'], str):
    472        raise TypeError
    473    if not isinstance(data['status_page']['status']['checksum']['invalid_value'], str):
    474        raise TypeError
    475
    476    if not isinstance(data['status_page']['status']['p7m']['enabled'], bool):
    477        raise TypeError
    478    if not isinstance(data['status_page']['status']['p7m']['message'], str):
    479        raise TypeError
    480    if not isinstance(data['status_page']['status']['p7m']['valid_value'], str):
    481        raise TypeError
    482    if not isinstance(data['status_page']['status']['p7m']['invalid_value'], str):
    483        raise TypeError
    484
    485    if not isinstance(data['status_page']['status']['assets']['enabled'], bool):
    486        raise TypeError
    487    if not isinstance(data['status_page']['status']['assets']['message'], str):
    488        raise TypeError
    489    if not isinstance(data['status_page']['status']['assets']['valid_value'], str):
    490        raise TypeError
    491    if not isinstance(data['status_page']['status']['assets']['invalid_value'], str):
    492        raise TypeError
    493
    494
    495def decode_invoice_files(file_group: dict, extract_attachments: bool = False) -> list:
    496    r"""Decode multiple invoice files."""
    497    invoice_files = list()
    498    for i in file_group:
    499        files = file_group[i]
    500        perm = permutations(files)
    501        files_perm = list(perm)
    502
    503        j = 0
    504        done = False
    505        while j < len(files_perm) and not done:
    506            # Try all permutations.
    507            metadata_file = files_perm[j][0]
    508            invoice_file = files_perm[j][1]
    509            status = decode_invoice_file(metadata_file, invoice_file, extract_attachments)
    510            if status['invoice_file'] != str():
    511                # Ignore unprocessed files.
    512                invoice_files.append(status)
    513
    514                # There is no need to try to invert the input files because
    515                # processing completed correctly.
    516                done = True
    517            j += 1
    518
    519    return invoice_files
    520
    521
    522def print_file(printer, file, job_name, proprieties):
    523    r"""Print a file with CUPS."""
    524    conn = cups.Connection()
    525    conn.printFile(printer, file, job_name, proprieties)
    526
    527
    528def print_invoice(file: dict, data: dict):
    529    r"""Print the invoice file."""
    530    validate_config_struct(data)
    531
    532    html_file = file['invoice_file'] + '.html'
    533    with tempfile.NamedTemporaryFile() as g:
    534        css = CSS(string=data['print']['css_string'])
    535        html = HTML(html_file)
    536        temp_name = g.name
    537        html.write_pdf(temp_name, stylesheets=[css])
    538        print_file(data['print']['printer'], temp_name, 'invoice', {'media': 'a4'})
    539
    540
    541def get_status_page(file: dict, data: dict):
    542    r"""Save and print the status page."""
    543    validate_config_struct(data)
    544
    545    html_file = file['invoice_file'] + '.html'
    546
    547    content = '<h1>' + pathlib.Path(html_file).stem + '</h1>'
    548    if data['status_page']['show']['info']['enabled']:
    549        content += '<h2>generated by <code>' + data['status_page']['show']['info']['url'] + '</code></h2>'
    550    if data['status_page']['show']['openssl_version']['enabled']:
    551        content += '<h2>' + subprocess.run(
    552            shlex.split('openssl version'),
    553            capture_output=True, shell=False).stdout.decode('UTF-8').rstrip() + '</h2> '
    554    if data['status_page']['status']['crypto']['enabled']:
    555        if file['valid_signature_and_signers_certificate']:
    556            content += '<h1>' + data['status_page']['status']['crypto']['message'] + ' ' + data['status_page']['status']['crypto']['valid_value'] + '</h1>'
    557        else:
    558            content += '<h1>' + data['status_page']['status']['crypto']['message'] + ' ' + data['status_page']['status']['crypto']['invalid_value'] + '</h1>'
    559    if data['status_page']['status']['checksum']['enabled']:
    560        if file['valid_checksum']:
    561            content += '<h1>' + data['status_page']['status']['checksum']['message'] + ' ' + data['status_page']['status']['checksum']['valid_value'] + '</h1>'
    562        else:
    563            content += '<h1>' + data['status_page']['status']['checksum']['message'] + ' ' + data['status_page']['status']['checksum']['invalid_value'] + '</h1>'
    564    if data['status_page']['status']['p7m']['enabled']:
    565        if file['file_type'] == 'p7m':
    566            content += '<h1>' + data['status_page']['status']['p7m']['message'] + ' ' + data['status_page']['status']['p7m']['valid_value'] + '</h1>'
    567        else:
    568            content += '<h1>' + data['status_page']['status']['p7m']['message'] + ' ' + data['status_page']['status']['p7m']['invalid_value'] + '</h1>'
    569    if data['status_page']['status']['assets']['enabled']:
    570        if file['valid_assets_checksum']:
    571            content += '<h1>' + data['status_page']['status']['assets']['message'] + ' ' + data['status_page']['status']['assets']['valid_value'] + '</h1>'
    572        else:
    573            content += '<h1>' + data['status_page']['status']['assets']['message'] + ' ' + data['status_page']['status']['assets']['invalid_value'] + '</h1>'
    574
    575    # Save and print.
    576    with tempfile.TemporaryDirectory() as tmpdirname:
    577        css = CSS(string=data['print']['css_string'])
    578        html = HTML(string=content)
    579        status_page_tmp_path = str(pathlib.Path(tmpdirname, 'status_page.pdf'))
    580        html.write_pdf(status_page_tmp_path, stylesheets=[css])
    581        if data['status_page']['file']['print']:
    582            print_file(data['print']['printer'], status_page_tmp_path, 'status_page',
    583                       {'media': 'a4'})
    584        if data['status_page']['file']['store']:
    585            dir = pathlib.Path(file['invoice_file']).parent
    586            shutil.move(
    587                status_page_tmp_path,
    588                str(
    589                    pathlib.Path(dir,
    590                                 file['invoice_file'] + '_status_page.pdf')))
    591
    592
    593if __name__ == '__main__':
    594    configuration_file = sys.argv[1]
    595    config = yaml.load(open(configuration_file, 'r'), Loader=yaml.SafeLoader)
    596
    597    validate_config_struct(config)
    598
    599    pathlib.Path(config['files']['destination_base_directory']).mkdir(
    600        mode=0o700, parents=True, exist_ok=True)
    601    file_group = get_attachments(config)
    602
    603    decoded_invoice_files = decode_invoice_files(file_group, config['invoice']['attachments']['extract'])
    604
    605    validate_decoded_invoice_files_struct(decoded_invoice_files)
    606    for f in decoded_invoice_files:
    607        if config['invoice']['file']['print']:
    608            print_invoice(f, config)
    609        get_status_page(f, config)
    610
    611        message = 'processed invoice = ' + pathlib.Path(
    612            f['invoice_file']).name
    613
    614        if config['notify']['gotify']['enabled']:
    615            m = config['notify']['gotify']['message'] + '\n' + message
    616            fpyutils.notify.send_gotify_message(
    617                config['notify']['gotify']['url'],
    618                config['notify']['gotify']['token'], m,
    619                config['notify']['gotify']['title'],
    620                config['notify']['gotify']['priority'])
    621        if config['notify']['email']['enabled']:
    622            fpyutils.notify.send_email(message,
    623                                       config['notify']['email']['smtp_server'],
    624                                       config['notify']['email']['port'],
    625                                       config['notify']['email']['sender'],
    626                                       config['notify']['email']['user'],
    627                                       config['notify']['email']['password'],
    628                                       config['notify']['email']['receiver'],
    629                                       config['notify']['email']['subject'])
    
  4. create a configuration file

    /home/jobs/scripts/by-user/myuser/archive_invoice_files.myuser.yaml#
      1#
      2# archive_invoice_files.myuser.conf
      3#
      4# Copyright (C) 2019-2022 Franco Masotti (franco \D\o\T masotti {-A-T-} tutanota \D\o\T com)
      5#
      6# This program is free software: you can redistribute it and/or modify
      7# it under the terms of the GNU General Public License as published by
      8# the Free Software Foundation, either version 3 of the License, or
      9# (at your option) any later version.
     10#
     11# This program is distributed in the hope that it will be useful,
     12# but WITHOUT ANY WARRANTY; without even the implied warranty of
     13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     14# GNU General Public License for more details.
     15#
     16# You should have received a copy of the GNU General Public License
     17# along with this program.  If not, see <https://www.gnu.org/licenses/>.
     18
     19# IMAP configuration.
     20certified_email:
     21    host: '<imap host>'
     22    port: 993
     23    username: '<email>'
     24    password: '<password>'
     25    mailbox: 'Inbox'
     26    subject_filter: 'POSTA CERTIFICATA:'
     27
     28files:
     29    # The full path where the invoice files are stored.
     30    destination_base_directory: '/home/myuser/invoices'
     31
     32    # A list of file names to ignore.
     33    ignore_attachments:
     34        - 'daticert.xml'
     35        - 'smime.p7s'
     36
     37print:
     38    # The printer name as reported by CUPS.
     39    printer: 'My_Printer'
     40
     41    # The CSS string used by WeasyPrint.
     42    css_string: 'body { font-size: 8pt; }; @page { size: A4; margin: 0cm; })])}'
     43
     44invoice:
     45    file:
     46        print: true
     47    attachments:
     48        extract: true
     49
     50        # TODO.
     51        # NOT IMPLEMENTED.
     52        print: true
     53
     54status page:
     55    file:
     56        # Store the status page.
     57        store: true
     58        print: true
     59    show:
     60        info:
     61            enabled: true
     62            url: 'https://docs.franco.net.eu.org/ftutorials/desktop/download/invoices.html'
     63        openssl version:
     64            enabled: true
     65    status:
     66        crypto:
     67            enabled: true
     68            message: 'Signature and/or certificare are'
     69            valid_value: 'valid [OK]'
     70            invalid_value: 'NOT valid! [WARNING]'
     71        schema:
     72            # TODO.
     73            # NOT IMPLEMENTED.
     74            enabled: true
     75            message: 'The scheme is'
     76            valid: 'valid [OK]'
     77            invalid: 'NOT valid! [WARNING]'
     78        checksum:
     79            enabled: true
     80            message: 'File integrity check is'
     81            valid_value: 'valid [OK]'
     82            invalid_value: 'NOT valid! [WARNING]'
     83        p7m:
     84            enabled: true
     85            message: 'The original file is a'
     86            valid_value: 'signed P7M'
     87            invalid_value: 'plain XML'
     88        assets:
     89            enabled: true
     90            message: 'Assets file integrity check is'
     91            valid_value: 'valid [OK]'
     92            invalid_value: 'NOT valid! [WARNING]'
     93
     94notify:
     95    email:
     96        enabled: false
     97        smtp_server: 'smtp.gmail.com'
     98        port: 465
     99        sender: 'myusername@gmail.com'
    100        user: 'myusername'
    101        password: 'my awesome password'
    102        receiver: 'myusername@gmail.com'
    103        subject: 'archive invoice(s)'
    104    gotify:
    105        priority: 5
    106        enabled: false
    107        url: '<gotify url>'
    108        token: '<app token>'
    109        title: 'archive invoice(s)'
    110        message: 'archive and print invoices for user myuser'
    
  5. use this Systemd service unit file

    /home/jobs/services/by-user/myuser/archive-invoice-files.myuser.service#
     1[Unit]
     2Description=archive and print myusers's invoices
     3Requires=network-online.target
     4After=network-online.target
     5
     6[Service]
     7Type=simple
     8ExecStart=/home/myuser/.local/venv/archive_invoice_files/bin/python3 /home/jobs/scripts/by-user/myuser/archive_invoice_files.py /home/jobs/scripts/by-user/myuser/archive_invoice_files.myuser.yaml
     9User=myuser
    10Group=myuser
    11
    12[Install]
    13WantedBy=multi-user.target
    
  6. use this Systemd timer unit file

    /home/jobs/services/by-user/myuser/archive-invoice-files.myuser.timer#
    1[Unit]
    2Description=Once a day archive and print myusers's invoices
    3
    4[Timer]
    5OnCalendar=*-*-* 6:00:00
    6Persistent=true
    7
    8[Install]
    9WantedBy=timers.target
    
  7. fix the permissions

    chown -R myuser:myuser /home/jobs/{scripts,services}/by-user/myuser
    chmod 700 -R /home/jobs/{scripts,services}/by-user/myuser
    
  8. run the deploy script

  9. go back to the desktop user create a new virtual environment

    exit
    export ENVIRONMENT_NAME="archive_invoice_files"
    python3 -m venv ~/.local/venv/"${ENVIRONMENT_NAME}"
    . ~/.local/venv/"${ENVIRONMENT_NAME}"/bin/activate
    pip3 install wheel
    pip3 install requests fpyutils==2.2.0 python-dateutil fattura-elettronica-reader==3.0.3 WeasyPrint==52.1 pycups lxml
    deactivate
    

Footnotes

1

https://software.franco.net.eu.org/frnmst/automated-tasks GNU GPLv3+, copyright (c) 2019-2022, Franco Masotti

2

https://github.com/markuz/scripts/blob/master/getmail.py GNU GPL v2+, Copyright (c) 2011 Marco Antonio Islas Cruz