2
\$\begingroup\$

I was presented with a task to come up with a script that generates a CSV with POSTAL codes via bruteforce (I'm in Brazil, so that means CEP to us).

Points to note:

  • I'm using an external library, but basically consuming a SOAP service provided by the Postal agency;
  • I have a list of cities I have to get the results for, with variable ranges to be tested;
  • I have 8 threads available for me, and using 7 to process and 1 to writing (Iw as scared of sincronism);

When formulating the solution, I had two paths:

  1. Send a city to each thread, and wait for that city to finish processing.
  2. Send the entire list of postal codes available for a city to all threads, and only grab the next city after it is all done.

Currently I opted for option 1, since I was thinking the number of requests and processing would be the same, but now it does seem a bit slow (it's been running for around 7 days and not done).

Problems:

  • There're cities with huge ranges to test (like Sao Paulo, with 6.5M), which means a thread stuck processing it;
  • but on the other hand, it's complicated to manage which cities have completed by going for option 2.

My input data is a CSV file, formatted like this (This is in Brazilian Portuguese):

Adamantina,17800-000 a 17809-999,Não codificada por logradouros,Total do município
Agudos,17120-001 a 17149-999,Codificado por logradouros,Total do município

My code is below. Some of the code can be ignored (like the logging statements):

import threading
import csv
import queue
import time
from datetime import timedelta

import requests
import pycep_correios
from loguru import logger
from pycep_correios import HOMOLOGACAO, PRODUCAO
from tqdm import tqdm


def read_csv(csv_path: str):
    """Utility function to read CSV files

    Arguments:
        csv_path {str} -- path to the CSV file

    Returns:
        list -- a list of lists where each CSV row is a list
    """
    with open(csv_path, 'r') as f:
        csv_reader = csv.reader(f)
        return [list(row) for row in csv_reader]


def recover_range(range_string: str):
    """Utility function to recover the numeric range from string

    Given that the string is in a specific format, 
    we split it and get the numeric range

    Arguments:
        range_string {str} -- string containing the range, must be in format 'xxxxx-xxx a xxxxx-xxx'

    Returns:
        list -- list with the numeric range ([from, to])
    """
    cep_range = [v.replace('-', '') for v in range_string.split(' ')]
    cep_range = [int(v) for v in cep_range if v.isnumeric()]
    return cep_range


class DataWriterThread(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        logger.info('Instantiating {} on {}', self.__class__.__name__,
                    self.name)
        self.queue = queue

    def write_to_file(self, content: dict):
        """Method to write content to the target CSV

        Arguments:
            content {dict} -- dictionary where each key is a column for the CSV
        """
        with open('output_sp.csv', 'a') as f:
            logger.info('{}: writing new register to CSV.', self.name)
            writer = csv.DictWriter(f, content.keys())
            writer.writerow(content)

    def run(self):
        while True:
            result = self.queue.get()
            self.write_to_file(result)
            self.queue.task_done()


class RequestThread(threading.Thread):
    def __init__(self, in_queue: queue.Queue, out_queue: queue.Queue):
        threading.Thread.__init__(self)
        logger.info('Instantiating {} on {}', self.__class__.__name__,
                    self.name)
        self.in_queue = in_queue
        self.out_queue = out_queue

    def run(self):
        while True:
            record = self.in_queue.get()
            for item in record:
                result = self.process(item)
                if result is not None:
                    self.out_queue.put(result)
            else:
                logger.success('{} has finished processing...', result['cidade'])
            self.in_queue.task_done()

    def process(self, record):
        """Processing method

        This method expects a CEP number as input, and will make a request 
        to the Correios API to check if it exists. In positive cases we 
        return the data, else None is returned.

        Arguments:
            record {int} -- CEP number

        Returns:
            dict -- Dictionary containing the valid CEP data, None if not found
        """
        cep = str(record)
        if len(cep) < 8:
            cep = '0{}'.format(cep)
        if pycep_correios.validar_cep(cep):
            try:
                address = pycep_correios.consultar_cep(cep=cep,
                                                        ambiente=HOMOLOGACAO)
                if address is not None:
                    logger.success('{}: {} is a valid CEP, saving...',
                                   self.name, cep)
                    return address
            except pycep_correios.excecoes.ExcecaoPyCEPCorreios as exc:
                logger.error('{}: Exception when processing: {} - [{}]',
                             self.name, cep, exc.message)
                return None
            except AttributeError as exc:
                logger.error('{}: CEP {} has AttributeError {}', self.name,
                             cep, exc)
            except requests.exceptions.ConnectionError:
                logger.warning(
                    '{}: Connection blocked, trying again in 60 seconds...',
                    self.name)
                time.sleep(60)  # After some time the SOAP service force disconnects
                # Recursive call to reprocess the item
                return self.process(cep)
        else:
            logger.error('{}: Invalid CEP {}', self.name, cep)
            return None


def main():
    lg = logger.add("events.log",
                    rotation="25 MB",
                    compression="zip",
                    format="{time} {level} {message}",
                    level="INFO")
    cep_list = read_csv('cep_sp.csv')  # read and preprocess ranges
    cep_list = [
        item for item in cep_list
        if item[1] and item[2] == 'Codificado por logradouros'
    ]
    cep_queue = queue.Queue()  # list of ceps to process
    result_queue = queue.Queue()  # results acquired

    for _ in range(3):  # 4 cores = 3 proc threads + 1 writer thread
        t = RequestThread(in_queue=cep_queue, out_queue=result_queue)
        t.daemon = True
        t.start()

    # spawn threads to print
    t = DataWriterThread(result_queue)
    t.daemon = True
    t.start()

    cep_processing_list = []
    for record in cep_list:
        cep_ranges = recover_range(record[1])
        if not cep_ranges:  # dealing with header rows
            continue
        else:
            cep_processing_list.append(
                [cep for cep in range(cep_ranges[0], cep_ranges[1])])

    # add paths to queue
    for record in tqdm(cep_processing_list):
        cep_queue.put(record)

    start_time = time.time()
    logger.info('Starting the script...')

    # wait for queue to get empty
    cep_queue.join()
    result_queue.join()

    logger.info('Finished. Done in {}.',
                str(timedelta(seconds=(time.time() - start_time))))
    logger.remove(lg)


if __name__ == "__main__":
    main()

# reference: https://stackoverflow.com/questions/11983938/python-appending-to-same-file-from-multiple-threads

The goals I want to achieve:

  • Improve efficiency overall;
  • Improve code to be cleaner and more compliant with Python guidelines (I tried my best);
  • Get some advice if I should split this code in functions and files more;
  • If possible, come up with a solution for the huge amount of data implied.

If more hardware is the only answer, there's nothing I can do. But I would like to do the best I can with the code. Thanks in advance.

EDIT: As this question has been around for some time and has not seem much activity besides the initial response I am marking it as solved, but if there're any pointers you would like to comment, you are free to do.

\$\endgroup\$
  • 1
    \$\begingroup\$ While unintuitive, you should know that the threading package does not actually do proper multithreading, it only simulates asynchronous threads. It's great for doing HTTP requests, and waiting for results. But you probably want to look at the multiprocessing package, which actually runs your code in parallel, and offers a speedup. \$\endgroup\$ – maxb Sep 3 at 12:50
  • \$\begingroup\$ That's interesting to know. I am quite new to multithreading in python, thanks for the head up. \$\endgroup\$ – inblank Sep 4 at 13:55
2
\$\begingroup\$

Type hints for return values

You add str as a parameter hint, but you're missing a return value hint - for read_csv for instance, it should be list.

You're also missing a type hint for queue.

Formatting leading zeros

Don't pre-convert cep to a string. Instead:

cep = '{:08d}'.format(cep)

f-strings

logger.success(f'{self.name}: {cep} is a valid CEP, saving...')

Recursion for retry

Don't! There's no need to recurse, and it's needlessly bloating your stack. Rather than recursing, use a simple loop.

continue logic

    if not cep_ranges:  # dealing with header rows
        continue
    else:
        cep_processing_list.append(
            [cep for cep in range(cep_ranges[0], cep_ranges[1])])

should be

if cep_ranges:
    cep_processing_list.append(
        [cep for cep in range(cep_ranges[0], cep_ranges[1])])
\$\endgroup\$
  • \$\begingroup\$ Thank you for your pointers, I was not aware loguru supports f-strings directly. The recursion strategy is a hack, and your suggestion seems pretty much smarter. \$\endgroup\$ – inblank Aug 30 at 1:26
  • 1
    \$\begingroup\$ An f-string, after formatting, turns into a regular string - so anything that accepts a str supports an f-string. \$\endgroup\$ – Reinderien Aug 30 at 1:33
  • \$\begingroup\$ Overall, what do you think of the logic? \$\endgroup\$ – inblank Aug 30 at 12:51

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.