scripts/plugin_support.py

"""Copied into each standalone plugin; no repository runtime dependency."""
import argparse
from decimal import Decimal, InvalidOperation
import json
import os
from pathlib import Path
import tempfile

def obj(value, field='input'):
    if not isinstance(value, dict): raise ValueError(f'{field} must be an object')
    return value

def text(value, field='value'):
    if not isinstance(value, str) or not value.strip(): raise ValueError(f'{field} must be a nonempty string')
    return value

def records(value, field, id_key='id', limit=10000):
    if not isinstance(value, list) or len(value) > limit: raise ValueError(f'{field} must be a list of at most {limit} objects')
    found = set()
    for row in value:
        obj(row, field)
        if id_key is not None:
            key = text(row.get(id_key), f'{field}.{id_key}')
            if key in found: raise ValueError(f'Duplicate {field}.{id_key}: {key}')
            found.add(key)
    return value

def amount(value, field='amount', nonnegative=True):
    if type(value) not in (str, int): raise ValueError(f'{field} must be a decimal string or integer')
    try: result = Decimal(value)
    except InvalidOperation: raise ValueError(f'Invalid {field}') from None
    if not result.is_finite() or (nonnegative and result < 0): raise ValueError(f'{field} must be finite' + (' and nonnegative' if nonnegative else ''))
    return result

def validate_json(value):
    try: json.dumps(value, allow_nan=False)
    except (TypeError, ValueError, RecursionError) as exc: raise ValueError('Input must contain finite JSON values') from exc

def main(run):
    parser = argparse.ArgumentParser(description='Aivana GmbH local plugin | https://www.aivana-gmbh.ai')
    parser.add_argument('--input', required=True, type=Path); parser.add_argument('--output', required=True, type=Path)
    args = parser.parse_args()
    try:
        if args.input.resolve() == args.output.resolve() or (args.output.exists() and args.input.samefile(args.output)): raise ValueError('Input and output must be different files')
        with args.input.open('rb') as handle: raw = handle.read(8 * 1024 * 1024 + 1)
        if len(raw) > 8 * 1024 * 1024: raise ValueError('Input exceeds 8 MiB')
        data = json.loads(raw.decode('utf-8-sig')); obj(data); validate_json(data); result = run(data)
        payload = json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + '\n'; args.output.parent.mkdir(parents=True, exist_ok=True)
        temporary = None
        try:
            with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=args.output.parent, delete=False) as handle: temporary = Path(handle.name); handle.write(payload)
            os.replace(temporary, args.output)
        finally:
            if temporary and temporary.exists(): temporary.unlink()
        return 0
    except (ValueError, TypeError, KeyError, OSError, RecursionError) as exc: parser.exit(2, f'Error: {exc}\n')