Job Tracker — Import CSV et intégrations

Section 11.3.17 : Job Tracker — Import CSV et intégrations

🎯 Objectif pédagogique

Ajouter l'import de données dans le Job Tracker : fichiers CSV, export LinkedIn, et intégration API. Vous serez capable de nettoyer et importer des données depuis différentes sources.

Importer ses données existantes

Marc a un fichier CSV avec 50 candidatures passées. Il doit pouvoir les importer dans le Job Tracker sans les re-saisir manuellement. Il veut aussi pouvoir importer ses contacts LinkedIn.

Import CSV avec nettoyage

# importer.py — Import de données depuis différentes sources
import csv
import io
import pandas as pd
import streamlit as st
from database import add_application, get_connection
from datetime import datetime

def parse_csv(file_content, delimiter=','):
    """Parse un fichier CSV et retourne un DataFrame nettoyé."""
    df = pd.read_csv(io.StringIO(file_content), delimiter=delimiter)
    
    # Normaliser les noms de colonnes
    df.columns = [col.strip().lower().replace(' ', '_') for col in df.columns]
    
    return df

def map_columns(df, mapping):
    """Mapper les colonnes du CSV vers les champs du Job Tracker."""
    mapped_data = []
    
    for _, row in df.iterrows():
        entry = \{\}
        for target_field, source_col in mapping.items():
            if source_col and source_col in row.index:
                entry[target_field] = str(row[source_col]).strip()
            else:
                entry[target_field] = None
        
        # Valider les champs obligatoires
        if entry.get('company') and entry.get('position'):
            mapped_data.append(entry)
    
    return mapped_data

def import_to_db(entries):
    """Importer les entrées nettoyées dans la base de données."""
    imported = 0
    skipped = 0
    
    for entry in entries:
        company = entry.pop('company')
        position = entry.pop('position')
        
        # Filtrer les valeurs None ou 'nan'
        clean_kwargs = \{
            k: v for k, v in entry.items() 
            if v and str(v).lower() not in ('none', 'nan', '')
        \}
        
        try:
            add_application(company, position, **clean_kwargs)
            imported += 1
        except Exception as e:
            print(f"Skipped \{company\}/\{position\}: \{e\}")
            skipped += 1
    
    return imported, skipped

def render_import_page():
    """Page d'import dans Streamlit."""
    st.title("⚙️ Import de données")
    
    tab1, tab2, tab3 = st.tabs(["📄 CSV", "💼 LinkedIn", "🔗 API"])
    
    with tab1:
        st.subheader("Import depuis un fichier CSV")
        
        uploaded_file = st.file_uploader(
            "Choisissez votre fichier CSV",
            type=['csv'],
            help="Format attendu : colonnes avec entreprise, poste, statut, date..."
        )
        
        if uploaded_file:
            content = uploaded_file.read().decode('utf-8')
            df = parse_csv(content)
            
            st.write(f"**\{len(df)\} lignes trouvées.** Colonnes détectées :")
            st.write(list(df.columns))
            
            # Prévisualisation
            st.dataframe(df.head(5))
            
            # Mapping des colonnes
            st.subheader("Mapper les colonnes")
            tracker_fields = ['company', 'position', 'status', 'salary_min', 
                            'salary_max', 'location', 'url', 'contact_email', 'notes']
            
            mapping = \{\}
            cols = st.columns(3)
            for i, field in enumerate(tracker_fields):
                with cols[i % 3]:
                    options = ['(ignorer)'] + list(df.columns)
                    selection = st.selectbox(
                        f"→ \{field\}",
                        options,
                        key=f"map_\{field\}"
                    )
                    if selection != '(ignorer)':
                        mapping[field] = selection
            
            if st.button("🚀 Importer"):
                if 'company' not in mapping or 'position' not in mapping:
                    st.error("Les champs 'company' et 'position' sont obligatoires.")
                else:
                    entries = map_columns(df, mapping)
                    imported, skipped = import_to_db(entries)
                    st.success(f"✅ \{imported\} candidatures importées, \{skipped\} ignorées.")
    
    with tab2:
        st.subheader("Import depuis LinkedIn")
        st.markdown("""
        **Comment exporter depuis LinkedIn :**
        1. Allez dans **Paramètres → Données & confidentialité**
        2. Cliquez sur **Obtenir une copie de vos données**
        3. Sélectionnez **Connexions** et/ou **Candidatures**
        4. Vous recevrez un ZIP par email
        5. Uploadez le CSV ici
        """)
        
        linkedin_file = st.file_uploader(
            "Fichier LinkedIn Connections.csv",
            type=['csv'],
            key="linkedin"
        )
        
        if linkedin_file:
            content = linkedin_file.read().decode('utf-8')
            df = parse_csv(content)
            st.dataframe(df.head(5))
            st.info("Mappez les colonnes dans l'onglet CSV ci-dessus.")
    
    with tab3:
        st.subheader("Import via API")
        st.markdown("""
        Vous pouvez aussi importer des données via l'API du Job Tracker.
        """)
        
        st.code("""
# Exemple : import via script Python
import requests

jobs = [
    \{"company": "Google", "position": "Data Analyst", "status": "applied"\},
    \{"company": "Meta", "position": "ML Engineer", "status": "interview"\},
]

for job in jobs:
    response = requests.post(
        "http://localhost:8501/api/import",
        json=job
    )
    print(response.json())
        """, language="python")

Gestion des doublons

# Ajouter dans database.py
def check_duplicate(company, position):
    """Vérifier si une candidature similaire existe déjà."""
    conn = get_connection()
    result = conn.execute(
        """SELECT id FROM applications 
           WHERE LOWER(company) = LOWER(?) AND LOWER(position) = LOWER(?)
           AND archived = 0""",
        (company, position)
    ).fetchone()
    conn.close()
    return result is not None

🏋️ Exercice pratique (20 minutes)

1. →Créez un fichier CSV test avec 10 candidatures (Company, Position, Status, Date)
2. →Ajoutez la page Import dans votre Job Tracker
3. →Importez le CSV et vérifiez dans le Kanban
4. →Bonus : ajoutez la détection de doublons avant import