🎯 Mini-projet — Workflow automatisé complet

Section 11.3.20 : 🎯 Mini-projet — Workflow automatisé complet

🎯 Objectif pédagogique

Combiner toutes les compétences de la Semaine 3 dans un workflow automatisé de bout en bout. Vous construirez un pipeline complet : collecte de données, traitement IA, stockage, notifications et dashboard.

Le projet : Job Search Automation Pipeline

Marc va assembler un pipeline complet qui :

1. →Collecte des offres d'emploi automatiquement
2. →Analyse chaque offre avec l'IA (classification, scoring)
3. →Stocke dans le Job Tracker (SQLite)
4. →Notifie pour les offres prioritaires
5. →Visualise les résultats dans le dashboard

Architecture du pipeline

Implémentation complète

# pipeline.py — Job Search Automation Pipeline
import json
import time
from datetime import datetime
from database import add_application, get_connection, init_db
from notifications import NotificationManager, send_daily_digest
from logger import setup_logger, alert_on_failure
import requests
import os

logger = setup_logger('pipeline', 'logs/pipeline.log')

# === ÉTAPE 1 : Collecte ===
@alert_on_failure
def collect_jobs(keywords, location="Paris"):
    """Collecter les offres depuis l'API France Travail."""
    logger.info(f"Collecting jobs: \{keywords\} in \{location\}")
    
    # API France Travail (ex Pôle Emploi)
    base_url = "https://api.francetravail.io/partenaire/offresdemploi/v2/offres/search"
    headers = \{
        "Authorization": f"Bearer \{os.environ.get('FT_API_TOKEN', 'demo')\}",
        "Accept": "application/json"
    \}
    
    all_jobs = []
    for keyword in keywords:
        params = \{
            "motsCles": keyword,
            "commune": "75056",  # Paris
            "distance": 30,
            "range": "0-49"
        \}
        
        try:
            response = requests.get(base_url, headers=headers, params=params, timeout=15)
            if response.status_code == 200:
                data = response.json()
                jobs = data.get('resultats', [])
                all_jobs.extend(jobs)
                logger.info(f"Keyword '\{keyword\}': \{len(jobs)\} jobs found")
            else:
                logger.warning(f"API returned \{response.status_code\} for '\{keyword\}'")
        except requests.RequestException as e:
            logger.error(f"Request failed for '\{keyword\}': \{e\}")
        
        time.sleep(1)  # Rate limiting
    
    # Dédupliquer par ID
    seen = set()
    unique_jobs = []
    for job in all_jobs:
        job_id = job.get('id')
        if job_id and job_id not in seen:
            seen.add(job_id)
            unique_jobs.append(job)
    
    logger.info(f"Total unique jobs: \{len(unique_jobs)\}")
    return unique_jobs

# === ÉTAPE 2 : Analyse IA ===
@alert_on_failure
def classify_job(job):
    """Classifier une offre avec l'IA (simulé ou réel)."""
    title = job.get('intitule', '')
    company = job.get('entreprise', \{\}).get('nom', 'Unknown')
    description = job.get('description', '')[:500]
    salary = job.get('salaire', \{\}).get('libelle', 'Non précisé')
    
    # Scoring basé sur des mots-clés (version sans API OpenAI)
    score = 0
    keywords_scores = \{
        'python': 3, 'data': 3, 'analyst': 2, 'ia': 3,
        'machine learning': 3, 'sql': 2, 'tableau': 1,
        'junior': -1, 'senior': 1, 'remote': 2,
        'télétravail': 2, 'cdi': 1, 'stage': -2
    \}
    
    text = (title + ' ' + description).lower()
    for keyword, value in keywords_scores.items():
        if keyword in text:
            score += value
    
    # Normaliser entre 1 et 5
    score = max(1, min(5, score))
    
    return \{
        'title': title,
        'company': company,
        'description': description[:200],
        'salary': salary,
        'score': score,
        'category': 'MATCH_PARFAIT' if score >= 4 else 'MATCH_PARTIEL' if score >= 2 else 'PAS_PERTINENT',
        'url': job.get('origineOffre', \{\}).get('urlOrigine', ''),
        'location': job.get('lieuTravail', \{\}).get('libelle', '')
    \}

