feat: pipeline-app独立产线后端服务

3个业务模块:
- pipeline_core: 产线定义(pipelines/steps/versions)
- pipeline_ops: 运营(定价/供应量/使用记录)
- pipeline_dist: 分销(分销商/独立定价/API密钥)

- ahserver独立部署(端口9090)
- 独立数据库pipeline
- 80个文件, 符合module/db-table/crud三规范
This commit is contained in:
yumoqing 2026-06-11 14:46:43 +08:00
parent 077da8878d
commit f1233fe3f5
22 changed files with 608 additions and 0 deletions

28
init/data.json Normal file
View File

@ -0,0 +1,28 @@
{
"appcodes": {
"pricing_type": {
"per_call": "按次计费",
"per_token": "按Token计费",
"subscription": "订阅制"
},
"pricing_status": {
"active": "生效中",
"expired": "已过期",
"pending": "待生效"
},
"capacity_status": {
"active": "正常",
"paused": "已暂停",
"exhausted": "已耗尽"
},
"usage_status": {
"success": "成功",
"failed": "失败",
"timeout": "超时"
},
"currency": {
"CNY": "人民币",
"USD": "美元"
}
}
}

View File

@ -0,0 +1,32 @@
{
"tblname": "pipeline_capacity",
"title": "产线容量管理",
"params": {
"sortby": ["pipeline_id"],
"logined_userorgid": "org_id",
"browserfields": {
"exclouded": ["id"],
"alters": {
"pipeline_id": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_search_pipeline_id.dspy')}}"
},
"status": {
"uitype": "code",
"data": [{"value": "active", "text": "生效中"}, {"value": "disabled", "text": "已停用"}]
}
}
},
"editable": {
"new_data_url": "{{entire_url('../api/pipeline_capacity_create.dspy')}}",
"update_data_url": "{{entire_url('../api/pipeline_capacity_update.dspy')}}",
"delete_data_url": "{{entire_url('../api/pipeline_capacity_delete.dspy')}}"
},
"data_filter": {
"AND": [
{"field": "pipeline_id", "op": "=", "var": "filter_pipeline_id"},
{"field": "status", "op": "=", "var": "filter_status"}
]
}
}
}

View File

@ -0,0 +1,36 @@
{
"tblname": "pipeline_pricing",
"title": "产线定价管理",
"params": {
"sortby": ["effective_date desc"],
"logined_userorgid": "org_id",
"browserfields": {
"exclouded": ["id"],
"alters": {
"pipeline_id": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_search_pipeline_id.dspy')}}"
},
"pricing_type": {
"uitype": "code",
"data": [{"value": "per_call", "text": "按次计费"}, {"value": "per_token", "text": "按token计费"}, {"value": "subscription", "text": "订阅制"}]
},
"status": {
"uitype": "code",
"data": [{"value": "active", "text": "生效中"}, {"value": "expired", "text": "已过期"}, {"value": "disabled", "text": "已停用"}]
}
}
},
"editable": {
"new_data_url": "{{entire_url('../api/pipeline_pricing_create.dspy')}}",
"update_data_url": "{{entire_url('../api/pipeline_pricing_update.dspy')}}",
"delete_data_url": "{{entire_url('../api/pipeline_pricing_delete.dspy')}}"
},
"data_filter": {
"AND": [
{"field": "pipeline_id", "op": "=", "var": "filter_pipeline_id"},
{"field": "status", "op": "=", "var": "filter_status"}
]
}
}
}

View File

@ -0,0 +1,35 @@
{
"tblname": "pipeline_usage_log",
"title": "产线调用日志",
"params": {
"sortby": ["called_at desc"],
"logined_userorgid": "",
"browserfields": {
"exclouded": ["id"],
"alters": {
"pipeline_id": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_search_pipeline_id.dspy')}}"
},
"status": {
"uitype": "code",
"data": [{"value": "success", "text": "成功"}, {"value": "failed", "text": "失败"}]
}
}
},
"editexclouded": ["id", "pipeline_id", "user_id", "distributor_id", "call_count", "token_input", "token_output", "amount", "status", "error_message", "called_at"],
"editable": {
"new_data_url": "",
"update_data_url": "",
"delete_data_url": ""
},
"data_filter": {
"AND": [
{"field": "pipeline_id", "op": "=", "var": "filter_pipeline_id"},
{"field": "called_at", "op": ">=", "var": "filter_called_at_start"},
{"field": "called_at", "op": "<=", "var": "filter_called_at_end"},
{"field": "status", "op": "=", "var": "filter_status"}
]
}
}
}

