Introduction aux Requêtes Asynchrones

L'utilisation de coroutines avec asyncio et aiohttp permet d'exécuter des opérations d'entrée/sortie de manière non bloquante et concurrente.

import asyncio
import aiohttp

liste_urls = [
    "https://example.com/image1.jpg",
    "https://example.com/image2.jpg",
    "https://example.com/image3.jpg"
]

async def telecharger_fichier(url):
    nom_fichier = url.split('/')[-1]
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as reponse:
            donnees = await reponse.read()
            with open(nom_fichier, 'wb') as fichier:
                fichier.write(donnees)

async def telechargements_concurrents():
    taches = []
    for url in liste_urls:
        taches.append(asyncio.create_task(telecharger_fichier(url)))
    await asyncio.gather(*taches)

if __name__ == '__main__':
    asyncio.run(telechargements_concurrents())

Scraping de Contenu Littéraire

L'apprcohe suivante illustre l'extraction structurée de chapitres de roman à partir d'une API.

import asyncio
import aiohttp
import aiofiles
import json

entetes = {"User-Agent": "Mozilla/5.0"}

async def extraire_chapitre(id_chapitre, id_livre, titre):
    donnees_requete = {
        "book_id": id_livre,
        "cid": f"{id_livre}|{id_chapitre}",
        "need_bookinfo": 1
    }
    url_api = f"http://api.exemple.com/getContent?data={json.dumps(donnees_requete)}"
    
    async with aiohttp.ClientSession() as session:
        async with session.get(url_api, headers=entetes) as reponse:
            contenu_json = await reponse.json()
            async with aiofiles.open(f"{titre}.txt", 'w', encoding='utf-8') as fichier:
                await fichier.write(contenu_json['data']['text'])

async def obtenir_sommaire(url_sommaire, identifiant_livre):
    import requests
    reponse = requests.get(url_sommaire, headers=entetes)
    catalogue = reponse.json()
    
    taches = []
    for element in catalogue['data']['chapters']:
        taches.append(asyncio.create_task(
            extraire_chapitre(element['cid'], identifiant_livre, element['title'])
        ))
    await asyncio.wait(taches)
    reponse.close()

if __name__ == '__main__':
    id_livre = "1234567890"
    url_catalogue = f"http://api.exemple.com/getCatalogue?data={json.dumps({'book_id': id_livre})}"
    asyncio.run(obtenir_sommaire(url_catalogue, id_livre))

Téléchargement de Séquences Vidéo

Cette méthode démontre le traitement de fichiers playlist M3U8 pour télécharger des segments vidéo.

import asyncio
import aiohttp
import aiofiles
import requests

entetes = {"User-Agent": "Mozilla/5.0"}

async def recuperer_segment(url_segment, nom_fichier, session):
    async with session.get(url_segment) as reponse:
        contenu = await reponse.read()
        async with aiofiles.open(f"video/{nom_fichier}", 'wb') as fichier:
            await fichier.write(contenu)
    print(f"Segment {nom_fichier} acquis")

async def traitement_m3u8():
    taches = []
    async with aiohttp.ClientSession() as session:
        async with aiofiles.open("playlist.m3u8", 'r', encoding='utf-8') as playlist:
            async for ligne in playlist:
                ligne = ligne.strip()
                if ligne.startswith('#'):
                    continue
                nom_segment = ligne.rsplit('/', 1)[-1]
                taches.append(asyncio.create_task(
                    recuperer_segment(ligne, nom_segment, session)
                ))
        await asyncio.gather(*taches)

def initialiser_telechargement():
    url_playlist = "https://cdn.exemple.com/video/playlist.m3u8"
    reponse = requests.get(url_playlist, headers=entetes)
    with open("playlist.m3u8", 'wb') as fichier:
        fichier.write(reponse.content)
    asyncio.run(traitement_m3u8())

if __name__ == '__main__':
    initialiser_telechargement()