import difflib
import re
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
[docs]
class AuditEntryFact(NamedTuple):
[docs]
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
[docs]
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,
),
)
[docs]
AUDIT_ENTRY_FACTORIES: dict[
str,
Callable[[AuditEntry, OrgRequest], tuple[AuditEntryFact, ...]],
] = {
'pages': page_audit_entry_facts,
}
[docs]
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 ()
[docs]
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 value is None:
return 'null'
if isinstance(value, str):
return value or '""'
return json.dumps(value, indent=2, sort_keys=True)
def word_diff(previous: str, current: str) -> tuple[Markup, Markup]:
token_pattern = re.compile(
r'\d{4}-\d{2}-\d{2}|[a-zA-Z0-9_]+|[^a-zA-Z0-9_]+'
)
previous_words = token_pattern.findall(previous)
current_words = token_pattern.findall(current)
previous_parts: list[Markup] = []
current_parts: list[Markup] = []
matcher = difflib.SequenceMatcher(
a=previous_words,
b=current_words,
autojunk=False,
)
for operation, start_a, end_a, start_b, end_b in matcher.get_opcodes():
old = previous_words[start_a:end_a]
new = current_words[start_b:end_b]
if operation == 'equal':
previous_parts.extend(Markup.escape(word) for word in old)
current_parts.extend(Markup.escape(word) for word in new)
continue
if old:
previous_parts.append(
Markup('<del class="diff-word-removed">{}</del>').format(
''.join(old)
)
)
if new:
current_parts.append(
Markup('<ins class="diff-word-added">{}</ins>').format(
''.join(new)
)
)
return Markup('').join(previous_parts), Markup('').join(current_parts)
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:
previous_text = display(previous)
current_text = display(current)
previous_diff, current_diff = word_diff(
previous_text,
current_text,
)
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),
previous_diff,
current_diff,
)
)
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,
)
[docs]
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,
)
[docs]
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
),
}