shopdb-flask/plugins/geenforce/importer.py

"""Parse on-share manifest.json / preinstall.json into draft DB rows.

The reverse of serializer.py. `build_scope` turns a parsed manifest dict into a
ManifestScope with its ordered entries and child rows (not yet committed).
`discover_share` walks a GE-Enforce share root and yields (scopename, phase,
manifest_dict) for common + every gea-shopfloor-* runtime manifest, skipping
`.bak` variants. `load_preinstall` reads a preinstall.json.

Import is a rebuild: the caller replaces a scope's draft rows so re-running is
idempotent.
"""

import json
import os

from .models import (
    ManifestScope, ManifestEntry, ManifestEntryPcType, ManifestEntryHostname,
    ManifestEntryMachineNumber, ManifestInUseCheck, ManifestInUseCheckProcess,
)
from .serializer import _SCALAR_FIELDS, _BOOL_FLAGS

# manifest key -> model attr (reverse of the serializer's scalar map).
_KEY_TO_ATTR = {mk: attr for attr, mk in _SCALAR_FIELDS}
_FLAG_TO_ATTR = {mk: attr for attr, mk in _BOOL_FLAGS}


def build_entry(entry_dict, sortorder):
    """Build a ManifestEntry (+ children) from one Applications[] entry."""
    entry = ManifestEntry(sortorder=sortorder,
                          name=entry_dict.get('Name'),
                          entrytype=entry_dict.get('Type'))
    if '_comment' in entry_dict:
        entry.comment = entry_dict['_comment']
    for key, attr in _KEY_TO_ATTR.items():
        if attr in ('name', 'entrytype'):
            continue
        if key in entry_dict and entry_dict[key] is not None:
            setattr(entry, attr, entry_dict[key])
    # RegValue stored as its raw JSON literal so DWord vs string typing survives.
    if 'RegValue' in entry_dict:
        entry.regvalue = json.dumps(entry_dict['RegValue'])
    for key, attr in _FLAG_TO_ATTR.items():
        if entry_dict.get(key):
            setattr(entry, attr, True)
    # Multi-value filters -> child rows (preserve order).
    for i, value in enumerate(entry_dict.get('PCTypes') or []):
        entry.pctypes.append(ManifestEntryPcType(sortorder=i, pctypevalue=value))
    for i, value in enumerate(entry_dict.get('TargetHostnames') or []):
        entry.hostnames.append(
            ManifestEntryHostname(sortorder=i, hostnamepattern=value))
    for i, value in enumerate(entry_dict.get('TargetMachineNumbers') or []):
        entry.machinenumbers.append(
            ManifestEntryMachineNumber(sortorder=i, machinenumber=str(value)))
    # InUseCheck object + Processes[].
    inuse = entry_dict.get('InUseCheck')
    if inuse:
        check = ManifestInUseCheck(behavior=inuse.get('Behavior'))
        for i, proc in enumerate(inuse.get('Processes') or []):
            check.processes.append(ManifestInUseCheckProcess(
                sortorder=i,
                processname=proc.get('Name'),
                exepath=proc.get('ExePath'),
                gracefulclosetimeoutsec=proc.get('GracefulCloseTimeoutSec')))
        entry.inusecheck = check
    return entry


def build_scope(scopename, phase, manifest_dict, iscommon=False):
    """Build a ManifestScope (+ entries) from a parsed manifest dict."""
    scope = ManifestScope(
        scopename=scopename,
        phase=phase,
        manifestversion=str(manifest_dict.get('Version', '1.0')),
        topcomment=manifest_dict.get('_comment'),
        site=manifest_dict.get('Site'),
        iscommon=iscommon)
    for i, entry_dict in enumerate(manifest_dict.get('Applications') or []):
        scope.entries.append(build_entry(entry_dict, i))
    return scope


def discover_share(shareroot):
    """Yield (scopename, phase, manifest_dict) for runtime manifests on a share.

    Reads `common/manifest.json` and every `gea-shopfloor-*/manifest.json`.
    Skips `.bak` / `.pre-*.bak` variants (only the exact `manifest.json`).
    """
    for name in sorted(os.listdir(shareroot)):
        if name != 'common' and not name.startswith('gea-shopfloor-'):
            continue
        path = os.path.join(shareroot, name, 'manifest.json')
        if not os.path.isfile(path):
            continue
        with open(path) as handle:
            manifest = json.load(handle)
        yield name, 'runtime', manifest


def load_manifest_file(path):
    """Parse a single manifest.json / preinstall.json file."""
    with open(path) as handle:
        return json.load(handle)