yumoqing
a72637b2a3
- Updated core modules: __init__, ai_config, ai_core, attachment, contract, enhanced_contract - Updated mysql.ddl.sql - Updated UI files: contract_detail, contract_edit, contract_list - Added new files: api/*.dspy, contract_list.dspy
58 lines
2.2 KiB
Python
58 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""List contract"""
|
|
import json
|
|
|
|
result = {'success': False, 'rows': [], 'total': 0}
|
|
|
|
try:
|
|
dbname = get_module_dbname('contract_management')
|
|
async with DBPools().sqlorContext(dbname) as sor:
|
|
where_clauses = []
|
|
where_ns = {}
|
|
|
|
customer_id = params_kw.get('customer_id', '')
|
|
status = params_kw.get('status', '')
|
|
|
|
if customer_id:
|
|
where_clauses.append("customer_id=${customer_id}$")
|
|
where_ns['customer_id'] = customer_id
|
|
if status:
|
|
where_clauses.append("status=${status}$")
|
|
where_ns['status'] = status
|
|
|
|
where_sql = " AND ".join(where_clauses)
|
|
where_prefix = " WHERE " if where_clauses else ""
|
|
|
|
count_sql = f"SELECT count(*) rcnt FROM contract{where_prefix}{where_sql}"
|
|
count_rows = await sor.sqlExe(count_sql, where_ns)
|
|
total = 0
|
|
if count_rows and len(count_rows) > 0:
|
|
r = count_rows[0]
|
|
if hasattr(r, 'keys'):
|
|
total = r.get('rcnt', 0)
|
|
elif isinstance(r, dict):
|
|
total = r.get('rcnt', 0)
|
|
elif hasattr(r, 'rcnt'):
|
|
total = r.rcnt
|
|
|
|
if total > 0:
|
|
ns = {'page': int(params_kw.get('page', 1)), 'rows': int(params_kw.get('rows', 20)), 'sort': params_kw.get('sort', 'created_at')}
|
|
sql = f"SELECT id, title, contract_number, contract_type, customer_id, amount, sign_date, start_date, end_date, status, owner_id, org_id, created_at, updated_at FROM contract{where_prefix}{where_sql}"
|
|
|
|
query_ns = dict(list(ns.items()) + list(where_ns.items()))
|
|
rows = await sor.sqlExe(sql, query_ns)
|
|
|
|
if isinstance(rows, dict):
|
|
result['rows'] = rows.get('rows', [])
|
|
result['total'] = rows.get('total', total)
|
|
elif rows:
|
|
result['rows'] = [dict(r) if hasattr(r, 'keys') else r for r in rows]
|
|
result['total'] = total
|
|
|
|
result['success'] = True
|
|
except Exception as e:
|
|
result['error'] = str(e)
|
|
|
|
return json.dumps(result, ensure_ascii=False, default=str)
|