Is your SAP data ready for AI? A readiness check you can run

"""Orchestrate course, Unit 7: profile SAP-shaped master data and write a readiness report.

Usage:
  python data_readiness.py --sample                     made-up products and suppliers (no account)
  python data_readiness.py --products p.csv --suppliers s.csv   your own CSV extracts
  python data_readiness.py --sandbox                    SAP Business Accelerator Hub sandbox (SAP_API_KEY in .env)

Writes readiness_report.md next to this script. Built-in modules only, except
--sandbox, which uses requests and python-dotenv from the Unit 1 setup.
"""
import argparse
import csv
import random
import re
import statistics
import sys
import unicodedata
from collections import Counter, defaultdict
from datetime import date
from difflib import SequenceMatcher
from pathlib import Path

HERE = Path(__file__).resolve().parent
DATA = HERE / "readiness_data"
REPORT = HERE / "readiness_report.md"
TODAY = date(2026, 10, 15)  # fixed so sample results are repeatable; use date.today() for real data

# 1. The readiness definition: what "good enough" means for ONE use case.
#    Use case: a procurement assistant that answers questions about products and suppliers.
#    Impact follows SAP MDG's scale: 1 low, 2 medium, 3 high.
USE_CASE = "Procurement assistant: answer questions about products and suppliers"
SPEC = {
    "products": {
        "key": "Product",
        "name": "ProductDescription",
        "complete": {"ProductDescription": 3, "ProductType": 3, "BaseUnit": 3, "ProductGroup": 2,
                     "GrossWeight": 1},
        "needed": ["Product", "ProductDescription", "ProductType", "ProductGroup", "BaseUnit",
                   "GrossWeight", "NetWeight", "WeightUnit", "IsMarkedForDeletion", "LastChangeDate"],
        "stale_years": 5,
    },
    "suppliers": {
        "key": "Supplier",
        "name": "SupplierName",
        "complete": {"SupplierName": 3, "Country": 3, "CityName": 2, "TaxNumber1|VATRegistration": 2,
                     "PostalCode": 1},
        "needed": ["Supplier", "SupplierName", "Country", "CityName", "PostalCode", "TaxNumber1",
                   "VATRegistration", "PurchasingIsBlocked", "PostingIsBlocked", "DeletionIndicator"],
        "personal": ["BirthDate", "IsNaturalPerson"],
    },
}
TO_KG = {"KG": 1.0, "G": 0.001, "LB": 0.45359237, "TO": 1000.0}  # your system's units may differ
TARGET, WARNING = 90.0, 75.0  # our thresholds, set with the business


def blank(value):
    return value is None or str(value).strip() == ""


def truthy(value):
    return str(value).strip().lower() in {"true", "x", "1", "yes"}


def number(value):
    try:
        return float(str(value).replace(",", "."))
    except ValueError:
        return None


def norm_name(text):
    """Lowercase, strip accents, punctuation and legal-form words, for duplicate matching."""
    text = unicodedata.normalize("NFKD", str(text)).encode("ascii", "ignore").decode().lower()
    text = re.sub(r"[^a-z0-9 ]", " ", text.replace(".", ""))
    stop = {"gmbh", "ag", "inc", "ltd", "llc", "co", "kg", "sa", "srl", "bv", "the", "and"}
    return " ".join(w for w in text.split() if w not in stop)


