"""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))