ZipeCode Scripts Python

get-duplicity-zipecode.py

import pymongo

print("start program --")

client = pymongo.MongoClient("mongodb://localhost/DataBaseName?ssl=true&authSource=admin&maxIdleTimeMS=120000&retrywrites=false"
                             , socketTimeoutMS=3600000)

db = client["DataBaseName"]
collection = db["ZipeCodeNumber"]
ceps = {}

# Itere sobre todos os documentos na coleção
for doc in collection.find():
    cep = doc["cep"]
    if cep in ceps:
        ceps[cep] += 1
    else:
        ceps[cep] = 1

# Itere sobre os ceps com contagens maiores do que 1 e imprima-os
for cep, count in ceps.items():
    if count > 1:
        print(cep)

print("finish program")

insert-zipecode.cs

import pandas as pd
from pymongo import MongoClient

print("start program --")

client = MongoClient("mongodb://localhost/DataBaseName?ssl=true&authSource=admin&maxIdleTimeMS=120000&retrywrites=false"
                             , socketTimeoutMS=3600000)
db = client["DataBaseName"]
collection = db["ZipeCodeNumber"]

csv_file = "/Users/vasconcellos/Downloads/ceps-somentenumero.csv"
data = pd.read_csv(csv_file, header=2, names=["cep", "nada"], dtype={"cep": str}, skiprows=1)

cepsDocuments=[]
# Insira os CEPs na coleção do MongoDB
for index, row in data.iterrows():
    cep = row["cep"]
    document = { "cep": cep}
    cepsDocuments.append(document)
    collection.insert_one(document)
    
    wasInserted = False
    while wasInserted: # Tente inserir o documento até que seja bem-sucedido
        try:
            collection.insert_one(document)
            wasInserted = True  # Inserção bem-sucedida, sair do loop while
        except Exception as e:
            print(f"Erro ao inserir o documento {document}: {e}. Tentando novamente...")

print("CEPs carregados com sucesso!")