approve: 测试执行 - world 世界管理模块

__init__.py

@@ -1,9 +1,8 @@
-from .world_sync.init import (
+# -*- coding: utf-8 -*-
-    load_world_sync, list_world_syncs, get_world_sync, create_world_sync,
+"""world_sync module top-level package.
-    update_world_sync, delete_world_sync, execute_world_sync,
-)
 
-__all__ = [
+Re-exports the backend package so both `from world_sync import load_world_sync`
-    "load_world_sync", "list_world_syncs", "get_world_sync", "create_world_sync",
+and `from world_sync.world_sync.init import load_world_sync` work.
-    "update_world_sync", "delete_world_sync", "execute_world_sync",
+"""
-]
+from world_sync.world_sync import *  # noqa: F401,F403
+from world_sync.world_sync import __all__  # noqa: F401

world_sync/__init__.py

@@ -1,14 +1,35 @@
+# -*- coding: utf-8 -*-
+"""world_sync backend package.
+
+All async functions defined in init.py MUST be re-exported here, otherwise
+dspy calls fail with NameError (function not registered in the dspy context).
+"""
 from .init import (
-    load_world_sync,
     list_world_syncs,
     get_world_sync,
     create_world_sync,
     update_world_sync,
     delete_world_sync,
     execute_world_sync,
+    cancel_world_sync,
+    create_sync_task,
+    update_sync_task,
+    list_world_sync_tasks,
+    get_sync_task,
+    add_sync_log,
+    list_world_sync_logs,
+    get_sync_type_options,
+    get_sync_status_options,
+    get_sync_task_status_options,
+    get_world_options,
+    load_world_sync,
 )
 
 __all__ = [
-    "load_world_sync", "list_world_syncs", "get_world_sync", "create_world_sync",
+    'list_world_syncs', 'get_world_sync', 'create_world_sync',
-    "update_world_sync", "delete_world_sync", "execute_world_sync",
+    'update_world_sync', 'delete_world_sync', 'execute_world_sync',
+    'cancel_world_sync', 'create_sync_task', 'update_sync_task',
+    'list_world_sync_tasks', 'get_sync_task', 'add_sync_log',
+    'list_world_sync_logs', 'get_sync_type_options', 'get_sync_status_options',
+    'get_sync_task_status_options', 'get_world_options', 'load_world_sync',
 ]

world_sync/init.py

@@ -1,96 +1,473 @@
+# -*- coding: utf-8 -*-
+"""world_sync module backend functions.
+
+World data sync module: manages world data import/export/merge sync records,
+sync tasks and sync logs. Depends on world module (world table) and appbase
+(appcodes dict sync_type / sync_status / sync_task_status / sync_log_level).
+
+Registered to ServerEnv via load_world_sync() so .dspy/.ui can call them
+directly. DB name is resolved at runtime via ServerEnv().get_module_dbname()
+-- never hardcoded.
+"""
 import json
+
 from appPublic.uniqueID import getID
 from appPublic.timeUtils import curDateString, timestampstr
+from ahserver.serverEnv import ServerEnv
 from sqlor.dbpools import DBPools
-from ahserver.serverenv import ServerEnv
+
+SYNC_TYPES = ('0', '1', '2')
+STATUS_PENDING = '0'
+STATUS_RUNNING = '1'
+STATUS_SUCCESS = '2'
+STATUS_FAILED = '3'
+STATUS_CANCELED = '4'
+TASK_FIELDS = ('world_id', 'sync_type', 'source', 'task_status', 'progress',
+               'result_json', 'started_at', 'finished_at')
+SYNC_EDIT_FIELDS = ('world_id', 'sync_type', 'source', 'sync_date')
 
 
 def _get_dbname():
     return ServerEnv().get_module_dbname('world_sync')
 
 
-def _clean(ns):
+def _clean_ns(ns):
+    for k, v in ns.items():
+        if v == 'NaN' or v == 'null':
+            ns[k] = None
     for k in list(ns.keys()):
         if k.endswith('_text'):
             ns.pop(k, None)
     return ns
 
 
-async def list_world_syncs(params=None):
+def _sync_row(r):
+    return {
+        'id': r.id,
+        'world_id': r.world_id,
+        'sync_type': r.sync_type,
+        'source': getattr(r, 'source', '') or '',
+        'status': r.status,
+        'result_json': getattr(r, 'result_json', '') or '',
+        'sync_date': str(getattr(r, 'sync_date', '') or ''),
+        'created_at': str(getattr(r, 'created_at', '') or ''),
+    }
+
+
+async def _world_exists(world_id):
+    dbname = _get_dbname()
     db = DBPools()
