import json
import sqlite3
import time
from datetime import datetime, timezone
import config_utils
import factory
def _fmt_ts(ts):
try:
dt = datetime.fromtimestamp(int(ts), tz=timezone.utc)
return dt.strftime('%Y-%m-%d %H:%M UTC')
except Exception:
return '-'
def _active_sessions_table():
try:
con = sqlite3.connect(config_utils.SESSIONS_DB, timeout=5)
rows = con.execute(
'SELECT session_id, email, access_level, created_at, last_seen'
' FROM sessions ORDER BY last_seen DESC'
).fetchall()
con.close()
except Exception:
rows = []
if not rows:
return '
No active sessions.
'
now = int(time.time())
trs = ''
for sid, email, access_level, created_at, last_seen in rows:
online = (now - int(last_seen)) < 300
badge = (
'Online'
if online else
'Offline'
)
btn = (
f''
)
trs += (
f''
f'| {factory.e(email)} | '
f'{factory.e(access_level)} | '
f'{badge} | '
f'{_fmt_ts(created_at)} | '
f'{_fmt_ts(last_seen)} | '
f'{btn} | '
f'
'
)
return (
''
''
''
''
''
''
''
'
' + trs + '
'
)
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