postmon/IbgeTracker.py at master · willianrocha/postmon · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import requests

from lxml.html import fromstring
from database import MongoDb as Database
from utils import slug


class IbgeTracker():

    def __init__(self):
        base_url = 'http://www.ibge.gov.br/home/geociencias/areaterritorial'
        self.url_ufs = base_url + '/principal.shtm'
        self.url_cidades = base_url + '/area.php?nome=%'

    def _request(self, url):
        response = requests.post(url)
        response.raise_for_status()
        return response.text

    def _get_info_ufs(self, siglas):
        texto = self._request(self.url_ufs)
        html = fromstring(texto)
        seletorcss_linhas = "div#miolo_interno > table > tr"
        linhas = html.cssselect(seletorcss_linhas)
        try:
            linhas.pop(0)  # a primeira é o cabeçalho
        except IndexError:
            pass
        infos = []
        for linha in linhas:
            seletorcss_celulas = "td"
            celulas = linha.cssselect(seletorcss_celulas)
            codigo_ibge = celulas[0].text_content()
            if codigo_ibge in siglas:
                sigla = siglas[codigo_ibge]
                infos.append({
                    'sigla': sigla,
                    'codigo_ibge': codigo_ibge,
                    'nome': celulas[1].text_content().strip(' (*)'),
                    'area_km2': celulas[2].text_content()
                })

        #  neste ponto, após a carga
        #  das cidades, a lista
        #  'infos' deve estar populada

        return infos

    def _get_info_cidades(self):
        texto = self._request(self.url_cidades)
        html = fromstring(texto)
        seletorcss_linhas = "div#miolo_interno > table > tr"
        linhas = html.cssselect(seletorcss_linhas)
        try:
            linhas.pop(0)  # a primeira é o cabeçalho
        except IndexError:
            pass
        infos = []
        for linha in linhas:
            seletorcss_celulas = "td"
            celulas = linha.cssselect(seletorcss_celulas)
            infos.append({
                'codigo_ibge_uf': celulas[0].text_content(),
                'sigla_uf': celulas[1].text_content(),
                'codigo_ibge': celulas[2].text_content(),
                'nome': celulas[3].text_content(),
                'area_km2': celulas[4].text_content()
            })
        return infos

    def _track_ufs(self, db, siglas):
        infos = self._get_info_ufs(siglas)
        for info in infos:
            db.insert_or_update_uf(info)

    def _track_cidades(self, db):
        infos = self._get_info_cidades()
        siglas = {}
        for info in infos:
            codigo_ibge_uf = info['codigo_ibge_uf']
            sigla_uf = info['sigla_uf']
            nome = info['nome']
            if codigo_ibge_uf not in siglas:
                siglas[codigo_ibge_uf] = sigla_uf

            # a chave única de uma cidade não
            # pode ser só o nome, pois
            # existem cidades com mesmo nome
            # em estados diferentes
            info['sigla_uf_nome_cidade'] = slug('%s_%s' % (sigla_uf, nome))

            db.insert_or_update_cidade(info)

        return siglas

    def track(self, db):
        """
        Atualiza as bases internas do mongo
        com os dados mais recentes do IBGE
        referente a ufs e cidades
        """
        siglas = self._track_cidades(db)
        # siglas é um dict cod_ibge -> sigla:
        # { '35': 'SP', '35': 'RJ', ... }
        self._track_ufs(db, siglas)


def _standalone():
    db = Database()
    ibge = IbgeTracker()
    ibge.track(db)


if __name__ == "__main__":
    _standalone()