world_snapshot/world_snapshot/init.py

# -*- coding: utf-8 -*-
"""world_snapshot module implementation (W-03 world snapshot).

Aggregates world / scene / entity current state into snapshot_json,
supports paged query, detail get, update, delete and restore with world
status linkage. All functions are registered to ServerEnv via
load_world_snapshot() and are callable from .dspy wrappers directly.
"""
import json

from appPublic.timeUtils import curDateString
from appPublic.uniqueID import getID
from ahserver.serverenv import ServerEnv
from sqlor.dbpools import DBPools

SNAPSHOT_TYPES = ('full', 'incremental')
SNAPSHOT_STATUS_ACTIVE = 'active'
SNAPSHOT_STATUS_RESTORED = 'restored'
WORLD_MODE_ACTIVE = 'active'


def _to_dict(row):
    """Convert a sqlor result row (DictObject) to a plain dict."""
    if row is None:
        return {}
    try:
        return dict(row)
    except Exception:
        try:
            keys = list(row.keys())
        except Exception:
            keys = [k for k in dir(row) if not k.startswith('_')]
        return {k: getattr(row, k, None) for k in keys}


def _module_dbname(env, module):
    try:
        return env.get_module_dbname(module)
    except Exception:
        return None


async def _get_world(env, world_id):
    """Read the world record (best effort, world module db)."""
    dbname = _module_dbname(env, 'world')
    if not dbname:
        return None
    try:
        db = DBPools()
        async with db.sqlorContext(dbname) as sor:
            rows = await sor.sqlExe('select * from world where id = ${wid}$', {'wid': world_id})
        if not rows:
            return None
        return _to_dict(rows[0])
    except Exception:
        return None


async def _fetch_child_rows(env, module, table, world_id):
    """Best-effort read of child rows (scene/entity) filtered by world_id."""
    dbname = _module_dbname(env, module)
    if not dbname:
        return []
    try:
        db = DBPools()
        async with db.sqlorContext(dbname) as sor:
            rows = await sor.sqlExe(
                'select * from ' + table + ' where world_id = ${wid}$ order by created_at desc',
                {'wid': world_id})
        return [_to_dict(r) for r in (rows or [])]
    except Exception:
        return []


async def _get_appcodes(parentid):
    """Read dict items from appcodes_kv (appbase first, module db fallback)."""
    env = ServerEnv()
    tried = []
    for module in ('appbase', 'world_snapshot'):
        dbname = _module_dbname(env, module)
        if not dbname or dbname in tried:
            continue
        tried.append(dbname)
        try:
            db = DBPools()
            async with db.sqlorContext(dbname) as sor:
                rows = await sor.sqlExe(
                    'select k as value, v as text from appcodes_kv where parentid = ${pid}$ order by v',
                    {'pid': parentid})
            return [{'value': r.value, 'text': r.text} for r in (rows or [])]
        except Exception:
            continue
    return []


async def create_snapshot(params):
    """Create a world snapshot: aggregate world/scene/entity current state."""
    params = params or {}
    world_id = params.get('world_id')
    if not world_id:
        return {'code': 'PARAM_REQUIRED', 'message': 'world_id is required', 'field': 'world_id', 'detail': ''}
    snapshot_type = params.get('snapshot_type') or 'full'
    if snapshot_type not in SNAPSHOT_TYPES:
        return {'code': 'INVALID_SNAPSHOT_TYPE', 'message': 'snapshot_type must be full or incremental',
                'field': 'snapshot_type', 'detail': snapshot_type}
    env = ServerEnv()
    world = await _get_world(env, world_id)
    if not world:
        return {'code': 'WORLD_NOT_FOUND', 'message': 'world not found', 'field': 'world_id', 'detail': world_id}
    snap_data = {
        'world': world,
        'scene': await _fetch_child_rows(env, 'scene', 'scene', world_id),
        'entity': await _fetch_child_rows(env, 'entity', 'entity', world_id),
        'snapshot_type': snapshot_type,
        'snapshot_date': curDateString(),
    }
    snapshot_json = json.dumps(snap_data, ensure_ascii=False)
    name = params.get('name')
    if not name:
        name = str(world.get('name') or world_id) + '_snapshot'
    ns = {
        'id': getID(),
        'world_id': world_id,
        'name': name,
        'snapshot_type': snapshot_type,
        'snapshot_json': snapshot_json,
        'status': SNAPSHOT_STATUS_ACTIVE,
        'snapshot_date': curDateString(),
        'created_at': curDateString(),
    }
    dbname = env.get_module_dbname('world_snapshot')
    db = DBPools()
    async with db.sqlorContext(dbname) as sor:
        await sor.C('world_snapshot', ns)
    return {'success': True, 'id': ns['id'], 'snapshot_type': snapshot_type,
            'snapshot_date': ns['snapshot_date']}


