#!/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 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 exists: {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()