# === ÉTAPE 3 : Stockage ===
@alert_on_failure
def store_classified_jobs(classified_jobs):
    """Stocker les offres classifiées dans le Job Tracker."""
    stored = 0
    
    for job in classified_jobs:
        if job['category'] == 'PAS_PERTINENT':
            continue
        
        try:
            status = 'wishlist' if job['score'] >= 4 else 'applied'
            add_application(
                company=job['company'],
                position=job['title'],
                status=status,
                location=job['location'],
                url=job['url'],
                notes=f"Score IA: \{job['score']\}/5 | \{job['category']\}\n\{job['description']\}"
            )
            stored += 1
        except Exception as e:
            logger.warning(f"Could not store \{job['title']\}: \{e\}")
    
    logger.info(f"Stored \{stored\} jobs in tracker")
    return stored

# === ÉTAPE 4 : Notifications ===
@alert_on_failure
def notify_high_priority(classified_jobs):
    """Notifier pour les offres prioritaires."""
    high_priority = [j for j in classified_jobs if j['score'] >= 4]
    
    if not high_priority:
        logger.info("No high-priority jobs found")
        return
    
    message = f"🌟 \{len(high_priority)\} offres prioritaires trouvées :\n\n"
    for job in high_priority:
        message += f"• \{job['title']\} — \{job['company']\} (\{job['location']\})\n"
        message += f"  Score: \{'⭐' * job['score']\} | Salaire: \{job['salary']\}\n\n"
    
    # Webhook (Make/Slack)
    webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    if webhook_url:
        requests.post(webhook_url, json=\{"text": message\}, timeout=10)
    
    logger.info(f"Notified for \{len(high_priority)\} high-priority jobs")
    print(message)

# === PIPELINE PRINCIPAL ===
@alert_on_failure
def run_pipeline():
    """Exécuter le pipeline complet."""
    start_time = time.time()
    logger.info("=" * 50)
    logger.info("PIPELINE START")
    
    # 1. Collecter
    keywords = ["data analyst", "python developer", "machine learning", "business intelligence"]
    raw_jobs = collect_jobs(keywords)
    
    # 2. Classifier
    classified = [classify_job(job) for job in raw_jobs]
    
    # 3. Stats
    categories = \{\}
    for job in classified:
        cat = job['category']
        categories[cat] = categories.get(cat, 0) + 1
    
    logger.info(f"Classification: \{categories\}")
    
    # 4. Stocker
    stored = store_classified_jobs(classified)
    
    # 5. Notifier
    notify_high_priority(classified)
    
    # 6. Rapport
    duration = time.time() - start_time
    logger.info(f"PIPELINE COMPLETE in \{duration:.1f\}s")
    logger.info(f"Collected: \{len(raw_jobs)\} | Stored: \{stored\} | Categories: \{categories\}")
    logger.info("=" * 50)
    
    return \{
        "collected": len(raw_jobs),
        "stored": stored,
        "categories": categories,
        "duration": round(duration, 1)
    \}

if __name__ == "__main__":
    init_db()
    result = run_pipeline()
    print(f"\n✅ Pipeline terminé : \{json.dumps(result, indent=2)\}")

Automatiser le lancement

# Lancer manuellement
python pipeline.py

# Lancer 2x/jour avec cron (Linux/Mac)
# crontab -e
0 8,18 * * * cd /path/to/job-tracker && python pipeline.py >> logs/cron.log 2>&1

# Windows Task Scheduler
# schtasks /create /sc daily /tn "JobPipeline" /tr "python pipeline.py" /st 08:00

# Ou via Make : Schedule (2x/jour) → Webhook → pipeline.py (Cloud Function)

Ce que Marc a appris en Semaine 3

📊 Bilan Semaine 3 — Automatisation & Outils IA :

✅ APIs REST — consommer, authentifier, webhooks
✅ Google Apps Script — automatiser Workspace
✅ Make — workflows visuels, routeurs, itérateurs, LLM
✅ Zapier — automatisations rapides, comparaison
✅ n8n — alternative open-source, self-hosted
✅ AI Coding Assistants — Copilot, Cursor
✅ Job Tracker — Kanban, notifications, dashboard, import
✅ Monitoring — logging, alertes, debugging
✅ Scaling — rate limiting, retry, architecture modulaire
✅ Pipeline complet — collecte → IA → stockage → notification

🎯 Compétences acquises :
- Conception de workflows automatisés multi-étapes
- Intégration d'APIs LLM dans des chaînes d'automatisation
- Construction d'outils personnels (Job Tracker)
- Monitoring et maintenance de systèmes automatisés

🏋️ Exercice pratique (45 minutes)

1. →Assemblez le pipeline complet (pipeline.py)
2. →Exécutez python pipeline.py et vérifiez les logs
3. →Vérifiez que les offres apparaissent dans le Kanban et le Dashboard
4. →Configurez le lancement automatique (cron ou Task Scheduler)
5. →Bonus : ajoutez l'analyse IA réelle avec l'API OpenAI