async def get_snapshot(params):
    """Get one snapshot record by id."""
    params = params or {}
    snapshot_id = params.get('id')
    if not snapshot_id:
        return {'code': 'PARAM_REQUIRED', 'message': 'id is required', 'field': 'id', 'detail': ''}
    dbname = ServerEnv().get_module_dbname('world_snapshot')
    db = DBPools()
    async with db.sqlorContext(dbname) as sor:
        rows = await sor.R('world_snapshot', {'id': snapshot_id})
    if not rows:
        return {'code': 'SNAPSHOT_NOT_FOUND', 'message': 'snapshot not found', 'field': 'id', 'detail': snapshot_id}
    return _to_dict(rows[0])


async def list_snapshots(params):
    """Paged snapshot list; count and data queries separated (P95 friendly)."""
    params = params or {}
    try:
        page = int(params.get('page') or 1)
    except Exception:
        page = 1
    try:
        page_size = int(params.get('page_size') or params.get('rows') or params.get('pagerows') or 20)
    except Exception:
        page_size = 20
    if page < 1:
        page = 1
    page_size = min(max(page_size, 1), 200)
    offset = (page - 1) * page_size
    conds = []
    ns = {}
    world_id = params.get('world_id')
    if world_id:
        conds.append('world_id = ${world_id}$')
        ns['world_id'] = world_id
    status = params.get('status')
    if status:
        conds.append('status = ${status}$')
        ns['status'] = status
    snapshot_type = params.get('snapshot_type')
    if snapshot_type:
        conds.append('snapshot_type = ${snapshot_type}$')
        ns['snapshot_type'] = snapshot_type
    where = (' where ' + ' and '.join(conds)) if conds else ''
    dbname = ServerEnv().get_module_dbname('world_snapshot')
    db = DBPools()
    async with db.sqlorContext(dbname) as sor:
        cnt = await sor.sqlExe('select count(*) as cnt from world_snapshot' + where, ns)
        total = int(cnt[0].cnt) if cnt else 0
        rows = await sor.sqlExe(
            'select id, world_id, name, snapshot_type, status, snapshot_date, created_at'
            ' from world_snapshot' + where +
            ' order by created_at desc limit ' + str(page_size) + ' offset ' + str(offset),
            ns)
    return {'list': [_to_dict(r) for r in (rows or [])], 'total': total,
            'page': page, 'page_size': page_size}


async def update_snapshot(params):
    """Partial update of a snapshot record (id required)."""
    params = params or {}
    snapshot_id = params.get('id')
    if not snapshot_id:
        return {'code': 'PARAM_REQUIRED', 'message': 'id is required', 'field': 'id', 'detail': ''}
    upd = {}
    for k, v in params.items():
        if k in ('id', 'created_at', 'snapshot_json'):
            continue
        if k.endswith('_text'):
            continue
        if v is None or v == '':
            continue
        upd[k] = v
    if not upd:
        return {'code': 'NO_FIELDS_TO_UPDATE', 'message': 'no fields to update', 'field': '', 'detail': ''}
    upd['id'] = snapshot_id
    dbname = ServerEnv().get_module_dbname('world_snapshot')
    db = DBPools()
    async with db.sqlorContext(dbname) as sor:
        await sor.U('world_snapshot', upd)
    return {'success': True, 'id': snapshot_id}


