premier commit

.gitignore

+__pycache__/
+bin/
+lib/
+data/
+pyvenv.cfg
+.env
+config.yml
+*.latest.txt

.gitlab-ci.yml

+variables:
+  DEBIAN_FRONTEND: noninteractive
+
+stages:
+  - install
+  - lint
+
+install:
+  stage: install
+  script:
+  - apt-get update -y
+  - apt-get dist-upgrade -y
+  - apt-get install --no-install-recommends -y python3 python3-virtualenv python-is-python3 
+  - virtualenv --python=/usr/bin/python3 .
+  - source bin/activate
+  - pip3 install -r requirements.txt
+  cache:
+    key: $CI_COMMIT_REF_SLUG
+    paths:
+    - bin/
+    - lib/
+    - pyvenv.cfg
+    - *.latest.txt
+
+lint:
+  stage: lint
+  needs: [install]
+  script:
+  - bin/pycodestyle --first --show-source --ignore=E501 *.py
+  cache:
+    key: $CI_COMMIT_REF_SLUG
+    paths:
+    - bin/
+    - lib/
+    - *.latest.txt

Attrap_bot.py

+import email
+import hashlib
+import json
+import logging
+import os
+import requests
+import smtplib
+import ssl
+
+import yaml
+
+from mastodon import Mastodon
+
+# On paramètre le niveau de verbosité des logs
+logging.basicConfig(level=os.environ.get('LOG_LEVEL', 'WARNING').upper())
+logger = logging.getLogger(__name__)
+
+
+class Attrap_bot:
+    class SMTP_sender:
+        port = 587
+        ssl = False
+        starttls = True
+
+        def __init__(self, hostname):
+            self.hostname = hostname
+
+        def set_username(self, username):
+            self.username = username
+
+        def set_password(self, password):
+            self.password = password
+
+        def set_port(self, port):
+            self.port = port
+
+        def set_ssl(self, ssl):
+            self.ssl = ssl
+            if ssl is True:
+                self.starttls = False
+
+        def set_starttls(self, starttls):
+            self.starttls = starttls
+            if starttls is True:
+                self.ssl = False
+
+        def send(self, email_from, email_to, email_subject, email_message):
+            message = email.message.EmailMessage()
+
+            message['Subject'] = email_subject
+            message['From'] = email_from
+            message['Date'] = email.utils.formatdate()
+            message.set_content(email_message)
+
+            if self.ssl:
+                smtp = smtplib.SMTP_SSL(self.hostname, self.port, context=ssl.create_default_context())
+            elif self.starttls:
+                smtp = smtplib.SMTP(self.hostname, self.port)
+                smtp.starttls(context=ssl.create_default_context())
+            else:
+                smtp = smtplib.SMTP(self.hostname, self.port)
+
+            if self.username:
+                smtp.login(self.username, self.password)
+
+            for address in email_to.split(','):
+                del message['To']
+                message['To'] = address
+                del message['Message-ID']
+                message['Message-ID'] = email.utils.make_msgid(domain=email_from.split('@')[-1])
+                smtp.send_message(message)
+
+            smtp.quit()
+
+    class Mastodon_sender:
+        mastodon = None
+
+        def __init__(self, instance, access_token):
+            self.instance = instance
+            self.access_token = access_token
+
+        def send(self, message):
+            if not self.mastodon:
+                self.mastodon = Mastodon(
+                    api_base_url=self.instance,
+                    access_token=self.access_token
+                )
+
+            self.mastodon.toot(message)
+
+    def __init__(self, config_file):
+        # On importe les fichiers de configuration et on surcharge la configuration par défaut
+        config_default_file = yaml.load(open('config.default.yml', 'r'), Loader=yaml.FullLoader)
+        config_file = yaml.load(open(config_file, 'r'), Loader=yaml.FullLoader)
+        self.config = config_default_file | config_file
+
+        logger.info('Démarrage de Attrap_bot')
+        logger.info(f"Source des données : {self.config['data_source']}")
+
+    def analyze(self):
+        for query in self.config['queries']:
+            query_id = query['id']
+            search = query['search']
+            administration = query['administration']
+            if not administration:
+                administration = ""
+
+            logger.info(f'Démarrage de la recherche {query_id}')
+            logger.info(f'Requête : {search}')
+            logger.info(f'Administration : {administration}')
+
+            # On récupère de quoi envoyer par mail et sur Mastodon
+            smtp_sender = None
+            mastodon_sender = None
+            if query.get('email') and query['email'].get('smtp'):
+                logger.debug('Configuration SMTP trouvée')
+                smtp_sender = self.get_smtp_sender(query['email']['smtp'])
+            if query.get('mastodon') and query['mastodon'].get('instance'):
+                logger.debug('Configuration Mastodon trouvée')
+                mastodon_sender = self.get_mastodon_sender(query['mastodon']['instance'])
+
+            # On fabrique l'URL de requête
+            request_url = self.config['data_source'].replace('{search}', search).replace('{administration}', administration)
+
+            # On ouvre le fichier de status
+            status_file_path = f'{query_id}.latest.txt'
+            if os.path.isfile(status_file_path):
+                status = open(status_file_path, 'r').read().strip()
+            else:
+                status = None
+
+            # On interroge l'API d'Attrap
+            response = json.loads(requests.get(request_url, timeout=(10, 120)).content)
+            raa = response['elements']
+
+            # Si le dernier RAA n'est pas celui connu, on analyse les résultats
+            latest = raa[0]
+            if latest['id'] != status:
+                email_message = ""
+
+                # On récupère les derniers RAA jusqu'à ce qu'on trouve le dernier analysé
+                raa_candidates = []
+                for result in raa:
+                    if result['id'] != status:
+                        raa_candidates.append(result)
+                    else:
+                        break
+
+                # Maintenant, on prend les derniers RAA à analyser, pour commencer par les plus anciens en premier
+                for result in raa_candidates[::-1]:
+                    status = result['id']
+                    raa_name = result['name']
+                    raa_date = result['date']
+                    raa_administration = result['administration']
+                    raa_administration_name = result['administration_name']
+                    raa_url = result['url']
+
+                    # On affiche le résultat dans la console
+                    if (self.config['console_output']):
+                        print(f'\033[92m{raa_name}\033[0m ({raa_date}) : {raa_url}')
+
+                    # On publie sur Mastodon
+                    if mastodon_sender:
+                        toot = f'{raa_name} ({raa_date})\n\n{raa_url}'
+                        if query['mastodon']['prefix'] and not query['mastodon']['prefix'] == '':
+                            prefix = query['mastodon']['prefix']
+                            prefix = prefix.replace('{id}', query_id)
+                            prefix = prefix.replace('{administration}', raa_administration)
+                            prefix = prefix.replace('{administration_name}', raa_administration_name)
+                            toot = f"{prefix}\n\n{toot}"
+                        if query['mastodon']['suffix'] and not query['mastodon']['suffix'] == '':
+                            suffix = query['mastodon']['suffix']
+                            suffix = suffix.replace('{id}', query_id)
+                            suffix = suffix.replace('{administration}', raa_administration)
+                            suffix = suffix.replace('{administration_name}', raa_administration_name)
+                            toot = f"{toot}\n\n{suffix}"
+                        mastodon_sender.send(toot)
+
+                    # On complète le mail
+                    if smtp_sender:
+                        if email_message == "":
+                            email_message = f"Attrap ({query_id})\nRequête : {search}"
+                        email_message = f"{email_message}\n\n{raa_administration_name} :\n{raa_name} ({raa_date})\nURL : {raa_url}"
+
+                # On envoie le mail si le message est prêt
+                if not email_message == "":
+                    smtp_sender.send(
+                        query['email']['from'],
+                        query['email']['to'],
+                        f"[Attrap] [{query_id}] Nouveaux éléments trouvés",
+                        email_message
+                    )
+
+                if os.path.isfile(status_file_path):
+                    os.remove(status_file_path)
+                status_file = open(status_file_path, 'w')
+                status_file.write(status)
+                status_file.close()
+
+    def get_mastodon_sender(self, config_id):
+        if self.config.get('mastodon'):
+            for i in self.config['mastodon']:
+                if i.get('id') and i['id'] == config_id and i['instance'] and i['access_token']:
+                    return Attrap_bot.Mastodon_sender(i['instance'], i['access_token'])
+        logger.warning(f"La configuration Mastodon est invalide (id: {query['mastodon']})")
+        return None
+
+    def get_smtp_sender(self, config_id):
+        if self.config.get('smtp'):
+            for i in self.config['smtp']:
+                if i.get('id') and i['id'] == config_id and i['hostname']:
+                    smtp_sender = Attrap_bot.SMTP_sender(i['hostname'])
+                    if i.get('username'):
+                        smtp_sender.set_username(i['username'])
+                    if i.get('password'):
+                        smtp_sender.set_password(i['password'])
+                    if i.get('port'):
+                        smtp_sender.set_port(i['port'])
+                    if i.get('ssl'):
+                        smtp_sender.set_ssl(i['ssl'])
+                    if i.get('starttls'):
+                        smtp_sender.set_starttls(i['starttls'])
+                    return smtp_sender
+        logger.warning(f"La configuration Mastodon est invalide (id: {query['mastodon']})")
+        return None

