Techniques d'Extraction de Données Web et de Stockage en Base de Données

Extraction des Prévisions Météorologiques

Exigences
Récupérer les prévisions météorologiques sur 7 jours pour une ville spécifique depuis un site météorologique chinois, puis stocker les données dans une base de données SQLite.

Méthodologie
Utilisation de la bibliothèque requests pour récupérer la page web, puis de BeautifulSoup pour analyser le HTML et extraire les données météorologiques des éléments li. Les données sont ensuite insérées dans une table SQLite avec sqlite3.

reponse = requests.get(url_meteo, headers=en_tetes) reponse.encoding = "utf-8"

analyseur_html = BeautifulSoup(reponse.text, "html.parser")

========== 2. Analyse des données (7 premiers jours) ==========

conteneur_jours = analyseur_html.find("div", id="7d").find("ul") donnees_collectees = []

for element_li in conteneur_jours.find_all("li")[:7]: date_prevue = element_li.find("h1").text.strip() description_meteo = element_li.find("p", class_="wea").text.strip() temperature_info = element_li.find("p", class_="tem").text.strip()

donnees_collectees.append((ville_cible, date_prevue, description_meteo, temperature_info))

</details><details><summary>Afficher le code</summary>```
# ========== 3. Stockage dans SQLite ==========
connexion_db = sqlite3.connect("meteo_stockee.db")
curseur_db = connexion_db.cursor()

curseur_db.execute("""
    CREATE TABLE IF NOT EXISTS previsions_meteo (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        ville TEXT,
        date TEXT,
        meteo_description TEXT,
        temperature TEXT
    )
""")

curseur_db.executemany(
    "INSERT INTO previsions_meteo (ville, date, meteo_description, temperature) VALUES (?, ?, ?, ?)",
    donnees_collectees
)

connexion_db.commit()

Extraction d'Informations Boursières via API

Exigences
Cibler et extraire des données boursières en utilisant des requêtes HTTP et le format JSON, puis les stocker dans une base de données.

Méthodologie
Identification de l'endpoint API pour les données boursières à l'aide des outils de développement du navigateur. Utilisation de requests avec des paramètres de pagination pour récupérer les données, suivie d'une analyse JSON et d'un stockage dans SQLite.

MAX_ARTICLES = 50 TAILLE_PAGE = 25 donnees_totales = [] page_actuelle = 1

print(f"Début de l'extraction, objectif maximal: {MAX_ARTICLES} articles")

while len(donnees_totales) < MAX_ARTICLES: parametres_requete = { "page": str(page_actuelle), "num": str(TAILLE_PAGE), "sort": "changepercent", "asc": "0", "node": "hs_a" }

reponse_api = requests.get(api_url, params=parametres_requete, headers=en_tetes_requete, timeout=10)
reponse_api.raise_for_status()
reponse_api.encoding = "utf-8"

# Extraction et analyse JSON
texte_brut = reponse_api.text.strip()
if not texte_brut:
    break

try:
    liste_actions = json.loads(texte_brut)
except json.JSONDecodeError:
    print("Erreur de décodage JSON, arrêt de la pagination.")
    break

if not liste_actions:
    break

articles_necessaires = MAX_ARTICLES - len(donnees_totales)
articles_a_ajouter = liste_actions[:articles_necessaires]
donnees_totales.extend(articles_a_ajouter)

print(f"Page {page_actuelle} récupérée ({len(articles_a_ajouter)} articles), total: {len(donnees_totales)}")

page_actuelle += 1

</details><details><summary>Afficher le code</summary>```
# Stockage dans SQLite
nom_fichier_db = "donnees_boursieres.db"
try:
    connexion = sqlite3.connect(nom_fichier_db)
    curseur = connexion.cursor()

    curseur.execute("""
        CREATE TABLE IF NOT EXISTS information_actions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            symbole TEXT,
            nom_entreprise TEXT,
            cours_actuel REAL,
            variation_pourcentage REAL,
            volume_echanges INTEGER,
            montant_total REAL,
            plus_haut REAL,
            plus_bas REAL
        )
    """)

    insertions = []
    for article in donnees_totales:
        insertions.append((
            article.get("symbol", ""),
            article.get("name", ""),
            article.get("trade", 0.0),
            article.get("changepercent", 0.0),
            article.get("volume", 0),
            article.get("amount", 0.0),
            article.get("high", 0.0),
            article.get("low", 0.0)
        ))

    curseur.executemany("""
        INSERT INTO information_actions (symbole, nom_entreprise, cours_actuel, variation_pourcentage,
                                         volume_echanges, montant_total, plus_haut, plus_bas)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """, insertions)

    connexion.commit()
    print(f"Données boursières enregistrées dans {nom_fichier_db}")
