280 lines
15 KiB
Python
280 lines
15 KiB
Python
"""Read-only INE XLSX adapter. Uses labels/relationships, never executes formulas/macros."""
|
||
import argparse
|
||
from collections import defaultdict
|
||
from decimal import Decimal, InvalidOperation
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path, PurePosixPath
|
||
import re
|
||
import unicodedata
|
||
import xml.etree.ElementTree as ET
|
||
from zipfile import ZipFile
|
||
|
||
NS = {"s": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
|
||
REL = "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"
|
||
FORMAT = "investmoz-demography/1"
|
||
|
||
|
||
def normalized(value):
|
||
return " ".join("".join(c for c in unicodedata.normalize("NFKD", value.casefold()) if not unicodedata.combining(c)).split())
|
||
|
||
|
||
def sheets(path):
|
||
with ZipFile(path) as archive:
|
||
if len(archive.infolist()) > 2000 or sum(i.file_size for i in archive.infolist()) > 150_000_000:
|
||
raise ValueError("Workbook exceeds safe expanded size")
|
||
if any("vbaProject" in i.filename or "externalLinks/" in i.filename for i in archive.infolist()):
|
||
raise ValueError("Macros/external links are not supported")
|
||
strings = []
|
||
if "xl/sharedStrings.xml" in archive.namelist():
|
||
strings = ["".join(x.itertext()) for x in ET.fromstring(archive.read("xl/sharedStrings.xml")).findall("s:si", NS)]
|
||
relationships = {x.attrib["Id"]: x.attrib["Target"] for x in ET.fromstring(archive.read("xl/_rels/workbook.xml.rels"))}
|
||
for sheet in ET.fromstring(archive.read("xl/workbook.xml")).find("s:sheets", NS):
|
||
target = relationships[sheet.attrib[REL]]
|
||
member = target.lstrip("/") if target.startswith("/") else str(PurePosixPath("xl") / target)
|
||
rows = []
|
||
for row in ET.fromstring(archive.read(member)).findall("s:sheetData/s:row", NS):
|
||
cells = {}
|
||
for cell in row:
|
||
value = cell.findtext("s:v", "", NS)
|
||
kind = cell.attrib.get("t")
|
||
if kind == "s":
|
||
value = strings[int(value)]
|
||
elif kind == "inlineStr":
|
||
value = "".join(cell.find("s:is", NS).itertext())
|
||
if value or cell.find("s:f", NS) is not None:
|
||
cells[re.sub(r"\d", "", cell.attrib["r"])] = {"value": value, "cell": cell.attrib["r"], "invalid": kind == "e", "formula": cell.findtext("s:f", None, NS)}
|
||
if cells:
|
||
rows.append(cells)
|
||
yield sheet.attrib["name"], rows
|
||
|
||
|
||
def age_group(label):
|
||
label = normalized(label).replace("–", "-").replace(" ", "")
|
||
if label == "total":
|
||
return "all"
|
||
match = re.search(r"(\d+)-(\d+)", label)
|
||
if match:
|
||
return match[1] + "-" + match[2]
|
||
if "65oumais" in label:
|
||
return "65+"
|
||
if re.fullmatch(r"\d+\+", label):
|
||
return label
|
||
return None
|
||
|
||
|
||
def population_columns(rows):
|
||
"""Two-level area/sex headers, with merged area headings carried forward."""
|
||
area_row = next((r for r in rows if any(normalized(c["value"]) in ("urbana", "populacao urbana") for c in r.values())), {})
|
||
sex_row = next((r for r in rows if any(normalized(c["value"]) == "homens" for c in r.values())), {})
|
||
result, area = {}, "all"
|
||
for column, cell in sorted(sex_row.items(), key=lambda x: (len(x[0]), x[0])):
|
||
heading = normalized(area_row.get(column, {}).get("value", ""))
|
||
if "urbana" in heading:
|
||
area = "urban"
|
||
elif "rural" in heading:
|
||
area = "rural"
|
||
elif heading in ("total", "populacao total"):
|
||
area = "all"
|
||
sex = {"total": "all", "homens": "male", "mulheres": "female"}.get(normalized(cell["value"]))
|
||
if sex:
|
||
result[column] = (sex, area)
|
||
return result
|
||
|
||
|
||
def parse(path, baseline_year=None):
|
||
path = Path(path)
|
||
report = {"format": FORMAT, "file_name": path.name, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
|
||
"source": "Instituto Nacional de Estatística de Moçambique (INE)", "publication_version": None,
|
||
"records": [], "issues": [], "sheets": [], "territories": [], "years": [], "territory_headers": {}}
|
||
seen = {}
|
||
|
||
def issue(code, sheet, cell="", **extra):
|
||
report["issues"].append(dict(code=code, sheet=sheet, cell=cell, **extra))
|
||
|
||
def add(sheet, cell, province, territory, level, year, indicator, sex="all", area="all", age="all", unit="pessoas", kind="projected"):
|
||
if cell and cell.get("formula") is not None and cell["value"] == "":
|
||
issue("invalid_numeric_or_formula", sheet, cell["cell"], reason="formula_without_numeric_cache")
|
||
return
|
||
if not cell or cell["value"] == "":
|
||
issue("missing_value", sheet, territory=territory, indicator=indicator, age_group=age, sex=sex, residence=area)
|
||
return
|
||
raw = cell["value"]
|
||
try:
|
||
value = Decimal(raw)
|
||
if cell["invalid"] or not value.is_finite() or value < 0:
|
||
raise InvalidOperation
|
||
except InvalidOperation:
|
||
issue("invalid_numeric_or_formula", sheet, cell["cell"])
|
||
return
|
||
if cell.get("formula") is not None:
|
||
issue("cached_formula_value", sheet, cell["cell"], formula=cell["formula"])
|
||
row = dict(province=province, territory=territory, level=level, year=year, indicator=indicator, sex=sex,
|
||
residence=area, age_group=age, unit=unit, data_kind=kind, original_value=raw,
|
||
numeric_value=format(value, "f"), sheet=sheet, cell=cell["cell"])
|
||
key = tuple(row[k] for k in ("province", "territory", "level", "year", "indicator", "sex", "residence", "age_group"))
|
||
if key in seen:
|
||
issue("duplicate_value" if seen[key] == raw else "conflicting_value", sheet, cell["cell"])
|
||
return
|
||
seen[key] = raw
|
||
report["records"].append(row)
|
||
|
||
workbook = list(sheets(path))
|
||
annual = any(re.fullmatch(r"20\d{2}", name) for name, _ in workbook)
|
||
province = None
|
||
for sheet, rows in workbook:
|
||
before = len(report["records"])
|
||
if annual:
|
||
if not re.fullmatch(r"20\d{2}", sheet):
|
||
report["sheets"].append(dict(name=sheet, records=0, state="index"))
|
||
continue
|
||
year = int(sheet)
|
||
columns = population_columns(rows)
|
||
if len(columns) != 9:
|
||
issue("unsupported_population_headers", sheet)
|
||
continue
|
||
territory = None
|
||
for row in rows:
|
||
label = next(iter(row.values()))["value"]
|
||
title = re.search(r"segundo idade\.\s*(.+),\s*(20\d{2})\s*$", label, re.I)
|
||
if title:
|
||
if int(title[2]) != year:
|
||
issue("title_year_mismatch", sheet, next(iter(row.values()))["cell"])
|
||
territory = None
|
||
continue
|
||
place = title[1].strip()
|
||
prefix = r"^(?:Distrito Municipal\s+|(?:Distrito|Cidade) (?:de|da)\s+)"
|
||
level = "district" if re.match(prefix, place, re.I) else "province"
|
||
territory = re.sub(prefix, "", place, flags=re.I)
|
||
if level == "province":
|
||
if province and normalized(province) != normalized(territory):
|
||
raise ValueError("Multiple provinces in annual workbook")
|
||
province = territory
|
||
if province:
|
||
header_key = "|".join([province, territory, level])
|
||
report["territory_headers"].setdefault(header_key, [])
|
||
if place not in report["territory_headers"][header_key]:
|
||
report["territory_headers"][header_key].append(place)
|
||
continue
|
||
age = age_group(row["A"]["value"]) if "A" in row else None
|
||
if territory and age and province:
|
||
for column, (sex, area) in columns.items():
|
||
add(sheet, row.get(column), province, territory, level, year, "population", sex, area, age)
|
||
else:
|
||
name = normalized(sheet)
|
||
if name.startswith("populacao_") or "grupo etario" in name:
|
||
columns = population_columns(rows)
|
||
is_age = "grupo etario" in name
|
||
explicit = [int(c["value"]) for r in rows[:2] for c in r.values() if re.fullmatch(r"20\d{2}", c["value"])]
|
||
year = explicit[0] if explicit else baseline_year
|
||
if not year:
|
||
issue("reference_year_required", sheet)
|
||
continue
|
||
if not explicit:
|
||
issue("year_from_import_metadata", sheet, year=year)
|
||
province = None
|
||
for row in rows[2:]:
|
||
if row.get("A"):
|
||
province = row["A"]["value"].strip()
|
||
age = age_group(row.get("B", {}).get("value", "")) if is_age else "all"
|
||
if province and age:
|
||
for column, (sex, area) in columns.items():
|
||
add(sheet, row.get(column), province, province, "province", year, "population", sex, area, age, kind="unspecified")
|
||
elif "vitais" in name:
|
||
years = re.findall(r"20\d{2}", " ".join(c["value"] for r in rows[:3] for c in r.values()))
|
||
if len(set(years)) != 1:
|
||
issue("reference_year_required", sheet)
|
||
continue
|
||
# Indicator names derive from the actual two-row headings. Units absent in the source remain unspecified.
|
||
heading = rows[1]
|
||
active = ""
|
||
for column, cell in sorted(rows[2].items(), key=lambda x: (len(x[0]), x[0])):
|
||
if column == "A":
|
||
continue
|
||
if column in heading:
|
||
active = normalized(heading[column]["value"])
|
||
detail = normalized(cell["value"])
|
||
if "esperanca" in active:
|
||
indicator, unit = "life_expectancy", "anos"
|
||
elif "infantil" in active:
|
||
indicator, unit = "infant_mortality", "unspecified"
|
||
else:
|
||
indicator = {"mortalidade": "crude_death_rate", "natalidade": "crude_birth_rate", "fecundidade": "fertility_rate", "exponencial": "population_growth_rate"}.get(detail)
|
||
unit = "unspecified"
|
||
if not indicator:
|
||
issue("unsupported_vital_header", sheet, cell["cell"])
|
||
continue
|
||
if unit == "unspecified":
|
||
issue("unit_not_stated", sheet, cell["cell"], indicator=indicator)
|
||
sex = {"homens": "male", "mulheres": "female"}.get(detail, "all")
|
||
for row in rows[3:]:
|
||
if "A" in row:
|
||
p = row["A"]["value"].strip()
|
||
add(sheet, row.get(column), p, p, "province", int(years[0]), indicator, sex, unit=unit, kind="unspecified")
|
||
else:
|
||
issue("unpopulated_indicator_catalogue" if not any(re.fullmatch(r"\d+(\.\d+)?", c["value"]) for r in rows for c in r.values()) else "unsupported_sheet", sheet)
|
||
report["sheets"].append(dict(name=sheet, records=len(report["records"]) - before))
|
||
reconcile(report)
|
||
report["territories"] = sorted({(r["province"], r["territory"], r["level"]) for r in report["records"]})
|
||
report["years"] = sorted({r["year"] for r in report["records"]})
|
||
report["record_count"] = len(report["records"])
|
||
report["issue_counts"] = dict(sorted(__import__("collections").Counter(i["code"] for i in report["issues"]).items()))
|
||
if not report["records"]:
|
||
raise ValueError("No supported demographic data found")
|
||
return report
|
||
|
||
|
||
def reconcile(report):
|
||
groups = defaultdict(dict)
|
||
territorial = defaultdict(dict)
|
||
ages = defaultdict(dict)
|
||
for r in report["records"]:
|
||
if r["indicator"] != "population":
|
||
continue
|
||
base = (r["province"], r["territory"], r["level"], r["year"], r["age_group"])
|
||
groups[base][(r["sex"], r["residence"])] = r
|
||
ages[base[:4] + (r["sex"], r["residence"])][r["age_group"]] = r
|
||
if r["age_group"] == r["sex"] == r["residence"] == "all":
|
||
territorial[(r["province"], r["year"])][(r["level"], r["territory"])] = r
|
||
|
||
def check(total, parts, code):
|
||
if total and all(parts):
|
||
delta = Decimal(total["numeric_value"]) - sum(Decimal(p["numeric_value"]) for p in parts)
|
||
# Binary Excel numeric storage leaves sub-micro-person residues; retain originals,
|
||
# but do not report these as demographic inconsistencies.
|
||
if abs(delta) > Decimal("0.000001"):
|
||
report["issues"].append(dict(code=code, sheet=total["sheet"], cell=total["cell"], delta=str(delta), territory=total["territory"], year=total["year"]))
|
||
|
||
for rows in groups.values():
|
||
for area in ("all", "urban", "rural"):
|
||
check(rows.get(("all", area)), [rows.get(("male", area)), rows.get(("female", area))], "sex_total_mismatch")
|
||
for sex in ("all", "male", "female"):
|
||
check(rows.get((sex, "all")), [rows.get((sex, "urban")), rows.get((sex, "rural"))], "residence_total_mismatch")
|
||
for rows in ages.values():
|
||
detail = [v for k, v in rows.items() if k != "all"]
|
||
if detail:
|
||
check(rows.get("all"), detail, "age_total_mismatch")
|
||
for rows in territorial.values():
|
||
provinces = [v for (kind, _), v in rows.items() if kind == "province"]
|
||
districts = [v for (kind, _), v in rows.items() if kind == "district"]
|
||
if provinces and districts:
|
||
check(provinces[0], districts, "district_total_mismatch")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
cli = argparse.ArgumentParser()
|
||
cli.add_argument("file")
|
||
cli.add_argument("--baseline-year", type=int)
|
||
cli.add_argument("--output", required=True)
|
||
cli.add_argument("--records-output")
|
||
args = cli.parse_args()
|
||
result = parse(args.file, args.baseline_year)
|
||
if args.records_output:
|
||
record_path = Path(args.records_output)
|
||
with record_path.open('w', encoding='utf-8', newline='\n') as target:
|
||
for record in result.pop('records'):
|
||
target.write(json.dumps(record, ensure_ascii=False, separators=(',', ':'))+'\n')
|
||
result['records_file_sha256'] = hashlib.sha256(record_path.read_bytes()).hexdigest()
|
||
Path(args.output).write_text(json.dumps(result, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
|
||
print(json.dumps({k: v for k, v in result.items() if k not in ("records", "issues")}, ensure_ascii=False))
|