285 lines
15 KiB
Python
285 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""定价重建落库脚本（2026-09-07，用户提供的阿里云百炼价目表）。
|
||
|
||
背景：今日 14:08:01 pricing_program/pricing_program_timing/product/product_category
|
||
四表被重建清空，23 个模型的 ppid 全部悬空。本脚本按价目表重建能对上的模型定价：
|
||
- qwen3.8-max 12/36/缓存1.5 元/百万tokens
|
||
- qwen3.8-flash 0.8/2.7/缓存0.1
|
||
- qwen3.7-flash 分三档（<=32k: 0.2/0.8/0.04; 32k~256k: 0.6/2.4/0.12; 256k~1m: 1.2/4.8/0.24）
|
||
- qwen3.7-text-embedding-flash 0.125 元/百万tokens（用户指令：模型先改名再按此定价）
|
||
- ali-deepseek-v4-pro 用 deepseek-v4-pro-0813 价（用户指令）：忙时(8-22点) 9/27/0.9，闲时(22-次日8点) 4.5/13.5/0.45
|
||
- wan3.0-video-prime 480P 0.45/720P 0.9/1080P 1.8 元/秒
|
||
- happyhorse-1.1-i2v/t2v/r2v 三模型定价相同（480P 0.45/720P 0.9/1080P 1.2）→ 共享一个 ppid
|
||
|
||
跳过（价目表对不上/用户明示后补）：ali-deepseek-v4-flash、qwen3-vl-embedding、
|
||
qwen3-vl-rerank、tongyi-embedding-vision-flash、doubao 两模型（豆包非百炼价目表）、
|
||
MiniMax 全系（minimax 价目表非本次提供，且已有 ppid 悬空——另行处理）。
|
||
|
||
幂等：pricing_program 按「名称+出处」查重复用；timing 每 ppid 只保留一条有效行
|
||
（内容相同跳过；不同且今天启用原地更新；历史行拉链）。
|
||
所有 YAML 已过引擎试算（/d/pipeline/tmp_pricing_dryrun*.py，金额逐项核对）。
|
||
|
||
用法：py3/bin/python tmp_rebuild_pricing.py [--dry]
|
||
"""
|
||
import asyncio
|
||
import json
|
||
import sys
|
||
import time
|
||
|
||
sys.path.insert(0, '/d/pipeline/pipeline-app')
|
||
|
||
import yaml # noqa: E402
|
||
from sqlor.dbpools import DBPools # noqa: E402
|
||
from appPublic.uniqueID import getID # noqa: E402
|
||
from appPublic.jsonConfig import getConfig # noqa: E402
|
||
|
||
DRY = '--dry' in sys.argv
|
||
DOC_QUOTE = '用户提供的阿里云百炼价目表（2026-09-07粘贴正文）'
|
||
|
||
# ── YAML 构造（与试算脚本完全一致）──
|
||
TOKEN_FIELDS = {
|
||
'price_factors': {'type': 'string', 'role': 'factor', 'label': '计价因子'},
|
||
'unit_prices': {'type': 'float', 'role': 'factor', 'label': '单位定价'},
|
||
'unit': {'type': 'string', 'role': 'factor', 'label': '计价单位'},
|
||
'uncache_tokens': {'type': 'int', 'role': 'factor', 'label': '非缓存输入Token',
|
||
'derived': 'prompt_tokens - prompt_tokens_details.cached_tokens'},
|
||
'cached_tokens': {'type': 'int', 'role': 'factor', 'label': '缓存Token',
|
||
'derived': 'prompt_tokens_details.cached_tokens'},
|
||
'completion_tokens': {'type': 'float', 'role': 'factor', 'label': '输出tokens'},
|
||
}
|
||
|
||
|
||
def token_yaml(u, c, ca):
|
||
return yaml.dump({'unit_values': {'百万': 1000000}, 'fields': TOKEN_FIELDS,
|
||
'pricings': [
|
||
{'price_factors': 'uncache_tokens', 'unit_prices': u, 'unit': '百万'},
|
||
{'price_factors': 'completion_tokens', 'unit_prices': c, 'unit': '百万'},
|
||
{'price_factors': 'cached_tokens', 'unit_prices': ca, 'unit': '百万'}]},
|
||
allow_unicode=True, sort_keys=False)
|
||
|
||
|
||
def tiered_token_yaml(tiers):
|
||
fields = dict(TOKEN_FIELDS)
|
||
fields['prompt_tokens'] = {'type': 'int', 'role': 'filter', 'label': 'prompt_tokens',
|
||
'value_mode': 'between'}
|
||
pricings = []
|
||
for fi, factor in enumerate(('uncache_tokens', 'completion_tokens', 'cached_tokens')):
|
||
filters = []
|
||
for i, (lo, hi, u, c, ca) in enumerate(tiers):
|
||
op = ('0 =~= %d' % hi) if lo == 0 else ('%d ~= %d' % (lo, hi))
|
||
filters.append({'prompt_tokens': op, 'value_mode': 'between',
|
||
'unit_prices': (u, c, ca)[fi]})
|
||
pricings.append({'price_factors': factor, 'unit_prices': tiers[-1][2 + fi],
|
||
'unit': '百万', 'filters': filters})
|
||
return yaml.dump({'unit_values': {'百万': 1000000}, 'fields': fields, 'pricings': pricings},
|
||
allow_unicode=True, sort_keys=False)
|
||
|
||
|
||
def hour_tiered_token_yaml():
|
||
"""deepseek-v4-pro-0813 忙闲时：忙时[8,22) 9/27/0.9；闲时[22,24)+[0,8) 4.5/13.5/0.45。
|
||
hour 维度由出账从 llm_usage.created_at 注入（pipeline-llm accounting.py 已改）。"""
|
||
fields = dict(TOKEN_FIELDS)
|
||
fields['hour'] = {'type': 'int', 'role': 'filter', 'label': '调用小时', 'value_mode': 'between'}
|
||
tiers = [('8 =~ 22', 9, 27, 0.9), ('22 =~ 24', 4.5, 13.5, 0.45), ('0 =~ 8', 4.5, 13.5, 0.45)]
|
||
pricings = []
|
||
for fi, factor in enumerate(('uncache_tokens', 'completion_tokens', 'cached_tokens')):
|
||
filters = [{'hour': rng, 'value_mode': 'between', 'unit_prices': (u, c, ca)[fi]}
|
||
for rng, u, c, ca in tiers]
|
||
pricings.append({'price_factors': factor, 'unit_prices': (9, 27, 0.9)[fi],
|
||
'unit': '百万', 'filters': filters})
|
||
return yaml.dump({'unit_values': {'百万': 1000000}, 'fields': fields, 'pricings': pricings},
|
||
allow_unicode=True, sort_keys=False)
|
||
|
||
|
||
def video_yaml(prices):
|
||
fields = {
|
||
'price_factors': {'type': 'string', 'role': 'factor', 'label': '计价因子'},
|
||
'unit_prices': {'type': 'float', 'role': 'factor', 'label': '单位定价'},
|
||
'unit': {'type': 'string', 'role': 'factor', 'label': '计价单位'},
|
||
'duration': {'type': 'float', 'role': 'factor', 'label': '时长'},
|
||
'SR': {'type': 'int', 'role': 'filter', 'label': '分辨率'},
|
||
}
|
||
pricings = [{'price_factors': 'duration', 'unit_prices': p, 'unit': '秒', 'SR': sr}
|
||
for sr, p in prices]
|
||
return yaml.dump({'unit_values': {'秒': 1}, 'fields': fields, 'pricings': pricings},
|
||
allow_unicode=True, sort_keys=False)
|
||
|
||
|
||
def embedding_yaml(price):
|
||
fields = {
|
||
'price_factors': {'type': 'string', 'role': 'factor', 'label': '计价因子'},
|
||
'unit_prices': {'type': 'float', 'role': 'factor', 'label': '单位定价'},
|
||
'unit': {'type': 'string', 'role': 'factor', 'label': '计价单位'},
|
||
'prompt_tokens': {'type': 'float', 'role': 'factor', 'label': '输入tokens'},
|
||
}
|
||
return yaml.dump({'unit_values': {'百万': 1000000}, 'fields': fields,
|
||
'pricings': [{'price_factors': 'prompt_tokens',
|
||
'unit_prices': price, 'unit': '百万'}]},
|
||
allow_unicode=True, sort_keys=False)
|
||
|
||
|
||
# 方案定义：(方案名, YAML, 文档原文quote, [挂载的模型名...])
|
||
PLANS = [
|
||
('qwen3.8-max 定价', token_yaml(12, 36, 1.5),
|
||
'qwen3.8-max 输入¥12/百万tokens 输出¥36 缓存命中¥1.5', ['qwen3.8-max']),
|
||
('qwen3.8-flash 定价', token_yaml(0.8, 2.7, 0.1),
|
||
'qwen3.8-flash 输入¥0.8/百万tokens 输出¥2.7 缓存命中¥0.1', ['qwen3.8-flash']),
|
||
('qwen3.7-flash 定价', tiered_token_yaml([
|
||
(0, 32000, 0.2, 0.8, 0.04),
|
||
(32000, 256000, 0.6, 2.4, 0.12),
|
||
(256000, 1000000, 1.2, 4.8, 0.24)]),
|
||
'qwen3.7-flash 输入<=32k:0.2/0.8/0.04; 32k<输入<=256k:0.6/2.4/0.12; 256k<输入<=1m:1.2/4.8/0.24 元/百万tokens',
|
||
['qwen3.7-flash']),
|
||
('qwen3.7-text-embedding-flash 定价', embedding_yaml(0.125),
|
||
'qwen3.7-text-embedding-flash 文本输入¥0.125/百万tokens', ['qwen3.7-text-embedding-flash']),
|
||
('ali-deepseek-v4-pro 定价(忙闲时)', hour_tiered_token_yaml(),
|
||
'deepseek-v4-pro-0813 忙时(8点-22点)输入¥9/输出¥27/缓存¥0.9; 闲时(22点-次日8点)¥4.5/¥13.5/¥0.45 元/百万tokens',
|
||
['ali-deepseek-v4-pro']),
|
||
('wan3.0-video-prime 定价', video_yaml([(480, 0.45), (720, 0.9), (1080, 1.8)]),
|
||
'wan3.0-video-prime 480P¥0.45/秒 720P¥0.9/秒 1080P¥1.8/秒', ['wan3.0-video-prime']),
|
||
('happyhorse-1.1 视频系列定价(共享)', video_yaml([(480, 0.45), (720, 0.9), (1080, 1.2)]),
|
||
'happyhorse-1.1-i2v/t2v/r2v 480P¥0.45/秒 720P¥0.9/秒 1080P¥1.2/秒',
|
||
['happyhorse-1.1-i2v', 'happyhorse-1.1-t2v', 'happyhorse-1.1-r2v']),
|
||
]
|
||
|
||
RENAME = ('qwen3.7-text-embedding', 'qwen3.7-text-embedding-flash')
|
||
|
||
|
||
async def main():
|
||
cfg = getConfig()
|
||
DBPools(cfg.databases)
|
||
db = DBPools()
|
||
log = []
|
||
async with db.sqlorContext('pipeline') as sor:
|
||
# 0. 改名（用户指令）：name 与 vendor_model_id 一起改，保持调用名与注册名一致
|
||
old, new = RENAME
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, name, vendor_model_id FROM llm_model WHERE name=${n}$ OR vendor_model_id=${n}$",
|
||
{"n": old})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if recs:
|
||
mid = recs[0].id
|
||
log.append('改名: %s → %s (model_id=%s, 原vmid=%s)' % (
|
||
old, new, mid, getattr(recs[0], 'vendor_model_id', '')))
|
||
if not DRY:
|
||
await sor.sqlExe(
|
||
"UPDATE llm_model SET name=${new}$, vendor_model_id=${new}$, updated_at=NOW() "
|
||
"WHERE id=${i}$", {"new": new, "i": mid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
else:
|
||
# 已改过？
|
||
recs2 = await sor.sqlExe("SELECT id FROM llm_model WHERE name=${n}$", {"n": new})
|
||
await sor.sqlExe("COMMIT", {})
|
||
log.append('改名: 未找到 %s（%s 存在=%s，视为已改）' % (old, new, bool(recs2)))
|
||
|
||
now = time.strftime('%Y-%m-%d')
|
||
for pname, yml, quote, model_names in PLANS:
|
||
desc = '定价出处：%s | 文档原文：%s' % (DOC_QUOTE, quote)
|
||
# 幂等：同名方案已存在则复用
|
||
recs = await sor.sqlExe(
|
||
"SELECT id FROM pricing_program WHERE name=${n}$ LIMIT 1", {"n": pname})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if recs:
|
||
ppid = recs[0].id
|
||
action = '复用方案'
|
||
# timing 幂等：只保留一条有效行
|
||
recs2 = await sor.sqlExe(
|
||
"SELECT id, pricing_data, enabled_date FROM pricing_program_timing "
|
||
"WHERE ppid=${p}$ AND expired_date='9999-12-31' ORDER BY enabled_date DESC LIMIT 1",
|
||
{"p": ppid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
cur = recs2[0] if recs2 else None
|
||
same = False
|
||
if cur is not None:
|
||
try:
|
||
same = (yaml.safe_load(cur.pricing_data or '{}') == yaml.safe_load(yml))
|
||
except Exception:
|
||
same = (cur.pricing_data or '') == yml
|
||
if same:
|
||
action = '复用方案+内容无变化（幂等跳过）'
|
||
elif cur is not None and str(cur.enabled_date)[:10] == now:
|
||
if not DRY:
|
||
await sor.sqlExe(
|
||
"UPDATE pricing_program_timing SET pricing_data=${y}$ WHERE id=${i}$",
|
||
{"y": yml, "i": cur.id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
action = '复用方案+原地更新（同日）'
|
||
elif cur is not None:
|
||
if not DRY:
|
||
await sor.sqlExe(
|
||
"UPDATE pricing_program_timing SET expired_date=${d}$ "
|
||
"WHERE ppid=${p}$ AND expired_date='9999-12-31'",
|
||
{"d": now, "p": ppid})
|
||
await sor.C('pricing_program_timing', {
|
||
'id': getID(), 'ppid': ppid, 'name': pname,
|
||
'pricing_data': yml, 'enabled_date': now,
|
||
'expired_date': '9999-12-31'})
|
||
await sor.sqlExe("COMMIT", {})
|
||
action = '复用方案+拉链更新'
|
||
else:
|
||
if not DRY:
|
||
await sor.C('pricing_program_timing', {
|
||
'id': getID(), 'ppid': ppid, 'name': pname,
|
||
'pricing_data': yml, 'enabled_date': now,
|
||
'expired_date': '9999-12-31'})
|
||
await sor.sqlExe("COMMIT", {})
|
||
action = '复用方案+新建时序'
|
||
if not DRY:
|
||
await sor.sqlExe("UPDATE pricing_program SET description=${d}$ WHERE id=${i}$",
|
||
{"d": desc[:1000], "i": ppid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
else:
|
||
ppid = getID()
|
||
if not DRY:
|
||
await sor.C('pricing_program', {
|
||
'id': ppid, 'name': pname, 'ownerid': '0', 'providerid': '',
|
||
'pricing_belong': '', 'description': desc[:1000], 'currency': 'CNY'})
|
||
await sor.C('pricing_program_timing', {
|
||
'id': getID(), 'ppid': ppid, 'name': pname,
|
||
'pricing_data': yml, 'enabled_date': now,
|
||
'expired_date': '9999-12-31'})
|
||
await sor.sqlExe("COMMIT", {})
|
||
action = '新建方案+时序'
|
||
# 挂模型
|
||
for mn in model_names:
|
||
recs3 = await sor.sqlExe(
|
||
"SELECT id, COALESCE(ppid,'') AS ppid FROM llm_model WHERE name=${n}$ AND status='active'",
|
||
{"n": mn})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs3:
|
||
log.append(' ⚠ 模型 %s 未找到（active）——跳过挂载' % mn)
|
||
continue
|
||
mid = recs3[0].id
|
||
old_ppid = recs3[0].ppid or ''
|
||
if old_ppid == ppid:
|
||
log.append(' %s ppid 已正确（%s）' % (mn, ppid))
|
||
continue
|
||
# 旧 ppid 悬空校验
|
||
dangling = ''
|
||
if old_ppid:
|
||
recs4 = await sor.sqlExe(
|
||
"SELECT id FROM pricing_program WHERE id=${p}$", {"p": old_ppid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
dangling = '' if recs4 else '（旧ppid悬空已清理）'
|
||
if not DRY:
|
||
await sor.sqlExe("UPDATE llm_model SET ppid=${p}$, updated_at=NOW() WHERE id=${i}$",
|
||
{"p": ppid, "i": mid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
log.append(' %s: ppid %s→%s %s [%s]' % (mn, old_ppid or '空', ppid, dangling, action))
|
||
|
||
# 清理：悬空 ppid 置空（模型指向已删除方案的）
|
||
recs5 = await sor.sqlExe(
|
||
"SELECT m.id, m.name, m.ppid FROM llm_model m WHERE m.ppid<>'' AND m.ppid IS NOT NULL "
|
||
"AND NOT EXISTS (SELECT 1 FROM pricing_program p WHERE p.id=m.ppid)", {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
for r in (recs5 or []):
|
||
log.append('悬空清理: %s ppid=%s → 置空' % (r.name, r.ppid))
|
||
if not DRY:
|
||
await sor.sqlExe("UPDATE llm_model SET ppid='' WHERE id=${i}$", {"i": r.id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
|
||
print(json.dumps({'dry': DRY, 'log': log}, ensure_ascii=False, indent=1))
|
||
|
||
|
||
asyncio.run(main())
|