diff --git a/__init__.py b/__init__.py index fedbe19..d1ed73e 100644 --- a/__init__.py +++ b/__init__.py @@ -1,9 +1,8 @@ -from .world_sync.init import ( - load_world_sync, list_world_syncs, get_world_sync, create_world_sync, - update_world_sync, delete_world_sync, execute_world_sync, -) +# -*- coding: utf-8 -*- +"""world_sync module top-level package. -__all__ = [ - "load_world_sync", "list_world_syncs", "get_world_sync", "create_world_sync", - "update_world_sync", "delete_world_sync", "execute_world_sync", -] +Re-exports the backend package so both `from world_sync import load_world_sync` +and `from world_sync.world_sync.init import load_world_sync` work. +""" +from world_sync.world_sync import * # noqa: F401,F403 +from world_sync.world_sync import __all__ # noqa: F401 diff --git a/world_sync/__init__.py b/world_sync/__init__.py index b29da27..d798b9f 100644 --- a/world_sync/__init__.py +++ b/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", - "update_world_sync", "delete_world_sync", "execute_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', ] diff --git a/world_sync/init.py b/world_sync/init.py index 3cec829..d4aa1c3 100644 --- a/world_sync/init.py +++ b/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: - 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", {}) + async with db.sqlorContext(dbname) as sor: + recs = await sor.R('world', {'id': world_id}) + return len(recs) > 0 -async def get_world_sync(ns): - rec_id = (ns or {}).get('id') - if not rec_id: +async def _run_sync(sor, world_id, sync_type): + """Execute the actual sync work inside the caller's sqlor context.""" + 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: - rows = await sor.sqlExe("select * from world_sync where id=${id}$", {'id': rec_id}) - return rows[0] if rows else None + async with db.sqlorContext(dbname) as sor: + recs = await sor.R('world_sync', {'id': sync_id}) + if not recs: + return None + return _sync_row(recs[0]) -async def create_world_sync(ns): - ns = _clean(dict(ns)) - ns['id'] = ns.get('id') or getID() - if not ns.get('sync_date'): - 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', '') +async def create_world_sync(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: - return {'success': False, 'message': '缺少 world_id'} - actions = {'0': '导入', '1': '导出', '2': '合并'} - if sync_type not in actions: - return {'success': False, 'message': '未知同步类型: %s' % sync_type} - action = actions[sync_type] - result = {'action': action, 'world_id': world_id, 'sync_type': sync_type, 'source': source, 'message': '世界%s完成' % action} - 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()} + return {'ok': False, 'message': 'world_id is required'} + if not sync_type or 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['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: - await sor.C('world_sync', rec) - return {'success': True, 'result': result} + async with db.sqlorContext(dbname) as sor: + await sor.C('world_sync', ns) + 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