DemonGatanjieu-Anomalous_Model_Browser/api/activity_log.py

"""Activity log: what Anomalous changed, newest first, kept in the ComfyUI user folder.

Two sources feed it. The middleware records this plugin's own write requests
(and Anomalous_TTS's) after they succeed: files such as models, covers, recipes,
materials, notes, presets, images and audio. The browser posts the canvas changes
it made (`POST /anomalous/activity`). Only the newest MAX_ENTRIES are kept and long
values are shortened, so the file stays small. Recording never fails a request:
the log is a convenience, not part of the action.
"""

import asyncio
import json
import logging
import os
import threading
import time
import uuid

from aiohttp import web
import folder_paths

from .path_utils import atomic_write_json

MAX_ENTRIES = 500
MAX_LOG_BYTES = 2 * 1024 * 1024  # long canvas entries could otherwise outgrow the JSON writer's limit
MAX_VALUE_CHARS = 400
DIFF_CONTEXT_CHARS = 60  # kept before the first difference of a long before/after pair
MAX_CHANGES = 60

# Write routes worth a line in the log → the action id the browser has words for. Scans are
# recorded when they end, with what they found (api/scan_report.py), not here.
ROUTE_ACTIONS = {
    '/anomalous/delete_model': 'model_delete',
    '/anomalous/update_metadata': 'model_edit',
    '/anomalous/set_custom_cover': 'model_cover',
    '/anomalous/upload_custom_cover': 'model_cover',
    '/anomalous/save_recipe': 'recipe_save',
    '/anomalous/update_recipe': 'recipe_edit',
    '/anomalous/delete_recipe': 'recipe_delete',
    '/anomalous/restore_recipe_version': 'recipe_restore',
    '/anomalous/set_recipe_gallery_cover': 'recipe_cover',
    '/anomalous/import_recipe_package_commit': 'recipe_import',
    '/anomalous/save_image_material': 'material_save',
    '/anomalous/save_prompt_plan': 'material_save',
    '/anomalous/update_material': 'material_edit',
    '/anomalous/delete_material': 'material_delete',
    '/anomalous/save_notebook': 'note_save',
    '/anomalous/delete_notebook': 'note_delete',
    '/anomalous/save_parameter': 'preset_save',
    '/anomalous/rename_parameter': 'preset_rename',
    '/anomalous/delete_parameter': 'preset_delete',
    '/anomalous/delete_gallery_image': 'image_delete',
    '/anomalous/delete_audio_gallery': 'audio_delete',
    '/anomalous/save_audio_preview': 'audio_save',
    '/anomalous/clear_cache': 'cache_clear',
    '/anomalous_tts/import/commit': 'voice_import',
    '/anomalous_tts/settings': 'voice_settings',
}
# Request fields that name what was changed, in order of preference.
TARGET_FIELDS = ('name', 'character', 'filename', 'target_filename', 'recipe_filename',
                 'notebook_filename', 'new_name', 'path', 'file', 'target')

# What a handler attaches to its request for the log: request[ACTIVITY_DETAIL] = {fields, files}.
# web.RequestKey is newer than the aiohttp some ComfyUI installs have; a plain key works there.
ACTIVITY_DETAIL = web.RequestKey("activity_detail", dict) if hasattr(web, "RequestKey") else "activity_detail"

_lock = threading.RLock()
_entries = None


def log_path():
    user_dir = folder_paths.get_user_directory() if hasattr(folder_paths, 'get_user_directory') else None
    if not user_dir:
        user_dir = os.path.join(folder_paths.base_path, 'user')
    directory = os.path.join(user_dir, 'anomalous')
    os.makedirs(directory, exist_ok=True)
    return os.path.join(directory, 'activity_log.json')


def _load():
    global _entries
    if _entries is None:
        try:
            with open(log_path(), encoding='utf-8') as source:
                data = json.load(source)
            _entries = [entry for entry in data if isinstance(entry, dict)] if isinstance(data, list) else []
        except (OSError, ValueError):
            _entries = []
    return _entries


def _encoded_size(entries):
    return len(json.dumps(entries, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))


