Consulta el estado de un paquete en serpost a partir del número de tracking

Pipfile

[[source]]
url = "https://pypi.python.org/simple"
verify_ssl = true
name = "pypi"

[dev-packages]
ptpython = "*"

[packages]
aiohttp = "*"
"beautifulsoup4" = "*"
lxml = "*"

[requires]
python_version = "3.6"

serpost.py

import asyncio
import aiohttp as aio
from datetime import datetime
from bs4 import BeautifulSoup

base_url = 'http://clientes.serpost.com.pe/prj_online/Web_Busqueda.aspx'

def process_summary(summary):
    field_dict = dict(
        año='RetornoCadena1',
        codigo='RetornoCadena2',
        estado='RetornoCadena3',
        num_aviso='RetornoCadena4',
        origen='RetornoCadena5',
        destino='RetornoCadena6',
        tipo='RetornoCadena7',
        observacion='RetornoCadena8'
    )

    return {k:summary[v] for k,v in field_dict.items()}

async def fetch_track_summary(session, code, year):
    res = await session.post(
        f'{base_url}/Consultar_Tracking',
        json=dict(Anio=str(year), Tracking=code))

    data = (await res.json()).get('d')

    if data is None:
        return None

    return process_summary(data[0])

def process_details(details):
    soup = BeautifulSoup(details['ResulQuery'], 'lxml')
    texts = [cell.text for row in soup.find_all('tr') for cell in row.find_all('td')]
    row_list = [texts[i:i+3] for i in range(0, len(texts), 3)]

    return [dict(
        destino=destino,
        fecha=datetime.strptime(fecha, '%d/%m/%Y'),
        descripcion=texto,
    ) for (destino, fecha, texto) in row_list]


async def fetch_track_details(session, destination, code, year):
    res = await session.post(
        f'{base_url}/Consultar_Tracking_Detalle',
        json=dict(Anio=str(year), Tracking=code, Destino=destination))

    data = (await res.json()).get('d')

    if data is None:
        return None

    return process_details(data)

async def track(session, code, year):
    summary = await fetch_track_summary(session, code, year)
    if summary is None:
        return None

    details = await fetch_track_details(session, summary['destino'], code, year)
    return dict(
        summary,
        historia=details,
    )

async def track_many(session, code_iter, year):
    return {
        code: asyncio.ensure_future(track(session, code, year))
        for code in code_iter
    }

if __name__ == '__main__':
    import argparse
    from pprint import pprint
    parser = argparse.ArgumentParser(
        description='Request state of packages to serpost API')
    parser.add_argument('tracking', metavar='T', type=str, nargs='+',
                   help='numero de tracking a consultar')
    parser.add_argument('--year', '-y', type=int,
                        default=datetime.now().year)
    parser.add_argument('--concurrency', '-c', type=int,
                        default=5, help='numero de peticiones paralelas')

    args = parser.parse_args()
    
    async def main():
        # limit maximun parallel petitions
        conn = aio.TCPConnector(limit_per_host=args.concurrency)
        async with aio.ClientSession() as session:
            tracking_list = await track_many(session, args.tracking, args.year)
            for code, tracking in tracking_list.items():
                data = await tracking
                if data is None:
                    print(f'\n{code}: Unavailable\n')
                else:
                    pprint(data)

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())