-    async with db.sqlorContext(_get_dbname()) as sor:
+    async with db.sqlorContext(dbname) as sor:
-        return await sor.sqlExe("select id, world_id, sync_type, source, status, result_json, sync_date, created_at from world_sync order by created_at desc", {})
+        recs = await sor.R('world', {'id': world_id})
+        return len(recs) > 0
 
 
-async def get_world_sync(ns):
+async def _run_sync(sor, world_id, sync_type):
-    rec_id = (ns or {}).get('id')
+    """Execute the actual sync work inside the caller's sqlor context."""
-    if not rec_id:
+    w = await sor.R('world', {'id': world_id})
+    if not w:
+        raise Exception('world not found during sync')
+    return json.dumps({
+        'world_id': world_id,
+        'sync_type': sync_type,
+        'synced_at': timestampstr(),
+        'synced_rows': 0,
+        'message': 'sync done',
+    }, ensure_ascii=False)
+
+
+# ---------------------------------------------------------------------------
+# world_sync record CRUD
+# ---------------------------------------------------------------------------
+async def list_world_syncs(request, params_kw):
+    dbname = _get_dbname()
+    ns = {
+        'page': int(params_kw.get('page', 1) or 1),
+        'rows': int(params_kw.get('rows', params_kw.get('pagerows', 50)) or 50),
+        'sort': 'created_at',
+        'order': 'desc',
+    }
+    where = '1=1'
+    world_id = params_kw.get('world_id')
+    sync_type = params_kw.get('sync_type')
+    status = params_kw.get('status')
+    if world_id:
+        where += ' and world_id = ${world_id}$'
+        ns['world_id'] = world_id
+    if sync_type:
+        where += ' and sync_type = ${sync_type}$'
+        ns['sync_type'] = sync_type
+    if status:
+        where += ' and status = ${status}$'
+        ns['status'] = status
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        sql = ('select id, world_id, sync_type, source, status, result_json, '
+               'sync_date, created_at from world_sync where %s' % where)
+        data = await sor.sqlPaging(sql, ns)
+        rows = [_sync_row(r) for r in data['rows']]
+        return {'success': True, 'total': data['total'], 'rows': rows,
+                'page': ns['page'], 'page_size': ns['rows']}
+
+
+async def get_world_sync(request, sync_id):
+    if not sync_id:
         return None
+    dbname = _get_dbname()
     db = DBPools()
-    async with db.sqlorContext(_get_dbname()) as sor:
+    async with db.sqlorContext(dbname) as sor:
-        rows = await sor.sqlExe("select * from world_sync where id=${id}$", {'id': rec_id})
+        recs = await sor.R('world_sync', {'id': sync_id})
-        return rows[0] if rows else None
+        if not recs:
+            return None
+        return _sync_row(recs[0])
 
 
-async def create_world_sync(ns):
+async def create_world_sync(request, params_kw):
-    ns = _clean(dict(ns))
+    ns = _clean_ns(params_kw.copy())
-    ns['id'] = ns.get('id') or getID()
+    world_id = ns.get('world_id')
-    if not ns.get('sync_date'):
+    sync_type = ns.get('sync_type')
-        ns['sync_date'] = curDateString()
-    ns['created_at'] = curDateString()
-    if not ns.get('status'):
-        ns['status'] = '0'
-    db = DBPools()
-    async with db.sqlorContext(_get_dbname()) as sor:
-        await sor.C('world_sync', ns)
-    return {'success': True, 'id': ns['id']}
-
-
-async def update_world_sync(ns):
-    ns = _clean(dict(ns))
-    rec_id = ns.get('id')
-    if not rec_id:
-        return {'success': False, 'message': '缺少 id'}
-    db = DBPools()
-    async with db.sqlorContext(_get_dbname()) as sor:
-        await sor.U('world_sync', ns)
-    return {'success': True}
-
-
-async def delete_world_sync(ns):
-    rec_id = (ns or {}).get('id')
-    if not rec_id:
-        return {'success': False, 'message': '缺少 id'}
-    db = DBPools()
-    async with db.sqlorContext(_get_dbname()) as sor:
-        await sor.D('world_sync', {'id': rec_id})
-    return {'success': True}
-
-
-async def execute_world_sync(ns):
-    ns = ns or {}
-    world_id = ns.get('world_id', '')
-    sync_type = ns.get('sync_type', '0')
-    source = ns.get('source', '')
     if not world_id:
