result = {'success': False, 'rows': [], 'total': 0, 'page': 1, 'page_size': 50} try: # 参数解析 try: page = int(params_kw.get('page', 1)) except (ValueError, TypeError): page = 1 if page < 1: page = 1 rows_per_page = int(params_kw.get('rows', params_kw.get('pagerows', 50))) offset = (page - 1) * rows_per_page sort_field = params_kw.get('sort', 'use_time desc') # 构建 WHERE 条件(只支持常用筛选字段,手动拼接避免框架开销) conditions = ['1=1'] ns = {} llmid = params_kw.get('llmid') if llmid: conditions.append('llmid=${llmid}$') ns['llmid'] = llmid status = params_kw.get('status') if status: conditions.append('status=${status}$') ns['status'] = status accounting_status = params_kw.get('accounting_status') if accounting_status: conditions.append('accounting_status=${accounting_status}$') ns['accounting_status'] = accounting_status userid = params_kw.get('userid') if userid: conditions.append('userid=${userid}$') ns['userid'] = userid userorgid = params_kw.get('userorgid') if userorgid: conditions.append('userorgid=${userorgid}$') ns['userorgid'] = userorgid start_date = params_kw.get('start_date') if start_date: conditions.append('use_date>=${start_date}$') ns['start_date'] = start_date end_date = params_kw.get('end_date') if end_date: conditions.append('use_date<=${end_date}$') ns['end_date'] = end_date where = 'WHERE ' + ' AND '.join(conditions) # 列表字段(不含 usages/ioinfo 大 TEXT) select_fields = 'id, llmid, use_date, use_time, userid, transno, responsed_seconds, finish_seconds, status, taskid, amount, cost, userorgid, ownerid, accounting_status' db = DBPools() dbname = get_module_dbname('llmage') async with db.sqlorContext(dbname) as sor: # count count_recs = await sor.sqlExe(f'SELECT count(*) as cnt FROM llmusage {where}', ns) total = count_recs[0].cnt if count_recs else 0 # data rows = await sor.sqlExe( f'SELECT {select_fields} FROM llmusage {where} ORDER BY {sort_field} LIMIT {rows_per_page} OFFSET {offset}', ns ) result['success'] = True result['total'] = total result['rows'] = rows if rows else [] result['page'] = page result['page_size'] = rows_per_page except Exception as e: debug(f'llmusage_list error: {format_exc()}') result['error'] = str(e) return json.dumps(result, ensure_ascii=False, default=str)