product_management/product_management/core.py

"""Product Management Core Business Logic - org_id isolated per reseller"""
import json
import time
import datetime
from html import escape as _html_escape
from appPublic.uniqueID import getID
import json, time

from appPublic.dictObject import DictObject
from appPublic.log import info, error, exception, debug
from appPublic.jsonConfig import getConfig
from sqlor.dbpools import DBPools
from ahserver.serverenv import ServerEnv
from accounting.consume import consume_accounting


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
        # 两遍建树（2026-09-06 根治 KeyError）：先给全部节点初始化 children，
        # 再挂父子关系。单遍写法隐含「父节点排序在子节点前」的假设——
        # ORDER BY sort_order ASC, name ASC 下子类别 sort_order 更小（如导入引擎
        # 给子类别 0、顶级类别 10/20）时，子先被处理，父还未初始化 children
        # → parent['children'].append 抛 KeyError → storefront 整页 500。
        nodes = [dict(r) for r in rows]
        for n in nodes:
            n['children'] = []
        node_map = {n['id']: n for n in nodes}
        tree = []

        for node in nodes:
            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
            # sqlor 占位符格式为 ${name}$；f-string 里花括号需转义，否则 ${key} 会被求值
            param_keys = []
            params = {'org_id': org_id}
            for i, cid in enumerate(all_ids):
                key = f'cid_{i}'
                param_keys.append('${' + 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 + 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

        return {
            'success': True,
            'data': {
                'product_info': product_info,
                'category_info': {
                    'name': product_info.get('category_name'),
                    'description': product_info.get('category_description')
                }
            }
        }

    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'
        }

    # ─── 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
        skipped_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 — skip existing, create new only
            # Incremental mode: only import products not already in the system
            seen_codes = set()
            for prod in products:
                target_cat_id = source_to_id.get(prod.get('source_category_id'))
                if not target_cat_id:
                    continue

                product_code = prod.get('product_code', '')
                resource_ref_id = prod.get('resource_ref_id', '')

                # Skip duplicate product_code in this batch
                if product_code in seen_codes:
                    continue
                seen_codes.add(product_code)

                # 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}
                    )

                # Fallback: check by product_code + org_id (unique constraint)
                if not existing_prod and product_code:
                    existing_prod = await sor.sqlExe(
                        """SELECT id FROM product
                           WHERE product_code = ${code}$ AND org_id = ${org_id}$""",
                        {'code': product_code, 'org_id': org_id}
                    )

                if existing_prod:
                    skipped_prods += 1
                    continue

                # 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,
                    'providerid': prod.get('providerid', ''),
                    '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} 个已有类别, {skipped_prods} 个已有产品'
        }

    # ─── Auto-sync a published llm model as product ───

    async def sync_llm_product(self, llm_id):
        """llmage 上架模型后自动同步为产品（幂等）。

        复用 import_categories_and_products 的增量导入（已存在的产品跳过），
        顺带补齐其他已上架但缺产品的模型。
        """
        if not llm_id:
            return {'success': False, 'error': '缺少 llm_id'}

        dbname = self._get_dbname()
        lmage_dbname = ServerEnv().get_module_dbname('llmage')

        # 1. llm 记录必须存在且已上架
        async with DBPools().sqlorContext(lmage_dbname) as sor:
            llm_rows = await sor.sqlExe(
                "SELECT id, status, ownerid FROM llm WHERE id = ${lid}$",
                {'lid': llm_id})
        if not llm_rows:
            return {'success': False, 'error': '模型不存在: %s' % llm_id}
        llm_rec = llm_rows[0]
        if getattr(llm_rec, 'status', '') != 'published':
            return {'success': False, 'error': '模型未上架，无需同步'}
        ownerid = getattr(llm_rec, 'ownerid', None) or '0'

        # 2. 幂等：产品已存在则跳过
        async with DBPools().sqlorContext(dbname) as sor:
            existing = await sor.sqlExe(
                """SELECT id FROM product
                   WHERE resource_ref_id = ${ref_id}$ AND org_id = ${org_id}$""",
                {'ref_id': llm_id, 'org_id': ownerid})
        if existing:
            return {'success': True, 'skipped': True, 'product_id': existing[0].id}

        # 3. 确定导入父类别：沿用已有 llmage 类别的父级，避免类别树分裂；
        #    完全没有 llmage 类别时才新建根类别
        async with DBPools().sqlorContext(dbname) as sor:
            cat_rows = await sor.sqlExe(
                """SELECT parent_id FROM product_category
                   WHERE resource_module = 'llmage' AND org_id = ${org_id}$
                   LIMIT 1""",
                {'org_id': ownerid})
            parent_category_id = None
            if cat_rows:
                parent_category_id = getattr(cat_rows[0], 'parent_id', None)
            if not parent_category_id:
                parent_category_id = getID()
                now = time.strftime('%Y-%m-%d %H:%M:%S')
                await sor.C('product_category', {
                    'id': parent_category_id,
                    'parent_id': '0',
                    'name': 'AI模型',
                    'description': 'llmage 模型产品类别',
                    'has_product': '1',
                    'product_type': 'llm_model',
                    'product_type_title': '',
                    'sort_order': '0',
                    'icon': '',
                    'status': '1',
                    'resource_module': 'llmage',
                    'org_id': ownerid,
                    'created_by': 'system',
                    'created_at': now,
                    'updated_at': now
                })

        # 4. 增量导入（跳过已有类别/产品）
        result = await self.import_categories_and_products(
            resource_module='llmage', org_id=ownerid,
            parent_category_id=parent_category_id, user_id='system')
        if not result or not result.get('success'):
            err = result.get('error', '导入失败') if result else '导入无返回'
            return {'success': False, 'error': err}

        # 5. 确认产品已落库
        async with DBPools().sqlorContext(dbname) as sor:
            new_prod = await sor.sqlExe(
                """SELECT id FROM product
                   WHERE resource_ref_id = ${ref_id}$ AND org_id = ${org_id}$""",
                {'ref_id': llm_id, 'org_id': ownerid})
        if new_prod:
            return {'success': True, 'product_id': new_prod[0].id}
        return {'success': False, 'error': '导入后仍未找到产品（模型可能不在已上架列表）'}

    # ─── 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)

    # 多价目表格化：常见英文维度名 → 中文表头润饰（YAML fields 无中文 label 时兜底，
    # 数据源头 label 由提取层生成，此处只做展示润饰不做语义裁决）
    _DIM_LABEL_POLISH = {
        'resolution': '分辨率', 'type': '类型', 'size': '尺寸',
        'quality': '质量', 'duration': '时长', 'style': '风格',
        'prompt_tokens': '输入长度(tokens)', 'total_tokens': '总长度(tokens)',
        'hour': '调用小时', 'SR': '分辨率',
    }
    # ppid_pairs 的通用挂载标签——对表格无区分度，不作为要素列
    _GENERIC_PRICE_LABELS = ('', '按量', '容量')

    def _build_price_table(self, prices, discount):
        """多价目定价表格（2026-09-09 用户要求）。

        病根：同一模型多条定价（qwen-image-3.0-pro 按 分辨率×类型 四档价）拼成
        一行文本时全是重复前缀「按次 ¥0.02 元/张 | ...」，客户看不出各价的适用
        条件。改为表格：表头 = 定价要素1..N + 价格，行 = 要素取值组合 → 价格。

        要素列生成规则（回答「什么条件用哪个价」）：
        - 计费方式列：prices[].label 有区分度（非 ''/按量/容量）时出现
          （产线 包月/包年 两行）
        - 定价维度列：定价 YAML role=filter 维度（filter_labels），按首次出现序
          （图像模型 分辨率×类型；视频模型 SR）
        - 计价因子列：factor 各行不同时出现（token 模型 输入/输出/缓存三因子）；
          全表同因子（如 flat=按次）时省略——价格列单位（元/张）已表达口径

        Returns:
            {'headers': [...], 'rows': [[要素值..., {'t': 价格文本, 'k': 'price'}...]]}
            k='orig' 的单元格是划线原价（让利时附）。.ui 用声明式控件渲染（2026-09-18）。
        """
        cols = []          # [(key, header)]
        seen = set()

        def _add_col(key, header):
            if key not in seen:
                seen.add(key)
                cols.append((key, header))

        if any((p.get('label') or '') not in self._GENERIC_PRICE_LABELS
               for p in prices):
            _add_col('__label__', '计费方式')
        for p in prices:
            for k in (p.get('filter_labels') or {}):
                header = self._DIM_LABEL_POLISH.get(str(k), '') or str(k)
                _add_col('f:' + str(k), header)
        factors = set((p.get('factor') or '') for p in prices)
        if len(factors) > 1 or not cols:
            _add_col('__factor__', '计价因子')

        # 表头兜底：无 label 的列按用户口径命名「定价要素N」
        headers = []
        for i, (k, h) in enumerate(cols):
            headers.append(h if str(h).strip() else '定价要素%d' % (i + 1))
        headers.append('价格')

        rows = []
        # 结构化返回（2026-09-18 用户纠正：价格表用 bricks 声明式控件渲染，
        # 后端不拼 HTML——Html 控件内联样式是暗色治理死角且违背声明式铁律）。
        # 单元格 = {'t': 文本, 'k': 'price'|'orig'|''}，.ui 按 k 上色/划线。
        for p in prices:
            vals = []
            for key, _h in cols:
                if key == '__label__':
                    v = p.get('label') or ''
                    v = '' if v in self._GENERIC_PRICE_LABELS else v
                elif key == '__factor__':
                    v = p.get('factor_label') or p.get('factor') or ''
                else:
                    v = (p.get('filter_labels') or {}).get(key[2:], '')
                vals.append(str(v if v is not None else ''))
            cells = [{'t': v, 'k': ''} for v in vals]
            cells.append({'t': '¥%.4g %s' % (p['amount'], p.get('unit_label') or '元'),
                          'k': 'price'})
            if discount < 1.0 - 1e-9:
                cells.append({'t': '¥%.4g' % p['original_price'], 'k': 'orig'})
            rows.append(cells)
        return {'headers': headers, 'rows': rows}

    async def get_customer_price_display(self, product_id=None, product_code=None,
                                         user_org_id=None):
        """客户视角展示价 = 资源真实定价 × 客户折扣率（2026-09-06 新增）。

        统一三类产品的对外报价口径（用户明确要求「产品显示的价格要用定价里
        价格×客户的折扣率」）：
            产线   pipeline_pricing_map → ppid → 单位价(元/月) × 折扣
            存储   storres_pricing_map   → ppid → 单位价(元/GB月) × 折扣
            模型   llm_model.ppid        → ppid → 单位价(元/百万tokens 等) × 折扣

        客户折扣率解析（discount 模块，专属优先 / '*' 兜底 / 无记录 1.0）：
        >1 = 加价卖给客户（如存储 1.5），<1 = 让利（如百炼模型 0.9）。

        未登录（user_org_id 空）时用 '*' 通用折扣报价——展示页对访客可见。

        Returns:
            {'success', 'prices': [{'label','unit_price','unit_label','amount','unit',
                                    'filter_labels','original_price'}],
             'pricing_text', 'discount', 'original_text',
             'price_table': {'headers': [...], 'rows': [[{'t','k'}...]]} 或 None（多价目时给出）}
        """
        iface, product, err = await self._get_product_interface(
            product_id, product_code)
        if err:
            return {'success': False, 'message': err}
        product = product or {}
        ref_id, err = self._get_ref_id(product)
        if err:
            return {'success': False, 'message': err}

        env = ServerEnv()
        product_type = product.get('product_type', '') or ''
        pid = product.get('id') or ''
        resellerid = product.get('org_id') or '0'

        # 1. 客户折扣率（未登录 → 传 '' 让 discount 模块只命中 '*' 兜底）
        discount = 1.0
        try:
            d = await env.get_min_product_discount(pid, resellerid, user_org_id or '')
            if d not in (None, '', 0):
                discount = float(d)
        except Exception:
            discount = 1.0

        # 2. 按产品类型解析 ppid 列表 [(标签, ppid, months 倍数)]
        #    months：产线年付=10（年费=月单价×10），其余类型为 1
        dbname = self._get_dbname()
        ppid_pairs = []
        try:
            async with DBPools().sqlorContext(dbname) as sor:
                if product_type == 'pipeline':
                    recs = await sor.sqlExe(
                        "SELECT charge_mode, ppid, months, valid_days FROM pipeline_pricing_map "
                        "WHERE pipeline_id=${rid}$ AND status='active' ORDER BY months",
                        {'rid': ref_id})
                    for r in (recs or []):
                        cm = str(getattr(r, 'charge_mode', '') or '')
                        months = int(getattr(r, 'months', 1) or 1)
                        ppid_pairs.append(('包年' if cm == 'year' else '包月',
                                           getattr(r, 'ppid', '') or '', months))
                elif product_type == 'workspace_storage':
                    recs = await sor.sqlExe(
                        "SELECT meter_mode, ppid FROM storres_pricing_map "
                        "WHERE spec_id=${rid}$ AND status='active'",
                        {'rid': ref_id})
                    seen = set()
                    for r in (recs or []):
                        ppid = getattr(r, 'ppid', '') or ''
                        if ppid and ppid not in seen:
                            seen.add(ppid)
                            ppid_pairs.append(('容量', ppid, 1))
                else:
                    # 模型类（pipeline_llm_model / llm_model）：ppid 挂模型表
                    recs = await sor.sqlExe(
                        "SELECT ppid FROM llm_model WHERE id=${rid}$", {'rid': ref_id})
                    if recs:
                        ppid = getattr(recs[0], 'ppid', '') or ''
                        if ppid:
                            ppid_pairs.append(('按量', ppid, 1))
                await sor.sqlExe("COMMIT", {})
        except Exception as e:
            debug(f'get_customer_price_display: ppid 解析失败 {pid}: {e}')

        if not ppid_pairs:
            return {'success': False, 'message': '未配置定价方案',
                    'prices': [], 'pricing_text': '未配置定价',
                    'discount': discount, 'original_text': ''}

        # 3. 取定价单位价 → × 折扣率（months>1 时单位价×months，如年付=月价×10）
        prices = []
        for label, ppid, months in ppid_pairs:
            if not ppid:
                continue
            try:
                pd = await env.get_pricing_display(ppid)
            except Exception:
                pd = None
            if not pd:
                continue
            mult = months if months and months > 1 else 1
            unit_suffix = '元/年' if mult > 1 else ''
            for item in (pd.get('items') or []):
                for pf in (item.get('price_factors') or []):
                    up = pf.get('unit_price')
                    if up is None:
                        continue
                    up = float(up)
                    total = up * mult
                    prices.append({
                        'label': label,
                        'factor': pf.get('factor', ''),
                        'factor_label': pf.get('label', ''),
                        'unit': pf.get('unit', ''),
                        'unit_label': unit_suffix or pf.get('unit_label', ''),
                        'filter_labels': item.get('filter_labels') or {},
                        'original_price': round(total, 6),
                        'amount': round(total * discount, 6),
                    })

        if not prices:
            return {'success': False, 'message': '定价数据为空',
                    'prices': [], 'pricing_text': '未配置定价',
                    'discount': discount, 'original_text': ''}

        # 3b. 多价目表格化（2026-09-09 用户要求）：同一模型多条定价（如
        #     qwen-image-3.0-pro 按 分辨率×类型 四档价）拼成一行文本时客户
        #     看不出各价适用条件（「按次 ¥0.02 | 按次 ¥0.02 | ...」），改为
        #     表格展示：表头 = 定价要素1..N + 价格，每行 = 一组要素取值对应的价格。
        #     单价目产品仍走文本（无歧义）。
        price_table = None
        if len(prices) > 1:
            price_table = self._build_price_table(prices, discount)

        # 4. 拼展示文本（仅折扣<1 让利时给原价划线，让客户看见优惠幅度；
        #    折扣≥1（无折扣/加价）不显示原价——2026-09-08 用户定夺：加价时
        #    「原价」低于现价还划线展示是误导，产线展示页/购买页同规则）
        #    多价目产品（如模型按量：输入/输出/缓存三档 tokens 价）必须带因子表头，
        #    否则客户看到裸的「¥2.1 元/百万 | ¥8.4 元/百万」不知道各是什么价
        #    （2026-09-08 用户反馈）。factor_label 来自定价 YAML fields 的 label。
        parts = []
        orig_parts = []
        for p in prices:
            head = p['label'] if p['label'] and p['label'] not in ('按量', '容量') else ''
            if not head and len(prices) > 1:
                head = p.get('factor_label') or p.get('factor') or ''
            seg = ('%s ' % head if head else '') + '¥%.4g %s' % (
                p['amount'], p['unit_label'] or '元')
            parts.append(seg.strip())
            if discount < 1.0 - 1e-9:
                orig_parts.append('%.4g' % p['original_price'])

        pricing_text = ' | '.join(parts)
        original_text = ''
        if orig_parts:
            original_text = '原价 %s %s' % (
                ' / '.join(orig_parts), prices[0]['unit_label'] or '元')
        if price_table:
            # 有表格时文本降为摘要（一行文本装不下要素条件，仅作降级展示）
            original_text = ''

        return {'success': True, 'prices': prices, 'pricing_text': pricing_text,
                'original_text': original_text, 'discount': discount,
                'price_table': price_table}

    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'}

        product = product or {}

        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()

        # 到期门禁：订阅型产品（账号/存储）使用/开通前校验订阅未到期。
        # 到期 → 阻断使用，须续费（走 purchase_realtime）后恢复。
        if product.get('product_type') in ('account', 'workspace_storage'):
            gate = await self.check_subscription_valid(product.get('id'), user_org_id)
            if not gate.get('valid'):
                return {'success': False, 'message': gate.get('reason', '订阅不可用'),
                        'status': 'EXPIRED', 'end_date': gate.get('end_date', '')}

        result = await fn(ref_id, user_id, user_org_id, request_data or {})
        if not isinstance(result, dict) or not result.get('success'):
            return result

        # 权益生效：资源模块返回 entitlement 时，写 product_subscription（订阅/配额）
        ent = result.get('entitlement')
        if ent:
            try:
                result['subscription_id'] = await self._grant_subscription(
                    product, ent, user_id, user_org_id)
            except Exception as e:
                result['message'] = (result.get('message', '') or '') + f'；订阅写入失败: {e}'
        return result

    async def _grant_subscription(self, product, entitlement, user_id, user_org_id):
        """写 product_subscription：订阅有效期 + 配额（账号工作空间数/存储容量GB）。

        幂等：同一产品+机构已有活跃订阅则延长有效期与配额，不重复建。
        续费语义：现有订阅未到期时从 end_date 起顺延（不丢剩余天数），
        已到期则从今天重新起算。
        """
        dbname = self._get_dbname()
        now = time.strftime('%Y-%m-%d %H:%M:%S')
        today = datetime.date.today()
        days = int(entitlement.get('duration_days', 0) or 0)

        quota_total = 0.0
        quota_unit = entitlement.get('quota_unit', '')
        if entitlement.get('workspace_max'):
            quota_total = float(entitlement['workspace_max'])
            quota_unit = quota_unit or 'workspace'
        elif entitlement.get('storage_gb'):
            quota_total = float(entitlement['storage_gb'])
            quota_unit = quota_unit or 'GB'
        elif entitlement.get('workspace_gb'):
            quota_total = float(entitlement['workspace_gb'])
            quota_unit = quota_unit or 'GB'

        sub_id = getID()
        async with DBPools().sqlorContext(dbname) as sor:
            existing = await sor.sqlExe(
                "SELECT id, end_date FROM product_subscription WHERE product_id=${pid}$ "
                "AND user_org_id=${org}$ AND status='1' LIMIT 1",
                {'pid': product.get('id'), 'org': user_org_id})
            if existing:
                sub_id = existing[0].id
                # 续费：未到期从 end_date 顺延，已到期从今天重算
                try:
                    old_end = datetime.date.fromisoformat(str(existing[0].end_date)[:10])
                except Exception:
                    old_end = today
                base = old_end if old_end >= today else today
                end_date = (base + datetime.timedelta(days=days)).isoformat()
                await sor.U('product_subscription', {
                    'id': sub_id,
                    'end_date': end_date, 'quota_total': quota_total,
                    'quota_unit': quota_unit, 'updated_at': now})
            else:
                end_date = (today + datetime.timedelta(days=days)).isoformat() if days else today.isoformat()
                await sor.C('product_subscription', {
                    'id': sub_id, 'product_id': product.get('id'),
                    'user_id': user_id, 'user_org_id': user_org_id,
                    'subscription_type': '1', 'status': '1',
                    'start_date': today.isoformat(), 'end_date': end_date,
                    'quota_total': quota_total, 'quota_used': 0,
                    'quota_unit': quota_unit, 'overflow_mode': '1', 'overflow_rate': 0,
                    'purchase_price': float(product.get('price', 0) or 0),
                    'purchase_currency': product.get('currency', 'CNY'),
                    'created_at': now, 'updated_at': now})
        return sub_id

    # ═══════════════════════════════════════════════════════════════
    # 实时购买（选 A：购买即记账）
    # ═══════════════════════════════════════════════════════════════
    async def purchase_realtime(self, product_id=None, product_code=None,
                                user_id=None, user_org_id=None,
                                charge_mode=None, quantity=1,
                                storage_gb=0, valid_months=1):
        """实时购买：点击支付 → 算价 → 余额预检 → 计费写用量 → 实时复式记账 → 开权益。

        账号产品（account）：charge_mode(trial/month/year) 决定有效期与定价。
        存储产品（workspace_storage）：storage_gb × valid_months 直购买断。
        任一步失败整体回滚（删除已写用量记录），保证不落半截账。
        返回 {'success','message','orderid','amount','subscription_id','end_date'}
        """
        env = ServerEnv()
        dbname = self._get_dbname()

        iface, product, err = await self._get_product_interface(product_id, product_code)
        if err:
            return {'success': False, 'message': err}
        product = product or {}
        product_type = product.get('product_type', '')
        ref_id, err = self._get_ref_id(product)
        if err:
            return {'success': False, 'message': err}

        if not user_id:
            user_id = await env.get_user()
        if not user_org_id:
            fn = getattr(env, 'get_userorgid', None)
            if callable(fn):
                user_org_id = await fn()
            user_org_id = user_org_id or self._get_current_org_id() or '0'

        quantity = int(quantity or 1)
        storage_gb = float(storage_gb or 0)
        valid_months = max(1, int(valid_months or 1))

        usage_rec = None
        charging_module = None  # 记录计费来源，失败时回滚用
        try:
            if product_type == 'account':
                # 解析 charge_mode（未传则取该规格缺省定价映射）
                charge_mode, valid_days = await self._resolve_account_charge(ref_id, charge_mode)
                from account_resource.init import account_charging as _acc_charging
                charging_module = 'account'
                usage_rec = await _acc_charging(
                    ref_id, charge_mode, user_id, user_org_id,
                    quantity=quantity)
                duration_days = int(valid_days or 0) * quantity
                entitlement = await self._build_account_entitlement(ref_id, charge_mode, duration_days)
                usage_data = {'charge_mode': charge_mode, 'duration': quantity}

            elif product_type == 'workspace_storage':
                if storage_gb <= 0:
                    return {'success': False, 'message': '请填写购买容量(GB)'}
                from storage_resource.init import storage_direct_charging as _sto_charging
                charging_module = 'storage'
                usage_rec = await _sto_charging(
                    ref_id, storage_gb, user_id, user_org_id,
                    valid_months=valid_months)
                duration_days = 30 * valid_months
                entitlement = {'storage_gb': storage_gb, 'quota_unit': 'GB',
                               'duration_days': duration_days}
                usage_data = {'meter_mode': 'direct', 'storage_gb': storage_gb,
                              'valid_months': valid_months}

            elif product_type == 'pipeline':
                # 产线使用费（订阅制）：无 usage 表，直接算价+实时记账+开权益
                # （2026-09-06 产线产品化：月付 months=1 / 年付 months=10=月费×10）
                from pipeline_core.pipeline_pricing import calculate_pipeline_amount
                charging_module = 'pipeline'
                cm = charge_mode or 'month'
                if cm not in ('month', 'year'):
                    return {'success': False, 'message': '产线只支持 charge_mode=month/year'}
                amount, ppid, months, duration_days = await calculate_pipeline_amount(ref_id, cm)
                entitlement = {'duration_days': duration_days, 'charge_mode': cm,
                               'pipeline_id': ref_id}
                usage_data = {'charge_mode': cm, 'months': months}
            else:
                return {'success': False,
                        'message': f'产品类型({product_type})暂不支持实时购买'}

            # ── 算价（售价，含客户折扣）用于余额预检 ──
            cost = await self.calculate_product_cost(
                product_id=product.get('id'), usage_data=usage_data,
                user_org_id=user_org_id)
            sell_amount = float((cost or {}).get('amount', 0) or 0)
            if not (cost or {}).get('success', True):
                raise Exception((cost or {}).get('message', '算价失败'))

            # ── 余额预检：客户资金账户余额 ≥ 售价 ──
            balance = await self._check_balance(user_org_id)
            if balance is None:
                raise Exception('账户未开通，请先完成开户')
            if balance < sell_amount:
                raise Exception(f'余额不足：可用 {balance:.2f}，应付 {sell_amount:.2f}，请先充值')

            # ── 实时复式记账（内部写 biz_order + 分录，余额不足会抛 AccountOverDraw）──
            if charging_module == 'pipeline':
                # 产线无 usage 表：直接走通用落账（product_accounting_generic
                # 内部再算一次价含客户折扣，与预检同口径）
                acc = await self.product_accounting_generic(
                    product.get('id'), usage_data, user_org_id, user_id)
            else:
                rec_obj = DictObject(**usage_rec) if isinstance(usage_rec, dict) else usage_rec
                if charging_module == 'account':
                    acc = await self.account_resource_accounting(rec_obj)
                else:
                    acc = await self.storage_resource_accounting(rec_obj)
            if not acc or not acc.get('success'):
                raise Exception(f'记账失败: {acc}')

            # ── 开权益：写订阅（配额 + 到期日），续费自动顺延 ──
            sub_id = await self._grant_subscription(product, entitlement, user_id, user_org_id)
            end_date = await self._get_subscription_end(sub_id)

            return {'success': True, 'message': '购买成功，已实时记账',
                    'orderid': acc.get('orderid', ''),
                    'amount': round(float(acc.get('customer_amount', sell_amount)), 2),
                    'subscription_id': sub_id, 'end_date': end_date,
                    'product_name': product.get('product_name', '')}

        except Exception as e:
            exception(f'purchase_realtime failed: {e}')
            # 回滚：删除已写的待记账用量记录，避免残留半截账
            # （产线 charging_module='pipeline' 无 usage 表，usage_rec 恒 None 天然跳过）
            if usage_rec is not None:
                try:
                    rid = usage_rec.get('id') if isinstance(usage_rec, dict) \
                        else getattr(usage_rec, 'id', '')
                    tbl = 'acctres_usage' if charging_module == 'account' else 'storres_usage'
                    async with DBPools().sqlorContext(dbname) as sor:
                        await sor.D(tbl, {'id': rid})
                except Exception:
                    pass
            return {'success': False, 'message': str(e)}

    async def _resolve_account_charge(self, spec_id, charge_mode):
        """解析账号 charge_mode 与 valid_days（未传则取缺省映射）。"""
        dbname = self._get_dbname()
        cond = "spec_id=${sid}$ AND status='active'"
        ns = {'sid': spec_id}
        if charge_mode:
            cond += " AND charge_mode=${cm}$"
            ns['cm'] = charge_mode
        else:
            cond += " AND is_default='1'"
        async with DBPools().sqlorContext(dbname) as sor:
            recs = await sor.sqlExe(
                f"SELECT charge_mode, valid_days FROM acctres_pricing_map WHERE {cond}", ns)
        if recs:
            return recs[0].charge_mode, int(recs[0].valid_days or 0)
        # 兜底
        return (charge_mode or 'month'), {'trial': 7, 'month': 30, 'year': 365}.get(charge_mode or 'month', 30)

    async def _build_account_entitlement(self, spec_id, charge_mode, duration_days):
        """账号权益：有效期 + 规格配额（工作空间数/容量/并发/成员）。"""
        env = ServerEnv()
        ent = {'duration_days': duration_days, 'charge_mode': charge_mode}
        try:
            spec = await env.get_account_spec(spec_id)
            if spec:
                ent['workspace_max'] = int(getattr(spec, 'workspace_max', 0) or 0)
                ent['workspace_gb'] = float(getattr(spec, 'workspace_gb', 0) or 0)
                ent['concurrent_task'] = int(getattr(spec, 'concurrent_task', 0) or 0)
                ent['member_max'] = int(getattr(spec, 'member_max', 0) or 0)
        except Exception:
            pass
        return ent

    async def _check_balance(self, user_org_id):
        """查客户资金账户余额。未开户返回 None。"""
        env = ServerEnv()
        try:
            acc_dbname = env.get_module_dbname('accounting')
            async with DBPools().sqlorContext(acc_dbname) as sor:
                return await env.getCustomerBalance(sor, user_org_id)
        except Exception as e:
            debug(f'_check_balance failed: {e}')
            return None

    async def _get_subscription_end(self, sub_id):
        """读订阅到期日（供购买结果展示）。"""
        dbname = self._get_dbname()
        try:
            async with DBPools().sqlorContext(dbname) as sor:
                recs = await sor.sqlExe(
                    "SELECT end_date FROM product_subscription WHERE id=${id}$",
                    {'id': sub_id})
                if recs:
                    return str(recs[0].end_date)[:10]
        except Exception:
            pass
        return ''

    async def get_my_subscriptions(self, user_org_id=None):
        """我的权益列表：当前机构所有订阅 + 到期状态。"""
        dbname = self._get_dbname()
        if not user_org_id:
            env = ServerEnv()
            user_org_id = await env.get_userorgid() if hasattr(env, 'get_userorgid') \
                else self._get_current_org_id()
        today = datetime.date.today().isoformat()
        async with DBPools().sqlorContext(dbname) as sor:
            recs = await sor.sqlExe(
                """SELECT s.*, p.product_name, p.product_code, p.product_type
                   FROM product_subscription s
                   LEFT JOIN product p ON s.product_id = p.id
                   WHERE s.user_org_id=${org}$ AND s.status='1'
                   ORDER BY s.end_date DESC""", {'org': user_org_id})
        result = []
        for r in recs:
            end = str(getattr(r, 'end_date', '') or '')[:10]
            expired = end < today if end else True
            result.append({
                'id': getattr(r, 'id', ''),
                'product_name': getattr(r, 'product_name', ''),
                'product_code': getattr(r, 'product_code', ''),
                'product_type': getattr(r, 'product_type', ''),
                'product_id': getattr(r, 'product_id', ''),
                'quota_total': float(getattr(r, 'quota_total', 0) or 0),
                'quota_used': float(getattr(r, 'quota_used', 0) or 0),
                'quota_unit': getattr(r, 'quota_unit', ''),
                'start_date': str(getattr(r, 'start_date', '') or '')[:10],
                'end_date': end,
                'expired': expired,
                'status_text': '已到期' if expired else '生效中',
            })
        return result

    # ═══════════════════════════════════════════════════════════════
    # 到期检查
    # ═══════════════════════════════════════════════════════════════
    async def check_subscription_valid(self, product_id, user_org_id):
        """到期门禁：订阅型产品使用前校验订阅未到期。

        返回 {'valid': bool, 'reason': str, 'end_date': str}。
        valid=False 时调用方必须阻断使用动作。无订阅记录视为未购买（阻断）。
        """
        dbname = self._get_dbname()
        today = datetime.date.today().isoformat()
        async with DBPools().sqlorContext(dbname) as sor:
            recs = await sor.sqlExe(
                "SELECT end_date FROM product_subscription WHERE product_id=${pid}$ "
                "AND user_org_id=${org}$ AND status='1' ORDER BY end_date DESC LIMIT 1",
                {'pid': product_id, 'org': user_org_id})
        if not recs:
            return {'valid': False, 'reason': '未购买该产品，请先购买', 'end_date': ''}
        end = str(getattr(recs[0], 'end_date', '') or '')[:10]
        if not end or end < today:
            return {'valid': False, 'reason': f'订阅已到期({end})，请续费后使用',
                    'end_date': end}
        return {'valid': True, 'reason': '', 'end_date': end}

    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):
        """计算消费费用：资源真实定价 + 客户折扣(售价) + 供应商折扣(成本)。

        三类折扣各自独立、归属不同环节，不取 min：
        - 客户折扣   discount 模块 get_min_product_discount → amount（客户应付售价）
        - 供应商折扣 supplychain 模块 calculate_sale_amounts → cost（平台进货成本）
        - 分销商折扣 分销场景单独用（calculate_sale_amounts 的 distribution_amount），
          不进本函数。

        返回：original_amount(资源真实定价)、amount(售价=原价×客户折扣)、
              cost(成本=原价×供应商折扣)、discount(客户折扣率)
        """
        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'}

        product = product or {}

        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()

        # 1. 资源层真实定价（原价，资源模块不碰折扣）
        result = await fn(ref_id, usage_data or {}, user_org_id)
        if not isinstance(result, dict):
            return result
        if result.get('success') is False:
            return result
        original_amount = float(result.get('original_amount',
                                           result.get('amount', 0) or 0))

        env = ServerEnv()
        resellerid = product.get('org_id') or '0'
        pid = product.get('id') or ''

        # 2. 客户折扣（discount 模块，售价侧）→ amount
        discount = 1.0
        if user_org_id:
            try:
                d = await env.get_min_product_discount(pid, resellerid, user_org_id)
                if d not in (None, '', 0):
                    discount = float(d)
            except Exception:
                pass
        amount = original_amount * discount

        # 3. 供应商成本（supplychain 模块，成本侧）→ cost
        #    供应商成本是平台进货成本，与客户折扣独立，不取 min。
        #    优先 product_supplier_mapping 供货价 supply_price（平台内/外部统一）；
        #    无供货价则按外部供应商合同折扣 supply_contract_items 算。
        cost = original_amount
        try:
            dbname = self._get_dbname()
            mapping = None
            async with DBPools().sqlorContext(dbname) as sor:
                maps = await sor.R('product_supplier_mapping', {'product_id': pid})
                if maps:
                    pref = [m for m in maps
                            if str(getattr(m, 'is_preferred', '')) == '1'] or maps
                    mapping = pref[0]

            if mapping is not None:
                # 首选供应商映射：优先直接供货价
                supply_price = getattr(mapping, 'supply_price', None)
                if supply_price not in (None, '', 0):
                    cost = float(supply_price)
                else:
                    supplier_id = (getattr(mapping, 'external_supplier_id', '')
                                   or product.get('providerid') or '')
                    if supplier_id:
                        sc = await env.calculate_sale_amounts(None, {
                            'resellerid': resellerid,
                            'supplier_id': supplier_id,
                            'productid': pid,
                            'prodtypeid': product.get('category_id') or '',
                            'quantity': 1,
                            'unit_price': original_amount,
                        })
                        if sc:
                            sc_data = json.loads(sc) if isinstance(sc, str) else sc
                            sc_data = sc_data.get('data', {}) if isinstance(sc_data, dict) else {}
                            supply_discount = float(sc_data.get('supply_discount', 1.0) or 1.0)
                            cost = original_amount * supply_discount
            else:
                # 无映射，回退 product.providerid（外部供应商走合同折扣）
                supplier_id = product.get('providerid') or ''
                if supplier_id:
                    sc = await env.calculate_sale_amounts(None, {
                        'resellerid': resellerid,
                        'supplier_id': supplier_id,
                        'productid': pid,
                        'prodtypeid': product.get('category_id') or '',
                        'quantity': 1,
                        'unit_price': original_amount,
                    })
                    if sc:
                        sc_data = json.loads(sc) if isinstance(sc, str) else sc
                        sc_data = sc_data.get('data', {}) if isinstance(sc_data, dict) else {}
                        supply_discount = float(sc_data.get('supply_discount', 1.0) or 1.0)
                        cost = original_amount * supply_discount
        except Exception:
            cost = original_amount

        result['amount'] = round(amount, 6)
        result['original_amount'] = round(original_amount, 6)
        result['cost'] = round(cost, 6)
        result['discount'] = discount
        return result

    async def llm_id_to_product_id(self, llm_id):
        """Convert LLM model ID (llmage.llm.id) to product ID (product.id).

        Uses resource_ref_id mapping: product.resource_ref_id = llm.id
        and product.product_type = 'llm_model'.
        """
        dbname = self._get_dbname()
        async with DBPools().sqlorContext(dbname) as sor:
            sql = """SELECT id FROM product
WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1"""
            recs = await sor.sqlExe(sql, {'llm_id': llm_id})
            if recs:
                return recs[0].id
            return None

    # ─── Product Accounting ───

    async def product_accounting(self, llmusage):
        """llmage 用量计费落账（薄封装：转换 llmusage → 通用参数 → generic）。"""
        llmid = llmusage.llmid
        userorgid = llmusage.userorgid
        userid = llmusage.userid

        product_id = await self.llm_id_to_product_id(llmid)
        if not product_id:
            raise Exception(f'llm({llmid}) has no product mapping')

        from llmage.utils import get_llmage_llm
        llm = await get_llmage_llm(llmid)
        if not llm:
            raise Exception(f'llm({llmid}) not found')
        ownerid = getattr(llm, 'ownerid', '0') or '0'
        providerid = getattr(llm, 'providerid', '0') or '0'

        usages = llmusage.usages
        if isinstance(usages, str):
            usages = json.loads(usages)

        result = await self.product_accounting_generic(
            product_id, usages, userorgid, userid,
            providerid=providerid, ownerid=ownerid,
            tenantid=getattr(llmusage, 'tenantid', None))

        # finalize_balance（llmage 专属，非关键）
        try:
            luid = getattr(llmusage, 'id', None)
            if luid:
                from llmage.balance import finalize_balance
                await finalize_balance(ServerEnv(), luid, result.get('customer_amount', 0),
                                       llmid=llmid)
        except Exception as e:
            exception(f'finalize_balance failed (non-critical): {e}')

        # 标记 llmusage accounted
        lmage_dbname = ServerEnv().get_module_dbname('llmage')
        async with DBPools().sqlorContext(lmage_dbname) as sor:
            await sor.U('llmusage', {'id': llmusage.id, 'accounting_status': 'accounted'})

        return result

    async def product_accounting_generic(self, product_id, usage_data, userorgid, userid,
                                         providerid=None, ownerid=None, tenantid=None):
        """通用商品计费落账（llm/账号/存储通用）。

        核心：算金额(calculate_product_cost，含客户折扣+供应商成本) →
              分销链各环节金额 → consume_accounting 落账。
        返回 {'success', 'orderid', 'customer_amount', 'supplier_cost', 'original_amount'}
        """
        env = ServerEnv()
        ownerid = ownerid or '0'
        providerid = providerid or '0'
        is_self_use = (userorgid or '0') == (ownerid or '0')

        # 1. 算金额（客户售价 amount + 供应商成本 cost）
        cost_result = await self.calculate_product_cost(
            product_id=product_id, usage_data=usage_data, user_org_id=userorgid)
        if not cost_result or not cost_result.get('success'):
            raise Exception(f'calculate_product_cost failed: {cost_result}')
        raw_cost = cost_result.get('amount', 0)
        original_amount = cost_result.get('original_amount', 0) or raw_cost
        supplier_cost = cost_result.get('cost', 0) or raw_cost

        # 2. 币种转换
        prod_currency = 'CNY'
        try:
            dbname = self._get_dbname()
            async with DBPools().sqlorContext(dbname) as sor:
                prod_rows = await sor.sqlExe(
                    "SELECT currency FROM product WHERE id=${pid}$", {'pid': product_id})
                if prod_rows:
                    prod_currency = prod_rows[0].currency or 'CNY'
        except Exception:
            pass

        user_currency = 'CNY'
        cost_in_user = raw_cost
        cost_base = raw_cost
        try:
            user_currency = await env.get_user_currency(userorgid)
            if prod_currency != user_currency:
                rate = await env.get_exchange_rate(prod_currency, user_currency, 'sell_rate')
                cost_in_user = round(raw_cost * rate, 2)
            cost_base = await env.convert_to_base(cost_in_user, user_currency, 'sell_rate')
        except Exception:
            pass

        # 3. 分销链
        seller_orgid = tenantid or userorgid
        try:
            from supplychain.init import get_distribution_chain as _get_chain
            chain = await _get_chain(ownerid, seller_orgid)
        except Exception:
            chain = [ownerid]

        customer_amount = cost_in_user
        biz_date = await env.get_business_date(None)
        timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
        orderid = getID()

        # 4. 构建 accounting_items（PAY/PAY*）
        accounting_items = []
        if len(chain) == 1:
            if not is_self_use:
                accounting_items.append(DictObject(
                    action='PAY', customerid=userorgid, resellerid=ownerid,
                    providerid=providerid, biz_date=biz_date, timestamp=timestamp,
                    productid=product_id, transamt=customer_amount,
                    currency=user_currency, base_amount=round(customer_amount, 2),
                    variable={"交易金额": customer_amount, "交易币种": user_currency, "交易手续费": 0}))
            accounting_items.append(DictObject(
                action='PAY*', customerid=ownerid, resellerid=ownerid,
                providerid=providerid, biz_date=biz_date, timestamp=timestamp,
                productid=product_id, transamt=supplier_cost,
                currency=prod_currency, base_amount=cost_base,
                variable={"采购成本": supplier_cost, "采购币种": prod_currency}))
        else:
            for i, orgid in enumerate(chain):
                is_last = (i == len(chain) - 1)
                if is_last:
                    if not is_self_use:
                        accounting_items.append(DictObject(
                            action='PAY', customerid=userorgid, resellerid=orgid,
                            providerid=providerid, biz_date=biz_date, timestamp=timestamp,
                            productid=product_id, transamt=customer_amount,
                            currency=user_currency, base_amount=round(customer_amount, 2),
                            variable={"交易金额": customer_amount, "交易币种": user_currency, "交易手续费": 0}))
                else:
                    next_orgid = chain[i + 1]
                    link_amount = raw_cost
                    try:
                        sc = await env.calculate_sale_amounts(None, {
                            'resellerid': next_orgid,
                            'sub_reseller_id': orgid,
                            'productid': product_id,
                            'prodtypeid': '',
                            'quantity': 1,
                            'unit_price': original_amount,
                        })
                        if sc:
                            scd = json.loads(sc) if isinstance(sc, str) else sc
                            scd = scd.get('data', {}) if isinstance(scd, dict) else {}
                            link_amount = float(scd.get('distribution_amount', raw_cost) or raw_cost)
                    except Exception:
                        pass
                    accounting_items.append(DictObject(
                        action='PAY*', customerid=orgid, resellerid=next_orgid,
                        providerid=providerid, biz_date=biz_date, timestamp=timestamp,
                        productid=product_id, transamt=link_amount,
                        currency=prod_currency, base_amount=round(link_amount, 2),
                        variable={"分销结算": link_amount, "结算币种": prod_currency}))
            accounting_items.append(DictObject(
                action='PAY*', customerid=chain[0], resellerid=chain[0],
                providerid=providerid, biz_date=biz_date, timestamp=timestamp,
                productid=product_id, transamt=supplier_cost,
                currency=prod_currency, base_amount=cost_base,
                variable={"采购成本": supplier_cost, "采购币种": prod_currency}))

        # 5. 写 biz_order + consume_accounting
        dbname = self._get_dbname()
        db = DBPools()
        config = getConfig()
        db.databases = config.databases
        async with db.sqlorContext(dbname) as sor:
            await sor.C('biz_order', {
                "id": orderid, "customerid": userorgid, "resellerid": ownerid,
                "order_date": biz_date, "order_status": "1", "business_op": "PAY",
                "amount": customer_amount, "currency": user_currency,
                "userid": userid, "productid": product_id})
            await sor.C('biz_orderdetail', {
                "id": getID(), "orderid": orderid, "productid": product_id,
                "product_cnt": 1, "trans_amount": customer_amount, "currency": user_currency})
            await consume_accounting(sor, orderid, accounting_items)

        return {'success': True, 'orderid': orderid, 'customer_amount': customer_amount,
                'supplier_cost': supplier_cost, 'original_amount': original_amount}

    async def _resource_ref_to_product_id(self, ref_id, product_type):
        """资源引用 id（spec.id）→ 产品 id（按 product_type 区分资源模块）。"""
        dbname = self._get_dbname()
        async with DBPools().sqlorContext(dbname) as sor:
            recs = await sor.sqlExe(
                "SELECT id FROM product WHERE resource_ref_id=${rid}$ "
                "AND product_type=${pt}$ LIMIT 1",
                {'rid': ref_id, 'pt': product_type})
            if recs:
                return recs[0].id
        return None

    async def account_resource_accounting(self, usage_record):
        """账号用量计费落账（acctres_usage → generic）。"""
        spec_id = getattr(usage_record, 'spec_id', '')
        userorgid = getattr(usage_record, 'userorgid', '') or '0'
        userid = getattr(usage_record, 'userid', '') or ''
        product_id = await self._resource_ref_to_product_id(spec_id, 'account')
        if not product_id:
            raise Exception(f'acctres spec({spec_id}) has no product mapping')
        usages = getattr(usage_record, 'usages', '{}')
        if isinstance(usages, str):
            usages = json.loads(usages)
        result = await self.product_accounting_generic(product_id, usages, userorgid, userid)
        from account_resource.init import mark_acctres_accounted
        await mark_acctres_accounted(getattr(usage_record, 'id', ''))
        return result

    async def storage_resource_accounting(self, usage_record):
        """存储用量计费落账（storres_usage → generic）。"""
        spec_id = getattr(usage_record, 'spec_id', '')
        userorgid = getattr(usage_record, 'userorgid', '') or '0'
        userid = getattr(usage_record, 'userid', '') or ''
        product_id = await self._resource_ref_to_product_id(spec_id, 'workspace_storage')
        if not product_id:
            raise Exception(f'storres spec({spec_id}) has no product mapping')
        usages = getattr(usage_record, 'usages', '{}')
        if isinstance(usages, str):
            usages = json.loads(usages)
        result = await self.product_accounting_generic(product_id, usages, userorgid, userid)
        from storage_resource.init import mark_storres_accounted
        await mark_storres_accounted(getattr(usage_record, 'id', ''))
        return result

    async def backend_accounting(self):
        """Background accounting loop — replaces llmage.backend_accounting."""
        import asyncio
        from datetime import datetime, timedelta
        from llmage.accounting import get_accounting_llmusages, llm_accoung_failed
        from llmage.utils import get_user_tpac
        
        info(f"product backend accounting started ...")
        last_backup_date = None
        while True:
            try:
                lus = await get_accounting_llmusages()
                info(f"accounting loop: got {len(lus)} records")
            except Exception as e:
                exception(f"get_accounting_llmusages failed: {e}")
                lus = []
            for lu in lus:
                try:
                    tpac = await get_user_tpac(lu.userid)
                    if tpac:
                        from llmage.accounting import tpac_accounting
                        debug(f'{lu.id=},{lu.userid=}, {tpac=}, go tpac')
                        await tpac_accounting(tpac, lu.userid, lu.llmid, lu.amount, lu.usages, lu.id, lu.model)
                    else:
                        debug(f'{lu.id=},{lu.userid=}, go product accounting')
                        await self.product_accounting(lu)
                    # Clean up failed records
                    try:
                        lmage_dbname = ServerEnv().get_module_dbname('llmage')
                        async with DBPools().sqlorContext(lmage_dbname) as sor:
                            await sor.execute(
                                "DELETE FROM llmusage_accounting_failed WHERE llmusageid=${luid}$",
                                {'luid': lu.id})
                    except Exception as e2:
                        debug(f'清理失败记录异常: {e2}')
                except Exception as e:
                    exception(f'{e}, {lu.id=}')
                    await llm_accoung_failed(lu.id, reason=str(e))
            
            # Daily backup
            today = datetime.now().strftime('%Y-%m-%d')
            if today != last_backup_date:
                yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
                last_backup_date = today
                try:
                    from llmage.accounting import backup_accounted_llmusage
                    debug(f'date changed to {today}, triggering backup')
                    await backup_accounted_llmusage(yesterday)
                except Exception as e:
                    exception(f'backup failed: {e}')
            
            await asyncio.sleep(10)