# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Shared extraction, validation and overlay of external filter TOML.
.. versionadded:: 2.4
:Author: MAFw developers
"""
from __future__ import annotations
import logging
import os
import tempfile
from collections.abc import Mapping, Sequence
from copy import deepcopy
from pathlib import Path
from typing import Any, cast
import tomlkit
log = logging.getLogger(__name__)
"""Logger for dormant filter scopes."""
_KEYS = {'__filter__', '__logic__', '__new_only__'}
"""Filter-related processor entries."""
[docs]
def read_filters(path: Path | str) -> tomlkit.TOMLDocument:
"""Read and validate an external filter file without loading plugins.
.. versionadded:: 2.4
"""
doc = tomlkit.parse(Path(path).expanduser().read_text(encoding='utf-8'))
validate_filters(doc)
return doc
[docs]
def validate_filters(config: Mapping[str, Any]) -> None:
"""Reject non-filter entries and validate existing filter expressions.
.. versionadded:: 2.4
"""
from mafw.steering.builder import SteeringBuilder
from mafw.steering.validation import _validate_filter_conditions, _validate_logic_expressions
def check(node: Any, path: str) -> None:
# Metadata is reserved at every filter level; ordinary names are conditions.
if isinstance(node, Mapping):
for key, value in node.items():
if key.startswith('__') and key not in {'__logic__', '__enable__', '__conditional__'}:
raise ValueError(f'Unknown filter metadata {path}.{key}')
if key == '__enable__' and not isinstance(value, bool):
raise ValueError(f'{path}.{key} must be boolean')
if key == '__logic__' and not isinstance(value, str):
raise ValueError(f'{path}.{key} must be a string')
if key == '__conditional__':
allowed = {
'name',
'__enable__',
'condition_field',
'condition_op',
'condition_value',
'then_field',
'then_op',
'then_value',
'else_field',
'else_op',
'else_value',
}
if not isinstance(value, list) or any(not isinstance(entry, Mapping) for entry in value):
raise ValueError(f'{path}.{key} must be an array of tables')
for entry in value:
if set(entry) - allowed:
raise ValueError(f'Unknown conditional metadata at {path}.{key}')
check(value, f'{path}.{key}')
elif isinstance(node, list):
for value in node:
check(value, path)
for scope, entries in config.items():
if scope == 'new_only':
if not isinstance(entries, bool):
raise ValueError('new_only must be boolean')
continue
if scope in {'DBConfiguration', 'StorageConfiguration', 'UserInterface'} or not isinstance(entries, Mapping):
raise ValueError(f'Non-filter entry: {scope}')
if not entries:
raise ValueError(f'Non-filter entry: {scope}')
for key, value in entries.items():
if key not in _KEYS:
raise ValueError(f'Non-filter entry: {scope}.{key}')
if key == '__new_only__' and not isinstance(value, bool):
raise ValueError(f'{scope}.{key} must be boolean')
if key == '__logic__' and not isinstance(value, str):
raise ValueError(f'{scope}.{key} must be a string')
if key == '__filter__':
if not isinstance(value, Mapping):
raise ValueError(f'{scope}.{key} must be a table')
for name, model in value.items():
if name != '__logic__' and not isinstance(model, Mapping):
raise ValueError(f'{scope}.{key}.{name} must be a table')
check(value, f'{scope}.{key}')
builder = SteeringBuilder.from_toml_text(tomlkit.dumps(dict(config)))
issues: list[Any] = []
_validate_filter_conditions(builder, issues)
_validate_logic_expressions(builder, issues)
if issues:
raise issues[0]
[docs]
def strip_filters(config: Mapping[str, Any]) -> Any:
"""Copy steering data with filters removed, retaining new-only settings.
.. versionadded:: 2.4
"""
result: Any = deepcopy(config)
for entries in result.values():
if isinstance(entries, Mapping):
for key in ('__filter__', '__logic__'):
if key in entries:
del cast(Any, entries)[key]
return result
[docs]
def drop_filters(config: Mapping[str, Any]) -> Any:
"""Copy steering data with filter definitions removed, retaining new-only settings.
.. versionadded:: 2.4
"""
return strip_filters(config)
[docs]
def overlay_filters(config: Mapping[str, Any], overlays: Sequence[Mapping[str, Any]]) -> Any:
"""Copy configuration and replace matching model definitions in order.
.. versionadded:: 2.4
"""
result: Any = deepcopy(config)
active: set[str] = set()
def collect(names: Any) -> None:
for name in names:
if name in active:
continue
active.add(name)
section = config.get(name, {})
if isinstance(section, Mapping):
collect(section.get('processors_to_run', []))
collect(config.get('processors_to_run', []))
for overlay in overlays:
validate_filters(overlay)
for scope, entries in overlay.items():
if scope == 'new_only':
result[scope] = deepcopy(entries)
continue
if scope not in active:
log.warning('Filter scope %s is not in processors_to_run', scope)
if scope not in result:
result[scope] = {}
target = result[scope]
for key, value in entries.items():
if key == '__filter__':
if key not in target:
target[key] = {}
# A model owns its field/conditional children; replace all of it.
for name, definition in value.items():
target[key][name] = deepcopy(definition)
else:
target[key] = deepcopy(value)
return result
[docs]
def atomic_write(path: Path | str, text: str) -> None:
"""Serialize safely through a temporary file beside the destination.
.. versionadded:: 2.4
"""
target = Path(path)
temporary: str | None = None
try:
with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=target.parent, delete=False) as handle:
temporary = handle.name
handle.write(text)
if target.exists():
os.chmod(temporary, target.stat().st_mode)
os.replace(temporary, target)
finally:
if temporary is not None and os.path.exists(temporary):
os.unlink(temporary)