import json from collections import defaultdict from datetime import datetime from flask import session from config_utils import ( collect_layout_tokens, get_dashboard_pending, load_all_groups, get_done_timestamps, _apply_changes_immediately, _find_cmd_in_queues, WEB_APP_DISPLAY_NAME, ) from factory import LEVEL_RANK, e, client_level, build_snap_val, snap_expand_row, load_icon def collect_tokens(cfg): tokens = collect_layout_tokens(cfg) tokens['GENERAL_APPLY_ON_SAVE'] = 'true' if session.get('apply_changes_immediately', False) else 'false' all_groups = load_all_groups() group_uuid_set = {g['uuid'] for g, _ in all_groups} pending_items = get_dashboard_pending() if pending_items: pgroups = defaultdict(list) for uuid, ts, cmd, user in pending_items: pgroups[cmd].append((uuid, user)) rows = '' for cmd, entries in pgroups.items(): users = ', '.join(sorted({u for _, u in entries if u and u != 'unknown'})) snap_uuids = [uuid for uuid, _ in entries if uuid in group_uuid_set] if snap_uuids: req_tags = ''.join( f'' f'{uuid[:8]}' f'{uuid[:8]}' f'{uuid[:8]}' '' for uuid in snap_uuids ) req_cell = f'
{req_tags}
' else: req_cell = '-' rows += ( '' f'{e(cmd)}' f'{e(users)}' f'{req_cell}' '' ) pending_html = ( '' '' '' '' '' f'{rows}
CommandUserRequired By
' ) else: pending_html = '

No pending actions.

' tokens['PENDING_ACTIONS_HTML'] = pending_html tokens['NO_PENDING'] = 'true' if not pending_items else '' tokens['NO_DISMISSIBLE_PENDING'] = 'true' if not any(c != 'fix problems' for _, _, c, _ in pending_items) else '' tokens['APPLY_WARNING'] = ( f'

{load_icon("arrow-left")} Applying actions will briefly disrupt connections as network services are restarted.

' if pending_items else '' ) done_ts_map = get_done_timestamps() if all_groups: is_manager = client_level() >= LEVEL_RANK['manager'] no_revert = set() for g, _ in all_groups: if g['reverts_group']: no_revert.add(g['uuid']) no_revert.add(g['reverts_group']) hist_rows = '' hist_onclick = ( 'onclick="if(event.target.type!==\'checkbox\')' 'this.nextElementSibling.hidden=!this.nextElementSibling.hidden"' ) for g, changes in all_groups: uuid = g['uuid'] applied_ts = done_ts_map.get(uuid) dt_str = datetime.fromtimestamp(applied_ts).strftime('%Y-%m-%d %H:%M') if applied_ts else '-' all_before_null = all(c['before'] is None for c in changes) all_after_null = all(c['after'] is None for c in changes) if g['reverts_group']: verb = 'Reverted' elif all_before_null: verb = 'Added' elif all_after_null: verb = 'Deleted' else: verb = 'Edited' item = g.get('item_value') or '' summary = f'{verb} {g["parent_path"]}: {item}' if item else f'{verb} {g["parent_path"]}' snap_tag = ( f'
' f'{e(uuid[:8])}' f'{e(uuid[:8])}' f'{e(uuid[:8])}' '
' ) snap_user = e(g.get('user', '')) cb_attrs = '' if is_manager else ('disabled title="Cannot revert"' if uuid in no_revert else '') hist_rows += ( f'' f'' f'{e(dt_str)}' f'{e(summary)}' f'{build_snap_val(changes)}' f'{snap_tag}' f'{snap_user}' '' f'{snap_expand_row(changes, 6)}' ) select_all = ( '' ) history_html = ( '' f'' '' '' '' '' '' '' f'{hist_rows}
{select_all}AppliedChangeFieldsChange IDUser
' ) else: history_html = '

No change history.

' tokens['CHANGE_HISTORY_HTML'] = history_html tokens['NO_HISTORY'] = 'true' if not all_groups else '' return tokens