Ce runbook couvre la couche infra et intégration : déployer, connecter, extraire, synchroniser. La sécurisation avancée et les extensions LLM feront l'objet d'un second volet. Deux entités fictives hébergées : Association des Petits Bouchons et Les Américains à la Lune.
Étape 1 / 6
Infrastructure VMware — 2 VMs Ubuntu
Disposer de deux VMs opérationnelles, réseau interne isolé, et passthrough USB scanner vers VM1. Point de départ de tout le reste.
Ubuntu 26.04 LTS confirmé sur VM1 (capture disponible).
Réseau VMware — configuration
# Sur chaque VM : vérifier l'interface réseau interne
ip addr show
# VM1 doit avoir une IP fixe ex: 192.168.100.10
# VM2 doit avoir une IP fixe ex: 192.168.100.20
# Test de connectivité inter-VMs
ping 192.168.100.20 # depuis VM1 vers VM2
Dans VMware Workstation : Edit → Virtual Network Editor → Ajouter VMnet en mode Host-only. Assigner ce VMnet aux deux VMs. Ajouter un second adaptateur NAT pour l'accès internet.
# VMware Workstation : VM → Removable Devices → [Scanner] → Connect
# Vérification dans VM1 :
lsusb
# Le scanner doit apparaître dans la liste
# Test SANE
sudo apt install -y sane-utils
scanimage -L
# Doit lister le périphérique USB connecté
Le passthrough USB nécessite que VMware Tools soit installé sur VM1. Sans lui, la détection USB est instable.
Déployer Nextcloud Hub sur VM1 via Docker, activer le module Recognize pour l'OCR automatique des documents scannés, créer les dossiers des deux entités fictives.
Interface Nextcloud accessible sur http://192.168.100.10:8080
Création des dossiers — deux entités fictives
# Via WebDAV — créer les dossiers Scans pour chaque entité
curl -X MKCOL \
-u "admin:${NC_ADMIN_PASS}" \
"http://localhost:8080/remote.php/dav/files/admin/Scans_PetitsBouchons"
curl -X MKCOL \
-u "admin:${NC_ADMIN_PASS}" \
"http://localhost:8080/remote.php/dav/files/admin/Scans_AmericainsLune"
Token d'application Nextcloud
# Interface Nextcloud → Paramètres → Sécurité → Créer un nouveau token
# Copier le token généré → stocker dans .env
# NC_APP_TOKEN=xxxx-xxxx-xxxx-xxxx-xxxx
# Ne jamais utiliser le mot de passe admin dans les scripts Python
Le module Recognize (OCR) s'installe depuis Apps → Recognize dans l'interface Nextcloud. Il traite automatiquement les PDFs déposés dans les dossiers surveillés.
Étape 3 / 6
Flux scan → GED automatisé
Automatiser le cycle complet : déclenchement du scanner via SANE, génération du PDF, dépôt automatique dans Nextcloud via WebDAV, avec pod Kubernetes dédié.
Args corrigés : une seule chaîne avec | pour exécuter les commandes en séquence.
Script de scan et dépôt WebDAV
scan_and_upload.sh
#!/bin/bash
# Usage : ./scan_and_upload.sh petits_bouchons
ENTITY=${1:-"petits_bouchons"}
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
OUTPUT="/tmp/scan_${ENTITY}_${TIMESTAMP}.pdf"
# 1. Scan via SANE
scanimage --format=pdf --resolution=300 > "$OUTPUT"
echo "[OK] Scan produit : $OUTPUT"
# 2. Upload vers Nextcloud via WebDAV
NC_FOLDER="Scans_${ENTITY}"
curl -T "$OUTPUT" \
-u "${NC_USER}:${NC_APP_TOKEN}" \
"http://${NC_HOST}:8080/remote.php/dav/files/${NC_USER}/${NC_FOLDER}/$(basename $OUTPUT)"
echo "[OK] Document déposé dans Nextcloud : ${NC_FOLDER}"
# 3. Nettoyage local
rm "$OUTPUT"
Déclenchement depuis Kubernetes
# Port-forward vers le pod scanner
kubectl port-forward pod/scan-usb 8081:8080
# Déclencher un scan via l'API REST
curl -X POST http://localhost:8081/scan \
-H "Content-Type: application/json" \
-d '{"entity":"petits_bouchons","device":"usb:001:003"}'
# Vérifier les logs du pod
kubectl logs scan-usb -f
Le port-forward est temporaire — à chaque redémarrage de Minikube il faut le relancer. Pour une démo continue, utiliser un Service NodePort.
Étape 4 / 6
Extraction Python — désignation / nom / ville / date
Script Python déployé sur VM2 qui interroge Nextcloud via WebDAV, télécharge les nouveaux PDFs, extrait les champs structurés par regex, et produit CSV + JSON.
Connexion Nextcloud via WebDAV
config_nextcloud.py
import os
from webdav3.client import Client
options = {
'webdav_hostname': f"http://{os.environ['NC_HOST']}:8080/remote.php/dav",
'webdav_login': os.environ['NC_USER'],
'webdav_password': os.environ['NC_APP_TOKEN']
}
client = Client(options)
Toutes les variables sensibles passent par des variables d'environnement. Jamais d'IP ou de token en dur dans le code.
Téléchargement des nouveaux documents
telecharger_lister.py
import time, os
from config_nextcloud import client
def get_new_documents(folder="/Scans_petits_bouchons", last_run_file="last_ts.txt"):
try:
with open(last_run_file, 'r') as f:
last_ts = float(f.read())
except:
last_ts = 0
items = client.list(folder)
new_files = []
for item in items:
if not item.endswith('.pdf'):
continue
props = client.info(folder + '/' + item)
modified = float(props.get('modified', 0))
if modified > last_ts:
local_path = f"./downloads/{item}"
os.makedirs("./downloads", exist_ok=True)
client.download_file(folder + '/' + item, local_path)
new_files.append(local_path)
with open(last_run_file, 'w') as f:
f.write(str(time.time()))
return new_files
Extraction des champs structurés
champs_cibles.py
import re
def extract_fields(text):
designation = re.search(r"D[ée]signation\s*:\s*(.+)", text, re.IGNORECASE)
designation = designation.group(1).strip() if designation else ""
nom = re.search(r"Nom\s*:\s*(.+)", text, re.IGNORECASE)
nom = nom.group(1).strip() if nom else ""
ville = re.search(r"Ville\s*:\s*(.+)", text, re.IGNORECASE)
ville = ville.group(1).strip() if ville else ""
# Format JJ/MM/AAAA ou AAAA-MM-JJ
date = re.search(r"\b(\d{2}/\d{2}/\d{4})\b", text)
if not date:
date = re.search(r"\b(\d{4}-\d{2}-\d{2})\b", text)
date = date.group(1) if date else ""
return {"designation": designation, "nom": nom, "ville": ville, "date": date}
Assemblage — main.py
main.py
import pdfplumber, pandas as pd, json
from telecharger_lister import get_new_documents
from champs_cibles import extract_fields
def extract_text_from_pdf(pdf_path):
with pdfplumber.open(pdf_path) as pdf:
return "".join(page.extract_text() or "" for page in pdf.pages)
def run_extraction(folder):
new_files = get_new_documents(folder=folder)
if not new_files:
print(f"[INFO] Aucun nouveau document dans {folder}")
return []
results = []
for pdf_file in new_files:
text = extract_text_from_pdf(pdf_file)
data = extract_fields(text)
data["source_file"] = pdf_file
data["entite"] = folder.replace("/Scans_", "")
results.append(data)
df = pd.DataFrame(results)
df.to_csv("extractions.csv", mode='a', header=not pd.io.common.file_exists("extractions.csv"), index=False)
with open("extractions.json", "w", encoding="utf-8") as f:
json.dump(results, f, indent=2, ensure_ascii=False)
print(f"[OK] {len(results)} document(s) extrait(s)")
return results
if __name__ == "__main__":
run_extraction("/Scans_petits_bouchons")
run_extraction("/Scans_AmericainsLune")
Étape 5 / 6
API FastAPI exposée via Kubernetes
Exposer les résultats d'extraction via une API REST FastAPI déployée dans un pod Kubernetes. Le pod API REST interroge main.py et sert les données structurées.
Les secrets Kubernetes sont encodés base64, non chiffrés. En production : Vault ou Sealed Secrets. Pour cette démo : acceptable.
Étape 6 / 6
Intégration M365 (Graph API) + Google Drive
Les documents extraits sont synchronisés vers Google Drive et une notification e-mail est envoyée via Microsoft Graph API. Simulation avec deux tenants sandbox.
Google Drive — dépôt automatique
google_drive_upload.py
import os, json
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
SCOPES = ['https://www.googleapis.com/auth/drive.file']
SERVICE_ACCOUNT_FILE = os.environ['GOOGLE_SA_JSON']
def upload_to_drive(local_file, folder_id, filename):
creds = service_account.Credentials.from_service_account_file(
SERVICE_ACCOUNT_FILE, scopes=SCOPES)
service = build('drive', 'v3', credentials=creds)
file_metadata = {'name': filename, 'parents': [folder_id]}
media = MediaFileUpload(local_file, mimetype='application/json')
file = service.files().create(
body=file_metadata, media_body=media, fields='id').execute()
print(f"[OK] Fichier déposé dans Drive : {file.get('id')}")
return file.get('id')
# Appel après extraction
upload_to_drive("extractions.json", os.environ['GDRIVE_FOLDER_ID'], "extractions_ged.json")
from main import run_extraction
from google_drive_upload import upload_to_drive
from graph_notify import send_notification
import os
ENTITES = [
("/Scans_petits_bouchons", "Association des Petits Bouchons"),
("/Scans_AmericainsLune", "Les Américains à la Lune")
]
for folder, label in ENTITES:
results = run_extraction(folder)
if results:
upload_to_drive("extractions.json", os.environ['GDRIVE_FOLDER_ID'],
f"extractions_{label.replace(' ','_')}.json")
send_notification(len(results), label)
Pipeline complet : Nextcloud → extraction Python → Google Drive → notification M365.