94 lines
4.7 KiB
Python
94 lines
4.7 KiB
Python
"""Read-only XLSX extraction. No spreadsheet or pilot database is changed."""
|
|
import argparse
|
|
import collections
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import xml.etree.ElementTree as ET
|
|
import zipfile
|
|
|
|
|
|
def prepare(source, destination):
|
|
content = Path(source).read_bytes()
|
|
digest = hashlib.sha256(content).hexdigest()
|
|
ns = {"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
|
|
with zipfile.ZipFile(source) as archive:
|
|
strings = ["".join(t.text or "" for t in item.iter("{" + ns["m"] + "}t"))
|
|
for item in ET.fromstring(archive.read("xl/sharedStrings.xml"))]
|
|
xml = ET.fromstring(archive.read("xl/worksheets/sheet1.xml"))
|
|
parsed = []
|
|
for row in xml.findall("m:sheetData/m:row", ns):
|
|
cells = {}
|
|
for cell in row:
|
|
value = cell.find("m:v", ns)
|
|
value = value.text if value is not None else "".join(cell.itertext())
|
|
cells["".join(filter(str.isalpha, cell.get("r")))] = strings[int(value)] if cell.get("t") == "s" else value
|
|
if any(cells.values()):
|
|
parsed.append((int(row.get("r")), cells))
|
|
expected = ["ID", "Provincia", "Distrito", "Posto_Administrativo", "Localidade", "Bairro_Urbano", "Aldeia_Quarteirao", "Designacao"]
|
|
if [parsed[0][1].get(c, "") for c in "ABCDEFGH"] != expected:
|
|
raise ValueError("Unexpected source columns; import stopped")
|
|
records = []
|
|
by_path = collections.defaultdict(list)
|
|
ids = set()
|
|
for row, cells in parsed[1:]:
|
|
raw = {key: cells.get(col, "") for col, key in zip("ABCDEFGH", expected)}
|
|
identifier = raw["ID"]
|
|
if not identifier.isdecimal() or identifier in ids:
|
|
raise ValueError("Missing or duplicate source ID; import stopped")
|
|
ids.add(identifier)
|
|
codes = [cells.get(c, "").strip() for c in "BCDEFG"]
|
|
positions = [i for i, code in enumerate(codes) if code]
|
|
level = max(positions) + 1 if positions else 0
|
|
issues = []
|
|
if not raw["Designacao"].strip():
|
|
issues.append("blank_designation")
|
|
if not level or positions != list(range(level)):
|
|
issues.append("incomplete_code_path")
|
|
record = {"id": identifier, "source_row": row, "raw": raw, "codes": codes,
|
|
"level": level, "name": raw["Designacao"], "parent_id": None, "issues": issues}
|
|
records.append(record)
|
|
by_path[tuple(codes)].append(record)
|
|
for record in records:
|
|
if len(by_path[tuple(record["codes"])]) > 1:
|
|
record["issues"].append("duplicate_code_path")
|
|
if record["level"] > 1:
|
|
parent = record["codes"].copy()
|
|
parent[record["level"] - 1] = ""
|
|
candidates = by_path.get(tuple(parent), [])
|
|
if len(candidates) == 1:
|
|
record["parent_id"] = candidates[0]["id"]
|
|
else:
|
|
record["issues"].append("missing_parent" if not candidates else "ambiguous_parent")
|
|
by_id = {r["id"]: r for r in records}
|
|
for record in sorted(records, key=lambda r: r["level"]):
|
|
parent = by_id.get(record["parent_id"])
|
|
if parent and not parent["selectable"]:
|
|
record["issues"].append("ancestor_requires_review")
|
|
record["selectable"] = not record["issues"]
|
|
manifest = {"schema_version": 1, "country": "MZ", "source_file": Path(source).name,
|
|
"source_sha256": digest, "source_sheet": "Sheet1", "edition": None,
|
|
"authority": "INE, conforme indicação do proprietário; edição não verificada",
|
|
"columns": expected, "records": len(records),
|
|
"selectable": sum(r["selectable"] for r in records),
|
|
"levels": dict(collections.Counter(r["level"] for r in records)),
|
|
"issues": dict(collections.Counter(issue for r in records for issue in r["issues"])),
|
|
"duplicate_code_groups": sum(len(v) > 1 for v in by_path.values())}
|
|
result = {"manifest": manifest, "records": records}
|
|
destination = Path(destination)
|
|
if destination.exists():
|
|
previous = json.loads(destination.read_text(encoding="utf-8"))
|
|
if previous["manifest"]["source_sha256"] != digest:
|
|
raise ValueError("Preserve previous catalogue: use a new destination for a new source revision")
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_text(json.dumps(result, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8")
|
|
print(json.dumps(manifest, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("source")
|
|
parser.add_argument("destination")
|
|
args = parser.parse_args()
|
|
prepare(args.source, args.destination)
|