def _trim(entries):
    """Drop the oldest entries until the list fits MAX_ENTRIES and MAX_LOG_BYTES."""
    del entries[MAX_ENTRIES:]
    while len(entries) > 1 and _encoded_size(entries) > MAX_LOG_BYTES:
        del entries[max(1, len(entries) * 9 // 10):]


def _text(value):
    return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)


def _short(value):
    text = _text(value)
    return text if len(text) <= MAX_VALUE_CHARS else text[:MAX_VALUE_CHARS] + '…'


def _short_values(item, keys):
    """`keys` of `item` shortened. A long before/after pair is cut around its first difference
    instead of at the start, so a change late in a long prompt still shows."""
    out = {key: _short(item[key]) for key in keys if key in item and item[key] is not None}
    if 'before' not in out or 'after' not in out:
        return out
    old, new = _text(item['before']), _text(item['after'])
    if len(old) <= MAX_VALUE_CHARS and len(new) <= MAX_VALUE_CHARS:
        return out
    same = 0
    while same < min(len(old), len(new)) and old[same] == new[same]:
        same += 1
    start = max(0, same - DIFF_CONTEXT_CHARS)

    def cut(text):
        return ('…' if start else '') + text[start:start + MAX_VALUE_CHARS] + ('…' if start + MAX_VALUE_CHARS < len(text) else '')
    out['before'], out['after'] = cut(old), cut(new)
    return out


def add_entry(source, action, target='', detail=None):
    """Append one entry (newest first) and save. Returns the entry."""
    entry = {
        'id': uuid.uuid4().hex[:12],
        'time': time.time(),
        'source': source,
        'action': action,
        'target': _short(target or ''),
    }
    if detail:
        entry['detail'] = detail
    with _lock:
        entries = _load()
        entries.insert(0, entry)
        _trim(entries)
        atomic_write_json(log_path(), entries)
    return entry


def list_entries(limit=100, source=None):
    with _lock:
        entries = _load()
        picked = [entry for entry in entries if not source or entry.get('source') == source]
        return picked[:limit]


def clear():
    global _entries
    with _lock:
        _entries = []
        atomic_write_json(log_path(), [])


def _target_of(body):
    if not isinstance(body, dict):
        return ''
    for field in TARGET_FIELDS:
        value = body.get(field)
        if isinstance(value, str) and value.strip():
            return os.path.basename(value.strip().rstrip('/\\')) or value.strip()
    return ''


def _failed(response):
    """True when a response reports failure, including the 200 + {"status": "error"} style."""
    if response.status >= 300:
        return True
    if getattr(response, 'content_type', '') != 'application/json' or not getattr(response, 'body', None):
        return False
    try:
        data = json.loads(response.body)
    except (TypeError, ValueError):
        return False
    return isinstance(data, dict) and (data.get('success') is False or data.get('ok') is False
                                       or data.get('status') in ('error', 'failed'))


async def _request_body(request):
    """The body the handler already read (aiohttp caches it); {} when it was not JSON or a form."""
    try:
        if request.content_type == 'application/json':
            return await request.json()
        if request.content_type in ('multipart/form-data', 'application/x-www-form-urlencoded'):
            return dict(await request.post())
    except (ValueError, UnicodeDecodeError, RuntimeError):
        pass
    return {}


def _clean_detail(detail):
    """What a handler attached for the log (request[ACTIVITY_DETAIL]): the fields it changed
    ({field, before, after}) and the files it moved, values shortened."""
    if not isinstance(detail, dict):
        return None
    cleaned = {}
    if isinstance(detail.get('fields'), list):
        cleaned['fields'] = [_short_values(item, ('field', 'before', 'after'))
                             for item in detail['fields'][:MAX_CHANGES] if isinstance(item, dict)]
    if isinstance(detail.get('files'), list):
        cleaned['files'] = [_short(name) for name in detail['files'][:MAX_CHANGES]]
    return {key: value for key, value in cleaned.items() if value} or None


@web.middleware
async def activity_middleware(request, handler):
    response = await handler(request)
    action = ROUTE_ACTIONS.get(request.path) if request.method == 'POST' else None
    if action and not _failed(response):
        try:
            target = _target_of(await _request_body(request))
            detail = _clean_detail(request.get(ACTIVITY_DETAIL))
            await asyncio.get_running_loop().run_in_executor(None, add_entry, 'file', action, target, detail)
        except Exception:  # noqa: BLE001 - the log must never fail the action it records
            logging.getLogger(__name__).warning('Anomalous activity log: could not record %s', request.path, exc_info=True)
    return response


def _clean_changes(changes):
    cleaned = []
    for change in changes[:MAX_CHANGES]:
        if not isinstance(change, dict):
            continue
        item = _short_values(change, ('kind', 'node', 'type', 'widget', 'before', 'after'))
        if item.get('kind') in ('changed', 'added', 'removed', 'opened'):
            cleaned.append(item)
    return cleaned


async def api_get_activity(request):
    try:
        limit = max(1, min(MAX_ENTRIES, int(request.query.get('limit', '100'))))
    except ValueError:
        limit = 100
    source = request.query.get('source') or None
    entries = await asyncio.get_running_loop().run_in_executor(None, list_entries, limit, source)
    return web.json_response({'entries': entries})


async def api_post_activity(request):
    """Canvas changes from the browser: {changes: [...], total, workflow, via?}; via "mcp" marks
    changes an AI app made through the MCP endpoint."""
    try:
        body = await request.json()
    except ValueError:
        return web.json_response({'error': 'Invalid JSON body'}, status=400)
    if not isinstance(body, dict) or not isinstance(body.get('changes'), list):
        return web.json_response({'error': 'changes must be a list'}, status=400)
    changes = _clean_changes(body['changes'])
    if not changes:
        return web.json_response({'error': 'No recognizable changes'}, status=400)
    total = body.get('total') if isinstance(body.get('total'), int) else len(changes)
    detail = {'changes': changes, 'total': max(total, len(changes))}
    if body.get('via') == 'mcp':
        detail['via'] = 'mcp'
    entry = await asyncio.get_running_loop().run_in_executor(
        None, add_entry, 'canvas', 'canvas', body.get('workflow') if isinstance(body.get('workflow'), str) else '', detail)
    return web.json_response({'entry': entry})


async def api_clear_activity(request):
    await asyncio.get_running_loop().run_in_executor(None, clear)
    return web.json_response({'ok': True})


def register_routes(app):
    if hasattr(app, 'middlewares') and activity_middleware not in app.middlewares:
        app.middlewares.append(activity_middleware)
    app.router.add_get('/anomalous/activity', api_get_activity)
    app.router.add_post('/anomalous/activity', api_post_activity)
    app.router.add_post('/anomalous/activity/clear', api_clear_activity)