Merge pull request #69 from samuelgrigolato/master

Serviço de entrega de dados IBGE

Author: aleborba

.gitignore

@@ -3,3 +3,4 @@
 *.swp
 run.py
 build
+celerybeat-schedule

IbgeTracker.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import requests
+from lxml.html import fromstring
+from database import MongoDb as Database
+
+
+class IbgeTracker():
+
+    def __init__(self):
+        self.url_ufs = 'http://www.ibge.gov.br/home/geociencias' + \
+                       '/areaterritorial/principal.shtm'
+        self.url_cidades = 'http://www.ibge.gov.br/home/geociencias' + \
+                           '/areaterritorial/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)
+        linhas.pop()  # a primeira é o cabeçalho
+        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(),
+                    '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)
+        linhas.pop()  # a primeira é o cabeçalho
+        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'] = '%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()

Makefile

@@ -5,4 +5,4 @@ test: pep8
 
 .PHONY: pep8
 pep8:
-	@flake8 * --ignore=F403,F401 --exclude=requirements.txt,*.pyc,*.md,COPYING,Makefile,*.wsgi
+	@flake8 * --ignore=F403,F401 --exclude=requirements.txt,*.pyc,*.md,COPYING,Makefile,*.wsgi,*celerybeat-schedule*

PostmonServer.py

@@ -53,6 +53,24 @@ def format_result(result):
     return result
 
 
+def _get_estado_info(db, sigla):
+    sigla = sigla.upper()
+    return db.get_one_uf(sigla, fields={'_id': False, 'sigla': False})
+
+
+def _get_cidade_info(db, sigla_uf, nome_cidade):
+    sigla_uf = sigla_uf.upper()
+    sigla_uf_nome_cidade = '%s_%s' % (sigla_uf, nome_cidade)
+    fields = {
+        '_id': False,
+        'sigla_uf': False,
+        'codigo_ibge_uf': False,
+        'sigla_uf_nome_cidade': False,
+        'nome': False
+    }
+    return db.get_one_cidade(sigla_uf_nome_cidade, fields=fields)
+
+
 @route('/cep/<cep:re:\d{5}-?\d{3}>')
 @app_v1.route('/cep/<cep:re:\d{5}-?\d{3}>')
 def verifica_cep(cep):
@@ -79,13 +97,51 @@ def verifica_cep(cep):
 
     if result:
         response.headers['Cache-Control'] = 'public, max-age=2592000'
+        sigla_uf = result['estado']
+        estado_info = _get_estado_info(db, sigla_uf)
+        if estado_info:
+            result['estado_info'] = estado_info
+        nome_cidade = result['cidade']
+        cidade_info = _get_cidade_info(db, sigla_uf, nome_cidade)
+        if cidade_info:
+            result['cidade_info'] = cidade_info
         return format_result(result)
     else:
         response.status = '404 O CEP %s informado nao pode ser '
         'localizado' % cep
         return
 
 
+@app_v1.route('/uf/<sigla>')
+def uf(sigla):
+    db = Database()
+    result = _get_estado_info(db, sigla)
+    if result:
+        response.headers['Cache-Control'] = 'public, max-age=2592000'
+        return format_result(result)
+    else:
+        message = '404 A sigla %s informada ' \
+                  'nao pode ser localizada'
+        response.status = message % sigla
+        print response.status
+        return
+
+
+@app_v1.route('/cidade/<sigla_uf>/<nome>')
+def cidade(sigla_uf, nome):
+    db = Database()
+    result = _get_cidade_info(db, sigla_uf, nome)
+    if result:
+        response.headers['Cache-Control'] = 'public, max-age=2592000'
+        return format_result(result)
+    else:
+        message = '404 A cidade %s (%s) informada ' \
+                  'nao pode ser localizada'
+        response.status = message % (sigla_uf, nome)
+        print response.status
+        return
+
+
 @app_v1.route('/rastreio/<provider>/<track>')
 def track_pack(provider, track):
     if provider == 'ect':

PostmonTaskScheduler.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+from datetime import timedelta
+from celery import Celery
+from celery.utils.log import get_task_logger
+from IbgeTracker import IbgeTracker
+from database import MongoDb as Database
+import os
+
+USERNAME = os.environ.get('POSTMON_DB_USER')
+PASSWORD = os.environ.get('POSTMON_DB_PASSWORD')
+if all((USERNAME, PASSWORD)):
+    broker_conn_string = 'mongodb://%s:%s@localhost:27017' \
+        % (USERNAME, PASSWORD)
+else:
+    broker_conn_string = 'mongodb://localhost:27017'
+
+print(broker_conn_string)
+
+app = Celery('postmon', broker=broker_conn_string)
+
+app.conf.update(
+    CELERY_TASK_SERIALIZER='json',
+    CELERY_ACCEPT_CONTENT=['json'],  # Ignore other content
+    CELERY_RESULT_SERIALIZER='json',
+    CELERY_TIMEZONE='America/Sao_Paulo',
+    CELERY_ENABLE_UTC=True,
+    CELERYBEAT_SCHEDULE={
+        'track_ibge_daily': {
+            'task': 'PostmonTaskScheduler.track_ibge',
+            'schedule': timedelta(days=1)  # útil para
+                                           # testes: timedelta(minutes=1)
+        }
+    }
+)
+
+logger = get_task_logger(__name__)
+
+
+@app.task
+def track_ibge():
+    logger.info('Iniciando tracking do IBGE...')
+    db = Database()
+    ibge = IbgeTracker()
+    ibge.track(db)
+    logger.info('Finalizou o tracking do IBGE')

