"""Offline preparation only. Originals live in TemporaryDirectory and are discarded. Requires Python 3.11+ and openpyxl. No visitor request runs this program. Outputs contain municipal aggregates and public business addresses, never patients. """ import csv import hashlib import io import json import pathlib import tempfile import unicodedata import urllib.request import xml.etree.ElementTree as ET import zipfile from collections import Counter from datetime import datetime, timezone import openpyxl ROOT = pathlib.Path(__file__).resolve().parents[2] OUT = ROOT / 'dist/saude/dados' UF = dict(zip('11 12 13 14 15 16 17 21 22 23 24 25 26 27 28 29 31 32 33 35 41 42 43 50 51 52 53'.split(), 'RO AC AM RR PA AP TO MA PI CE RN PB PE AL SE BA MG ES RJ SP PR SC RS MS MT GO DF'.split())) KINDS = ['samunab', 'samunaa', 'samumoto', 'samuemb', 'samuaer'] IBGE = 'https://geoftp.ibge.gov.br/organizacao_do_territorio/estrutura_territorial/divisao_territorial/2024/DTB_2024.zip' PHARMACY = 'https://www.gov.br/saude/pt-br/composicao/sectics/farmacia-popular/publicacoes/farmacias-participante-do-programa-farmacia-popular' def normalize(value): return ''.join(c for c in unicodedata.normalize('NFKD', value).upper() if c.isalnum()) def integer(value): if value in ('', None): return None n = float(value) if n < 0 or not n.is_integer(): raise ValueError(f'Invalid count: {value}') return int(n) def write(directory, name, value): (directory / name).write_text(json.dumps(value, ensure_ascii=False, separators=(',', ':')) + '\n') def prepare(): sources = [] with tempfile.TemporaryDirectory(prefix='saude-') as temporary: def fetch(url): with urllib.request.urlopen(url, timeout=45) as response: data = response.read(8 * 1024 * 1024 + 1) if len(data) > 8 * 1024 * 1024: raise ValueError('Source exceeds 8 MiB download budget') path = pathlib.Path(temporary) / str(len(sources)) path.write_bytes(data) sources.append({'url': url, 'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()}) return path # Official DTB supplies the full seven-digit identifier, including check digit. with zipfile.ZipFile(fetch(IBGE)) as outer: with zipfile.ZipFile(io.BytesIO(outer.read('RELATORIO_DTB_BRASIL_2024_MUNICIPIOS.ods'))) as ods: tree = ET.fromstring(ods.read('content.xml')) ns = {'t': 'urn:oasis:names:tc:opendocument:xmlns:table:1.0'} municipalities = [] for row in tree.findall('.//t:table-row', ns): cells = [''.join(c.itertext()).strip() for c in row.findall('t:table-cell', ns)] if len(cells) > 8 and len(cells[7]) == 7 and cells[7].isdigit(): municipalities.append({'id': cells[7], 'name': cells[8], 'uf': UF[cells[0]]}) assert len(municipalities) == 5571 by_code = {m['id'][:6]: m for m in municipalities} by_name = {(m['uf'], normalize(m['name'])): m for m in municipalities} assert len(by_name) == len(by_code) == len(municipalities) states = {uf: {'municipalities': {}, 'samuState': {}, 'pharmacyTotal': 0, 'unmatchedPharmacies': 0} for uf in UF.values()} for m in municipalities: states[m['uf']]['municipalities'][m['id']] = {'samu': {}, 'pharmacies': []} updates = {} national = {} # Five independently published categories. Do not add the overlapping "basic+advanced". for kind in KINDS: with zipfile.ZipFile(fetch(f'https://demas-dados-abertos.s3.amazonaws.com/csv/{kind}.csv.zip')) as archive: info = archive.getinfo(kind + '.csv') if info.file_size > 20 * 1024 * 1024: raise ValueError('Unexpected expanded source size') with archive.open(info) as raw: rows = csv.DictReader(io.TextIOWrapper(raw, encoding='utf-8-sig')) seen = set() for row in rows: if row['sg_granularidade'] != 'MN': raise ValueError('Unexpected granularity') m = by_code[row['co_ibge']] assert row['sg_uf'] == m['uf'] period = row['co_anomes'] key = (m['id'], period) if key in seen: raise ValueError('Duplicate municipality-period') seen.add(key) state = states[m['uf']] state['municipalities'][m['id']]['samu'].setdefault(period, {})[kind] = integer(row['vl_indicador_calculado_mun']) total = integer(row['vl_indicador_calculado_uf']) totals = state['samuState'].setdefault(period, {}) if kind in totals and totals[kind] != total: raise ValueError('Inconsistent repeated state total') totals[kind] = total # Native state value, never summed repeated totals. brazil = national.setdefault(period, {}) brazil_total = integer(row['vl_indicador_calculado_br']) if kind in brazil and brazil[kind] != brazil_total: raise ValueError('Inconsistent repeated national total') brazil[kind] = brazil_total # Native Brazil value, retained once. updates[kind] = max(updates.get(kind, ''), row['dt_atualizacao']) for period, totals in national.items(): for kind, value in totals.items(): if value is None or any(s['samuState'].get(period, {}).get(kind) is None for s in states.values()): raise ValueError('Missing national/state count for reconciliation') if sum(s['samuState'][period][kind] for s in states.values()) != value: raise ValueError('National count differs from sum of native state totals') page = fetch(PHARMACY + '/view').read_text() # This dated snapshot was independently confirmed on its publication page and workbook. publication_date = '2026-10-02' assert '02/10/2026' in page, 'Publication page changed: review source' with open(fetch(PHARMACY + '/@@download/file'), 'rb') as raw: workbook = openpyxl.load_workbook(raw, read_only=True) rows = iter(workbook.active.values) next(rows); next(rows) assert next(rows)[0] == 'Data : 02/10/2026', 'Publication changed: review date before updating' assert tuple(next(rows)) == ('CNPJ', 'Razão Social', 'Endereço', 'Bairro', 'Município', 'UF') unmatched = Counter() seen = set() for cnpj, name, address, district, city, uf in rows: if not cnpj: continue if cnpj in seen: raise ValueError('Duplicate establishment identifier') seen.add(cnpj) # Used only in memory; not retained. state = states[uf] state['pharmacyTotal'] += 1 m = by_name.get((uf, normalize(city))) if m is None: unmatched[(uf, city)] += 1 state['unmatchedPharmacies'] += 1 continue # No fuzzy matching or invented municipality code. state['municipalities'][m['id']]['pharmacies'].append([str(name).strip(), str(address).strip(), str(district or '').strip()]) workbook.close() meta = {'schema': 1, 'preparedAt': datetime.now(timezone.utc).isoformat(), 'municipalityReference': 'DTB 2024 — 31/12/2024', 'pharmacyPublication': publication_date, 'samuUpdates': updates, 'samuBrazil': national, 'sources': sources, 'unmatchedPharmacies': [{'uf': uf, 'name': name, 'count': count} for (uf, name), count in sorted(unmatched.items())], 'pharmacyRecords': len(seen), 'municipalities': len(municipalities), 'retention': 'Public business name/address/district and municipal/state/national counts only. Retain latest valid snapshot; review after 90 days. Originals discarded after preparation; historical Git revisions retain derived snapshots.'} # Stage all outputs first; a failed source never overwrites the last valid snapshot. stage = pathlib.Path(temporary) / 'derived' stage.mkdir() write(stage, 'municipios.json', sorted(municipalities, key=lambda m: (m['name'], m['uf']))) write(stage, 'meta.json', meta) for uf, state in states.items(): write(stage, uf + '.json', state) OUT.mkdir(parents=True, exist_ok=True) for file in stage.iterdir(): file.replace(OUT / file.name) print(json.dumps({'municipalities': len(municipalities), 'pharmacies': len(seen), 'unmatched': sum(unmatched.values()), 'sources': len(sources)})) if __name__ == '__main__': prepare()