pipeline_core/pipeline_core/pipeline_pricing.py

"""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