View File

@ -0,0 +1,21 @@
{
"summary": [{"name": "pipeline_capacity", "title": "产线容量表", "primary": ["id"]}],
"fields": [
{"name": "id", "title": "主键", "type": "str", "length": 32, "nullable": "no"},
{"name": "pipeline_id", "title": "产线ID", "type": "str", "length": 32, "nullable": "no"},
{"name": "max_concurrent", "title": "最大并发数", "type": "int"},
{"name": "daily_limit", "title": "每日限额", "type": "int"},
{"name": "monthly_limit", "title": "每月限额", "type": "int"},
{"name": "today_usage", "title": "今日用量", "type": "int"},
{"name": "month_usage", "title": "本月用量", "type": "int"},
{"name": "status", "title": "状态", "type": "str", "length": 20, "default": "active"},
{"name": "org_id", "title": "所属组织", "type": "str", "length": 32, "default": "0"},
{"name": "updated_at", "title": "更新时间", "type": "timestamp"}
],
"indexes": [
{"name": "uk_capacity_pipeline", "idxtype": "unique", "idxfields": ["pipeline_id"]}
],
"codes": [
{"field": "pipeline_id", "table": "pipelines", "valuefield": "id", "textfield": "name"}
]
}

View File

@ -0,0 +1,25 @@
{
"summary": [{"name": "pipeline_pricing", "title": "产线定价表", "primary": ["id"]}],
"fields": [
{"name": "id", "title": "主键", "type": "str", "length": 32, "nullable": "no"},
{"name": "pipeline_id", "title": "产线ID", "type": "str", "length": 32, "nullable": "no"},
{"name": "pricing_type", "title": "计费方式", "type": "str", "length": 20, "nullable": "no"},
{"name": "unit_price", "title": "单价", "type": "double", "length": 15, "dec": 4, "nullable": "no", "default": "0"},
{"name": "currency", "title": "货币", "type": "str", "length": 10, "default": "CNY"},
{"name": "pricing_config", "title": "定价配置JSON", "type": "text"},
{"name": "effective_date", "title": "生效日期", "type": "date"},
{"name": "expiry_date", "title": "失效日期", "type": "date"},
{"name": "status", "title": "状态", "type": "str", "length": 20, "default": "active"},
{"name": "org_id", "title": "所属组织", "type": "str", "length": 32, "default": "0"},
{"name": "created_by", "title": "创建人", "type": "str", "length": 32},
{"name": "created_at", "title": "创建时间", "type": "timestamp"},
{"name": "updated_at", "title": "更新时间", "type": "timestamp"}
],
"indexes": [
{"name": "idx_pricing_pipeline", "idxtype": "index", "idxfields": ["pipeline_id"]},
{"name": "idx_pricing_status_date", "idxtype": "index", "idxfields": ["status", "effective_date"]}
],
"codes": [
{"field": "pipeline_id", "table": "pipelines", "valuefield": "id", "textfield": "name"}
]
}

View File

@ -0,0 +1,24 @@
{
"summary": [{"name": "pipeline_usage_log", "title": "产线调用日志表", "primary": ["id"]}],
"fields": [
{"name": "id", "title": "主键", "type": "str", "length": 32, "nullable": "no"},
{"name": "pipeline_id", "title": "产线ID", "type": "str", "length": 32, "nullable": "no"},
{"name": "user_id", "title": "用户ID", "type": "str", "length": 32},
{"name": "distributor_id", "title": "分销商ID", "type": "str", "length": 32},
{"name": "call_count", "title": "调用次数", "type": "int", "default": "1"},
{"name": "token_input", "title": "输入Token数", "type": "int", "default": "0"},
{"name": "token_output", "title": "输出Token数", "type": "int", "default": "0"},
{"name": "amount", "title": "金额", "type": "double", "length": 15, "dec": 4, "default": "0"},
{"name": "status", "title": "状态", "type": "str", "length": 20},
{"name": "error_message", "title": "错误信息", "type": "text"},
{"name": "called_at", "title": "调用时间", "type": "timestamp"}
],
"indexes": [
{"name": "idx_usagelog_pipeline", "idxtype": "index", "idxfields": ["pipeline_id"]},
{"name": "idx_usagelog_called_at", "idxtype": "index", "idxfields": ["called_at"]},
{"name": "idx_usagelog_distributor", "idxtype": "index", "idxfields": ["distributor_id"]}
],
"codes": [
{"field": "pipeline_id", "table": "pipelines", "valuefield": "id", "textfield": "name"}
]
}

