#!/usr/bin/env python3 """ SODH - structure proposed for the borrowed domains =================================================== Gives the nine TO_ARBITRATE concepts a home: a Business Object and a Sub-Domain each, per the structure arbitrated with Bastien on 2026-07-27. USAGE python3 scripts/apply_structure_sodh.py # dry run python3 scripts/apply_structure_sodh.py --apply # rewrite, .bak kept STILL A PROPOSAL Everything created here stays TO_ARBITRATE. DD-04, DD-10, DD-16 and DD-21 own these perimeters; the DGO is proposing a coherent structure, not ratifying one. Replacing padding with an equally unilateral structure would trade one problem for a tidier version of the same problem. The flag comes off when Helene Puchot, Anas El Kesri and Gaelle Seret say so, and OW-007 blocks publication until then. IDENTIFIERS ARE FINAL FROM CREATION Identifiers encode the sub-domain (BO-04.02-001 lives in SD-04.02), so the upstream scripts now mint them correctly rather than having this one renumber afterwards. Renaming after creation broke idempotence: the script that created an object stopped recognising its own output and made a second copy on the next run. COMPOUND SUB-DOMAINS SPLIT TOO 'Customer & Outlet' and 'Marketing Entity & Fiscal Calendar' have the same defect as the Business Objects that were split earlier: a name joined by '&' has no single subject. Same rule, one level up. """ import os import re import shutil import sys REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) TTL = os.path.join(REPO, "instances", "sodh.ttl") W = 78 # No IRI renaming here. Objects are created with their final identifiers by # the upstream scripts: renumbering after the fact broke idempotence, because # the creating script then no longer recognised its own output. # ---- sub-domains ---------------------------------------------------------- SD_RENAME = { "ex:SD_04_01": "Customer Structure", "ex:SD_16_01": "Currency & Exchange Rate", "ex:SD_16_02": "Fiscal Calendar", "ex:SD_21_01": "Standard Calendar", } SD_NEW = [ ("ex:SD_04_02", "SD-04.02", "Outlet & Points of Sale", "ex:DD_04"), ("ex:SD_04_03", "SD-04.03", "Geography", "ex:DD_04"), ("ex:SD_16_03", "SD-16.03", "Organisation Structure", "ex:DD_16"), ] # ---- business objects ----------------------------------------------------- BO_RENAME = { "ex:BO_10_01_001": "Product Hierarchy", "ex:BO_04_01_001": "Customer Hierarchy", "ex:BO_21_01_001": "Analysis Period", } BO_NEW = [ ("ex:BO_04_03_001", "BO-04.03-001", "Country", "ex:DD_04", "ex:ST_DD_04", "ex:SD_04_03", "ex:BC_04_03_001"), ("ex:BO_16_02_001", "BO-16.02-001", "Fiscal Period", "ex:DD_16", "ex:ST_DD_16", "ex:SD_16_02", "ex:BC_16_02_001"), ("ex:BO_16_03_001", "BO-16.03-001", "Marketing Entity", "ex:DD_16", "ex:ST_DD_16", "ex:SD_16_03", "ex:BC_16_03_001"), ] # ---- where each object finally sits --------------------------------------- BELONGS = { "ex:BO_10_01_001": "ex:SD_10_01", "ex:BO_04_01_001": "ex:SD_04_01", "ex:BO_04_02_001": "ex:SD_04_02", "ex:BO_04_03_001": "ex:SD_04_03", "ex:BO_16_01_001": "ex:SD_16_01", "ex:BO_16_01_002": "ex:SD_16_01", "ex:BO_16_02_001": "ex:SD_16_02", "ex:BO_16_03_001": "ex:SD_16_03", "ex:BO_21_01_001": "ex:SD_21_01", } # ---- subject of each borrowed object -------------------------------------- ABOUT = { "ex:BO_10_01_001": "ex:BC_10_01_001", # Product Hierarchy -> Product "ex:BO_04_01_001": "ex:BC_04_01_001", # Customer Hierarchy -> Customer "ex:BO_04_02_001": "ex:BC_04_01_002", # Outlet -> Outlet "ex:BO_04_03_001": "ex:BC_04_03_001", # Country -> Country "ex:BO_16_01_001": "ex:BC_16_01_001", # Currency -> Currency "ex:BO_16_01_002": "ex:BC_16_01_002", # Exchange Rate -> Exchange Rate "ex:BO_16_02_001": "ex:BC_16_02_001", # Fiscal Period -> Fiscal Period "ex:BO_16_03_001": "ex:BC_16_03_001", # Marketing Entity -> Marketing Entity "ex:BO_21_01_001": "ex:BC_21_01_001", # Analysis Period -> Calendar Date } MARKETING_NOTE = ("Named Marketing Entity, not Management Entity: the element is " "Marketing Entity Code and the column is MARKETING_ENTITY_CD. " "To be confirmed by DD-16.") def sd_block(i, ident, name, domain): return ('%s a pr:SubDomain ;\n' ' pr:hasIdentifier "%s" ; pr:hasName "%s" ;\n' ' pr:belongsTo %s ; pr:owningDomain %s ;\n' ' pr:arbitrationStatus "TO_ARBITRATE" ;\n' ' pr:hasStatus "DRAFT" ; pr:hasVersion "1.1" ; ' 'pr:hasSource "DGO proposal, split of a compound sub-domain" .' % (i, ident, name, domain, domain)) def bo_block(i, ident, name, domain, steward, sd, bc): return ('%s a pr:BusinessObject ;\n' ' pr:hasIdentifier "%s" ; pr:hasName "%s" ;\n' ' pr:owningDomain %s ; pr:ownedBy %s ;\n' ' pr:belongsTo %s ; pr:monitoredBy %s ;\n' ' pr:aboutConcept %s ;\n' ' pr:arbitrationStatus "TO_ARBITRATE" ;\n' ' pr:hasStatus "DRAFT" ; pr:hasVersion "1.1" ; ' 'pr:hasSource "DGO proposal, pending ratification by the owning domain" .' % (i, ident, name, domain, steward, sd, steward, bc)) def _split_clauses(body): """Split a Turtle predicate list on ';' that sit outside string literals.""" out, buf, i, n = [], [], 0, len(body) while i < n: c = body[i] if c == '"': j = i + 1 while j < n and body[j] != '"': j += 2 if body[j] == "\\" else 1 j = min(j + 1, n) buf.append(body[i:j]); i = j; continue if c == ";": out.append("".join(buf)); buf = []; i += 1; continue buf.append(c); i += 1 out.append("".join(buf)) return out def set_prop(text, subject, prop, value): """ Set or replace a single-valued property inside a subject block. Clause-level, not regex-level: a property can sit mid-line (pr:hasIdentifier "X" ; pr:hasName "Y" ;) and its value can contain a semicolon inside the literal. A line-anchored pattern silently misses both and APPENDS instead of replacing, which produces a second value and a cardinality violation -- exactly what happened on the first run here. """ pat = re.compile(r'(^%s a pr:\w+ ;)(.*?)(\.\s*\n)' % re.escape(subject), re.S | re.M) m = pat.search(text) if not m: return text, False kept = [c for c in _split_clauses(m.group(2)) if c.strip() and not re.match(r'\s*%s\s' % re.escape(prop), c)] kept.insert(0, "\n %s %s " % (prop, value)) body = " ;".join(kept) if not body.endswith("\n"): body = body.rstrip() + "\n " return text[:m.start()] + m.group(1) + body + m.group(3) + text[m.end():], True def main(): apply_changes = "--apply" in sys.argv if not os.path.exists(TTL): print("Not found: %s" % TTL) sys.exit(2) text = original = open(TTL, encoding="utf-8").read() report = [] # 2. sub-domains for sd, name in SD_RENAME.items(): text, ok = set_prop(text, sd, "pr:hasName", '"%s"' % name) if ok: report.append("renamed %s -> %s" % (sd.replace("ex:SD_", "SD-"), name)) todo_sd = [s for s in SD_NEW if not re.search(r'^%s a ' % re.escape(s[0]), text, re.M)] if todo_sd: anchor = re.search(r'\n(?=ex:BC_\w+ a pr:BusinessConcept ;|ex:BO_\w+ a pr:BusinessObject ;)', text) blocks = "\n".join(sd_block(*s) for s in todo_sd) text = (text[:anchor.start()] + "\n\n" + "# --- sub-domains from the v1.1 structure proposal --------------------\n" + "# 'Customer & Outlet' and 'Marketing Entity & Fiscal Calendar' had the\n" + "# same defect as the compound Business Objects: a name joined by '&'\n" + "# has no single subject. Same rule, one level up.\n" + blocks + "\n" + text[anchor.start():]) report.append("created %d sub-domains%s" % (len(todo_sd), "" if todo_sd else " (already present)")) # 3. new business objects todo_bo = [b for b in BO_NEW if not re.search(r'^%s a ' % re.escape(b[0]), text, re.M)] if todo_bo: anchor = re.search(r'\n(?=ex:DO_\w+ a pr:DataObject ;|ex:M_\w+ a pr:Metric ;)', text) blocks = "\n".join(bo_block(*b) for b in todo_bo) text = (text[:anchor.start()] + "\n\n" + "# --- business objects for the borrowed concepts -----------------------\n" + "# Country, Fiscal Period and Marketing Entity had a concept but nothing\n" + "# to hold it. Proposed to DD-04 and DD-16, hence TO_ARBITRATE.\n" + blocks + "\n" + text[anchor.start():]) report.append("created %d business objects%s" % (len(todo_bo), "" if todo_bo else " (already present)")) # 4. renames, placement, subject for bo, name in BO_RENAME.items(): text, _ = set_prop(text, bo, "pr:hasName", '"%s"' % name) report.append("renamed %d business objects to the agreed nomenclature" % len(BO_RENAME)) n_b = n_a = 0 for bo, sd in BELONGS.items(): text, ok = set_prop(text, bo, "pr:belongsTo", sd) n_b += ok for bo, bc in ABOUT.items(): text, ok = set_prop(text, bo, "pr:aboutConcept", bc) n_a += ok report.append("placed %d objects in their sub-domain, %d subjects confirmed" % (n_b, n_a)) # 5. the naming caveat, recorded on the object rather than in a side note text, ok = set_prop(text, "ex:BC_16_03_001", "pr:hasBusinessRule", '"%s"' % MARKETING_NOTE) if ok: report.append("recorded the Marketing/Management naming caveat on the concept") # ---- report print() print("SODH STRUCTURE PROPOSAL %s" % ("APPLY" if apply_changes else "DRY RUN")) print("=" * W) for line in report: print(" " + line) print("=" * W) defined = set(re.findall(r'^(ex:\w+) a ', text, re.M)) dangling = sorted(set(re.findall(r'\bex:\w+', text)) - defined) print(" dangling references: %s" % (", ".join(dangling) if dangling else "none")) for cls in ("SubDomain", "BusinessObject", "BusinessConcept"): ids = re.findall(r'^(ex:\w+) a pr:%s\b' % cls, text, re.M) dup = [i for i in set(ids) if ids.count(i) > 1] print(" %-16s %3d duplicates %d" % (cls, len(ids), len(dup))) doubled = [] for m in re.finditer(r'^(ex:\w+) a pr:\w+ ;(.*?)\.\s*\n', text, re.S | re.M): for prop in ("pr:hasName", "pr:hasIdentifier", "pr:belongsTo", "pr:aboutConcept"): if len(re.findall(r'%s\s' % re.escape(prop), m.group(2))) > 1: doubled.append("%s/%s" % (m.group(1), prop)) print(" duplicated single-valued properties: %s" % (", ".join(doubled) if doubled else "none")) arb = len(re.findall(r'pr:arbitrationStatus "TO_ARBITRATE"', text)) print(" TO_ARBITRATE %3d (nothing publishes until the domains ratify)" % arb) print("=" * W) if text != original and apply_changes: shutil.copy2(TTL, TTL + ".bak") open(TTL, "w", encoding="utf-8").write(text) print(" written, backup at %s.bak" % os.path.basename(TTL)) elif text != original: print(" dry run -- re-run with --apply to write") print() if __name__ == "__main__": main()