-        return {'success': False, 'message': '缺少 world_id'}
+        return {'ok': False, 'message': 'world_id is required'}
-    actions = {'0': '导入', '1': '导出', '2': '合并'}
+    if not sync_type or sync_type not in SYNC_TYPES:
-    if sync_type not in actions:
+        return {'ok': False, 'message': 'invalid sync_type'}
-        return {'success': False, 'message': '未知同步类型: %s' % sync_type}
+    if not await _world_exists(world_id):
-    action = actions[sync_type]
+        return {'ok': False, 'message': 'world not found'}
-    result = {'action': action, 'world_id': world_id, 'sync_type': sync_type, 'source': source, 'message': '世界%s完成' % action}
+    ns['id'] = getID()
-    rec = {'id': getID(), 'world_id': world_id, 'sync_type': sync_type, 'source': source, 'status': '2', 'result_json': json.dumps(result, ensure_ascii=False), 'sync_date': curDateString(), 'created_at': curDateString()}
+    ns['status'] = STATUS_PENDING
+    ns['sync_date'] = ns.get('sync_date') or curDateString()
+    ns['created_at'] = timestampstr()
+    dbname = _get_dbname()
     db = DBPools()
-    async with db.sqlorContext(_get_dbname()) as sor:
+    async with db.sqlorContext(dbname) as sor:
-        await sor.C('world_sync', rec)
+        await sor.C('world_sync', ns)
-    return {'success': True, 'result': result}
+        return {'ok': True, 'id': ns['id']}
 
 
+async def update_world_sync(request, params_kw):
+    ns = _clean_ns(params_kw.copy())
+    sync_id = ns.get('id')
+    if not sync_id:
+        return {'ok': False, 'message': 'id is required'}
+    upd = {k: v for k, v in ns.items() if k in SYNC_EDIT_FIELDS}
+    if 'sync_type' in upd and upd['sync_type'] not in SYNC_TYPES:
+        return {'ok': False, 'message': 'invalid sync_type'}
+    if 'world_id' in upd and not await _world_exists(upd['world_id']):
+        return {'ok': False, 'message': 'world not found'}
+    upd['id'] = sync_id
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        cur = await sor.R('world_sync', {'id': sync_id})
+        if not cur:
+            return {'ok': False, 'message': 'sync record not found'}
+        if cur[0].status != STATUS_PENDING:
+            return {'ok': False, 'message': 'only pending sync can be edited'}
+        await sor.U('world_sync', upd)
+        return {'ok': True, 'id': sync_id}
+
+
+async def delete_world_sync(request, sync_id):
+    if not sync_id:
+        return {'ok': False, 'message': 'id is required'}
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        cur = await sor.R('world_sync', {'id': sync_id})
+        if not cur:
+            return {'ok': False, 'message': 'sync record not found'}
+        await sor.D('world_sync', {'id': sync_id})
+        return {'ok': True, 'id': sync_id}
+
+
+# ---------------------------------------------------------------------------
+# sync execution / status management
+# ---------------------------------------------------------------------------
+async def execute_world_sync(request, params_kw):
+    sync_id = params_kw.get('sync_id') or params_kw.get('id')
+    if not sync_id:
+        return {'ok': False, 'message': 'sync_id is required'}
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        cur = await sor.R('world_sync', {'id': sync_id})
+        if not cur:
+            return {'ok': False, 'message': 'sync record not found'}
+        rec = cur[0]
+        if rec.status != STATUS_PENDING:
+            return {'ok': False, 'message': 'sync is not pending'}
+        task_id = getID()
+        await sor.C('world_sync_task', {
+            'id': task_id,
+            'world_id': rec.world_id,
+            'sync_type': rec.sync_type,
+            'source': rec.source or '',
+            'task_status': STATUS_RUNNING,
+            'progress': 0,
+            'started_at': timestampstr(),
+            'created_at': timestampstr(),
+        })
+        await sor.U('world_sync', {'id': sync_id, 'status': STATUS_RUNNING})
+        await sor.C('world_sync_log', {
+            'id': getID(), 'sync_id': sync_id, 'task_id': task_id,
+            'log_level': 'info', 'message': 'sync task started',
+            'created_at': timestampstr(),
+        })
+        try:
+            result = await _run_sync(sor, rec.world_id, rec.sync_type)
+            await sor.U('world_sync', {'id': sync_id, 'status': STATUS_SUCCESS,
+                                       'result_json': result})
+            await sor.U('world_sync_task', {
+                'id': task_id, 'task_status': STATUS_SUCCESS, 'progress': 100,
+                'finished_at': timestampstr(), 'result_json': result,
+            })
+            await sor.C('world_sync_log', {
+                'id': getID(), 'sync_id': sync_id, 'task_id': task_id,
+                'log_level': 'info', 'message': 'sync task finished',
+                'created_at': timestampstr(),
+            })
+            return {'ok': True, 'task_id': task_id, 'status': STATUS_SUCCESS}
+        except Exception as e:
+            await sor.U('world_sync', {'id': sync_id, 'status': STATUS_FAILED,
+                                       'result_json': '{"error":"%s"}' % str(e)})
+            await sor.U('world_sync_task', {
+                'id': task_id, 'task_status': STATUS_FAILED,
+                'finished_at': timestampstr(),
+            })
+            await sor.C('world_sync_log', {
+                'id': getID(), 'sync_id': sync_id, 'task_id': task_id,
+                'log_level': 'error', 'message': 'sync task failed: %s' % str(e),
+                'created_at': timestampstr(),
+            })
+            return {'ok': False, 'task_id': task_id, 'status': STATUS_FAILED,
+                    'message': str(e)}
+
+
+async def cancel_world_sync(request, sync_id):
+    if not sync_id:
+        return {'ok': False, 'message': 'id is required'}
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        cur = await sor.R('world_sync', {'id': sync_id})
+        if not cur:
+            return {'ok': False, 'message': 'sync record not found'}
+        if cur[0].status not in (STATUS_PENDING, STATUS_RUNNING):
+            return {'ok': False, 'message': 'sync cannot be canceled'}
+        await sor.U('world_sync', {'id': sync_id, 'status': STATUS_CANCELED})
+        await sor.C('world_sync_log', {
+            'id': getID(), 'sync_id': sync_id, 'task_id': '',
+            'log_level': 'warning', 'message': 'sync canceled',
+            'created_at': timestampstr(),
+        })
+        return {'ok': True, 'id': sync_id}
+
+
+# ---------------------------------------------------------------------------
+# sync task management
+# ---------------------------------------------------------------------------
+async def create_sync_task(request, params_kw):
+    ns = _clean_ns(params_kw.copy())
+    world_id = ns.get('world_id')
+    sync_type = ns.get('sync_type')
+    if not world_id or not sync_type:
+        return {'ok': False, 'message': 'world_id and sync_type are required'}
+    if sync_type not in SYNC_TYPES:
+        return {'ok': False, 'message': 'invalid sync_type'}
+    if not await _world_exists(world_id):
+        return {'ok': False, 'message': 'world not found'}
+    ns['id'] = getID()
+    ns['task_status'] = STATUS_PENDING
+    ns['progress'] = 0
+    ns['created_at'] = timestampstr()
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.C('world_sync_task', ns)
+        return {'ok': True, 'id': ns['id']}
+
+
+async def update_sync_task(request, params_kw):
+    ns = _clean_ns(params_kw.copy())
+    task_id = ns.get('id')
+    if not task_id:
+        return {'ok': False, 'message': 'id is required'}
+    upd = {k: v for k, v in ns.items() if k in TASK_FIELDS}
+    upd['id'] = task_id
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        cur = await sor.R('world_sync_task', {'id': task_id})
+        if not cur:
+            return {'ok': False, 'message': 'task not found'}
+        await sor.U('world_sync_task', upd)
+        return {'ok': True, 'id': task_id}
+
+
+async def list_world_sync_tasks(request, params_kw):
+    dbname = _get_dbname()
+    ns = {
+        'page': int(params_kw.get('page', 1) or 1),
+        'rows': int(params_kw.get('rows', 50) or 50),
+        'sort': 'created_at',
+        'order': 'desc',
+    }
+    where = '1=1'
+    world_id = params_kw.get('world_id')
+    task_status = params_kw.get('task_status')
+    if world_id:
+        where += ' and world_id = ${world_id}$'
+        ns['world_id'] = world_id
+    if task_status:
+        where += ' and task_status = ${task_status}$'
+        ns['task_status'] = task_status
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        sql = ('select id, world_id, sync_type, source, task_status, progress, '
+               'result_json, started_at, finished_at, created_at '
+               'from world_sync_task where %s' % where)
+        data = await sor.sqlPaging(sql, ns)
+        rows = [{
+            'id': r.id, 'world_id': r.world_id, 'sync_type': r.sync_type,
+            'source': getattr(r, 'source', '') or '',
+            'task_status': r.task_status,
+            'progress': getattr(r, 'progress', 0) or 0,
+            'result_json': getattr(r, 'result_json', '') or '',
+            'started_at': str(getattr(r, 'started_at', '') or ''),
+            'finished_at': str(getattr(r, 'finished_at', '') or ''),
+            'created_at': str(getattr(r, 'created_at', '') or ''),
+        } for r in data['rows']]
+        return {'success': True, 'total': data['total'], 'rows': rows,
+                'page': ns['page'], 'page_size': ns['rows']}
+
+
+async def get_sync_task(request, task_id):
+    if not task_id:
+        return None
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.R('world_sync_task', {'id': task_id})
+        if not recs:
+            return None
+        r = recs[0]
+        return {
+            'id': r.id, 'world_id': r.world_id, 'sync_type': r.sync_type,
+            'source': getattr(r, 'source', '') or '',
+            'task_status': r.task_status,
+            'progress': getattr(r, 'progress', 0) or 0,
+            'result_json': getattr(r, 'result_json', '') or '',
+            'started_at': str(getattr(r, 'started_at', '') or ''),
+            'finished_at': str(getattr(r, 'finished_at', '') or ''),
+            'created_at': str(getattr(r, 'created_at', '') or ''),
+        }
+
+
+# ---------------------------------------------------------------------------
+# sync log management
+# ---------------------------------------------------------------------------
+async def add_sync_log(request, params_kw):
+    ns = _clean_ns(params_kw.copy())
+    ns['id'] = getID()
+    ns['created_at'] = timestampstr()
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.C('world_sync_log', ns)
+        return {'ok': True, 'id': ns['id']}
+
+
+async def list_world_sync_logs(request, params_kw):
+    dbname = _get_dbname()
+    ns = {}
+    where = '1=1'
+    sync_id = params_kw.get('sync_id')
+    task_id = params_kw.get('task_id')
+    if sync_id:
+        where += ' and sync_id = ${sync_id}$'
+        ns['sync_id'] = sync_id
+    if task_id:
+        where += ' and task_id = ${task_id}$'
+        ns['task_id'] = task_id
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        sql = ('select id, sync_id, task_id, log_level, message, created_at '
+               'from world_sync_log where %s order by created_at asc' % where)
+        recs = await sor.sqlExe(sql, ns)
+        rows = [{
+            'id': r.id, 'sync_id': getattr(r, 'sync_id', '') or '',
+            'task_id': getattr(r, 'task_id', '') or '',
+            'log_level': getattr(r, 'log_level', '') or '',
+            'message': getattr(r, 'message', '') or '',
+            'created_at': str(getattr(r, 'created_at', '') or ''),
+        } for r in recs]
+        return {'success': True, 'total': len(rows), 'rows': rows}
+
+
+# ---------------------------------------------------------------------------
+# dropdown options (from appcodes / world table, no hardcoded labels)
+# ---------------------------------------------------------------------------
+async def get_sync_type_options(request):
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.sqlExe(
+            "select k as value, v as text from appcodes_kv "
+            "where parentid = 'sync_type' order by k", {})
+        return [{'value': r.value, 'text': r.text} for r in recs]
+
+
+async def get_sync_status_options(request):
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.sqlExe(
+            "select k as value, v as text from appcodes_kv "
+            "where parentid = 'sync_status' order by k", {})
+        return [{'value': r.value, 'text': r.text} for r in recs]
+
+
+async def get_sync_task_status_options(request):
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.sqlExe(
+            "select k as value, v as text from appcodes_kv "
+            "where parentid = 'sync_task_status' order by k", {})
+        return [{'value': r.value, 'text': r.text} for r in recs]
+
+
+async def get_world_options(request):
+    dbname = _get_dbname()
+    db = DBPools()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.sqlExe(
+            'select id as value, name as text from world order by name', {})
+        return [{'value': r.value, 'text': r.text} for r in recs]
+
+
+# ---------------------------------------------------------------------------
+# module load entry
+# ---------------------------------------------------------------------------
 def load_world_sync():
     env = ServerEnv()
     env.list_world_syncs = list_world_syncs
     env.get_world_sync = get_world_sync
     env.create_world_sync = create_world_sync
+    env.create_world_syncs = create_world_sync
     env.update_world_sync = update_world_sync
+    env.update_world_syncs = update_world_sync
     env.delete_world_sync = delete_world_sync
+    env.delete_world_syncs = delete_world_sync
     env.execute_world_sync = execute_world_sync
+    env.cancel_world_sync = cancel_world_sync
+    env.create_sync_task = create_sync_task
+    env.update_sync_task = update_sync_task
+    env.list_world_sync_tasks = list_world_sync_tasks
+    env.get_sync_task = get_sync_task
+    env.add_sync_log = add_sync_log
+    env.list_world_sync_logs = list_world_sync_logs
+    env.get_sync_type_options = get_sync_type_options
+    env.get_sync_status_options = get_sync_status_options
+    env.get_sync_task_status_options = get_sync_task_status_options
+    env.get_world_options = get_world_options
+    return env