README.md

+# Attrap_bot
+
+Un robot de veille des Recueils des actes administratifs (RAA) qui utilise les données en ligne de [Attrap](https://attrap.fr).
+
+## Installation
+
+Il est recommandé d'utiliser virtualenv :
+
+```bash
+virtualenv --python=/usr/bin/python3 .
+source bin/activate
+pip3 install -r requirements.txt
+```
+
+## Configuration
+
+Vous pouvez configurer le robot avec un fichier de configuration ou avec des variables d'environnement.
+
+### Fichier de configuration
+
+Le fichier `config.default.yml` contient les paramètres par défaut, notamment la source des données. Copiez-le vers `config.yml` et adaptez le fichier pour surcharger les paramètres par défaut.
+
+Les requêtes sont spécifiées dans la section `queries`, les paramètres pour envoyer les résultats par email dans `smtp` et ceux pour envoyer les résultats sur Mastodon dans `mastodon`. Exemple :
+
+```yaml
+queries:
+  - id: vsa-paris
+    search: "\"traitement algorithmique\" AND 2023-380"
+    administration: "ppparis,pref75"
+    email:
+      from: "attrap@example.org"
+      to: "vsa@example.org"
+      smtp: example_smtp
+  - id: vsa
+    search: "\"traitement algorithmique\" AND 2023-380"
+    administration:
+    mastodon:
+      prefix: "[{administration_name}]"
+      suffix: "#{id} #{administration}"
+      instance: mastodon_social
+
+smtp:
+  - id: example_smtp
+    hostname: "smtp.example.org"
+    port: 587
+    username: "attrap@example.org"
+    password: "secr3t"
+    ssl: False
+    starttls: True
+
+mastodon:
+  - id: mastodon_social
+    instance: "mastodon.social"
+    access_token: "a-secret-access-token-here"
+```
+
+### Variables d'environnement
+
+TODO
+
+## Développement
+
+Pour modifier le niveau de verbosité des logs, vous pouvez configurer la variable `LOG_LEVEL` à `DEBUG`, `INFO`, `WARNING` (par défaut) ou `ERROR`.
+
+## Licence
+
+[CeCILL_V2.1-fr](https://cecill.info/licences/Licence_CeCILL_V2.1-fr.html) (voir le fichier `LICENSE`)

cli.py

+#!/usr/bin/env python3
+
+import argparse
+
+from Attrap_bot import Attrap_bot
+
+parser = argparse.ArgumentParser(
+    prog='cli.py',
+    description='Recherche les derniers RAA correspondant à une certaine requête et envoie les résultats par email et sur Mastodon'
+)
+
+parser.add_argument(
+    '--config', '-c',
+    action='store',
+    help='Fichier de configuration (par défaut: config.yml)'
+)
+
+args = parser.parse_args()
+
+if args.config:
+    config_file = args.config
+else:
+    config_file = 'config.yml'
+
+bot = Attrap_bot(config_file)
+bot.analyze()

config.default.yml

+data_source: "https://attrap.fr/api/v1/search?s={search}&administration={administration}&sort=desc&size=50"

requirements.txt

+Mastodon.py>=1.8.1
+pycodestyle>=2.12.1
+pyyaml>=6.0.2
+requests>=2.32.3