Source code for org.views.audit

from onegov.core.custom import json
from onegov.core.elements import Link
from onegov.core.orm.audit import AuditEntry, AuditEntryCollection
from onegov.core.security import Secret
from onegov.org import _, OrgApp
from onegov.org.layout import AuditTrailLayout
from onegov.page import Page
from markupsafe import Markup


from typing import Any, NamedTuple, TYPE_CHECKING

if TYPE_CHECKING:
    from collections.abc import Callable

    from onegov.core.types import RenderData
    from onegov.org.request import OrgRequest




class AuditEntryFact(NamedTuple):


    label: str



    value: object






def page_snapshot_subpage_count(
    snapshot: dict[str, object],
) -> int | None:
    children = snapshot.get('children')
    if not isinstance(children, list):
        return None

    count = 0
    remaining = children.copy()
    while remaining:
        child = remaining.pop()
        if not isinstance(child, dict):
            continue
        count += 1
        grandchildren = child.get('children')
        if isinstance(grandchildren, list):
            remaining.extend(grandchildren)
    return count





def page_audit_entry_facts(
    entry: AuditEntry,
    request: OrgRequest,
) -> tuple[AuditEntryFact, ...]:
    count = page_snapshot_subpage_count(entry.snapshot)
    if count is None:
        return ()
    return (
        AuditEntryFact(
            request.translate(_('Contained subpages')),
            count,
        ),
    )





AUDIT_ENTRY_FACTORIES: dict[
    str,
    Callable[[AuditEntry, OrgRequest], tuple[AuditEntryFact, ...]],
] = {
    'pages': page_audit_entry_facts,
}





def audit_entry_facts(
    entry: AuditEntry,
    request: OrgRequest,
) -> tuple[AuditEntryFact, ...]:
    factory = AUDIT_ENTRY_FACTORIES.get(entry.target_table)
    return factory(entry, request) if factory else ()





def audit_snapshot_diff(
    previous_snapshot: dict[str, Any],
    snapshot: dict[str, Any],
    previous_label: str,
    snapshot_label: str,
) -> Markup:
    missing = object()

    def display(value: object) -> str:
        if value is missing:
            return '—'
        if isinstance(value, str):
            return value or '""'
        return json.dumps(value, indent=2, sort_keys=True)

    def changes(
        previous: Any,
        current: Any,
        path: tuple[str, ...] = (),
    ) -> list[Markup]:
        if type(previous) is type(current) and previous == current:
            return []
        rows: list[Markup] = []
        if isinstance(previous, dict) and isinstance(current, dict):
            for key in sorted(previous.keys() | current.keys()):
                rows.extend(
                    changes(
                        previous.get(key, missing),
                        current.get(key, missing),
                        (*path, key),
                    )
                )
        else:
            rows.append(
                Markup(
                    '<tr><th scope="row">{}</th>'
                    '<td class="diff_sub"><pre>{}</pre></td>'
                    '<td class="diff_add"><pre>{}</pre></td></tr>'
                ).format(
                    ' / '.join(path),
                    display(previous),
                    display(current),
                )
            )
        return rows

    return Markup(
        '<table class="diff"><thead><tr><td></td>'
        '<th scope="col">{}</th><th scope="col">{}</th>'
        '</tr></thead><tbody>{}</tbody></table>'
    ).format(
        previous_label,
        snapshot_label,
        Markup('').join(changes(previous_snapshot, snapshot)),
    )



@OrgApp.html(
    model=AuditEntryCollection,
    template='audit_trail.pt',
    permission=Secret,
)


def view_audit_trail(
    self: AuditEntryCollection,
    request: OrgRequest,
    layout: AuditTrailLayout | None = None,
) -> RenderData:
    entries = self.batch
    filters = [
        Link(
            text=request.translate(_('All')),
            active=self.operation is None,
            url=request.link(self.for_operation(None)),
        ),
        Link(
            text=request.translate(_('Created')),
            active=self.operation == 'insert',
            url=request.link(self.for_operation('insert')),
        ),
        Link(
            text=request.translate(_('Changed')),
            active=self.operation == 'update',
            url=request.link(self.for_operation('update')),
        ),
        Link(
            text=request.translate(_('Deleted')),
            active=self.operation == 'delete',
            url=request.link(self.for_operation('delete')),
        ),
    ]
    return {
        'layout': layout or AuditTrailLayout(self, request),
        'title': _('Audit Trail'),
        'entries': entries,
        'filters': filters,
        'entry_facts': {
            entry.id: audit_entry_facts(entry, request) for entry in entries
        },
    }



@OrgApp.html(
    model=AuditEntry,
    template='audit_entry.pt',
    permission=Secret,
)


def view_audit_entry(
    self: AuditEntry,
    request: OrgRequest,
    layout: AuditTrailLayout | None = None,
) -> RenderData:
    layout = layout or AuditTrailLayout(self, request)
    layout.breadcrumbs.append(Link(str(self.id), request.link(self)))
    current_page = None
    if self.target_table == 'pages' and self.target_id.isdigit():
        current_page = request.session.get(Page, int(self.target_id))
    operation = request.translate(
        {
            'insert': _('Created'),
            'update': _('Changed'),
            'delete': _('Deleted'),
        }.get(self.operation, self.operation)
    )
    return {
        'layout': layout,
        'title': _('Audit Trail Entry'),
        'entry': self,
        'operation': operation,
        'current_page': current_page,
        'diff': (
            audit_snapshot_diff(
                self.previous_snapshot,
                self.snapshot,
                request.translate(_('Before update')),
                request.translate(_('After update')),
            )
            if self.previous_snapshot
            else None
        ),
        'snapshot': json.dumps(self.snapshot, indent=2),
        'previous_snapshot': (
            json.dumps(self.previous_snapshot, indent=2)
            if self.previous_snapshot
            else None
        ),
    }