data-meta-model/scripts/migrate_instances_v1_3.py

#!/usr/bin/env python3
"""
SODH - migration v1.2 -> v1.3
==============================
USAGE
    python3 scripts/migrate_instances_v1_3.py            # dry run
    python3 scripts/migrate_instances_v1_3.py --apply    # rewrite, .bak kept

1. references -> qualifiedBy. The name now says why the link exists rather
   than how it is implemented.

2. hasSource removed from every Business Object. Same argument as for the
   Business Concepts in v1.2, and the values showed why it was needed: across
   the 15 objects the property held physical locations ("SODH Gold /
   F_SELL_OUT"), migration notes ("MDM Product -> SODH [concepts TBC by
   DD-10]") and arbitration remarks ("DGO proposal, pending ratification"),
   all under one name. A Business Object is a governance grouping; its data
   lives in the Data Objects that represent it, and those carry the system.

3. Short labels where the long name is unwieldy on screen. The Data Product
   was called "Sell Out Data Hub - Gold" on its own card and "SODH Gold"
   everywhere else, because the viewers each kept a display dictionary.

4. Instance versions aligned on the model version. Business Objects were
   still stamped 0.6 while the model had moved to 1.3.
"""
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

SHORT_LABELS = {
    "ex:DP_06_01_001": "SODH Gold",
    "ex:DC_06_01_001": "SODH Gold contract",
    "ex:DI_06_01_001": "Consumption schema",
    "ex:DI_06_01_002": "Panel extract",
    "ex:DO_06_01_001": "Sell Out Volume",
    "ex:DO_06_01_003": "Sell Out Distribution",
    "ex:DO_10_01_001": "Product",
    "ex:DO_04_01_001": "Customer",
    "ex:DO_16_01_001": "Currency",
    "ex:DO_21_01_001": "Calendar",
}
MODEL_VERSION = "1.3"


def clauses(body):
    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):
    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 clauses(m.group(2))
            if c.strip() and not re.match(r'\s*%s\s' % re.escape(prop), c)]
    if value is not None:
        kept.insert(0, "\n    %s %s " % (prop, value))
    body = " ;".join(kept).rstrip() + "\n    "
    return text[:m.start()] + m.group(1) + body + m.group(3) + text[m.end():], True


def has_prop(text, subject, prop):
    m = re.search(r'^%s a pr:\w+ ;(.*?)\.\s*\n' % re.escape(subject), text, re.S | re.M)
    return bool(m) and any(re.match(r'\s*%s\s' % re.escape(prop), c)
                           for c in clauses(m.group(1)))


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 = []

    text, n = re.subn(r'\bpr:references\b', 'pr:qualifiedBy', text)
    report.append("references -> qualifiedBy: %d occurrence(s)" % n)

    n_bo = 0
    for m in list(re.finditer(r'^(ex:BO_\w+) a pr:BusinessObject ;', text, re.M)):
        if has_prop(text, m.group(1), "pr:hasSource"):
            text, _ = set_prop(text, m.group(1), "pr:hasSource", None)
            n_bo += 1
    report.append("hasSource removed from %d business objects" % n_bo)

    n_sl = 0
    for subj, label in SHORT_LABELS.items():
        if re.search(r'^%s a ' % re.escape(subj), text, re.M):
            text, ok = set_prop(text, subj, "pr:hasShortLabel", '"%s"' % label)
            n_sl += ok
    report.append("short label set on %d objects" % n_sl)

    n_v = 0
    for m in list(re.finditer(r'^(ex:\w+) a pr:\w+ ;', text, re.M)):
        if has_prop(text, m.group(1), "pr:hasVersion"):
            text, ok = set_prop(text, m.group(1), "pr:hasVersion",
                                '"%s"' % MODEL_VERSION)
            n_v += ok
    report.append("version aligned on %s for %d objects" % (MODEL_VERSION, n_v))

    print()
    print("SODH MIGRATION v1.2 -> v1.3   %s" % ("APPLY" if apply_changes else "DRY RUN"))
    print("=" * W)
    for line in report:
        print("  " + line)
    print("=" * W)
    left_bo = len([1 for m in re.finditer(r'^(ex:BO_\w+) a pr:BusinessObject ;', text, re.M)
                   if has_prop(text, m.group(1), "pr:hasSource")])
    doubled = []
    for m in re.finditer(r'^(ex:\w+) a pr:\w+ ;(.*?)\.\s*\n', text, re.S | re.M):
        for p in ("pr:hasName", "pr:hasShortLabel", "pr:hasVersion", "pr:hasSource"):
            if len(re.findall(r'(?<![\w:])%s\s' % re.escape(p), m.group(2))) > 1:
                doubled.append("%s/%s" % (m.group(1), p))
    print("  business objects still carrying a source: %d   (must be 0)" % left_bo)
    print("  references left: %d   (must be 0)" % len(re.findall(r'pr:references', text)))
    print("  duplicated single-valued properties: %s"
          % (", ".join(doubled) if doubled else "none"))
    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()