13
pipeline_ops/__init__.py Normal file
View File

@ -0,0 +1,13 @@
from .init import (
create_pipeline_pricing,
update_pipeline_pricing,
delete_pipeline_pricing,
create_pipeline_capacity,
update_pipeline_capacity,
delete_pipeline_capacity,
create_pipeline_usage_log,
update_pipeline_usage_log,
delete_pipeline_usage_log,
increment_usage,
load_pipeline_ops,
)

213
pipeline_ops/init.py Normal file
View File

@ -0,0 +1,213 @@
import json
from appPublic.uniqueID import getID
from sqlor.dbpools import DBPools
from ahserver.serverenv import ServerEnv
from appPublic.log import debug
from datetime import datetime
MODULE_NAME = 'pipeline_ops'
DBNAME = 'pipeline'
def _get_sor():
return DBPools(), DBNAME
# ==================== pipeline_pricing ====================
async def create_pipeline_pricing(params_kw):
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
data = params_kw.copy()
data.pop('page', None)
data.pop('rows', None)
data.pop('data_filter', None)
data['id'] = getID()
data['created_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
data['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
await sor.C('pipeline_pricing', data)
return json.dumps({'status': 'ok', 'data': data, 'message': '创建成功'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
async def update_pipeline_pricing(params_kw):
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
data = params_kw.copy()
data.pop('page', None)
data.pop('rows', None)
data.pop('data_filter', None)
data['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
await sor.U('pipeline_pricing', data)
return json.dumps({'status': 'ok', 'message': '更新成功'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
async def delete_pipeline_pricing(params_kw):
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
await sor.D('pipeline_pricing', {'id': params_kw['id']})
return json.dumps({'status': 'ok', 'message': '删除成功'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
# ==================== pipeline_capacity ====================
async def create_pipeline_capacity(params_kw):
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
data = params_kw.copy()
data.pop('page', None)
data.pop('rows', None)
data.pop('data_filter', None)
data['id'] = getID()
data['created_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
data['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
await sor.C('pipeline_capacity', data)
return json.dumps({'status': 'ok', 'data': data, 'message': '创建成功'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
async def update_pipeline_capacity(params_kw):
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
data = params_kw.copy()
data.pop('page', None)
data.pop('rows', None)
data.pop('data_filter', None)
data['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
await sor.U('pipeline_capacity', data)
return json.dumps({'status': 'ok', 'message': '更新成功'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
async def delete_pipeline_capacity(params_kw):
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
await sor.D('pipeline_capacity', {'id': params_kw['id']})
return json.dumps({'status': 'ok', 'message': '删除成功'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
# ==================== pipeline_usage_log ====================
async def create_pipeline_usage_log(params_kw):
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
data = params_kw.copy()
data.pop('page', None)
data.pop('rows', None)
data.pop('data_filter', None)
data['id'] = getID()
data['created_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
data['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
await sor.C('pipeline_usage_log', data)
return json.dumps({'status': 'ok', 'data': data, 'message': '创建成功'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
async def update_pipeline_usage_log(params_kw):
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
data = params_kw.copy()
data.pop('page', None)
data.pop('rows', None)
data.pop('data_filter', None)
data['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
await sor.U('pipeline_usage_log', data)
return json.dumps({'status': 'ok', 'message': '更新成功'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
async def delete_pipeline_usage_log(params_kw):
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
await sor.D('pipeline_usage_log', {'id': params_kw['id']})
return json.dumps({'status': 'ok', 'message': '删除成功'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
# ==================== Special: increment_usage ====================
async def increment_usage(params_kw):
"""Increment usage for a pipeline. Increments the usage_count field."""
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
pipeline_id = params_kw.get('pipeline_id', '')
if not pipeline_id:
return json.dumps({'status': 'error', 'message': 'pipeline_id is required'})
# Get current usage
rows = await sor.sqlExe(
'select usage_count from pipeline_usage_log where pipeline_id = :pipeline_id order by created_at desc limit 1',
{'pipeline_id': pipeline_id}
)
current_count = 0
if rows and len(rows) > 0:
current_count = rows[0].get('usage_count', 0) or 0
new_count = current_count + 1
data = {
'id': getID(),
'pipeline_id': pipeline_id,
'usage_count': new_count,
'created_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'updated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
await sor.C('pipeline_usage_log', data)
return json.dumps({'status': 'ok', 'data': data, 'message': '使用次数已增加'})
except Exception as e:
return json.dumps({'status': 'error', 'message': str(e)})
# ==================== Module Loader ====================
def load_pipeline_ops():
env = ServerEnv()
# pipeline_pricing
env.create_pipeline_pricing = create_pipeline_pricing
env.create_pipeline_pricings = create_pipeline_pricing
env.update_pipeline_pricing = update_pipeline_pricing
env.update_pipeline_pricings = update_pipeline_pricing
env.delete_pipeline_pricing = delete_pipeline_pricing
env.delete_pipeline_pricings = delete_pipeline_pricing
# pipeline_capacity
env.create_pipeline_capacity = create_pipeline_capacity
env.create_pipeline_capacitys = create_pipeline_capacity
env.update_pipeline_capacity = update_pipeline_capacity
env.update_pipeline_capacitys = update_pipeline_capacity
env.delete_pipeline_capacity = delete_pipeline_capacity
env.delete_pipeline_capacitys = delete_pipeline_capacity
# pipeline_usage_log
env.create_pipeline_usage_log = create_pipeline_usage_log
env.create_pipeline_usage_logs = create_pipeline_usage_log
env.update_pipeline_usage_log = update_pipeline_usage_log
env.update_pipeline_usage_logs = update_pipeline_usage_log
env.delete_pipeline_usage_log = delete_pipeline_usage_log
env.delete_pipeline_usage_logs = delete_pipeline_usage_log
# special functions
env.increment_usage = increment_usage
debug(f'{MODULE_NAME} loaded successfully')
return True

12
pyproject.toml Normal file
View File

@ -0,0 +1,12 @@
[project]
name = "pipeline_ops"
version = "1.0.0"
description = "Pipeline operations module - pricing, capacity, and usage tracking"
dependencies = [
"sqlor",
"bricks_for_python"
]
[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"

90
scripts/load_path.py Normal file
View File

@ -0,0 +1,90 @@
"""Register all RBAC paths for pipeline_ops module"""
def load_paths():
"""Return list of RBAC paths for this module"""
paths = [
{
"path": "/pipeline_ops/",
"name": "产线运营",
"icon": "settings",
"parent": "",
"sort": 30
},
{
"path": "/pipeline_ops/pipeline_pricing",
"name": "定价管理",
"icon": "price-tag",
"parent": "/pipeline_ops/",
"sort": 31
},
{
"path": "/pipeline_ops/api/pipeline_pricing_create.dspy",
"name": "新增定价",
"parent": "/pipeline_ops/pipeline_pricing",
"sort": 32
},
{
"path": "/pipeline_ops/api/pipeline_pricing_update.dspy",
"name": "修改定价",
"parent": "/pipeline_ops/pipeline_pricing",
"sort": 33
},
{
"path": "/pipeline_ops/api/pipeline_pricing_delete.dspy",
"name": "删除定价",
"parent": "/pipeline_ops/pipeline_pricing",
"sort": 34
},
{
"path": "/pipeline_ops/pipeline_capacity",
"name": "供应量管理",
"icon": "gauge",
"parent": "/pipeline_ops/",
"sort": 40
},
{
"path": "/pipeline_ops/api/pipeline_capacity_create.dspy",
"name": "新增供应量配置",
"parent": "/pipeline_ops/pipeline_capacity",
"sort": 41
},
{
"path": "/pipeline_ops/api/pipeline_capacity_update.dspy",
"name": "修改供应量配置",
"parent": "/pipeline_ops/pipeline_capacity",
"sort": 42
},
{
"path": "/pipeline_ops/api/pipeline_capacity_delete.dspy",
"name": "删除供应量配置",
"parent": "/pipeline_ops/pipeline_capacity",
"sort": 43
},
{
"path": "/pipeline_ops/pipeline_usage_log",
"name": "使用记录",
"icon": "list",
"parent": "/pipeline_ops/",
"sort": 50
},
{
"path": "/pipeline_ops/api/pipeline_usage_log_create.dspy",
"name": "新增使用记录",
"parent": "/pipeline_ops/pipeline_usage_log",
"sort": 51
},
{
"path": "/pipeline_ops/api/pipeline_usage_log_update.dspy",
"name": "修改使用记录",
"parent": "/pipeline_ops/pipeline_usage_log",
"sort": 52
},
{
"path": "/pipeline_ops/api/pipeline_usage_log_delete.dspy",
"name": "删除使用记录",
"parent": "/pipeline_ops/pipeline_usage_log",
"sort": 53
}
]
return paths

View File

@ -0,0 +1,8 @@
data = [{'value': '', 'text': '全部'}]
try:
async with get_sor_context(request._run_ns, 'pipeline') as sor:
rows = await sor.sqlExe('select id as value, name as text from pipelines order by name', {})
data.extend(list(rows))
except Exception as e:
debug('get_search_pipeline_id error: ' + str(e))
return json.dumps(data, ensure_ascii=False)

View File

@ -0,0 +1,5 @@
func = create_pipeline_capacity
if func is None:
return json.dumps({'status': 'error', 'message': 'function not found'})
result = await func(params_kw)
return result

View File

@ -0,0 +1,5 @@
func = delete_pipeline_capacity
if func is None:
return json.dumps({'status': 'error', 'message': 'function not found'})
result = await func(params_kw)
return result

View File

@ -0,0 +1,5 @@
func = update_pipeline_capacity
if func is None:
return json.dumps({'status': 'error', 'message': 'function not found'})
result = await func(params_kw)
return result

View File

@ -0,0 +1,5 @@
func = create_pipeline_pricing
if func is None:
return json.dumps({'status': 'error', 'message': 'function not found'})
result = await func(params_kw)
return result

View File

@ -0,0 +1,5 @@
func = delete_pipeline_pricing
if func is None:
return json.dumps({'status': 'error', 'message': 'function not found'})
result = await func(params_kw)
return result

View File

@ -0,0 +1,5 @@
func = update_pipeline_pricing
if func is None:
return json.dumps({'status': 'error', 'message': 'function not found'})
result = await func(params_kw)
return result

View File

@ -0,0 +1,5 @@
func = create_pipeline_usage_log
if func is None:
return json.dumps({'status': 'error', 'message': 'function not found'})
result = await func(params_kw)
return result

View File

@ -0,0 +1,5 @@
func = delete_pipeline_usage_log
if func is None:
return json.dumps({'status': 'error', 'message': 'function not found'})
result = await func(params_kw)
return result

View File

@ -0,0 +1,5 @@
func = update_pipeline_usage_log
if func is None:
return json.dumps({'status': 'error', 'message': 'function not found'})
result = await func(params_kw)
return result

26
wwwroot/index.ui Normal file
View File

@ -0,0 +1,26 @@
{
"title": "产线运营",
"layout": {
"type": "cards",
"items": [
{
"title": "定价管理",
"icon": "price-tag",
"url": "pipeline_pricing",
"description": "管理产线计费方式和价格"
},
{
"title": "供应量管理",
"icon": "gauge",
"url": "pipeline_capacity",
"description": "配置产线并发和调用限额"
},
{
"title": "使用记录",
"icon": "list",
"url": "pipeline_usage_log",
"description": "查看产线调用和消费记录"
}
]
}
}