9115435441
- import_categories_and_products now: 1. Calls resource module for standardized data 2. Categories: skip if exists (name+parent+org), else create 3. Products: update if exists (resource_ref_id+org), else create - product.json: added resource_ref_id field for resource module internal ID mapping
1441 lines
59 KiB
Python
1441 lines
59 KiB
Python
"""Product Management Core Business Logic - org_id isolated per reseller"""
|
||
import json
|
||
import time
|
||
import datetime
|
||
from appPublic.uniqueID import getID
|
||
from appPublic.log import info, error, exception
|
||
from sqlor.dbpools import DBPools
|
||
from ahserver.serverenv import ServerEnv
|
||
|
||
|
||
MODULE_NAME = "product_management"
|
||
|
||
|
||
class ProductManager:
|
||
"""Core manager for product catalog, category tree, and operator configs.
|
||
|
||
All operations are scoped to org_id (reseller institution).
|
||
Different resellers have completely independent category trees and products.
|
||
"""
|
||
|
||
def _get_dbname(self):
|
||
"""Get module database name from ServerEnv."""
|
||
env = ServerEnv()
|
||
return env.get_module_dbname(MODULE_NAME)
|
||
|
||
def _get_current_org_id(self):
|
||
"""Get current user's organization ID from ServerEnv."""
|
||
env = ServerEnv()
|
||
return getattr(env, 'orgid', None) or getattr(env, 'org_id', '0')
|
||
|
||
async def get_category_tree(self, org_id=None):
|
||
"""Get full category tree for a specific org (reseller)."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
|
||
dbname = self._get_dbname()
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
sql = """SELECT * FROM product_category
|
||
WHERE org_id = ${org_id}$ AND status = '1'
|
||
ORDER BY sort_order ASC, name ASC"""
|
||
rows = await sor.sqlExe(sql, {'org_id': org_id})
|
||
rows = rows or []
|
||
|
||
# Build tree
|
||
nodes = [dict(r) for r in rows]
|
||
node_map = {n['id']: n for n in nodes}
|
||
tree = []
|
||
|
||
for node in nodes:
|
||
node['children'] = []
|
||
parent_id = node.get('parent_id', '0')
|
||
if parent_id == '0' or parent_id not in node_map:
|
||
tree.append(node)
|
||
else:
|
||
parent = node_map.get(parent_id)
|
||
if parent:
|
||
parent['children'].append(node)
|
||
|
||
return {'success': True, 'tree': tree}
|
||
|
||
async def get_products_by_category(self, category_id, org_id=None, status='1'):
|
||
"""Get all products under a category for a specific org (reseller)."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
|
||
dbname = self._get_dbname()
|
||
|
||
# Get all sub-category IDs recursively within same org
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
all_ids = [category_id]
|
||
queue = [category_id]
|
||
while queue:
|
||
parent = queue.pop(0)
|
||
children = await sor.sqlExe(
|
||
"SELECT id FROM product_category WHERE parent_id = ${pid}$ AND org_id = ${org_id}$",
|
||
{'pid': parent, 'org_id': org_id}
|
||
)
|
||
for c in (children or []):
|
||
cid = c['id']
|
||
all_ids.append(cid)
|
||
queue.append(cid)
|
||
|
||
if not all_ids:
|
||
return {'success': True, 'products': [], 'total': 0}
|
||
|
||
# Use IN clause with proper parameterization
|
||
param_keys = []
|
||
params = {'org_id': org_id}
|
||
for i, cid in enumerate(all_ids):
|
||
key = f'cid_{i}'
|
||
param_keys.append(f'${key}$')
|
||
params[key] = cid
|
||
|
||
placeholders = ','.join(param_keys)
|
||
sql = f"""SELECT p.*, pc.name as category_name
|
||
FROM product p
|
||
LEFT JOIN product_category pc ON p.category_id = pc.id AND p.org_id = pc.org_id
|
||
WHERE p.category_id IN ({placeholders})
|
||
AND p.org_id = ${org_id}$
|
||
AND p.status = ${status}$
|
||
ORDER BY p.sort_order ASC, p.created_at DESC"""
|
||
params['status'] = status
|
||
|
||
rows = await sor.sqlExe(sql, params)
|
||
|
||
today = datetime.date.today().isoformat()
|
||
products = []
|
||
for r in (rows or []):
|
||
r = dict(r)
|
||
enabled = str(r.get('enabled_date', '') or '')
|
||
expired = str(r.get('expired_date', '') or '')
|
||
r['is_active'] = True
|
||
if enabled and enabled > today:
|
||
r['is_active'] = False
|
||
if expired and expired < today:
|
||
r['is_active'] = False
|
||
# Parse extra_json
|
||
extra_str = r.get('extra_json', '')
|
||
if extra_str:
|
||
try:
|
||
r['extra_parsed'] = json.loads(extra_str)
|
||
except:
|
||
r['extra_parsed'] = {}
|
||
products.append(r)
|
||
|
||
return {'success': True, 'products': products, 'total': len(products)}
|
||
|
||
async def get_product_brief(self, product_id=None, product_code=None, category_id=None, org_id=None):
|
||
"""Get product brief for current org (reseller)."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
|
||
dbname = self._get_dbname()
|
||
|
||
conditions = ["p.status = '1'", "p.org_id = ${org_id}$"]
|
||
params = {'org_id': org_id}
|
||
|
||
if product_id:
|
||
conditions.append("p.id = ${product_id}$")
|
||
params['product_id'] = product_id
|
||
elif product_code:
|
||
conditions.append("p.product_code = ${product_code}$")
|
||
params['product_code'] = product_code
|
||
|
||
if category_id:
|
||
conditions.append("p.category_id = ${category_id}$")
|
||
params['category_id'] = category_id
|
||
|
||
where_clause = " AND ".join(conditions)
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
sql = f"""SELECT p.id, p.product_code, p.product_name, p.category_id,
|
||
pc.name as category_name, p.brief_intro,
|
||
p.price, p.currency, p.enabled_date, p.expired_date,
|
||
p.status, p.product_type, p.extra_json
|
||
FROM product p
|
||
LEFT JOIN product_category pc ON p.category_id = pc.id AND p.org_id = pc.org_id
|
||
WHERE {where_clause}
|
||
ORDER BY p.sort_order ASC, p.created_at DESC"""
|
||
rows = await sor.sqlExe(sql, params)
|
||
|
||
today = datetime.date.today().isoformat()
|
||
result = []
|
||
for r in (rows or []):
|
||
r = dict(r)
|
||
enabled = str(r.get('enabled_date', '') or '')
|
||
expired = str(r.get('expired_date', '') or '')
|
||
r['is_active'] = True
|
||
if enabled and enabled > today:
|
||
r['is_active'] = False
|
||
if expired and expired < today:
|
||
r['is_active'] = False
|
||
result.append(r)
|
||
|
||
return {'success': True, 'data': result, 'total': len(result)}
|
||
|
||
async def get_product_detail(self, product_id=None, product_code=None, org_id=None, user_id=None):
|
||
"""Get product detail for current org (reseller).
|
||
|
||
Returns product_info + category_info + operator_config + extra_parsed.
|
||
"""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
if not user_id:
|
||
env = ServerEnv()
|
||
try:
|
||
user_id = await env.get_user()
|
||
except:
|
||
user_id = 'anonymous'
|
||
|
||
dbname = self._get_dbname()
|
||
|
||
conditions = ["p.org_id = ${org_id}$"]
|
||
params = {'org_id': org_id}
|
||
if product_id:
|
||
conditions.append("p.id = ${product_id}$")
|
||
params['product_id'] = product_id
|
||
elif product_code:
|
||
conditions.append("p.product_code = ${product_code}$")
|
||
params['product_code'] = product_code
|
||
|
||
if not conditions:
|
||
return {'success': False, 'error': 'Missing product_id or product_code'}
|
||
|
||
where_clause = " AND ".join(conditions)
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
sql = f"""SELECT p.*, pc.name as category_name, pc.description as category_description
|
||
FROM product p
|
||
LEFT JOIN product_category pc ON p.category_id = pc.id AND p.org_id = pc.org_id
|
||
WHERE {where_clause}"""
|
||
rows = await sor.sqlExe(sql, params)
|
||
if not rows:
|
||
return {'success': False, 'error': 'Product not found or no access'}
|
||
|
||
product_info = dict(rows[0])
|
||
|
||
# Parse extra_json
|
||
extra_parsed = {}
|
||
extra_str = product_info.get('extra_json', '')
|
||
if extra_str:
|
||
try:
|
||
extra_parsed = json.loads(extra_str)
|
||
except:
|
||
extra_parsed = {'_raw': extra_str}
|
||
product_info['extra_parsed'] = extra_parsed
|
||
|
||
# Get operator config for this category
|
||
config_sql = """SELECT * FROM product_type_config
|
||
WHERE category_id = ${category_id}$
|
||
AND org_id = ${org_id}$
|
||
AND enabled_flg = '1'
|
||
AND (operator_id = ${user_id}$ OR operator_id = '0')
|
||
ORDER BY created_at DESC LIMIT 1"""
|
||
config_rows = await sor.sqlExe(config_sql, {
|
||
'category_id': product_info['category_id'],
|
||
'org_id': org_id,
|
||
'user_id': user_id
|
||
})
|
||
|
||
operator_config = {}
|
||
if config_rows:
|
||
operator_config = dict(config_rows[0])
|
||
config_json = operator_config.get('config_json', '')
|
||
if config_json:
|
||
try:
|
||
operator_config['config_parsed'] = json.loads(config_json)
|
||
except:
|
||
operator_config['config_parsed'] = {}
|
||
|
||
return {
|
||
'success': True,
|
||
'data': {
|
||
'product_info': product_info,
|
||
'category_info': {
|
||
'name': product_info.get('category_name'),
|
||
'description': product_info.get('category_description')
|
||
},
|
||
'operator_config': operator_config
|
||
}
|
||
}
|
||
|
||
async def purchase_product(self, product_id, quantity=1, purchase_data=None, org_id=None, user_id=None):
|
||
"""Purchase a product within current org (reseller)."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
if not user_id:
|
||
env = ServerEnv()
|
||
try:
|
||
user_id = await env.get_user()
|
||
except:
|
||
return {'success': False, 'message': 'User not authenticated'}
|
||
|
||
if not product_id:
|
||
return {'success': False, 'message': 'Missing product_id'}
|
||
|
||
dbname = self._get_dbname()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
quantity = int(quantity) if quantity else 1
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
sql = """SELECT * FROM product WHERE id = ${product_id}$ AND status = '1' AND org_id = ${org_id}$"""
|
||
rows = await sor.sqlExe(sql, {'product_id': product_id, 'org_id': org_id})
|
||
if not rows:
|
||
return {'success': False, 'message': 'Product not found or no access'}
|
||
|
||
product = dict(rows[0])
|
||
|
||
today = datetime.date.today().isoformat()
|
||
enabled = str(product.get('enabled_date', '') or '')
|
||
expired = str(product.get('expired_date', '') or '')
|
||
if enabled and enabled > today:
|
||
return {'success': False, 'message': 'Product not yet enabled'}
|
||
if expired and expired < today:
|
||
return {'success': False, 'message': 'Product has expired'}
|
||
|
||
order_id = getID()
|
||
order_data = {
|
||
'id': order_id,
|
||
'product_id': product_id,
|
||
'product_code': product.get('product_code', ''),
|
||
'product_name': product.get('product_name', ''),
|
||
'buyer_id': user_id,
|
||
'buyer_org_id': org_id,
|
||
'quantity': quantity,
|
||
'unit_price': float(product.get('price', 0)),
|
||
'total_price': float(product.get('price', 0)) * quantity,
|
||
'currency': product.get('currency', 'CNY'),
|
||
'purchase_data': purchase_data or '{}',
|
||
'status': 'pending',
|
||
'created_at': now,
|
||
'updated_at': now
|
||
}
|
||
|
||
try:
|
||
await sor.C('purchase_orders', order_data)
|
||
except Exception:
|
||
pass
|
||
|
||
return {
|
||
'success': True,
|
||
'order_id': order_id,
|
||
'message': 'Purchase request submitted'
|
||
}
|
||
|
||
async def use_product(self, product_id, order_id=None, use_data=None, org_id=None, user_id=None):
|
||
"""Use a product within current org (reseller)."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
if not user_id:
|
||
env = ServerEnv()
|
||
try:
|
||
user_id = await env.get_user()
|
||
except:
|
||
return {'success': False, 'message': 'User not authenticated'}
|
||
|
||
if not product_id:
|
||
return {'success': False, 'message': 'Missing product_id'}
|
||
|
||
dbname = self._get_dbname()
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
sql = """SELECT * FROM product WHERE id = ${product_id}$ AND status = '1' AND org_id = ${org_id}$"""
|
||
rows = await sor.sqlExe(sql, {'product_id': product_id, 'org_id': org_id})
|
||
if not rows:
|
||
return {'success': False, 'message': 'Product not found or no access'}
|
||
|
||
product = dict(rows[0])
|
||
|
||
# Parse extra_json
|
||
extra_parsed = {}
|
||
extra_str = product.get('extra_json', '')
|
||
if extra_str:
|
||
try:
|
||
extra_parsed = json.loads(extra_str)
|
||
except:
|
||
pass
|
||
|
||
# Verify purchase (if table exists)
|
||
try:
|
||
purchase_sql = """SELECT * FROM purchase_orders
|
||
WHERE product_id = ${product_id}$
|
||
AND buyer_id = ${user_id}$
|
||
AND buyer_org_id = ${org_id}$
|
||
AND status IN ('active', 'pending')"""
|
||
purchases = await sor.sqlExe(purchase_sql, {
|
||
'product_id': product_id,
|
||
'user_id': user_id,
|
||
'org_id': org_id
|
||
})
|
||
if not purchases and not order_id:
|
||
return {'success': False, 'message': 'Product not purchased'}
|
||
except:
|
||
pass
|
||
|
||
return {
|
||
'success': True,
|
||
'data': {
|
||
'product_info': {
|
||
'id': product['id'],
|
||
'name': product['product_name'],
|
||
'code': product['product_code'],
|
||
'product_type': product.get('product_type', '')
|
||
},
|
||
'extra_parsed': extra_parsed
|
||
},
|
||
'message': 'Product use successful'
|
||
}
|
||
|
||
async def get_operator_config(self, category_id, org_id=None, user_id=None):
|
||
"""Get operator configuration for a category within current org."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
if not user_id:
|
||
env = ServerEnv()
|
||
try:
|
||
user_id = await env.get_user()
|
||
except:
|
||
user_id = 'anonymous'
|
||
|
||
dbname = self._get_dbname()
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
sql = """SELECT * FROM product_type_config
|
||
WHERE category_id = ${category_id}$
|
||
AND org_id = ${org_id}$
|
||
AND enabled_flg = '1'
|
||
AND (operator_id = ${user_id}$ OR operator_id = '0')
|
||
ORDER BY created_at DESC"""
|
||
rows = await sor.sqlExe(sql, {
|
||
'category_id': category_id,
|
||
'org_id': org_id,
|
||
'user_id': user_id
|
||
})
|
||
|
||
configs = []
|
||
for r in (rows or []):
|
||
r = dict(r)
|
||
config_json = r.get('config_json', '')
|
||
if config_json:
|
||
try:
|
||
r['config_parsed'] = json.loads(config_json)
|
||
except:
|
||
r['config_parsed'] = {}
|
||
configs.append(r)
|
||
|
||
return {'success': True, 'configs': configs}
|
||
|
||
async def set_operator_config(self, category_id, config_name, config_json, org_id=None, user_id=None):
|
||
"""Create or update operator configuration within current org."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
if not user_id:
|
||
env = ServerEnv()
|
||
try:
|
||
user_id = await env.get_user()
|
||
except:
|
||
return {'success': False, 'message': 'User not authenticated'}
|
||
|
||
if not config_name:
|
||
return {'success': False, 'message': 'Missing config_name'}
|
||
|
||
dbname = self._get_dbname()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
try:
|
||
json.loads(config_json)
|
||
except:
|
||
return {'success': False, 'message': 'Invalid config_json format'}
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
# Verify category belongs to org
|
||
cat_check = await sor.sqlExe(
|
||
"SELECT id FROM product_category WHERE id = ${category_id}$ AND org_id = ${org_id}$",
|
||
{'category_id': category_id, 'org_id': org_id}
|
||
)
|
||
if not cat_check:
|
||
return {'success': False, 'message': 'Category not found or no access'}
|
||
|
||
existing = await sor.sqlExe(
|
||
"""SELECT id FROM product_type_config
|
||
WHERE category_id = ${category_id}$
|
||
AND org_id = ${org_id}$
|
||
AND operator_id = ${user_id}$
|
||
AND config_name = ${config_name}$""",
|
||
{'category_id': category_id, 'org_id': org_id,
|
||
'user_id': user_id, 'config_name': config_name}
|
||
)
|
||
|
||
if existing:
|
||
config_id = existing[0]['id']
|
||
await sor.U('product_type_config', {
|
||
'id': config_id,
|
||
'config_json': config_json,
|
||
'updated_at': now
|
||
})
|
||
return {'success': True, 'id': config_id, 'message': 'Config updated'}
|
||
else:
|
||
config_id = getID()
|
||
await sor.C('product_type_config', {
|
||
'id': config_id,
|
||
'operator_id': user_id,
|
||
'org_id': org_id,
|
||
'category_id': category_id,
|
||
'config_name': config_name,
|
||
'config_json': config_json,
|
||
'enabled_flg': '1',
|
||
'created_by': user_id,
|
||
'created_at': now,
|
||
'updated_at': now
|
||
})
|
||
return {'success': True, 'id': config_id, 'message': 'Config created'}
|
||
|
||
# ─── Resource Binding ───
|
||
|
||
async def bind_resource(self, product_id, resource_type, resource_ref_id,
|
||
resource_ref_name='', quota=0, quota_unit='',
|
||
overflow_product_id=None, org_id=None):
|
||
"""Bind a resource to a product."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
dbname = self._get_dbname()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
# Verify product exists and belongs to org
|
||
prod = await sor.sqlExe(
|
||
"SELECT id FROM product WHERE id=${pid}$ AND org_id=${oid}$",
|
||
{'pid': product_id, 'oid': org_id})
|
||
if not prod:
|
||
return {'success': False, 'message': 'Product not found'}
|
||
|
||
rid = getID()
|
||
await sor.C('product_resource', {
|
||
'id': rid,
|
||
'product_id': product_id,
|
||
'resource_type': resource_type,
|
||
'resource_ref_id': resource_ref_id,
|
||
'resource_ref_name': resource_ref_name,
|
||
'quota': float(quota) if quota else 0,
|
||
'quota_unit': quota_unit,
|
||
'priority': 1,
|
||
'overflow_product_id': overflow_product_id or '',
|
||
'status': '1',
|
||
'created_at': now,
|
||
'updated_at': now
|
||
})
|
||
return {'success': True, 'id': rid}
|
||
|
||
async def unbind_resource(self, product_resource_id, org_id=None):
|
||
"""Unbind a resource from product (cascade delete suppliers)."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
dbname = self._get_dbname()
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
# Verify ownership via product
|
||
check = await sor.sqlExe(
|
||
"""SELECT pr.id FROM product_resource pr
|
||
JOIN product p ON pr.product_id=p.id
|
||
WHERE pr.id=${id}$ AND p.org_id=${oid}$""",
|
||
{'id': product_resource_id, 'oid': org_id})
|
||
if not check:
|
||
return {'success': False, 'message': 'Resource binding not found'}
|
||
|
||
# Cascade delete suppliers
|
||
await sor.sqlExe(
|
||
"DELETE FROM product_resource_supplier WHERE product_resource_id=${id}$",
|
||
{'id': product_resource_id})
|
||
await sor.D('product_resource', {'id': product_resource_id})
|
||
return {'success': True}
|
||
|
||
async def get_product_resources(self, product_id, org_id=None):
|
||
"""Get all resource bindings for a product, with suppliers."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
dbname = self._get_dbname()
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
resources = await sor.sqlExe(
|
||
"""SELECT pr.* FROM product_resource pr
|
||
JOIN product p ON pr.product_id=p.id
|
||
WHERE pr.product_id=${pid}$ AND p.org_id=${oid}$
|
||
ORDER BY pr.priority ASC""",
|
||
{'pid': product_id, 'oid': org_id})
|
||
resources = [dict(r) for r in (resources or [])]
|
||
|
||
# Attach suppliers for each resource
|
||
for res in resources:
|
||
suppliers = await sor.sqlExe(
|
||
"""SELECT prs.*, s.supplier_name
|
||
FROM product_resource_supplier prs
|
||
LEFT JOIN supplychain.suppliers s ON prs.supplier_org_id=s.org_id
|
||
WHERE prs.product_resource_id=${rid}$
|
||
ORDER BY prs.priority ASC, prs.weight DESC""",
|
||
{'rid': res['id']})
|
||
res['suppliers'] = [dict(s) for s in (suppliers or [])]
|
||
|
||
return {'success': True, 'resources': resources}
|
||
|
||
async def add_supplier_to_resource(self, product_resource_id, supplier_org_id,
|
||
priority=1, weight=100, org_id=None):
|
||
"""Add a supplier to a product resource binding."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
dbname = self._get_dbname()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
# Verify resource belongs to org's product
|
||
check = await sor.sqlExe(
|
||
"""SELECT pr.id FROM product_resource pr
|
||
JOIN product p ON pr.product_id=p.id
|
||
WHERE pr.id=${rid}$ AND p.org_id=${oid}$""",
|
||
{'rid': product_resource_id, 'oid': org_id})
|
||
if not check:
|
||
return {'success': False, 'message': 'Resource binding not found'}
|
||
|
||
sid = getID()
|
||
try:
|
||
await sor.C('product_resource_supplier', {
|
||
'id': sid,
|
||
'product_resource_id': product_resource_id,
|
||
'supplier_org_id': supplier_org_id,
|
||
'priority': int(priority),
|
||
'weight': int(weight),
|
||
'status': '1',
|
||
'created_at': now
|
||
})
|
||
except Exception as e:
|
||
return {'success': False, 'message': f'Duplicate supplier: {e}'}
|
||
return {'success': True, 'id': sid}
|
||
|
||
async def remove_supplier_from_resource(self, prs_id, org_id=None):
|
||
"""Remove a supplier from a resource binding."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
dbname = self._get_dbname()
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
check = await sor.sqlExe(
|
||
"""SELECT prs.id FROM product_resource_supplier prs
|
||
JOIN product_resource pr ON prs.product_resource_id=pr.id
|
||
JOIN product p ON pr.product_id=p.id
|
||
WHERE prs.id=${id}$ AND p.org_id=${oid}$""",
|
||
{'id': prs_id, 'oid': org_id})
|
||
if not check:
|
||
return {'success': False, 'message': 'Not found'}
|
||
await sor.D('product_resource_supplier', {'id': prs_id})
|
||
return {'success': True}
|
||
|
||
async def update_supplier_priority(self, prs_id, priority=None, weight=None, org_id=None):
|
||
"""Update supplier priority/weight."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
dbname = self._get_dbname()
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
data = {'id': prs_id}
|
||
if priority is not None:
|
||
data['priority'] = int(priority)
|
||
if weight is not None:
|
||
data['weight'] = int(weight)
|
||
if len(data) <= 1:
|
||
return {'success': False, 'message': 'No fields to update'}
|
||
await sor.U('product_resource_supplier', data)
|
||
return {'success': True}
|
||
|
||
async def set_overflow_product(self, product_resource_id, overflow_product_id, org_id=None):
|
||
"""Set overflow product for a resource binding."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
dbname = self._get_dbname()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
await sor.U('product_resource', {
|
||
'id': product_resource_id,
|
||
'overflow_product_id': overflow_product_id or '',
|
||
'updated_at': now
|
||
})
|
||
return {'success': True}
|
||
|
||
# ─── Subscriptions ───
|
||
|
||
async def subscribe_product(self, product_id, user_id, user_org_id,
|
||
start_date, end_date, org_id=None):
|
||
"""Create a subscription for a monthly/quantity product."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
dbname = self._get_dbname()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
# Get product
|
||
prod = await sor.sqlExe(
|
||
"SELECT * FROM product WHERE id=${pid}$ AND status='1'",
|
||
{'pid': product_id})
|
||
if not prod:
|
||
return {'success': False, 'message': 'Product not found'}
|
||
product = dict(prod[0])
|
||
|
||
# Get resource binding for quota
|
||
resources = await sor.sqlExe(
|
||
"SELECT * FROM product_resource WHERE product_id=${pid}$ AND status='1' LIMIT 1",
|
||
{'pid': product_id})
|
||
quota_total = 0
|
||
quota_unit = ''
|
||
overflow_product_id = ''
|
||
if resources:
|
||
res = dict(resources[0])
|
||
quota_total = float(res.get('quota', 0))
|
||
quota_unit = res.get('quota_unit', '')
|
||
overflow_product_id = res.get('overflow_product_id', '')
|
||
|
||
# Get overflow rate from overflow product's price
|
||
overflow_rate = 0
|
||
if overflow_product_id:
|
||
op = await sor.sqlExe(
|
||
"SELECT price FROM product WHERE id=${oid}$",
|
||
{'oid': overflow_product_id})
|
||
if op:
|
||
overflow_rate = float(op[0].get('price', 0))
|
||
|
||
sub_id = getID()
|
||
sub_type = '1' # monthly by default
|
||
if product.get('product_type') == 'llm_monthly':
|
||
sub_type = '1'
|
||
elif product.get('product_type') in ('llm_model', 'compute'):
|
||
sub_type = '2'
|
||
|
||
await sor.C('product_subscription', {
|
||
'id': sub_id,
|
||
'product_id': product_id,
|
||
'user_id': user_id,
|
||
'user_org_id': user_org_id,
|
||
'subscription_type': sub_type,
|
||
'status': '1',
|
||
'start_date': start_date,
|
||
'end_date': end_date,
|
||
'quota_total': quota_total,
|
||
'quota_used': 0,
|
||
'quota_unit': quota_unit,
|
||
'overflow_mode': '1',
|
||
'overflow_rate': overflow_rate,
|
||
'purchase_price': float(product.get('price', 0)),
|
||
'purchase_currency': product.get('currency', 'CNY'),
|
||
'created_at': now,
|
||
'updated_at': now
|
||
})
|
||
return {'success': True, 'id': sub_id, 'quota_total': quota_total,
|
||
'overflow_rate': overflow_rate}
|
||
|
||
async def get_subscriptions(self, filters=None, org_id=None):
|
||
"""List subscriptions with optional filters."""
|
||
if not org_id:
|
||
org_id = self._get_current_org_id()
|
||
dbname = self._get_dbname()
|
||
filters = filters or {}
|
||
|
||
conditions = []
|
||
ns = {}
|
||
if filters.get('product_id'):
|
||
conditions.append("ps.product_id=${product_id}$")
|
||
ns['product_id'] = filters['product_id']
|
||
if filters.get('user_id'):
|
||
conditions.append("ps.user_id=${user_id}$")
|
||
ns['user_id'] = filters['user_id']
|
||
if filters.get('user_org_id'):
|
||
conditions.append("ps.user_org_id=${uoid}$")
|
||
ns['uoid'] = filters['user_org_id']
|
||
if filters.get('status'):
|
||
conditions.append("ps.status=${status}$")
|
||
ns['status'] = filters['status']
|
||
|
||
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
sql = f"""SELECT ps.*, p.product_name, p.product_type
|
||
FROM product_subscription ps
|
||
LEFT JOIN product p ON ps.product_id=p.id
|
||
{where}
|
||
ORDER BY ps.created_at DESC"""
|
||
rows = await sor.sqlExe(sql, ns)
|
||
return {'success': True, 'rows': [dict(r) for r in (rows or [])]}
|
||
|
||
async def get_subscription_detail(self, subscription_id):
|
||
"""Get subscription detail with quota usage."""
|
||
dbname = self._get_dbname()
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
rows = await sor.sqlExe(
|
||
"""SELECT ps.*, p.product_name, p.product_type, p.brief_intro
|
||
FROM product_subscription ps
|
||
LEFT JOIN product p ON ps.product_id=p.id
|
||
WHERE ps.id=${sid}$""",
|
||
{'sid': subscription_id})
|
||
if not rows:
|
||
return {'success': False, 'message': 'Not found'}
|
||
|
||
sub = dict(rows[0])
|
||
total = float(sub.get('quota_total', 0))
|
||
used = float(sub.get('quota_used', 0))
|
||
remaining = max(0, total - used)
|
||
pct = round((used / total * 100), 2) if total > 0 else 0
|
||
|
||
sub['quota_remaining'] = remaining
|
||
sub['quota_percentage'] = pct
|
||
return {'success': True, 'data': sub}
|
||
|
||
async def cancel_subscription(self, subscription_id, org_id=None):
|
||
"""Cancel a subscription."""
|
||
dbname = self._get_dbname()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
await sor.U('product_subscription', {
|
||
'id': subscription_id,
|
||
'status': '3', 'updated_at': now
|
||
})
|
||
return {'success': True}
|
||
|
||
async def expire_subscriptions(self):
|
||
"""Batch expire: set status='2' for subscriptions past end_date."""
|
||
dbname = self._get_dbname()
|
||
today = datetime.date.today().isoformat()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
result = await sor.sqlExe(
|
||
"""UPDATE product_subscription
|
||
SET status='2', updated_at=${now}$
|
||
WHERE status='1' AND end_date < ${today}$""",
|
||
{'today': today, 'now': now})
|
||
return {'success': True, 'expired_count': result}
|
||
|
||
# ─── Product Use Engine (CORE) ───
|
||
|
||
async def product_use(self, product_id, user_id, user_org_id,
|
||
used_amount, used_unit, resource_ref_id=None,
|
||
source_ref_table=None, source_ref_id=None):
|
||
"""Core consumption engine: route supplier, calc cost, log usage."""
|
||
dbname = self._get_dbname()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
today = datetime.date.today().isoformat()
|
||
used_amount = float(used_amount)
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
# Step 1: Get product
|
||
prod = await sor.sqlExe(
|
||
"SELECT * FROM product WHERE id=${pid}$ AND status='1'",
|
||
{'pid': product_id})
|
||
if not prod:
|
||
return {'success': False, 'message': 'Product not found or inactive'}
|
||
product = dict(prod[0])
|
||
product_type = product.get('product_type', '')
|
||
|
||
# Step 2: Check subscription for monthly products
|
||
subscription = None
|
||
billing_mode = '2' # default pay-per-use
|
||
subscription_id = None
|
||
sell_price = 0
|
||
remaining_quota = None
|
||
|
||
is_monthly = product_type in ('llm_monthly',)
|
||
|
||
if is_monthly:
|
||
subs = await sor.sqlExe(
|
||
"""SELECT * FROM product_subscription
|
||
WHERE product_id=${pid}$ AND user_id=${uid}$
|
||
AND user_org_id=${uoid}$ AND status='1'
|
||
AND start_date <= ${today}$ AND end_date >= ${today}$
|
||
ORDER BY created_at ASC LIMIT 1""",
|
||
{'pid': product_id, 'uid': user_id,
|
||
'uoid': user_org_id, 'today': today})
|
||
if subs:
|
||
subscription = dict(subs[0])
|
||
subscription_id = subscription['id']
|
||
quota_total = float(subscription.get('quota_total', 0))
|
||
quota_used = float(subscription.get('quota_used', 0))
|
||
remaining = quota_total - quota_used
|
||
|
||
if remaining >= used_amount:
|
||
# Within quota
|
||
billing_mode = '1'
|
||
remaining_quota = remaining - used_amount
|
||
# Update quota
|
||
await sor.U('product_subscription', {
|
||
'id': subscription_id,
|
||
'quota_used': quota_used + used_amount,
|
||
'updated_at': now
|
||
})
|
||
else:
|
||
# Overflow: use up remaining, rest is overage
|
||
if remaining > 0:
|
||
await sor.U('product_subscription', {
|
||
'id': subscription_id,
|
||
'quota_used': quota_total,
|
||
'status': '4',
|
||
'updated_at': now
|
||
})
|
||
billing_mode = '2'
|
||
overage = used_amount - remaining
|
||
else:
|
||
billing_mode = '2'
|
||
overage = used_amount
|
||
|
||
overflow_rate = float(subscription.get('overflow_rate', 0))
|
||
if overflow_rate <= 0:
|
||
overflow_rate = float(product.get('price', 0))
|
||
sell_price = overage * overflow_rate
|
||
remaining_quota = 0
|
||
|
||
if not is_monthly or billing_mode == '2':
|
||
# Pay-per-use: sell_price = amount × product price
|
||
if sell_price == 0:
|
||
sell_price = used_amount * float(product.get('price', 0))
|
||
|
||
# Step 3: Route supplier
|
||
supplier_org_id = None
|
||
product_resource_id = None
|
||
resource_type = ''
|
||
|
||
res_cond = "product_id=${pid}$ AND status='1'"
|
||
res_ns = {'pid': product_id}
|
||
if resource_ref_id:
|
||
res_cond += " AND resource_ref_id=${rrid}$"
|
||
res_ns['rrid'] = resource_ref_id
|
||
|
||
resources = await sor.sqlExe(
|
||
f"SELECT * FROM product_resource WHERE {res_cond} ORDER BY priority ASC",
|
||
res_ns)
|
||
|
||
if resources:
|
||
res = dict(resources[0])
|
||
product_resource_id = res['id']
|
||
resource_type = res.get('resource_type', '')
|
||
if not resource_ref_id:
|
||
resource_ref_id = res.get('resource_ref_id', '')
|
||
|
||
# Get suppliers ordered by priority, weight
|
||
suppliers = await sor.sqlExe(
|
||
"""SELECT * FROM product_resource_supplier
|
||
WHERE product_resource_id=${rid}$ AND status='1'
|
||
ORDER BY priority ASC, weight DESC""",
|
||
{'rid': product_resource_id})
|
||
if suppliers:
|
||
supplier_org_id = dict(suppliers[0]).get('supplier_org_id')
|
||
|
||
# Step 4: Calculate cost from supplier_resource_price
|
||
unit_cost = 0
|
||
total_cost = 0
|
||
|
||
if supplier_org_id and resource_ref_id:
|
||
sc_dbname = 'supplychain'
|
||
price_sql = (
|
||
"SELECT * FROM " + sc_dbname + ".supplier_resource_price"
|
||
" WHERE supplier_org_id=${soid}$"
|
||
" AND resource_ref_id=${rrid}$"
|
||
" AND status='1'"
|
||
" AND effective_date <= ${today}$"
|
||
" AND (expiry_date IS NULL OR expiry_date >= ${today}$)"
|
||
" ORDER BY effective_date DESC LIMIT 1"
|
||
)
|
||
price_rows = await sor.sqlExe(price_sql,
|
||
{'soid': supplier_org_id, 'rrid': resource_ref_id, 'today': today})
|
||
if price_rows:
|
||
pricing = dict(price_rows[0])
|
||
# For LLM: use input_price if available, else unit_price
|
||
if resource_type.startswith('llm'):
|
||
ip = pricing.get('input_price')
|
||
unit_cost = float(ip) if ip else float(pricing.get('unit_price', 0))
|
||
else:
|
||
unit_cost = float(pricing.get('unit_price', 0))
|
||
total_cost = used_amount * unit_cost
|
||
|
||
# Step 5: Write usage log
|
||
log_id = getID()
|
||
await sor.C('product_usage_log', {
|
||
'id': log_id,
|
||
'product_id': product_id,
|
||
'subscription_id': subscription_id or '',
|
||
'user_id': user_id,
|
||
'user_org_id': user_org_id,
|
||
'product_resource_id': product_resource_id or '',
|
||
'supplier_org_id': supplier_org_id or '',
|
||
'resource_type': resource_type,
|
||
'resource_ref_id': resource_ref_id or '',
|
||
'used_amount': used_amount,
|
||
'used_unit': used_unit,
|
||
'unit_cost': unit_cost,
|
||
'total_cost': total_cost,
|
||
'sell_price': sell_price,
|
||
'billing_mode': billing_mode,
|
||
'source_ref_table': source_ref_table or '',
|
||
'source_ref_id': source_ref_id or '',
|
||
'use_time': now,
|
||
'created_at': now
|
||
})
|
||
|
||
return {
|
||
'success': True,
|
||
'log_id': log_id,
|
||
'billing_mode': billing_mode,
|
||
'used_amount': used_amount,
|
||
'total_cost': round(total_cost, 6),
|
||
'sell_price': round(sell_price, 6),
|
||
'supplier_org_id': supplier_org_id,
|
||
'remaining_quota': remaining_quota
|
||
}
|
||
|
||
# ─── Usage Logs & Stats ───
|
||
|
||
async def get_usage_logs(self, filters=None, page=1, page_size=50):
|
||
"""Query usage logs with filters."""
|
||
dbname = self._get_dbname()
|
||
filters = filters or {}
|
||
page = int(page) or 1
|
||
page_size = int(page_size) or 50
|
||
|
||
conditions = []
|
||
ns = {}
|
||
for key in ('product_id', 'subscription_id', 'user_id', 'user_org_id',
|
||
'supplier_org_id', 'billing_mode', 'resource_type'):
|
||
if filters.get(key):
|
||
conditions.append(f"pul.{key}=${key}$")
|
||
ns[key] = filters[key]
|
||
if filters.get('start_date'):
|
||
conditions.append("pul.use_time >= ${start_date}$")
|
||
ns['start_date'] = filters['start_date']
|
||
if filters.get('end_date'):
|
||
conditions.append("pul.use_time <= ${end_date}$")
|
||
ns['end_date'] = filters['end_date']
|
||
|
||
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
||
offset = (page - 1) * page_size
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
count_sql = f"SELECT COUNT(*) as cnt FROM product_usage_log pul {where}"
|
||
cnt_rows = await sor.sqlExe(count_sql, ns)
|
||
total = cnt_rows[0]['cnt'] if cnt_rows else 0
|
||
|
||
sql = f"""SELECT pul.*, p.product_name, s.supplier_name
|
||
FROM product_usage_log pul
|
||
LEFT JOIN product p ON pul.product_id=p.id
|
||
LEFT JOIN supplychain.suppliers s ON pul.supplier_org_id=s.org_id
|
||
{where}
|
||
ORDER BY pul.use_time DESC
|
||
LIMIT {page_size} OFFSET {offset}"""
|
||
rows = await sor.sqlExe(sql, ns)
|
||
|
||
return {'success': True, 'rows': [dict(r) for r in (rows or [])],
|
||
'total': total, 'page': page}
|
||
|
||
async def get_usage_stats(self, filters=None):
|
||
"""Aggregate usage stats by product/supplier/date."""
|
||
dbname = self._get_dbname()
|
||
filters = filters or {}
|
||
group_by = filters.get('group_by', 'product_id')
|
||
|
||
conditions = []
|
||
ns = {}
|
||
if filters.get('start_date'):
|
||
conditions.append("use_time >= ${start_date}$")
|
||
ns['start_date'] = filters['start_date']
|
||
if filters.get('end_date'):
|
||
conditions.append("use_time <= ${end_date}$")
|
||
ns['end_date'] = filters['end_date']
|
||
|
||
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
sql = f"""SELECT {group_by},
|
||
COUNT(*) as usage_count,
|
||
SUM(used_amount) as total_used,
|
||
SUM(total_cost) as total_cost,
|
||
SUM(sell_price) as total_revenue,
|
||
SUM(sell_price) - SUM(total_cost) as total_profit
|
||
FROM product_usage_log
|
||
{where}
|
||
GROUP BY {group_by}
|
||
ORDER BY total_cost DESC"""
|
||
rows = await sor.sqlExe(sql, ns)
|
||
|
||
return {'success': True, 'stats': [dict(r) for r in (rows or [])]}
|
||
|
||
async def check_quota(self, subscription_id):
|
||
"""Check quota status for a subscription."""
|
||
dbname = self._get_dbname()
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
rows = await sor.sqlExe(
|
||
"SELECT * FROM product_subscription WHERE id=${sid}$",
|
||
{'sid': subscription_id})
|
||
if not rows:
|
||
return {'success': False, 'message': 'Not found'}
|
||
|
||
sub = dict(rows[0])
|
||
total = float(sub.get('quota_total', 0))
|
||
used = float(sub.get('quota_used', 0))
|
||
remaining = max(0, total - used)
|
||
pct = round((used / total * 100), 2) if total > 0 else 0
|
||
|
||
return {
|
||
'success': True,
|
||
'quota_total': total,
|
||
'quota_used': used,
|
||
'quota_remaining': remaining,
|
||
'quota_percentage': pct,
|
||
'status': sub.get('status'),
|
||
'overflow_mode': sub.get('overflow_mode'),
|
||
'overflow_rate': float(sub.get('overflow_rate', 0))
|
||
}
|
||
|
||
# ─── Product Import from Resource Modules ───
|
||
|
||
async def import_categories_and_products(self, resource_module, org_id,
|
||
parent_category_id, user_id):
|
||
"""Import sub-categories and products from a resource module.
|
||
|
||
1. Calls resource module's load_product_category_product() to get standardized data
|
||
2. For categories: skip if exists (by name+parent_id+org_id), else create
|
||
3. For products: update if exists (by resource_ref_id+org_id), else create
|
||
"""
|
||
import importlib
|
||
import time
|
||
from appPublic.uniqueID import getID
|
||
|
||
# Step 1: Get standardized data from resource module
|
||
try:
|
||
mod = importlib.import_module(f'{resource_module}.init')
|
||
load_fn = getattr(mod, 'load_product_category_product', None)
|
||
except Exception as e:
|
||
return {'success': False, 'error': f'无法加载资源模块 "{resource_module}": {e}'}
|
||
|
||
if load_fn is None:
|
||
return {'success': False, 'error': f'资源模块 "{resource_module}" 未提供 load_product_category_product 函数'}
|
||
|
||
try:
|
||
result = await load_fn(parent_category_id)
|
||
except Exception as e:
|
||
return {'success': False, 'error': f'调用 {resource_module}.load_product_category_product() 失败: {e}'}
|
||
|
||
if not result or not result.get('success'):
|
||
return {'success': False, 'error': result.get('error', '资源模块未返回有效数据') if result else '资源模块未返回数据'}
|
||
|
||
categories = result.get('categories', [])
|
||
products = result.get('products', [])
|
||
|
||
if not categories and not products:
|
||
return {'success': False, 'error': '资源模块返回数据为空'}
|
||
|
||
dbname = self._get_dbname()
|
||
now = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
source_to_id = {} # source_id -> product_category.id
|
||
created_cats = 0
|
||
skipped_cats = 0
|
||
created_prods = 0
|
||
updated_prods = 0
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
# Step 2: Process categories — skip existing, create new
|
||
for cat in categories:
|
||
existing = await sor.sqlExe(
|
||
"""SELECT id FROM product_category
|
||
WHERE name = ${name}$ AND parent_id = ${parent_id}$ AND org_id = ${org_id}$""",
|
||
{'name': cat['name'], 'parent_id': parent_category_id, 'org_id': org_id}
|
||
)
|
||
if existing:
|
||
source_to_id[cat['source_id']] = existing[0].id
|
||
skipped_cats += 1
|
||
continue
|
||
|
||
new_id = getID()
|
||
source_to_id[cat['source_id']] = new_id
|
||
await sor.C('product_category', {
|
||
'id': new_id,
|
||
'parent_id': parent_category_id,
|
||
'name': cat['name'],
|
||
'description': cat.get('description', ''),
|
||
'has_product': '1',
|
||
'product_type': cat.get('product_type', ''),
|
||
'product_type_title': cat.get('product_type_title', ''),
|
||
'sort_order': str(cat.get('sort_order', 0)),
|
||
'icon': '',
|
||
'status': '1',
|
||
'resource_module': resource_module,
|
||
'org_id': org_id,
|
||
'created_by': user_id,
|
||
'created_at': now,
|
||
'updated_at': now
|
||
})
|
||
created_cats += 1
|
||
|
||
# Step 3: Process products — update existing, create new
|
||
for prod in products:
|
||
target_cat_id = source_to_id.get(prod.get('source_category_id'))
|
||
if not target_cat_id:
|
||
continue
|
||
|
||
resource_ref_id = prod.get('resource_ref_id', '')
|
||
|
||
# Check if product already exists by resource_ref_id + org_id
|
||
existing_prod = None
|
||
if resource_ref_id:
|
||
existing_prod = await sor.sqlExe(
|
||
"""SELECT id FROM product
|
||
WHERE resource_ref_id = ${ref_id}$ AND org_id = ${org_id}$""",
|
||
{'ref_id': resource_ref_id, 'org_id': org_id}
|
||
)
|
||
|
||
if existing_prod:
|
||
# Update existing product
|
||
await sor.U('product', {
|
||
'id': existing_prod[0].id,
|
||
'product_name': prod.get('product_name', ''),
|
||
'product_code': prod.get('product_code', ''),
|
||
'product_type': prod.get('product_type', ''),
|
||
'brief_intro': prod.get('brief_intro', ''),
|
||
'category_id': target_cat_id,
|
||
'sort_order': str(prod.get('sort_order', 0)),
|
||
'updated_at': now
|
||
})
|
||
updated_prods += 1
|
||
else:
|
||
# Create new product
|
||
prod_id = getID()
|
||
await sor.C('product', {
|
||
'id': prod_id,
|
||
'category_id': target_cat_id,
|
||
'product_code': prod.get('product_code', ''),
|
||
'resource_ref_id': resource_ref_id,
|
||
'product_name': prod.get('product_name', ''),
|
||
'product_type': prod.get('product_type', ''),
|
||
'brief_intro': prod.get('brief_intro', ''),
|
||
'status': '1',
|
||
'price_type': '1',
|
||
'price': '0',
|
||
'currency': 'CNY',
|
||
'sort_order': str(prod.get('sort_order', 0)),
|
||
'org_id': org_id,
|
||
'created_by': user_id,
|
||
'created_at': now,
|
||
'updated_at': now
|
||
})
|
||
created_prods += 1
|
||
|
||
return {
|
||
'success': True,
|
||
'message': f'导入完成: 新增 {created_cats} 个类别, {created_prods} 个产品; '
|
||
f'跳过 {skipped_cats} 个已有类别; 更新 {updated_prods} 个已有产品'
|
||
}
|
||
|
||
# ─── Resource Module Interface Dispatcher ───
|
||
|
||
async def _get_product_interface(self, product_id=None, product_code=None):
|
||
"""Resolve product to its resource module's interface dict.
|
||
|
||
Returns: (interface_dict, product_record, error_message)
|
||
"""
|
||
dbname = self._get_dbname()
|
||
conditions = ["status = '1'"]
|
||
params = {}
|
||
if product_id:
|
||
conditions.append("id = ${pid}$")
|
||
params['pid'] = product_id
|
||
elif product_code:
|
||
conditions.append("product_code = ${code}$")
|
||
params['code'] = product_code
|
||
else:
|
||
return None, None, '需要指定 product_id 或 product_code'
|
||
|
||
where = " AND ".join(conditions)
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
rows = await sor.sqlExe(
|
||
f"SELECT * FROM product WHERE {where}", params)
|
||
if not rows:
|
||
return None, None, '产品不存在或已下线'
|
||
product = dict(rows[0])
|
||
|
||
# Get resource_module from category
|
||
cat_id = product.get('category_id')
|
||
if cat_id:
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
cat_rows = await sor.sqlExe(
|
||
"SELECT resource_module FROM product_category WHERE id = ${cid}$",
|
||
{'cid': cat_id})
|
||
if cat_rows and cat_rows[0].get('resource_module'):
|
||
resource_module = cat_rows[0]['resource_module']
|
||
else:
|
||
return None, product, '产品类别未绑定资源模块'
|
||
else:
|
||
return None, product, '产品无类别信息'
|
||
|
||
# Get interface from ServerEnv or importlib
|
||
env = ServerEnv()
|
||
iface = getattr(env, 'product_interface', None)
|
||
if iface and iface.get('module_name') == resource_module:
|
||
return iface, product, None
|
||
|
||
# Fallback: dynamic import
|
||
import importlib
|
||
try:
|
||
mod = importlib.import_module(f'{resource_module}.init')
|
||
# Trigger load if not loaded yet
|
||
load_fn_name = f'load_{resource_module}'
|
||
load_fn = getattr(mod, load_fn_name, None)
|
||
if load_fn:
|
||
load_fn()
|
||
iface = getattr(env, 'product_interface', None)
|
||
if iface and iface.get('module_name') == resource_module:
|
||
return iface, product, None
|
||
except Exception as e:
|
||
return None, product, f'加载资源模块 {resource_module} 失败: {e}'
|
||
|
||
return None, product, f'资源模块 {resource_module} 未注册 product_interface'
|
||
|
||
def _get_ref_id(self, product):
|
||
"""Extract resource_ref_id from product record, with fallback error."""
|
||
ref_id = product.get('resource_ref_id')
|
||
if not ref_id:
|
||
return None, '产品未绑定资源模块内部ID(resource_ref_id为空)'
|
||
return ref_id, None
|
||
|
||
async def get_product_display_info(self, product_id=None, product_code=None):
|
||
"""获取产品定价展示信息（通过资源模块接口）。"""
|
||
iface, product, err = await self._get_product_interface(
|
||
product_id, product_code)
|
||
if err:
|
||
return {'success': False, 'message': err}
|
||
|
||
fn = iface.get('get_product_display')
|
||
if not fn:
|
||
return {'success': False, 'message': '资源模块未实现 get_product_display'}
|
||
|
||
ref_id, err = self._get_ref_id(product)
|
||
if err:
|
||
return {'success': False, 'message': err}
|
||
return await fn(ref_id)
|
||
|
||
async def check_product_availability(self, product_id=None, product_code=None,
|
||
user_org_id=None):
|
||
"""检查产品可用性（通过资源模块接口）。"""
|
||
iface, product, err = await self._get_product_interface(
|
||
product_id, product_code)
|
||
if err:
|
||
return {'available': False, 'reason': err}
|
||
|
||
fn = iface.get('check_product_availability')
|
||
if not fn:
|
||
return {'available': False, 'reason': '资源模块未实现 check_product_availability'}
|
||
|
||
ref_id, err = self._get_ref_id(product)
|
||
if err:
|
||
return {'available': False, 'reason': err}
|
||
return await fn(ref_id, user_org_id)
|
||
|
||
async def check_product_consumable(self, product_id=None, product_code=None,
|
||
user_id=None, user_org_id=None):
|
||
"""消费前综合预检（通过资源模块接口）。"""
|
||
iface, product, err = await self._get_product_interface(
|
||
product_id, product_code)
|
||
if err:
|
||
return {'consumable': False, 'reason': err,
|
||
'min_balance': 0, 'pricing_available': False}
|
||
|
||
fn = iface.get('check_product_consumable')
|
||
if not fn:
|
||
return {'consumable': False, 'reason': '资源模块未实现 check_product_consumable',
|
||
'min_balance': 0, 'pricing_available': False}
|
||
|
||
ref_id, err = self._get_ref_id(product)
|
||
if err:
|
||
return {'consumable': False, 'reason': err,
|
||
'min_balance': 0, 'pricing_available': False}
|
||
|
||
if not user_id:
|
||
env = ServerEnv()
|
||
user_id = await env.get_user()
|
||
if not user_org_id:
|
||
user_org_id = self._get_current_org_id()
|
||
|
||
return await fn(ref_id, user_id, user_org_id)
|
||
|
||
async def execute_product(self, product_id=None, product_code=None,
|
||
request_data=None, user_id=None, user_org_id=None):
|
||
"""执行产品服务（通过资源模块接口）。"""
|
||
iface, product, err = await self._get_product_interface(
|
||
product_id, product_code)
|
||
if err:
|
||
return {'success': False, 'message': err, 'status': 'FAILED'}
|
||
|
||
fn = iface.get('execute_product_service')
|
||
if not fn:
|
||
return {'success': False, 'message': '资源模块未实现 execute_product_service',
|
||
'status': 'FAILED'}
|
||
|
||
ref_id, err = self._get_ref_id(product)
|
||
if err:
|
||
return {'success': False, 'message': err, 'status': 'FAILED'}
|
||
|
||
if not user_id:
|
||
env = ServerEnv()
|
||
user_id = await env.get_user()
|
||
if not user_org_id:
|
||
user_org_id = self._get_current_org_id()
|
||
|
||
return await fn(ref_id, user_id, user_org_id, request_data or {})
|
||
|
||
async def execute_product_stream(self, product_id=None, product_code=None,
|
||
request_data=None, user_id=None,
|
||
user_org_id=None):
|
||
"""执行产品服务-流式（通过资源模块接口，返回异步生成器）。"""
|
||
iface, product, err = await self._get_product_interface(
|
||
product_id, product_code)
|
||
if err:
|
||
async def _err_gen():
|
||
yield {'chunk': None, 'usage_data': None, 'done': True, 'error': err}
|
||
return _err_gen()
|
||
|
||
fn = iface.get('execute_product_service_stream')
|
||
if not fn:
|
||
async def _noimpl_gen():
|
||
yield {'chunk': None, 'usage_data': None, 'done': True,
|
||
'error': '资源模块未实现 execute_product_service_stream'}
|
||
return _noimpl_gen()
|
||
|
||
ref_id, err = self._get_ref_id(product)
|
||
if err:
|
||
async def _ref_err_gen():
|
||
yield {'chunk': None, 'usage_data': None, 'done': True, 'error': err}
|
||
return _ref_err_gen()
|
||
|
||
if not user_id:
|
||
env = ServerEnv()
|
||
user_id = await env.get_user()
|
||
if not user_org_id:
|
||
user_org_id = self._get_current_org_id()
|
||
|
||
return fn(ref_id, user_id, user_org_id, request_data or {})
|
||
|
||
async def calculate_product_cost(self, product_id=None, product_code=None,
|
||
usage_data=None, user_org_id=None):
|
||
"""计算消费费用（通过资源模块接口）。"""
|
||
iface, product, err = await self._get_product_interface(
|
||
product_id, product_code)
|
||
if err:
|
||
return {'success': False, 'message': err}
|
||
|
||
fn = iface.get('calculate_product_cost')
|
||
if not fn:
|
||
return {'success': False, 'message': '资源模块未实现 calculate_product_cost'}
|
||
|
||
ref_id, err = self._get_ref_id(product)
|
||
if err:
|
||
return {'success': False, 'message': err}
|
||
|
||
if not user_org_id:
|
||
user_org_id = self._get_current_org_id()
|
||
|
||
return await fn(ref_id, usage_data or {}, user_org_id)
|