feat(pipeline): 产线产品化接入 product_management——计费层+产品接口+定价映射表

产线使用费（订阅制）照 account_resource/storage_resource 范式接入产品层：
- models/pipeline_pricing_map.json: pipeline_id × charge_mode(month/year) → ppid
  （月付 months=1/valid_days=30；年付 months=10/valid_days=365，即月费×10 付10个月用12个月）
- pipeline_pricing.py: 定价映射 CRUD + calculate_pipeline_amount
  （buffered_charging(ppid, {months}) 算原价；无定价方案/金额0 均抛异常不静默）
- product_interface.py: 7 个标准接口（load_product_category_product 导入用；
  get_product_display 展示月/年双档价；execute_product_service 开订阅权益）
- init.py load_pipeline_core: 注册 env.product_interface（module_name='pipeline_core'）
  + env.pipeline_pricing_map_save/get_pipeline_ppid/calculate_pipeline_amount

资源主表复用 pipelines（status='published' 才可售），不建 usage 表——
订阅购买是同步实时记账（purchase_realtime → product_accounting_generic 写 biz_order）。

用户定稿的商业模型（2026-09-06）：账号免费只管并发配额；产线单独包月
（商机100/投标150/开发200，年付=月×10）；模型按量加价；存储 1元/GB月。

models/pipeline_pricing_map.json

