linuxrouter/docker/routlin-dash/app/pages/accountmanage/view.py

import json
import sqlite3
import time
from datetime import datetime
import config_utils
import factory


def _fmt_ts(ts, now):
    try:
        dt  = datetime.fromtimestamp(int(ts))
        ago = config_utils.relative_time(int(ts), now)
        return f'{dt.strftime("%Y-%m-%d %H:%M")} ({ago} ago)'
    except Exception:
        return '-'


_LEVEL_INT_TO_STR = {0: 'nothing', 1: 'viewer', 2: 'administrator', 3: 'manager'}


def _active_sessions_table():
    try:
        con  = sqlite3.connect(config_utils.ACCOUNTS_DB, timeout=5)
        con.row_factory = sqlite3.Row
        rows = con.execute(
            '''SELECT s.session_id, a.email, a.access_level,
                      s.session_started_ts, s.last_seen_ts
               FROM sessions s
               JOIN accounts a ON a.account_id = s.account_id
               ORDER BY s.last_seen_ts DESC'''
        ).fetchall()
        con.close()
    except Exception:
        rows = []

    if not rows:
        return '<p class="text-muted" style="margin:0">No active sessions.</p>'

    now = int(time.time())
    trs = ''
    for row in rows:
        sid          = row['session_id']
        email        = row['email']
        access_level = _LEVEL_INT_TO_STR.get(row['access_level'], 'viewer')
        started_ts   = row['session_started_ts']
        last_seen    = row['last_seen_ts']
        online   = (now - int(last_seen)) < 300
        ago      = config_utils.relative_time(int(last_seen), now)
        tip      = factory.e(f'Last seen {ago} ago')
        badge_cls = 'badge-enabled' if online else 'badge-disabled'
        badge_lbl = 'Online' if online else 'Offline'
        badge = (
            f'<span class="tooltip-wrap" data-tooltip="{tip}">'
            f'<span class="badge {badge_cls}">{badge_lbl}</span>'
            f'</span>'
        )
        btn = (
            f'<form method="post" action="/action/accountmanage/session_invalidate"'
            f' style="display:inline;margin:0">'
            f'<input type="hidden" name="session_id" value="{factory.e(sid)}">'
            f'<button type="submit" class="btn btn-danger btn-sm">Invalidate</button>'
            f'</form>'
        )
        trs += (
            f'<tr>'
            f'<td class="table-cell">{factory.e(email)}</td>'
            f'<td class="table-cell">{factory.e(access_level)}</td>'
            f'<td class="table-cell">{badge}</td>'
            f'<td class="table-cell">{_fmt_ts(started_ts, now)}</td>'
            f'<td class="table-cell">{btn}</td>'
            f'</tr>'
        )
    return (
        '<table class="data-table"><thead><tr>'
        '<th class="table-header">Email</th>'
        '<th class="table-header">Access Level</th>'
        '<th class="table-header">Status</th>'
        '<th class="table-header">Logged In</th>'
        '<th class="table-header"></th>'
        '</tr></thead><tbody>' + trs + '</tbody></table>'
    )


def collect_tokens(cfg):
    tokens = config_utils.collect_layout_tokens(cfg)
    tokens['ACCOUNT_LEVEL_OPTIONS'] = json.dumps([
        {'value': 'viewer',        'label': 'Viewer (read-only access to live data)'},
        {'value': 'administrator', 'label': 'Administrator (can modify configuration)'},
        {'value': 'manager',       'label': 'Manager (full access including account management)'},
    ])
    tokens['ACTIVE_SESSIONS_TABLE'] = _active_sessions_table()
    content = factory.load_json(f'{factory.PAGES_DIR}/accountmanage/content.json')
    for table_item in factory.iter_table_items(content.get('items', [])):
        ds = table_item.get('datasource', '')
        tokens[factory.table_token_key(ds)] = factory.build_table(table_item, tokens, config_utils.load_datasource(ds))
    return tokens