# 2. Sample data: SAP-shaped, made up, with problems planted on purpose.
def make_sample():
    rng = random.Random(7)
    groups = {"PUMPS": ("Centrifugal pump", 12, 60), "VALVES": ("Ball valve", 0.4, 6),
              "SEALS": ("Shaft seal", 0.05, 0.6), "MOTORS": ("Electric motor", 8, 45)}
    products = []
    for i in range(60):
        group = list(groups)[i % 4]
        label, low, high = groups[group]
        gross = round(rng.uniform(low, high), 2)
        products.append({
            "Product": f"TG-{1000 + i}", "ProductType": "FERT" if group != "SEALS" else "HAWA",
            "ProductGroup": group, "BaseUnit": "EA", "ProductDescription": f"{label} DN{20 + 5 * i}",
            "GrossWeight": gross, "NetWeight": round(gross * 0.9, 2), "WeightUnit": "KG",
            "IsMarkedForDeletion": "false",
            "LastChangeDate": f"{rng.randint(2022, 2026)}-{rng.randint(1, 12):02d}-15"})
    p = products
    p[3]["ProductDescription"] = ""                           # missing description
    p[7]["ProductDescription"] = ""
    p[11]["BaseUnit"] = ""                                    # missing base unit
    p[14]["WeightUnit"] = ""                                  # weight without unit
    p[19]["GrossWeight"], p[19]["NetWeight"] = 20.0, 22.5     # net heavier than gross
    p[24]["GrossWeight"], p[24]["NetWeight"], p[24]["WeightUnit"] = 25000, 22500, "G"  # grams: fine once converted
    p[28]["GrossWeight"], p[28]["NetWeight"] = 34000, 30600   # grams typed into a KG field
    p[30]["ProductGroup"] = ""                                # missing group
    p[34]["LastChangeDate"] = "2014-03-02"                    # stale
    p[38]["LastChangeDate"] = "2016-11-20"
    for i in range(2, 60, 8):
        p[i]["GrossWeight"] = p[i]["NetWeight"] = ""         # trading goods often lack weights
    p[42]["IsMarkedForDeletion"] = "true"                     # deleted: out of scope
    dup = dict(p[5], Product="TG-1100", ProductDescription=p[5]["ProductDescription"].upper())
    products.append(dup)                                      # same item created twice

    cities = [("DE", "Hamburg", "20095"), ("DE", "Munich", "80331"), ("US", "Houston", "77002"),
              ("FR", "Lyon", "69002"), ("IN", "Pune", "411001")]
    first = ["Nordwerk", "Baltic", "Gulf", "Rhone", "Deccan", "Alpen", "Lone Star", "Hanse"]
    second = ["Pumpen", "Valves", "Seal Supply", "Moteurs", "Castings"]
    suppliers = []
    for i in range(40):
        country, city, postal = cities[i % 5]
        base = f"{first[i % 8]} {second[i // 8]}"
        suppliers.append({
            "Supplier": f"{17300000 + i}", "SupplierName": base + (" GmbH" if country == "DE" else ""),
            "SupplierAccountGroup": "SUPL", "Country": country, "CityName": city, "PostalCode": postal,
            "TaxNumber1": "", "VATRegistration": f"{country}{rng.randint(100000000, 999999999)}",
            "IsNaturalPerson": "", "BirthDate": "", "PurchasingIsBlocked": "false",
            "PostingIsBlocked": "false", "DeletionIndicator": "false"})
    s = suppliers
    s[4]["Country"] = ""                                      # missing country
    s[9]["CityName"], s[9]["PostalCode"] = "", ""             # missing address
    s[13]["VATRegistration"] = ""                             # no tax number at all
    for i in range(24, 40, 5):
        s[i]["VATRegistration"], s[i]["TaxNumber1"] = "", ""  # one country's suppliers were loaded without tax data
    s[16]["PostalCode"] = "2009"                              # DE postal code too short
    s[21]["PurchasingIsBlocked"] = "true"                     # blocked: out of scope
    s[27]["DeletionIndicator"] = "true"
    s[33].update(SupplierName="Maria Keller", IsNaturalPerson="X", BirthDate="1979-04-12")  # a person
    suppliers.append(dict(s[0], Supplier="17300100", SupplierName="Nordwerk Pumpen G.m.b.H."))  # near duplicate
    suppliers.append(dict(s[18], Supplier="17300101", SupplierName=s[18]["SupplierName"] + " Co"))
    suppliers.append(dict(s[7], Supplier="17300102", SupplierName="Hanse Dichtungstechnik"))  # same tax number,
    # different name: a rename, a branch, or a typo. Only a person can tell.

    DATA.mkdir(exist_ok=True)
    for name, rows in (("products", products), ("suppliers", suppliers)):
        with open(DATA / f"{name}.csv", "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=list(rows[0]))
            writer.writeheader()
            writer.writerows(rows)
    return DATA / "products.csv", DATA / "suppliers.csv"


# 3. Optional: read the SAP sandbox (A_Product with descriptions, A_Supplier).
def fetch_sandbox():
    try:
        import requests
        from dotenv import load_dotenv
    except ImportError:
        sys.exit("--sandbox needs requests and python-dotenv: pip install requests python-dotenv")
    import os
    load_dotenv()
    key = os.environ.get("SAP_API_KEY")
    if not key:
        sys.exit("SAP_API_KEY is not set. Add it to .env, or run with --sample.")
    base = "https://sandbox.api.sap.com/s4hanacloud/sap/opu/odata/sap/"
    jobs = {
        "products": ("API_PRODUCT_SRV/A_Product", {
            "$select": "Product,ProductType,ProductGroup,BaseUnit,GrossWeight,NetWeight,WeightUnit,"
                       "IsMarkedForDeletion,LastChangeDate,to_Description/ProductDescription,"
                       "to_Description/Language",
            "$expand": "to_Description"}),
        "suppliers": ("API_BUSINESS_PARTNER/A_Supplier", {
            "$select": "Supplier,SupplierName,SupplierAccountGroup,TaxNumber1,VATRegistration,"
                       "IsNaturalPerson,BirthDate,PurchasingIsBlocked,PostingIsBlocked,DeletionIndicator"}),
    }
    DATA.mkdir(exist_ok=True)
    paths = []
    for name, (path, params) in jobs.items():
        params = {**params, "$top": "200", "$format": "json"}
        try:
            resp = requests.get(base + path, params=params, headers={"APIKey": key, "Accept": "application/json"},
                                timeout=60)
        except requests.RequestException as err:
            sys.exit(f"Could not reach sandbox.api.sap.com ({type(err).__name__}). Try --sample.")
        if resp.status_code != 200:
            sys.exit(f"Sandbox returned HTTP {resp.status_code} for {path}. Check SAP_API_KEY, or try --sample.")
        rows = []
        for item in resp.json()["d"]["results"]:
            item.pop("__metadata", None)
            texts = item.pop("to_Description", None)
            if texts is not None:
                english = [t["ProductDescription"] for t in texts.get("results", []) if t.get("Language") == "EN"]
                item["ProductDescription"] = english[0] if english else ""
            item = {k: ("" if v is None else v) for k, v in item.items()}
            for k, v in item.items():  # OData V2 dates arrive as /Date(ms)/
                match = re.fullmatch(r"/Date\((-?\d+)[^)]*\)/", str(v))
                if match:
                    item[k] = date.fromtimestamp(int(match.group(1)) / 1000).isoformat() if int(match.group(1)) > 0 else ""
            rows.append(item)
        if not rows:
            sys.exit(f"The sandbox returned no {name}.")
        out = DATA / f"sandbox_{name}.csv"
        with open(out, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=list(rows[0]))
            writer.writeheader()
            writer.writerows(rows)
        print(f"Saved {len(rows)} {name} to {out.name}")
        paths.append(out)
    return paths


# 4. Rules. Each rule returns (dimension, impact, rule name, ok keys, not-ok keys).
def load(path):
    try:
        with open(path, newline="", encoding="utf-8-sig") as f:
            return list(csv.DictReader(f))
    except FileNotFoundError:
        sys.exit(f"File not found: {path}")


def check_dataset(kind, rows):
    spec = SPEC[kind]
    key = spec["key"]
    columns = set(rows[0]) if rows else set()
    if key not in columns:
        sys.exit(f"{kind}: the key column {key} is missing. Columns found: {sorted(columns)}")
    findings = {"missing_columns": [], "out_of_scope": [], "outliers": [], "duplicates": [], "personal": {}}

    # Records that are deleted or blocked are excluded before scoring: the assistant must not use them.
    flags = [c for c in ("IsMarkedForDeletion", "DeletionIndicator", "PurchasingIsBlocked", "PostingIsBlocked")
             if c in columns]
    in_scope = []
    for r in rows:
        hit = [c for c in flags if truthy(r.get(c))]
        (findings["out_of_scope"].append((r[key], ", ".join(hit))) if hit else in_scope.append(r))
    rules = []

    def rule(dimension, impact, name, test):
        ok, bad = [], []
        for r in in_scope:
            verdict = test(r)
            if verdict is None:  # rule not applicable to this record
                continue
            (ok if verdict else bad).append(r[key])
        rules.append((dimension, impact, name, ok, bad))

    # Completeness: one rule per field the use case needs. "A|B" means either field counts.
    for field, impact in spec["complete"].items():
        options = field.split("|")
        present = [o for o in options if o in columns]
        if not present:
            findings["missing_columns"].append(field)
            rules.append(("Completeness", impact, f"{field} filled", [], [r[key] for r in in_scope]))
            continue
        rule("Completeness", impact, f"{field} filled", lambda r, p=present: any(not blank(r.get(o)) for o in p))

    if kind == "products":
        rule("Validity", 3, "Weight has a weight unit",
             lambda r: None if blank(r.get("GrossWeight")) else not blank(r.get("WeightUnit")))
        rule("Validity", 2, "Weight unit is known to this script",
             lambda r: None if blank(r.get("WeightUnit")) else r["WeightUnit"].upper() in TO_KG)
        rule("Consistency", 2, "Net weight not above gross weight",
             lambda r: None if None in (number(r.get("NetWeight") or ""), number(r.get("GrossWeight") or ""))
             else number(r["NetWeight"]) <= number(r["GrossWeight"]))
        if "LastChangeDate" in columns:
            limit = TODAY.year - spec["stale_years"]
            rule("Timeliness", 1, f"Changed since {limit}",
                 lambda r: None if blank(r.get("LastChangeDate")) else int(r["LastChangeDate"][:4]) >= limit)
        # Outliers: weight in kg, compared within the product group (robust modified z-score).
        by_group = defaultdict(list)
        for r in in_scope:
            gross, unit = number(r.get("GrossWeight") or ""), str(r.get("WeightUnit", "")).upper()
            if gross is not None and unit in TO_KG:
                by_group[r.get("ProductGroup") or "(none)"].append((r[key], gross * TO_KG[unit]))
        flagged = set()
        for group, items in by_group.items():
            if len(items) < 5:
                continue
            values = [v for _, v in items]
            med = statistics.median(values)
            mad = statistics.median(abs(v - med) for v in values)
            for k, v in items:
                score = 0.6745 * (v - med) / mad if mad else 0.0
                if abs(score) > 3.5:
                    flagged.add(k)
                    findings["outliers"].append((k, group, round(v, 2), round(score, 1)))
        rule("Accuracy", 2, "Weight plausible within its group", lambda r: r[key] not in flagged)

    if kind == "suppliers":
        rule("Validity", 1, "DE postal code has 5 digits",
             lambda r: None if r.get("Country") != "DE" or blank(r.get("PostalCode"))
             else bool(re.fullmatch(r"\d{5}", r["PostalCode"].strip())))
        tax_field = "VATRegistration" if "VATRegistration" in columns else None
        if tax_field:
            counts = Counter(r[tax_field] for r in in_scope if not blank(r.get(tax_field)))
            rule("Uniqueness", 3, "Tax number used by one supplier only",
                 lambda r: None if blank(r.get(tax_field)) else counts[r[tax_field]] == 1)

    # Uniqueness by name: near-identical names inside the same group or country.
    block = "ProductGroup" if kind == "products" else "Country"
    buckets = defaultdict(list)
    for r in in_scope:
        if not blank(r.get(spec["name"])):
            buckets[r.get(block, "")].append((r[key], r[spec["name"]], norm_name(r[spec["name"]])))
    dup_keys = set()
    for items in buckets.values():
        for i in range(len(items)):
            for j in range(i + 1, len(items)):
                if re.findall(r"\d+", items[i][2]) != re.findall(r"\d+", items[j][2]):
                    continue  # "pump DN20" and "pump DN40" are different products
                ratio = SequenceMatcher(None, items[i][2], items[j][2]).ratio()
                if ratio >= 0.9:
                    findings["duplicates"].append((items[i][0], items[j][0], items[i][1], items[j][1], round(ratio, 2)))
                    dup_keys.update({items[i][0], items[j][0]})
    rule("Uniqueness", 3, "Name not a near-duplicate of another record", lambda r: r[key] not in dup_keys)

    # Personal data: which personal fields are present, filled, and needed by the use case.
    for field in spec.get("personal", []):
        if field in columns:
            filled = [r[key] for r in rows if not blank(r.get(field))]
            findings["personal"][field] = (filled, field in spec["needed"])
    findings["unneeded_columns"] = sorted(c for c in columns if c not in spec["needed"])
    return rules, findings, len(rows), len(in_scope)


# 5. Scores, the MDG way: rule score = OK / (OK + Not OK); dimension = impact-weighted average.
def score(rules):
    dims = defaultdict(list)
    for dimension, impact, name, ok, bad in rules:
        if ok or bad:
            dims[dimension].append((impact, 100.0 * len(ok) / (len(ok) + len(bad))))
    result = {}
    for dimension, items in dims.items():
        weight = sum(i for i, _ in items)
        result[dimension] = sum(i * s for i, s in items) / weight if weight else None
    overall = statistics.mean(v for v in result.values() if v is not None) if result else 0.0
    return result, overall


def verdict(value, rules):
    """An average can hide a blocker, so any high-impact rule below target also means 'fix first'."""
    blockers = [name for _, impact, name, ok, bad in rules if impact == 3 and bad
                and 100 * len(ok) / (len(ok) + len(bad)) < TARGET]
    if value < WARNING:
        return "Not ready"
    if value < TARGET or blockers:
        return "Ready after fixes"
    return "Ready"


def write_report(results, source):
    lines = [f"# Data readiness report", "",
             f"Use case: {USE_CASE}  ", f"Data: {source}  ", f"Run: {TODAY.isoformat()}  ",
             f"Thresholds: target {TARGET:.0f}, warning {WARNING:.0f} (agreed with the business)", ""]
    lines += ["## Scores", "", "| Dataset | Records | In scope | " + " | ".join(DIMENSIONS) + " | Overall | Verdict |",
              "| --- | ---: | ---: | " + " | ".join("---:" for _ in DIMENSIONS) + " | ---: | --- |"]
    for kind, (rules, findings, total, scope, dims, overall) in results.items():
        cells = [f"{dims[d]:.0f}" if dims.get(d) is not None else "n/a" for d in DIMENSIONS]
        lines.append(f"| {kind} | {total} | {scope} | " + " | ".join(cells) + f" | {overall:.0f} | {verdict(overall, rules)} |")
    for kind, (rules, findings, total, scope, dims, overall) in results.items():
        lines += ["", f"## {kind.capitalize()}: what to fix", "", "| Rule | Dimension | Impact | Score | Failing records (first 5) |",
                  "| --- | --- | ---: | ---: | --- |"]
        for dimension, impact, name, ok, bad in sorted(rules, key=lambda x: (-x[1], len(x[3]) / max(1, len(x[3]) + len(x[4])))):
            if bad:
                lines.append(f"| {name} | {dimension} | {impact} | {100 * len(ok) / (len(ok) + len(bad)):.0f} | {', '.join(bad[:5])} |")
        notes = []
        if findings["missing_columns"]:
            notes.append(f"Columns the use case needs but the extract lacks: {', '.join(findings['missing_columns'])}.")
        for a, b, na, nb, ratio in findings["duplicates"][:5]:
            notes.append(f"Possible duplicate: {a} \"{na}\" and {b} \"{nb}\" (name similarity {ratio}).")
        for k, group, kg, z in findings["outliers"][:5]:
            notes.append(f"Unusual weight: {k} in group {group} weighs {kg} kg (modified z-score {z}).")
        if findings["out_of_scope"]:
            notes.append(f"Excluded as deleted or blocked: {len(findings['out_of_scope'])} "
                         f"({', '.join(f'{k} [{why}]' for k, why in findings['out_of_scope'][:5])}).")
        for field, (filled, needed) in findings["personal"].items():
            if filled:
                notes.append(f"Personal data: {field} is filled for {len(filled)} record(s) ({', '.join(filled[:5])}). "
                             + ("The use case needs it." if needed else "The use case does not need it: drop it from the extract."))
        if "IsNaturalPerson" in findings["personal"] and findings["personal"]["IsNaturalPerson"][0]:
            notes.append("Records flagged as natural persons: the name and address are personal data too. "
                         "Agree purpose, access and retention with your data protection officer.")
        if findings["unneeded_columns"]:
            notes.append(f"Columns the use case does not need (leave them out of the AI's view): "
                         f"{', '.join(findings['unneeded_columns'])}.")
        if notes:
            lines += [""] + [f"- {n}" for n in notes]
    lines += ["", "## Fill in by hand (profiling can't answer these)", "",
              "| Question | Answer |", "| --- | --- |",
              "| Source system and release (S/4HANA, ECC, BW, other) | |",
              "| Access path (released API, released CDS view, data product, extraction tool) | |",
              "| Business owner of each dataset | |",
              "| Who may see which records (authorizations to respect) | |",
              "| Personal data: purpose, retention and blocking rules | |",
              "| How fixes get back into SAP (MDG, mass change, owner) | |"]
    REPORT.write_text("\n".join(lines) + "\n", encoding="utf-8")


DIMENSIONS = ["Completeness", "Validity", "Consistency", "Uniqueness", "Accuracy", "Timeliness"]


def main():
    parser = argparse.ArgumentParser(description="Profile SAP-shaped master data for one AI use case.")
    parser.add_argument("--sample", action="store_true", help="write and profile made-up sample data")
    parser.add_argument("--sandbox", action="store_true", help="read products and suppliers from the SAP sandbox")
    parser.add_argument("--products", help="CSV with product master data")
    parser.add_argument("--suppliers", help="CSV with supplier master data")
    args = parser.parse_args()

    if args.sample:
        products, suppliers = make_sample()
        source = "made-up sample (readiness_data/)"
    elif args.sandbox:
        products, suppliers = fetch_sandbox()
        source = "SAP Business Accelerator Hub sandbox"
    elif args.products or args.suppliers:
        products, suppliers = args.products, args.suppliers
        source = ", ".join(p for p in (products, suppliers) if p)
    else:
        parser.print_help()
        sys.exit("\nChoose --sample, --sandbox, or give --products and/or --suppliers.")

    results = {}
    for kind, path in (("products", products), ("suppliers", suppliers)):
        if not path:
            continue
        rows = load(path)
        if not rows:
            print(f"{kind}: the file has no rows, nothing to profile.")
            continue
        rules, findings, total, scope = check_dataset(kind, rows)
        dims, overall = score(rules)
        results[kind] = (rules, findings, total, scope, dims, overall)
        print(f"\n{kind}: {total} records, {scope} in scope")
        for d in DIMENSIONS:
            if dims.get(d) is not None:
                print(f"  {d:<13}{dims[d]:6.1f}")
        print(f"  {'Overall':<13}{overall:6.1f}  -> {verdict(overall, rules)}")
    if results:
        write_report(results, source)
        print(f"\nReport written to {REPORT.name}")


if __name__ == "__main__":
    main()

python data_readiness.py --sample

products: 61 records, 60 in scope
  Completeness   97.5
  Validity       98.9
  Consistency    98.1
  Uniqueness     96.7
  Accuracy       96.7
  Timeliness     96.7
  Overall        97.4  -> Ready

suppliers: 43 records, 41 in scope
  Completeness   96.5
  Validity       93.8
  Uniqueness     86.8
  Overall        92.3  -> Ready after fixes

Report written to readiness_report.md

cd ..
git add unit07/data_readiness.py unit07/readiness_report.md
git commit -m "Unit 7: data readiness report"