except sqlite3.Error as e:
    print(f"Erreur de base de données: {e}")
finally:
    if connexion:
        connexion.close()

Extraction du Classement des Universités

Exigences
Récupérer les informations de classement universitaire depuis un site de classement académique, gérer les données JSONP avec des codes abrégés, et stocker le tout dans une base de données.

Méthodologie
Accès au fichier JavaScript sous-jacent contenant les données, utilisation d'expressions régulières pour extraire les informations pertinentes, et application d'une table de correspondance pour convertir les codes en valeurs lisibles avant le stockage.

Extraction des données depuis le JS

def analyser_donnees_universitaires(chemin_js="payload_univ.js", carte_scores=None): with open(chemin_js, "r", encoding="utf-8") as f: contenu_js = f.read()

# Recherche du bloc de données
match_bloc = re.search(r'univData:\s*\[(.*?)\]', contenu_js, re.DOTALL)
if not match_bloc:
    return []

chaine_donnees = match_bloc.group(1)
resultats = []

# Extraction individuelle des universités
for correspondance in re.finditer(r'\{[^}]*\}', chaine_donnees):
    texte_objet = correspondance.group(0)

    nom_cn = re.search(r'univNameCn:"([^"]*)"', texte_objet)
    province_code = re.search(r'province:"([^"]*)"', texte_objet)
    categorie_code = re.search(r'univCategory:"([^"]*)"', texte_objet)
    score_brut = re.search(r'score:([^,}]+)', texte_objet)

    nom_final = nom_cn.group(1) if nom_cn else "Inconnu"
    province_finale = province_code.group(1) if province_code else ""
    categorie_finale = categorie_code.group(1) if categorie_code else ""
    score_val = score_brut.group(1).strip('"') if score_brut else ""

    # Conversion des codes si nécessaire
    if carte_scores and score_val in carte_scores:
        score_val = carte_scores[score_val]

    try:
        score_numerique = float(score_val)
    except ValueError:
        score_numerique = 0.0

    resultats.append({
        "nom": nom_final,
        "province": province_finale,
        "categorie": categorie_finale,
        "score": score_numerique
    })

return resultats

Stockage final

def sauvegarder_classement(donnees, fichier_db="classement_univ.db"): connexion = sqlite3.connect(fichier_db) curseur = connexion.cursor()

curseur.execute("""
    CREATE TABLE IF NOT EXISTS classement_universites (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        nom_universite TEXT,
        province TEXT,
        categorie TEXT,
        score_final REAL
    )
""")

curseur.execute("DELETE FROM classement_universites")

for entree in donnees:
    curseur.execute("""
        INSERT INTO classement_universites (nom_universite, province, categorie, score_final)
        VALUES (?, ?, ?, ?)
    """, (entree["nom"], entree["province"], entree["categorie"], entree["score"]))

connexion.commit()
connexion.close()

</details>**Résultats**  
Les classements universitaires sont extraits, transformés et stockés, malgré la complexité des données JSONP et des codes variables. L'utilisation d'un fichier de correspondance facilite la conversion des données abrégées.

Étiquettes: Python requests sqlite3 beautifulsoup4 JSON parsing web data extraction

Publié le 14 juin à 20h35