angie-conv-image/jinja2/j2common.py

#!/usr/bin/env python3
import importlib
import json
import os
import os.path
import sys
import yaml
import jinja2

ME = sys.argv[0]

J2_MODULES_DEFAULT = 'os os.path sys netaddr pathlib psutil re wcmatch'
J2_SUFFIX = '.j2'
J2_CFG_PATHS = [
    '/angie/j2cfg',
    '/etc/angie/j2cfg',
]
J2_CFG_EXTS = [
    'yml',
    'yaml',
    'json',
]
J2_SEARCH_PATH = [
    '/etc/angie',
    '/run/angie',
    '/',
]
J2_EXT_LIST = [
    'jinja2.ext.do',
    'jinja2.ext.loopcontrols',
]

J2_MODULES = sorted(set(
    os.getenv('NGX_JINJA_MODULES', J2_MODULES_DEFAULT).split(sep=' ')
))

J2_CONFIG = os.getenv('NGX_JINJA_CONFIG', '')

J2_KWARGS = {}


def merge_dict_from_file(filename):
    if (not filename) or (str(filename) == ''):
        return False
    if not os.path.exists(filename):
        return False
    if not os.path.isfile(filename):
        print(
            f'{ME}: not a file, skipping: {filename}',
            file=sys.stderr)
        return False
    if filename.endswith('.yml') or filename.endswith('.yaml'):
        with open(filename, mode='r', encoding='utf-8') as fx:
            x = yaml.safe_load(fx)
            J2_KWARGS['cfg'] = J2_KWARGS['cfg'] | x
        return True
    if filename.endswith('.json'):
        with open(filename, mode='r', encoding='utf-8') as fx:
            x = json.load(fx)
            J2_KWARGS['cfg'] = J2_KWARGS['cfg'] | x
        return True
    print(
        f'{ME}: non-recognized name extension: {filename}',
        file=sys.stderr)
    return False


def merge_dict_default():
    for base in J2_CFG_PATHS:
        for full in [base + '.' + ext for ext in J2_CFG_EXTS]:
            if merge_dict_from_file(full):
                break
            continue


def render_error(msg, fail=True) -> bool:
    if fail:
        raise ValueError(msg)
    print(f'{ME}: {msg}', file=sys.stderr)
    return False


def render_file(file_in, file_out=None, fail=True):
    if (not file_in) or (str(file_in) == ''):
        return render_error(
            'argument "file_in" is empty',
            fail)
    if not os.path.exists(file_in):
        return render_error(
            f'file is missing: {file_in}',
            fail)
    if not os.path.isfile(file_in):
        return render_error(
            f'not a file: {file_in}',
            fail)

    f_out = file_out
    if not f_out:
        if not file_in.endswith(J2_SUFFIX):
            return render_error(
                f'input file name extension mismatch: {file_in}',
                fail)
        f_out = os.path.splitext(file_in)[0]

    dirs = J2_SEARCH_PATH.copy()
    for d in [os.path.dirname(file_in), os.getcwd()]:
        if d not in dirs:
            dirs.insert(0, d)

    j2_loader = jinja2.ChoiceLoader([
        jinja2.FileSystemLoader(
            d,
            encoding='utf-8',
            followlinks=True,
        ) for d in dirs
    ])

    j2_environ = jinja2.Environment(
        loader=j2_loader,
        extensions=J2_EXT_LIST,
    )
    j2_template = j2_environ.get_template(file_in)
    j2_stream = j2_template.stream(**J2_KWARGS)
    j2_stream.disable_buffering()

    if os.path.lexists(f_out):
        if os.path.islink(f_out) or (not os.path.isfile(f_out)):
            return render_error(
                f'output file is not safely writable: {f_out}',
                fail)
    if os.path.exists(f_out):
        if os.path.samefile(file_in, f_out):
            return render_error(
                f'unable to process template inplace: {file_in}',
                fail)
    j2_stream.dump(f_out, encoding='utf-8')
    return True


def init():
    kwa = {}
    for m in J2_MODULES:
        kwa[m] = importlib.import_module(m)
    kwa['env'] = os.environ
    kwa['cfg'] = {}
    global J2_KWARGS
    J2_KWARGS = kwa

    if J2_CONFIG != '':
        if os.path.isfile(J2_CONFIG):
            merge_dict_from_file(J2_CONFIG)
        else:
            print(
                f'{ME}: J2_CONFIG does not exist, skipping: {J2_CONFIG}',
                file=sys.stderr
            )
            merge_dict_default()
    else:
        merge_dict_default()