ppparis.py

#!/usr/bin/env python3

import os, sys, time, re
import subprocess
from bs4 import BeautifulSoup
import argparse
from urllib.parse import unquote
import logging

from selenium import webdriver
from pyvirtualdisplay import Display

from pdfminer.high_level import extract_text

# Config
__RAA_PAGE = 'https://www.prefecturedepolice.interieur.gouv.fr/actualites-et-presse/arretes/accueil-arretes'
__USER_AGENT = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'
__headless_mode = True
__WAITING_TIME = int(os.getenv('WAITING_TIME') or 15)
__LIST = os.getenv('LIST') or 'vidéoprotection,caméras,captation,aéronef'
__DATA_DIR = os.path.dirname(os.path.abspath(__file__))+'/data/ppparis/'

# Fonctions
def print_output(data):
  print(data)
  data = data.replace('\033[92m', '')
  data = data.replace('\033[0m', '')
  data = data.replace('\033[1m', '')
  f = open(os.path.dirname(os.path.abspath(__file__))+'/output.log','a')
  f.write(data+"\n")
  f.close()

def start_display():
  display = Display(visible=False, size=(1024, 768))
  display.start()

def stop_display():
  display.stop()

def get_html(url):
  browser.get(url)
  time.sleep(__WAITING_TIME)
  page_content = browser.page_source

  return page_content

def download_file(url, dest):
  browser.get(url)
  time.sleep(__WAITING_TIME)

def parse_pdf(filename, name, date):
  if not os.path.isfile(__DATA_DIR+filename):
    logging.warning(f'ATTENTION: le fichier {filename} n\'existe pas')
  else:
    text = extract_text(__DATA_DIR+filename)
    found = False
    for keyword in __LIST.split(','):
      if re.search(keyword, text, re.IGNORECASE|re.MULTILINE):
        if not found:
          print_output(f'\033[92m{name}\033[0m ({date})')
          found = True
        print_output(f'    Le terme \033[1m{keyword}\033[0m a été trouvé.')

    # Remplace le PDF par un fichier vide, afin de savoir la prochaine fois qu'il a déjà été analysé
    f = open(__DATA_DIR+filename,'w')
    f.write('')
    f.close()
    if found:
      print_output('')

# Début du script
parser = argparse.ArgumentParser(prog='ppparis.py', description='Télécharge les RAA de la Préfecture de police de Paris et recherche des mots-clés')
parser.add_argument('-n', '--noheadless', action='store_true', help='ne lance pas le navigateur en mode headless (pratique pour débugguer ou en dehors d\'une CI)')
parser.add_argument('-w', '--waiting-time', type=int, action='store', help='délai (en secondes) d\'attente de chargement d\'une page (par défaut : 15)')
parser.add_argument('-l', '--list', action='store', help='liste des termes recherchés, séparés par une virgule (par défaut : vidéoprotection,caméras,captation,aéronef)')
parser.add_argument('-v', action='store_true', help='relève le niveau de verbosité à INFO')
parser.add_argument('-vv', action='store_true', help='relève le niveau de verbosité à DEBUG')
args = parser.parse_args()

if args.v:
  logging.basicConfig(level=logging.INFO)

if args.vv:
  logging.basicConfig(level=logging.DEBUG)

if args.noheadless:
  __headless_mode = False

if not __headless_mode:
  logging.debug('Mode noheadless')

if args.waiting_time:
  __WAITING_TIME = args.waiting_time

logging.debug(f'WAITING_TIME: {__WAITING_TIME}')

if args.list:
  __LIST = args.list

logging.debug(f'Termes recherchés: {__LIST}')

# On crée le dossier de téléchargement
os.makedirs(__DATA_DIR, exist_ok=True)

# On démarre le navigateur
webdriver_options = webdriver.ChromeOptions()
webdriver_options.add_argument("--no-sandbox")
webdriver_options.add_argument("--disable-extensions")
webdriver_options.add_argument("--disable-gpu")
webdriver_options.add_argument("--disable-dev-shm-usage")
webdriver_options.add_argument("--use_subprocess")
webdriver_options.add_argument("--disable-blink-features=AutomationControlled")
webdriver_options.add_argument(f"--user-agent={__USER_AGENT}")
prefs = {
"download.default_directory":__DATA_DIR,
"download.prompt_for_download":False,
"plugins.always_open_pdf_externally":True
}
webdriver_options.add_experimental_option("prefs",prefs)
if __headless_mode:
  webdriver_options.add_argument("--headless")
  webdriver_options.add_argument("--window-size=1024,768")
  start_display()
else:
  webdriver_options.add_argument("--start-maximized")

browser = webdriver.Chrome(options=webdriver_options)

# Téléchargement des RAA
page_content = get_html(__RAA_PAGE)

# On charge le parser
soup = BeautifulSoup(page_content, 'html.parser')

# Pour chaque balise a, on regarde si c'est un PDF, et si oui on le parse
for a in soup.find_all('a', href=True):
  if a['href'].endswith('.pdf'):
    if a['href'].startswith('/'):
      url = 'https://www.prefecturedepolice.interieur.gouv.fr'+a['href']
    else:
      url = a['href']

    name = a.find('span').get_text()
    date = a.find('div', class_="field--type-datetime").get_text()

    filename = unquote(url.split('/')[-1])

    # Si le fichier n'a pas été téléchargé, on le télécharge et on le parse
    if not os.path.isfile(__DATA_DIR+filename):
      logging.info(f'Nouveau fichier : {name} ({date}). URL : {url}')
      try:
        download_file(url, __DATA_DIR+filename)
      except Exception:
        logging.warning('ATTENTION: Impossible de télécharger le fichier '+url)


      cmd = ['ocrmypdf', '-l', 'eng+fra', '--output-type', 'pdfa', '--redo-ocr', __DATA_DIR+filename, __DATA_DIR+filename]
      logging.debug(f'Lancement de ocrmypdf: {cmd}')
      try:
        output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
      except subprocess.CalledProcessError as exc:
        if not exc.returncode == 6:
          logging.warning('ATTENTION : Impossible d\'OCRiser le document', exc.returncode, exc.output)

      parse_pdf(filename, name, date)

# On arrête le navigateur
browser.quit()
if __headless_mode:
  stop_display()