README.md

@@ -20,6 +20,7 @@ As dependências estão listadas no arquivo requirements.txt.
 * nosetests
 * webtest
 * packtrack
+* celery
 
 Rodando testes
 ----------------
@@ -41,6 +42,21 @@ ou
 
 Caso queira rodar em outra porta, basta passá-la como parametro no chamado do _standalone
 
+Para rodar o [Scheduler](#scheduler):
+
+	$ celery worker -B -A PostmonTaskScheduler -l info
+
+Recomenda-se a utilização do [Supervisord](http://supervisord.org/) para manter o Celery rodando. Exemplo de configuração para o _supervisord.conf_:
+
+	[program:celeryd]
+	command=celery worker -B -A PostmonTaskScheduler -l info 
+	directory=POSTMON_HOME/repositorio
+	stdout_logfile=POSTMON_LOG_DIR/celeryd.log
+	stderr_logfile=POSTMON_LOG_DIR/celeryd_err.log
+	autostart=true
+	autorestart=true
+	startsecs=10
+	stopwaitsecs=600
 
 MongoDB com autenticação
 ------------------------
@@ -60,3 +76,38 @@ Agora que seu Mongo está com password exporte as variaveis de ambiente.
 export POSTMON_DB_USER=admin
 export POSTMON_DB_PASSWORD=123456
 ```
+
+Scheduler
+---------
+
+O Postmon conta com um scheduler baseado na ferramenta [Celery](http://www.celeryproject.org/). Até o momento, a única funcionalidade implementada nessa estrutura é a rotina de coleta de dados do [IBGE](#ibge).
+
+O Celery usa, como Broker, a mesma instância do MongoDB utilizada no módulo de CEP.
+
+O comando apresentado na seção [rodando a aplicação](#rodando-a-aplicação-localmente-na-porta-9876) pode ser quebrado em dois caso seja necessário rodar o Celery Worker separado do Celery Beat. Para mais informações sobre essa questão leia a [documentação do Celery](http://docs.celeryproject.org/en/latest/).
+
+Além do Broker, o Celery Beat depende internamente de uma base de dados, criada automaticamente na primeira execução, onde são armazenadas informações sobre os schedules. Por padrão essa base fica em um arquivo chamado _celerybeat_schedule_, criado no diretório onde o Celery Beat foi executado. Esse local pode ser alterado através do switch -s, conforme exemplo abaixo:
+
+	$ celery worker -B -A PostmonTaskScheduler -l info -s /novo/caminho/para/arquivo/celerybeat_schedule
+
+IBGE
+-------------
+
+O Postmon fornece as seguintes informações extraídas do site do IBGE:
+
+* Código do município/UF
+* Área territorial (em km²)
+
+Essas informações estão presentes nos atributos *estado_info* e *cidade_info* da rota de busca de _cep_, bem como nas seguintes rotas:
+
+* /uf/{sigla-uf}
+* /cidade/{sigla-uf}/{nome-cidade}
+
+Exemplos:
+
+* /uf/SP
+* /cidade/SP/São Paulo
+* /cidade/SP/Araraquara
+* /cidade/RJ/Macaé
+
+A rotina de atualização desses dados está configurada para rodar diariamente.

database.py

@@ -25,6 +25,16 @@ def __init__(self, address='localhost'):
     def get_one(self, cep, **kwargs):
         return self._db.ceps.find_one({'cep': cep}, **kwargs)
 
+    def get_one_uf(self, sigla, **kwargs):
+        return self._db.ufs.find_one({'sigla': sigla}, **kwargs)
+
+    def get_one_cidade(self, sigla_uf_nome_cidade, **kwargs):
+        spec = {'sigla_uf_nome_cidade': sigla_uf_nome_cidade}
+        return self._db.cidades.find_one(spec, **kwargs)
+
+    def get_one_uf_by_nome(self, nome, **kwargs):
+        return self._db.ufs.find_one({'nome': nome}, **kwargs)
+
     def insert_or_update(self, obj, **kwargs):
 
         update = {'$set': obj}
@@ -33,5 +43,14 @@ def insert_or_update(self, obj, **kwargs):
 
         self._db.ceps.update({'cep': obj['cep']}, update, upsert=True)
 
+    def insert_or_update_uf(self, obj, **kwargs):
+        update = {'$set': obj}
+        self._db.ufs.update({'sigla': obj['sigla']}, update, upsert=True)
+
+    def insert_or_update_cidade(self, obj, **kwargs):
+        update = {'$set': obj}
+        chave = 'sigla_uf_nome_cidade'
+        self._db.cidades.update({chave: obj[chave]}, update, upsert=True)
+
     def remove(self, cep):
         self._db.ceps.remove({'cep': cep})

requirements.txt

@@ -7,3 +7,4 @@ webtest==1.4.3
 packtrack==0.1.0
 xmltodict
 flake8
+celery[mongodb]