async def delete_snapshot(params):
    """Delete a snapshot record by id."""
    params = params or {}
    snapshot_id = params.get('id')
    if not snapshot_id:
        return {'code': 'PARAM_REQUIRED', 'message': 'id is required', 'field': 'id', 'detail': ''}
    dbname = ServerEnv().get_module_dbname('world_snapshot')
    db = DBPools()
    async with db.sqlorContext(dbname) as sor:
        await sor.D('world_snapshot', {'id': snapshot_id})
    return {'success': True, 'id': snapshot_id}


async def restore_snapshot(params):
    """Validate snapshot json, mark snapshot restored, link world status.

    World linkage goes through the world module's own set_world_mode()
    contract when available (keeps mode transition validation intact).
    """
    params = params or {}
    snapshot_id = params.get('id')
    if not snapshot_id:
        return {'code': 'PARAM_REQUIRED', 'message': 'id is required', 'field': 'id', 'detail': ''}
    env = ServerEnv()
    dbname = env.get_module_dbname('world_snapshot')
    db = DBPools()
    async with db.sqlorContext(dbname) as sor:
        rows = await sor.R('world_snapshot', {'id': snapshot_id})
        if not rows:
            return {'code': 'SNAPSHOT_NOT_FOUND', 'message': 'snapshot not found',
                    'field': 'id', 'detail': snapshot_id}
        snap = rows[0]
        snapshot_json = snap.snapshot_json or ''
        try:
            data = json.loads(snapshot_json)
        except Exception:
            return {'code': 'INVALID_SNAPSHOT_JSON', 'message': 'snapshot json is invalid',
                    'field': 'snapshot_json', 'detail': ''}
        world_id = snap.world_id
        await sor.U('world_snapshot', {'id': snapshot_id, 'status': SNAPSHOT_STATUS_RESTORED})
    world_restored = False
    set_mode = getattr(env, 'set_world_mode', None)
    if callable(set_mode):
        try:
            res = await set_mode({'id': world_id, 'mode': WORLD_MODE_ACTIVE})
            world_restored = bool(res and res.get('success'))
        except Exception:
            world_restored = False
    return {'success': True, 'id': snapshot_id, 'world_id': world_id,
            'world_restored': world_restored, 'data': data}


async def list_worlds_for_snapshot():
    """Dropdown source: worlds for the snapshot picker."""
    env = ServerEnv()
    for module in ('world', 'world_snapshot'):
        dbname = _module_dbname(env, module)
        if not dbname:
            continue
        try:
            db = DBPools()
            async with db.sqlorContext(dbname) as sor:
                rows = await sor.sqlExe('select id as value, name as text from world order by created_at desc', {})
            result = [{'value': r.value, 'text': r.text} for r in (rows or [])]
            if result:
                return result
        except Exception:
            continue
    return []


async def list_snapshot_types():
    return await _get_appcodes('snapshot_type')


async def list_snapshot_statuses():
    return await _get_appcodes('snapshot_status')


def load_world_snapshot(env=None):
    """Register all world_snapshot functions to ServerEnv."""
    if env is None:
        env = ServerEnv()
    env.create_snapshot = create_snapshot
    env.create_snapshots = create_snapshot
    env.update_snapshot = update_snapshot
    env.update_snapshots = update_snapshot
    env.delete_snapshot = delete_snapshot
    env.delete_snapshots = delete_snapshot
    env.get_snapshot = get_snapshot
    env.list_snapshots = list_snapshots
    env.restore_snapshot = restore_snapshot
    env.list_worlds_for_snapshot = list_worlds_for_snapshot
    env.list_snapshot_types = list_snapshot_types
    env.list_snapshot_statuses = list_snapshot_statuses
    return env