@@ -0,0 +1,28 @@
+{
+    "summary": [
+        {
+            "name": "pipeline_pricing_map",
+            "title": "产线计费映射（pipeline_id × charge_mode → ppid，照 acctres_pricing_map 范式）",
+            "primary": ["id"]
+        }
+    ],
+    "fields": [
+        {"name": "id", "title": "ID", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "pipeline_id", "title": "产线ID(pipelines.id)", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "charge_mode", "title": "计费方式(month/year)", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "months", "title": "计费月数(月付=1,年付=10即月费×10)", "type": "int", "nullable": "no", "default": "1"},
+        {"name": "valid_days", "title": "权益有效天数(月付30/年付365)", "type": "int", "nullable": "no", "default": "30"},
+        {"name": "ppid", "title": "定价项目ID", "type": "str", "length": 32, "nullable": "yes"},
+        {"name": "is_default", "title": "缺省方式", "type": "str", "length": 1, "nullable": "yes", "default": "0"},
+        {"name": "status", "title": "状态", "type": "str", "length": 16, "nullable": "yes", "default": "active"},
+        {"name": "org_id", "title": "所属机构", "type": "str", "length": 32, "nullable": "yes"},
+        {"name": "created_by", "title": "创建人", "type": "str", "length": 32, "nullable": "yes"},
+        {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "yes"},
+        {"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "yes"}
+    ],
+    "indexes": [
+        {"name": "idx_ppm_pipeline", "idxtype": "index", "idxfields": ["pipeline_id"]},
+        {"name": "idx_ppm_mode", "idxtype": "unique", "idxfields": ["pipeline_id", "charge_mode"]},
+        {"name": "idx_ppm_ppid", "idxtype": "index", "idxfields": ["ppid"]}
+    ]
+}

pipeline_core/init.py

@@ -18,5 +18,32 @@ def load_pipeline_core():
     """注册函数到 ServerEnv"""
     # 2026-09-04：旧 llm 表 CRUD 已移除——
     # 模型注册/管理统一收敛到模型治理模块（pipeline-llm），旧 `llm` 表停用。
+
+    # 2026-09-06：产线产品化——注册 product_management 资源模块接口
+    # （照 account_resource/storage_resource 范式；env.product_interface 是单例，
+    # 产品层按 product_category.resource_module 动态 import + 调本函数重设）。
+    from ahserver.serverenv import ServerEnv
+    from .product_interface import (
+        get_product_display, check_product_availability, check_product_consumable,
+        execute_product_service, execute_product_service_stream,
+        calculate_product_cost, load_product_category_product)
+    env = ServerEnv()
+    env.product_interface = {
+        'module_name': MODULE_NAME,
+        'get_product_display': get_product_display,
+        'check_product_availability': check_product_availability,
+        'check_product_consumable': check_product_consumable,
+        'execute_product_service': execute_product_service,
+        'execute_product_service_stream': execute_product_service_stream,
+        'calculate_product_cost': calculate_product_cost,
+        'load_product_category_product': load_product_category_product,
+    }
+    # 产线计费函数（订阅购买/定价解析），供产品层与运维使用
+    from .pipeline_pricing import (
+        pipeline_pricing_map_save, get_pipeline_ppid, calculate_pipeline_amount)
+    env.pipeline_pricing_map_save = pipeline_pricing_map_save
+    env.get_pipeline_ppid = get_pipeline_ppid
+    env.calculate_pipeline_amount = calculate_pipeline_amount
+
     debug(f'[{MODULE_NAME}] module loaded')
     return True

pipeline_core/pipeline_pricing.py

@@ -0,0 +1,169 @@
+"""pipeline_core 产线计费层 — 产线使用费（订阅制）
+
+照 account_resource / storage_resource 范式（资源产品化五层）：
+    资源主表:   pipelines（id=bidding_general 等，无 ppid）
+    定价映射:   pipeline_pricing_map（pipeline_id × charge_mode → ppid）
+    计费:       buffered_charging(ppid, {'months': N}) → 月费 × N
+                （月付 months=1；年付 months=10，即"年费=月费×10"送2个月）
+
+产线不建 usage 表：订阅购买是同步实时记账（purchase_realtime →
+product_accounting_generic 写 biz_order），订单即审计记录。
+
+定价单位约定（pricing_program_timing YAML）：
+    unit_values: {月: 1}
+    pricings: [{price_factors: months, unit_prices: 100, unit: 月}]
+一个 ppid 服务一条产线（月付/年付共用，靠 months 因子区分）。
+"""
+import datetime
+import json
+import logging
+
+from ahserver.serverenv import ServerEnv
+from appPublic.uniqueID import getID
+from sqlor.dbpools import DBPools
+
+logger = logging.getLogger(__name__)
+
+MODULE_NAME = 'pipeline_core'
+RESOURCE_TYPE = 'pipeline'   # product_resource.resource_type / product.product_type
+
+
+def _get_dbname():
+    env = ServerEnv()
+    return env.get_module_dbname(MODULE_NAME)
+
+
+def _now():
+    return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+
+
+def _val(rec, key, default=None):
+    if isinstance(rec, dict):
+        return rec.get(key, default)
+    return getattr(rec, key, default)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# 资源主表（pipelines，只读——产线定义由能力包管理）
+# ═══════════════════════════════════════════════════════════════════
+
+async def pipeline_res_list(ns=None):
+    """可售卖产线列表（status='published'）。"""
+    ns = ns or {}
+    sql = "SELECT * FROM pipelines WHERE status='published'"
+    params = {}
+    if ns.get('org_id'):
+        sql += " AND org_id=${org_id}$"
+        params['org_id'] = ns['org_id']
+    sql += " ORDER BY id"
+    async with DBPools().sqlorContext(_get_dbname()) as sor:
+        recs = await sor.sqlExe(sql, params)
+        await sor.sqlExe("COMMIT", {})
+    return recs or []
+
+
+async def get_pipeline_res(pipeline_id):
+    async with DBPools().sqlorContext(_get_dbname()) as sor:
+        recs = await sor.sqlExe(
+            "SELECT * FROM pipelines WHERE id=${pid}$", {'pid': pipeline_id})
+        await sor.sqlExe("COMMIT", {})
+    return recs[0] if recs else None
+
+
+# ═══════════════════════════════════════════════════════════════════
+# 定价映射（pipeline_pricing_map）
+# ═══════════════════════════════════════════════════════════════════
+
+async def pipeline_pricing_map_list(ns=None):
+    ns = ns or {}
+    sql = "SELECT * FROM pipeline_pricing_map WHERE 1=1"
+    params = {}
+    if ns.get('pipeline_id'):
+        sql += " AND pipeline_id=${pid}$"
+        params['pid'] = ns['pipeline_id']
+    if ns.get('status'):
+        sql += " AND status=${st}$"
+        params['st'] = ns['status']
+    sql += " ORDER BY pipeline_id, charge_mode"
+    async with DBPools().sqlorContext(_get_dbname()) as sor:
+        recs = await sor.sqlExe(sql, params)
+        await sor.sqlExe("COMMIT", {})
+    return recs or []
+
+
+async def get_pipeline_ppid(pipeline_id, charge_mode=None):
+    """解析产线定价方案。charge_mode 空时取 is_default='1'。
+
+    返回 (ppid, months, valid_days)；无映射返回 ('', 0, 0)。
+    """
+    sql = ("SELECT ppid, months, valid_days FROM pipeline_pricing_map "
+           "WHERE pipeline_id=${pid}$ AND status='active'")
+    params = {'pid': pipeline_id}
+    if charge_mode:
+        sql += " AND charge_mode=${cm}$"
+        params['cm'] = charge_mode
+    else:
+        sql += " AND is_default='1'"
+    async with DBPools().sqlorContext(_get_dbname()) as sor:
+        recs = await sor.sqlExe(sql, params)
+        await sor.sqlExe("COMMIT", {})
+    if recs:
+        return (_val(recs[0], 'ppid', '') or '',
+                int(_val(recs[0], 'months', 1) or 1),
+                int(_val(recs[0], 'valid_days', 30) or 30))
+    return '', 0, 0
+
+
+async def pipeline_pricing_map_save(data):
+    """幂等 upsert（按 pipeline_id+charge_mode 唯一）。"""
+    data = dict(data or {})
+    pid = data.get('pipeline_id', '')
+    cm = data.get('charge_mode', '')
+    if not pid or not cm:
+        raise Exception('pipeline_pricing_map_save: 缺 pipeline_id/charge_mode')
+    async with DBPools().sqlorContext(_get_dbname()) as sor:
+        recs = await sor.sqlExe(
+            "SELECT id FROM pipeline_pricing_map WHERE pipeline_id=${p}$ AND charge_mode=${c}$",
+            {'p': pid, 'c': cm})
+        if recs:
+            data['id'] = _val(recs[0], 'id')
+            data['updated_at'] = _now()
+            await sor.U('pipeline_pricing_map', data)
+        else:
+            data.setdefault('id', getID())
+            data.setdefault('status', 'active')
+            data.setdefault('is_default', '0')
+            data.setdefault('org_id', '0')
+            data['created_at'] = _now()
+            data['updated_at'] = _now()
+            await sor.C('pipeline_pricing_map', data)
+    return data['id']
+
+
+# ═══════════════════════════════════════════════════════════════════
+# 计费（订阅购买，同步）
+# ═══════════════════════════════════════════════════════════════════
+
+async def calculate_pipeline_amount(pipeline_id, charge_mode='month'):
+    """产线使用费原价 = 定价引擎(ppid, {months}) 结果。
+
+    返回 (amount, ppid, months, valid_days)。无定价方案抛异常（不静默0）。
+    """
+    ppid, months, valid_days = await get_pipeline_ppid(pipeline_id, charge_mode)
+    if not ppid:
+        # 未指定 charge_mode 映射时回退缺省
+        ppid, months, valid_days = await get_pipeline_ppid(pipeline_id)
+    if not ppid:
+        raise Exception('产线 %s 未配置定价方案（pipeline_pricing_map 无 ppid）' % pipeline_id)
+    env = ServerEnv()
+    charging = getattr(env, 'buffered_charging', None)
+    if charging is None:
+        raise Exception('pricing 模块未加载，无法计算产线使用费')
+    usages = {'months': months, 'charge_mode': charge_mode}
+    prices = await charging(ppid, usages)
+    if prices is None:
+        raise Exception('产线定价计算返回空（ppid=%s）' % ppid)
+    amount = float(sum(getattr(p, 'amount', 0) or 0 for p in prices))
+    if amount <= 0:
+        raise Exception('产线定价计算金额为0（ppid=%s, months=%s）——检查定价 YAML' % (ppid, months))
+    return amount, ppid, months, valid_days

pipeline_core/product_interface.py

@@ -0,0 +1,194 @@
+"""pipeline_core 产品接口 — 供 product_management 调用（照 account_resource 范式）。
+
+    resource_ref_id = pipelines.id（如 bidding_general）
+    定价维度 = charge_mode（month/year），映射表 pipeline_pricing_map
+
+产线是「订阅制使用费」产品：付月费/年费获得该产线的使用资格，
+模型/存储消耗另按量计费（各自独立产品），互不混淆。
+"""
+from ahserver.serverenv import ServerEnv
+
+from .pipeline_pricing import (
+    RESOURCE_TYPE,
+    calculate_pipeline_amount,
+    get_pipeline_ppid,
+    get_pipeline_res,
+    pipeline_pricing_map_list,
+    pipeline_res_list,
+)
+
+
+async def _resolve_pipeline(resource_ref_id):
+    p = await get_pipeline_res(resource_ref_id)
+    if not p:
+        return None, '产线 %s 不存在' % resource_ref_id
+    return p, None
+
+
+async def load_product_category_product(parent_category_id):
+    """返回产线类别 + 产线产品，供 product_management 导入。
+
+    类别一个（产线服务），产品 = pipelines 表全部 published 产线。
+    """
+    pls = await pipeline_res_list()
+    if not pls:
+        return {'success': False, 'error': '没有已发布(published)的产线'}
+
+    categories = [{
+        'source_id': 'pipelines',
+        'name': '产线服务',
+        'description': '产线使用费（订阅制，按月/按年）',
+        'product_type': RESOURCE_TYPE,
+        'product_type_title': '产线',
+        'sort_order': 0,
+    }]
+
+    products = []
+    for i, p in enumerate(pls):
+        pid = getattr(p, 'id', '')
+        name = getattr(p, 'name', '') or pid
+        desc = getattr(p, 'description', '') or ''
+        products.append({
+            'source_category_id': 'pipelines',
+            'resource_ref_id': pid,
+            'product_code': pid,
+            'product_name': name,
+            'product_type': RESOURCE_TYPE,
+            'brief_intro': (desc or '')[:200],
+            'sort_order': i * 10,
+        })
+
+    return {'success': True, 'categories': categories, 'products': products}
+
+
+async def get_product_display(resource_ref_id):
+    """产线定价展示（月费/年费两档，含权益有效期）。"""
+    p, err = await _resolve_pipeline(resource_ref_id)
+    if err:
+        return {'success': False, 'message': err}
+
+    env = ServerEnv()
+    texts = []
+    detail = {}
+    for cm, label in (('month', '包月'), ('year', '包年')):
+        try:
+            amount, ppid, months, valid_days = await calculate_pipeline_amount(
+                resource_ref_id, cm)
+            unit = '元/月' if cm == 'month' else '元/年'
+            texts.append('%s ¥%.2f %s' % (label, amount, unit))
+            detail[cm] = {'amount': round(amount, 2), 'ppid': ppid,
+                          'months': months, 'valid_days': valid_days}
+        except Exception as e:
+            texts.append('%s 定价未配置' % label)
+            detail[cm] = {'error': str(e)}
+
+    return {
+        'success': True,
+        'pricing_text': ' | '.join(texts),
+        'pricing_detail': detail,
+        'extra_info': {
+            'pipeline_id': resource_ref_id,
+            'pipeline_name': getattr(p, 'name', ''),
+            'pipeline_type': getattr(p, 'pipeline_type', ''),
+        },
+    }
+
+
+async def check_product_availability(resource_ref_id, user_org_id=None):
+    p, err = await _resolve_pipeline(resource_ref_id)
+    if err:
+        return {'available': False, 'reason': err}
+    if getattr(p, 'status', '') != 'published':
+        return {'available': False, 'reason': '产线未发布（status=%s）' % getattr(p, 'status', '')}
+    return {'available': True, 'reason': ''}
+
+
+async def check_product_consumable(resource_ref_id, user_id, user_org_id):
+    """购买前预检：产线可用 + 定价已配 + 客户余额。"""
+    avail = await check_product_availability(resource_ref_id, user_org_id)
+    if not avail.get('available'):
+        return {'consumable': False, 'reason': avail.get('reason', ''),
+                'min_balance': 0, 'pricing_available': False}
+
+    maps = await pipeline_pricing_map_list({'pipeline_id': resource_ref_id})
+    pricing_available = any(getattr(m, 'ppid', '') for m in maps)
+    if not pricing_available:
+        return {'consumable': False, 'reason': '产线未配置定价方案',
+                'min_balance': 0, 'pricing_available': False}
+
+    env = ServerEnv()
+    balance = None
+    if user_org_id and hasattr(env, 'getCustomerBalance'):
+        try:
+            from sqlor.dbpools import DBPools
+            acc_dbname = env.get_module_dbname('accounting')
+            async with DBPools().sqlorContext(acc_dbname) as sor:
+                balance = await env.getCustomerBalance(sor, user_org_id)
+        except Exception:
+            balance = None   # accounting 未接入 → 不拦
+    if balance is not None and float(balance) <= 0:
+        return {'consumable': False,
+                'reason': '客户资金账户余额不足（%.2f），请先充值' % float(balance),
+                'min_balance': 0, 'pricing_available': True}
+    return {'consumable': True, 'reason': '',
+            'min_balance': 0, 'pricing_available': True}
+
+
+async def execute_product_service(resource_ref_id, user_id, user_org_id, request_data):
+    """产线订阅开通。权益 = 该产线的使用资格，有效期按 charge_mode。"""
+    p, err = await _resolve_pipeline(resource_ref_id)
+    if err:
+        return {'success': False, 'message': err, 'status': 'FAILED'}
+
+    charge_mode = (request_data or {}).get('charge_mode', '') or 'month'
+    if charge_mode not in ('month', 'year'):
+        return {'success': False, 'message': '产线只支持 charge_mode=month/year',
+                'status': 'FAILED'}
+
+    try:
+        amount, ppid, months, valid_days = await calculate_pipeline_amount(
+            resource_ref_id, charge_mode)
+    except Exception as e:
+        return {'success': False, 'message': str(e), 'status': 'FAILED'}
+
+    return {'success': True,
+            'message': '产线订阅开通成功',
+            'status': 'SUCCESS',
+            'amount': round(amount, 4),
+            'entitlement': {
+                'charge_mode': charge_mode,
+                'duration_days': valid_days,
+                'pipeline_id': resource_ref_id,
+                'pipeline_name': getattr(p, 'name', ''),
+            }}
+
+
+async def execute_product_service_stream(resource_ref_id, user_id, user_org_id, request_data):
+    """产线订阅无流式场景，返回单条结果。"""
+    r = await execute_product_service(resource_ref_id, user_id, user_org_id, request_data)
+
+    async def _gen():
+        yield {'chunk': r, 'usage_data': None, 'done': True, 'error': None}
+    return _gen()
+
+
+async def calculate_product_cost(resource_ref_id, usage_data, user_org_id=None):
+    """计算产线使用费原价。usage_data 含 charge_mode（month/year）。
+
+    资源层只算真实定价（原价），客户折扣由产品层按 product_id 计算。
+    """
+    p, err = await _resolve_pipeline(resource_ref_id)
+    if err:
+        return {'success': False, 'message': err}
+
+    usage_data = usage_data or {}
+    charge_mode = usage_data.get('charge_mode', '') or 'month'
+    try:
+        amount, ppid, months, valid_days = await calculate_pipeline_amount(
+            resource_ref_id, charge_mode)
+    except Exception as e:
+        return {'success': False, 'message': str(e)}
+
+    return {'success': True, 'amount': round(amount, 6),
+            'original_amount': round(amount, 6),
+            'cost': 0.0, 'discount': 1.0, 'pricing_program_id': ppid}