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.