sqlinsight-pro/installer/msi-tables.py

#!/usr/bin/env python3
"""Bridges what wixl does not support, so installer/SQLInsight.wxs stays the single source (WiX 3 syntax).

  msi-tables.py strip <in.wxs> <out.wxs>   Removes Component and Feature conditions and WixVariable, which wixl cannot parse.
  msi-tables.py apply <in.wxs> <file.msi>  Writes them, Permanent/NeverOverwrite and ServiceInstall/@Vital into the MSI.
"""
import os
import subprocess
import sys
import tempfile
import xml.etree.ElementTree as ET

NS = "{http://schemas.microsoft.com/wix/2006/wi}"
PERMANENT = 0x10
ERROR_CONTROL_VITAL = 0x8000
NEVER_OVERWRITE = 0x80


def components(wxs):
    tree = ET.parse(wxs)
    for component in tree.iter(NS + "Component"):
        condition = component.find(NS + "Condition")
        yield tree, component, (condition.text or "").strip() if condition is not None else None, condition


def strip(source, target):
    tree = ET.parse(source)
    ET.register_namespace("", NS[1:-1])
    for feature in tree.iter(NS + "Feature"):
        for condition in feature.findall(NS + "Condition"):
            feature.remove(condition)
    for component in tree.iter(NS + "Component"):
        for condition in component.findall(NS + "Condition"):
            component.remove(condition)
    # WixVariable (license text for WiX 3's UI) is not known to wixl, which reads License.rtf from the working folder.
    for parent in tree.iter():
        for variable in parent.findall(NS + "WixVariable"):
            parent.remove(variable)
    tree.write(target, encoding="utf-8", xml_declaration=True)


def dump_table(msi, name):
    with tempfile.TemporaryDirectory() as folder:
        subprocess.run(["msidump", "-t", "-d", folder, os.path.abspath(msi)], check=True, capture_output=True, cwd=folder)
        with open(os.path.join(folder, f"{name}.idt"), encoding="utf-8", errors="replace") as idt:
            lines = idt.read().splitlines()
    header = lines[0].split("\t")
    return {row.split("\t")[0]: dict(zip(header, row.split("\t"))) for row in lines[3:] if row}


def tree_of(wxs):
    return ET.parse(wxs)


def apply(source, msi):
    table = dump_table(msi, "Component")
    queries = []
    for _, component, condition, _ in components(source):
        cid = component.get("Id")
        if cid not in table:
            sys.exit(f"Komponenten {cid} findes ikke i {msi}")
        attributes = int(table[cid]["Attributes"] or 0)
        if component.get("Permanent") == "yes":
            attributes |= PERMANENT
        if component.get("NeverOverwrite") == "yes":
            attributes |= NEVER_OVERWRITE
        sets = [f"`Attributes` = {attributes}"]
        if condition:
            if "'" in condition:
                sys.exit(f"Betingelsen for {cid} må ikke indeholde apostroffer")
            sets.append(f"`Condition` = '{condition}'")
        if attributes != int(table[cid]["Attributes"] or 0) or condition:
            queries.append(f"UPDATE `Component` SET {', '.join(sets)} WHERE `Component` = '{cid}'")
    # RegistryValue/@Type="expandable" becomes REG_EXPAND_SZ only when the stored value starts with "#%".
    registry = dump_table(msi, "Registry")
    for value in tree_of(source).iter(NS + "RegistryValue"):
        if value.get("Type") == "expandable":
            rows = [r for r in registry.values() if r["Name"] == value.get("Name") and r["Value"] == value.get("Value")]
            if len(rows) != 1:
                sys.exit(f"Registry-værdien {value.get('Name')} blev ikke fundet i {msi}")
            queries.append(f"UPDATE `Registry` SET `Value` = '#%{value.get('Value')}' WHERE `Registry` = '{rows[0]['Registry']}'")

    # Feature/Condition: the Condition table, which wixl does not write.
    feature_conditions = [(f.get("Id"), c.get("Level"), (c.text or "").strip())
                          for f in tree_of(source).iter(NS + "Feature") for c in f.findall(NS + "Condition")]
    if feature_conditions:
        queries.append("CREATE TABLE `Condition` (`Feature_` CHAR(38) NOT NULL, `Level` SHORT NOT NULL, `Condition` CHAR(255) PRIMARY KEY `Feature_`, `Level`)")
        for feature, level, condition in feature_conditions:
            queries.append(f"INSERT INTO `Condition` (`Feature_`, `Level`, `Condition`) VALUES ('{feature}', {int(level)}, '{condition}')")

    # ServiceInstall/@Vital: a service that cannot be installed (wrong account …) must fail the installation.
    for service in tree_of(source).iter(NS + "ServiceInstall"):
        if service.get("Vital") == "yes":
            control = {"ignore": 0, "normal": 1, "critical": 3}[service.get("ErrorControl", "normal")]
            queries.append(f"UPDATE `ServiceInstall` SET `ErrorControl` = {control | ERROR_CONTROL_VITAL} WHERE `ServiceInstall` = '{service.get('Id')}'")
    for query in queries:
        subprocess.run(["msibuild", msi, "-q", query], check=True)
    print(f"{len(queries)} rækker opdateret")


if __name__ == "__main__":
    {"strip": strip, "apply": apply}[sys.argv[1]](sys.argv[2], sys.argv[3])