From e134cde44e73c34eea1770f34c973f992a20f065 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sun, 6 Sep 2026 16:55:30 +0800 Subject: [PATCH] =?UTF-8?q?feat(pipeline):=20=E4=BA=A7=E7=BA=BF=E4=BA=A7?= =?UTF-8?q?=E5=93=81=E5=8C=96=E6=8E=A5=E5=85=A5=20product=5Fmanagement?= =?UTF-8?q?=E2=80=94=E2=80=94=E8=AE=A1=E8=B4=B9=E5=B1=82+=E4=BA=A7?= =?UTF-8?q?=E5=93=81=E6=8E=A5=E5=8F=A3+=E5=AE=9A=E4=BB=B7=E6=98=A0?= =?UTF-8?q?=E5=B0=84=E8=A1=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 产线使用费（订阅制）照 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 | 28 +++++ pipeline_core/init.py | 27 ++++ pipeline_core/pipeline_pricing.py | 169 +++++++++++++++++++++++++ pipeline_core/product_interface.py | 194 +++++++++++++++++++++++++++++ 4 files changed, 418 insertions(+) create mode 100644 models/pipeline_pricing_map.json create mode 100644 pipeline_core/pipeline_pricing.py create mode 100644 pipeline_core/product_interface.py diff --git a/models/pipeline_pricing_map.json b/models/pipeline_pricing_map.json new file mode 100644 index 0000000..8ae06cd --- /dev/null +++ b/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"]} + ] +} diff --git a/pipeline_core/init.py b/pipeline_core/init.py index 2a05082..ee2d421 100644 --- a/pipeline_core/init.py +++ b/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 diff --git a/pipeline_core/pipeline_pricing.py b/pipeline_core/pipeline_pricing.py new file mode 100644 index 0000000..90a169e --- /dev/null +++ b/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 diff --git a/pipeline_core/product_interface.py b/pipeline_core/product_interface.py new file mode 100644 index 0000000..4c45e00 --- /dev/null +++ b/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}