Python Parsing JSON Responses : Techniques de traitement des données du dictionnaire

Python s'amuse avec les données IP de proxy : désassemblage pratique d'ensembles JSON

Les crawlers savent que traiter des données JSON renvoyées par des proxy IP est comme ouvrir une boîte aveugle - vous ne savez jamais quel genre de format bizarre le serveur va vous bourrer. Aujourd'hui, nous allons prendre la réponse de l'API d'ipipgo comme étude de cas, et vous enseigner quelques principes de base.Il est bon sur le terrain.Le dictionnaire traite des voies sauvages.

import requests
from json import JSONDecodeError

def grab_proxies() :
    try : resp = requests.get('', timeout=5)
        resp = requests.get('https://api.ipipgo.com/proxy', timeout=5)
        data = resp.json().get('data', {})
        return data['ips'] if 'ips' in data else []
    except JSONDecodeError : print("Je ne suis pas sûr que ce soit le cas.
        print("Le serveur a renvoyé des données erronées !")
        return []

Vous voyez ? Deux points essentiels se cachent dans cette opération de base :capture des exceptionsrépondre en chantantValeur par défaut Réglage. De nombreux débutants prennent data['ips'] directement et sans réfléchir, pour ensuite voir le programme s'évanouir sur place lorsqu'ils rencontrent une pompe côté serveur renvoyant des données vides.

Les mille couches de l'emboîtement des dictionnaires

Les données de proxy IP d'ipipgo sont souvent accompagnées de plusieurs couches d'imbrication, comme celle-ci :

{
  "node" : {
    "east-china" : [
      {"ip" : "1.1.1.1", "expire" : "2024-08-01"}, {"ip" : "2.2.2.2", "expire" : "2024-08-02"}
      {"ip" : "2.2.2.2", "expire" : "2024-08-02"}
    ]
  }
}

À ce stade, ne vous précipitez pas pour utiliser des boucles pour durcir votre jeu, essayez cette opération tarte :

def extract_ips(raw_data) : [ return [
    return [
        item['ip']
        for region in raw_data.get('node', {}).values()
        pour item dans région
        if isinstance(region, list)
    ]

dépense ou fraisDérivés du dictionnaire + jugement de typeLa double assurance, peu importe comment les données changent, peut être aussi stable qu'un vieux chien. En particulier, l'ipipgo insère parfois des informations de débogage dans le nœud, sans filtrer les minutes d'isinstance pour signaler les erreurs.

Conseils pour l'entretien de la piscine Dynamic Proxy

N'utilisez pas la liste IP directement après l'avoir obtenue.Test de survie. De nombreux frères ont indiqué que le proxy IP avec l'utilisation de l'utilisation de l'échec, en fait, à cause de l'échec de faire un bon travail de prétraitement :

def check_alive(ip_list).
    working_ips = []
    pour ip dans ip_list.
        try.
            test_resp = requests.get('http://httpbin.org/ip',
                                   proxies={'http' : f'http://{ip}'},
                                   timeout=3)
            if ip in test_resp.text.
                working_ips.append(ip)
        except.
            continue
    return working_ips

Voici un conseil : utilisez l'interface httpbin.org/ip pour vérifier que le retour contient l'IP actuellement utilisée, ce qui est beaucoup plus fiable que de simplement regarder le code d'état de la réponse. En particulier avec les proxys éphémères d'ipipgo, cette étape de test ne devrait jamais être omise.

L'heure de l'assurance qualité : déminage des pièges les plus courants

Q : Que dois-je faire si je rencontre toujours des erreurs d'analyse JSON ?
R : 80% du contenu de la réponse est contaminé. Utilisez d'abord resp.text pour imprimer les données brutes et voir si elles sont entrecoupées de pages d'erreur HTML. Dans ce cas, il est recommandé de contacter l'assistance technique d'ipipgo, dont la stabilité de l'API est considérée comme la meilleure du secteur.

Q : L'adresse IP que j'ai obtenue interrompt toujours la connexion ?
R : Vérifiez trois points : 1. s'il faut passer par la vérification du proxy 2. le site cible n'a pas bloqué le proxy 3. le réseau local n'a pas de restrictions. Nous recommandons d'utiliser le forfait de facturation au volume d'ipipgo, leur pool d'IP est mis à jour fréquemment, et le taux de survie est supérieur au forfait mensuel de plus de 30%.

Q : Comment gérez-vous les demandes simultanées des agents ?
R : N'utilisez pas directement l'aversion pour le multithread ! Il est recommandé d'utiliser la stratégie de pooling de connexions + IP polling. Le package d'entreprise d'ipipgo prend en charge les appels d'API à haute concurrence, avec la bibliothèque aiohttp pour le traitement asynchrone, il n'y a pas de problème pour gérer des centaines de requêtes par seconde.

Conseils pratiques : planification intelligente de l'IP

Enfin, j'aimerais vous faire part d'un jeu de haut niveau : le changement dynamique d'agents en fonction de scénarios d'entreprise :

de random import choice

classe ProxyManager.
    def __init__(self).
        self.ips = []
        self.last_update = 0

    def refresh(self) : if time.time() - self.last_update = 0 : self.last_update = 0
        if time.time() - self.last_update > 300 : mise à jour toutes les 5 minutes
            self.ips = grab_proxies()
            self.last_update = time.time()

    def get_ip(self).
        self.refresh()
        return choice(self.ips) if self.ips else None

Ce planificateur met en œuvre leMise à jour automatique + sélection aléatoireLa double garantie. Le proxy tunnel dynamique d'ipipgo permet d'éviter efficacement que l'IP soit bloquée par le site web cible. La technologie de routage intelligente d'ipipgo permet d'attribuer automatiquement la ligne optimale en fonction du type d'entreprise, ce qui est beaucoup plus simple que la commutation manuelle.

En fin de compte, le traitement des données IP des serveurs mandataires est un travail méticuleux. Utilisez ces conseils, et avec un fournisseur de services fiable comme ipipgo, vous avez la garantie de faire décoller directement l'efficacité de votre crawler. Si vous ne comprenez pas, n'hésitez pas à laisser un message, nous allons travailler ensemble !