chore(#49): track seed script and geography CSV in git
Allowlist scripts/ in .gitignore (was caught by [Ss]cripts venv rule) so seed_geography.py and colombia_municipios.csv can be committed. This simplifies deployment since the files no longer need to be copied separately to the server.
This commit is contained in:
2
.gitignore
vendored
2
.gitignore
vendored
@@ -321,7 +321,7 @@ pyrightconfig.json
|
||||
[Ll]ib
|
||||
[Ll]ib64
|
||||
[Ll]ocal
|
||||
[Ss]cripts
|
||||
Scripts/
|
||||
pyvenv.cfg
|
||||
pip-selfcheck.json
|
||||
|
||||
|
||||
1121
scripts/data/colombia_municipios.csv
Normal file
1121
scripts/data/colombia_municipios.csv
Normal file
File diff suppressed because it is too large
Load Diff
188
scripts/seed_geography.py
Normal file
188
scripts/seed_geography.py
Normal file
@@ -0,0 +1,188 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Seed geografía de Colombia (países, departamentos y municipios).
|
||||
|
||||
Requisitos (instalar ad-hoc, NO son dependencia del proyecto):
|
||||
pip install pycountry requests
|
||||
|
||||
Uso:
|
||||
python scripts/seed_geography.py http://localhost:7000
|
||||
|
||||
El script obtiene el país y los departamentos de Colombia desde
|
||||
`pycountry` (ISO 3166-1/3166-2) y los municipios desde el dataset DANE
|
||||
incluido en `scripts/data/colombia_municipios.csv`. No consulta internet
|
||||
en tiempo de ejecución.
|
||||
"""
|
||||
import argparse
|
||||
import csv
|
||||
import getpass
|
||||
import os
|
||||
import sys
|
||||
import unicodedata
|
||||
|
||||
import requests
|
||||
|
||||
try:
|
||||
import pycountry
|
||||
except ModuleNotFoundError:
|
||||
print(
|
||||
"pycountry no está instalado. Ejecuta: pip install pycountry",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
TOKEN_URL = "/api/token/"
|
||||
SEED_GEOGRAPHY_URL = "/don_confiao/api/seed_geography"
|
||||
|
||||
MUNICIPALITIES_CSV = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"data",
|
||||
"colombia_municipios.csv",
|
||||
)
|
||||
|
||||
# Nombres DANE que difieren de los nombres oficiales de pycountry.
|
||||
DEPARTMENT_ALIASES = {
|
||||
"BOGOTA": "Distrito Capital de Bogotá",
|
||||
"SAN ANDRES": "San Andrés, Providencia y Santa Catalina",
|
||||
}
|
||||
|
||||
|
||||
def get_credentials():
|
||||
username = input("Usuario: ")
|
||||
password = getpass.getpass("Contraseña: ")
|
||||
return username, password
|
||||
|
||||
|
||||
def get_token(domain, username, password):
|
||||
url = domain.rstrip("/") + TOKEN_URL
|
||||
response = requests.post(
|
||||
url, json={"username": username, "password": password}
|
||||
)
|
||||
if response.status_code != 200:
|
||||
print(
|
||||
f"Error al obtener token: {response.status_code} {response.text}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
return response.json()["access"]
|
||||
|
||||
|
||||
def _normalize(name):
|
||||
return "".join(
|
||||
char
|
||||
for char in unicodedata.normalize("NFD", name)
|
||||
if unicodedata.category(char) != "Mn"
|
||||
).upper()
|
||||
|
||||
|
||||
def get_department_mapping(subdivisions):
|
||||
by_normalized = {
|
||||
_normalize(subdivision.name): subdivision.name
|
||||
for subdivision in subdivisions
|
||||
}
|
||||
|
||||
csv_department_names = set()
|
||||
with open(MUNICIPALITIES_CSV, newline="", encoding="utf-8") as f:
|
||||
for row in csv.DictReader(f):
|
||||
csv_department_names.add(row["department"])
|
||||
|
||||
mapping = dict(DEPARTMENT_ALIASES)
|
||||
for name in csv_department_names:
|
||||
if name in mapping:
|
||||
continue
|
||||
mapped = by_normalized.get(_normalize(name))
|
||||
if mapped is None:
|
||||
print(
|
||||
f"Departamento '{name}' no encontrado en pycountry",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
mapping[name] = mapped
|
||||
return mapping
|
||||
|
||||
|
||||
def read_municipalities(mapping):
|
||||
municipalities = []
|
||||
with open(MUNICIPALITIES_CSV, newline="", encoding="utf-8") as f:
|
||||
for row in csv.DictReader(f):
|
||||
municipality = {
|
||||
"name": row["municipality"],
|
||||
"department": mapping[row["department"]],
|
||||
}
|
||||
if row.get("latitude"):
|
||||
municipality["latitude"] = float(row["latitude"])
|
||||
if row.get("longitude"):
|
||||
municipality["longitude"] = float(row["longitude"])
|
||||
municipalities.append(municipality)
|
||||
return municipalities
|
||||
|
||||
|
||||
def build_payload():
|
||||
country = pycountry.countries.get(alpha_2="CO")
|
||||
subdivisions = pycountry.subdivisions.get(country_code="CO")
|
||||
|
||||
departments = [
|
||||
{"name": subdivision.name, "code": subdivision.code}
|
||||
for subdivision in subdivisions
|
||||
]
|
||||
mapping = get_department_mapping(subdivisions)
|
||||
municipalities = read_municipalities(mapping)
|
||||
|
||||
return {
|
||||
"country": {"name": country.name, "code": country.alpha_2},
|
||||
"departments": departments,
|
||||
"municipalities": municipalities,
|
||||
}
|
||||
|
||||
|
||||
def seed_geography(domain, token, payload):
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
full_url = domain.rstrip("/") + SEED_GEOGRAPHY_URL
|
||||
response = requests.post(full_url, headers=headers, json=payload)
|
||||
if response.status_code != 200:
|
||||
print(
|
||||
f"Error al sembrar geografía: {response.status_code} {response.text}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
return response.json()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description=(
|
||||
"Alimenta países, departamentos y municipios de Colombia "
|
||||
"usando pycountry y el dataset DANE incluido en scripts/data/."
|
||||
)
|
||||
)
|
||||
parser.add_argument(
|
||||
"domain", help="Dominio del backend (ej: http://localhost:7000)"
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
username, password = get_credentials()
|
||||
token = get_token(args.domain, username, password)
|
||||
print("Token obtenido correctamente.")
|
||||
|
||||
payload = build_payload()
|
||||
print(
|
||||
"Preparando siembra: 1 país, "
|
||||
f"{len(payload['departments'])} departamentos, "
|
||||
f"{len(payload['municipalities'])} municipios."
|
||||
)
|
||||
|
||||
print("Enviando geografía...")
|
||||
result = seed_geography(args.domain, token, payload)
|
||||
print(
|
||||
f" [OK] {result['country']}: "
|
||||
f"{result['total_departments']} departamentos, "
|
||||
f"{result['total_municipalities']} municipios."
|
||||
)
|
||||
print(
|
||||
f"\nResumen: {result['departments_created']} departamentos y "
|
||||
f"{result['municipalities_created']} municipios creados "
|
||||
f"(re-ejecutar es idempotente)."
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user