87 lines
3.4 KiB
Python
87 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
||
"""独立异步记账进程 — 产线模型用量出账（三方账）。
|
||
|
||
职责分工（2026-09 定夺）：
|
||
- web 服务器：同步记账（账号/存储实时购买当场落账）
|
||
- 本进程：异步记账（模型调用时只记成本侧 + pending 流水，
|
||
本进程扫 llm_usage 出三方账：客户付/商户营收/供应商成本）
|
||
|
||
只加载记账链路所需模块（无 HTTP、无 RBAC 路由）。
|
||
用法：
|
||
py3/bin/python app/pipeline_accounting_worker.py -w <workdir>
|
||
部署：由 start.sh 以 nohup 启动，pid 文件 pipeline-accounting.pid。
|
||
"""
|
||
import os, sys, asyncio, argparse, logging
|
||
|
||
logger = logging.getLogger("pipeline.accounting_worker")
|
||
|
||
|
||
def init_worker():
|
||
"""初始化：配置 → DB → ServerEnv → 记账链路模块（与主应用同一套加载，裁剪到必需）。"""
|
||
app_dir = os.path.dirname(os.path.abspath(__file__))
|
||
root_dir = os.path.dirname(app_dir)
|
||
sys.path.insert(0, root_dir)
|
||
sys.path.insert(0, app_dir)
|
||
|
||
from appPublic.jsonConfig import getConfig
|
||
from appPublic.folderUtils import ProgramPath
|
||
from sqlor.dbpools import DBPools
|
||
from ahserver.serverenv import ServerEnv
|
||
from appPublic.event_dispatcher import EventDispatcher
|
||
|
||
config = getConfig(root_dir, NS={'workdir': root_dir, 'ProgramPath': ProgramPath()})
|
||
DBPools(config.databases)
|
||
env = ServerEnv()
|
||
env.event_dispatcher = EventDispatcher()
|
||
|
||
def get_module_dbname(mname):
|
||
"""All modules use the pipeline database."""
|
||
return 'pipeline'
|
||
env.get_module_dbname = get_module_dbname
|
||
|
||
# 记账链路必需模块（顺序同主应用：appbase/pricing 先于资源/产品模块）
|
||
from appbase.init import load_appbase
|
||
from pricing.init import load_pricing
|
||
from accounting.init import load_accounting
|
||
from discount.init import load_discount
|
||
from supplychain.init import load_supplychain
|
||
from product_management import load_product_management
|
||
from pipeline_llm.init import load_llm
|
||
|
||
load_appbase() # get_business_date（product_accounting_generic 依赖；
|
||
# 2026-09-05 实测缺失致出账 'NoneType' not callable）
|
||
load_pricing() # buffered_charging（定价引擎）
|
||
load_accounting() # consume_accounting / 余额 / 业务日期
|
||
load_discount() # get_min_product_discount（客户折扣→售价）
|
||
load_supplychain() # calculate_sale_amounts / get_distribution_chain（供应商折扣→成本）
|
||
load_product_management() # product_accounting_generic（三方账落账）
|
||
load_llm() # pipeline_llm（流水扫描 + 结算分流）
|
||
logger.info("[accounting_worker] 记账链路模块加载完成")
|
||
|
||
|
||
async def _main():
|
||
from pipeline_llm.accounting import llm_usage_accounting_loop
|
||
await llm_usage_accounting_loop()
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument('-w', '--workdir', default=os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
args = parser.parse_args()
|
||
os.chdir(args.workdir)
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s[%(levelname)s][%(name)s]%(message)s')
|
||
logger.info("[accounting_worker] 启动（workdir=%s）", args.workdir)
|
||
|
||
init_worker()
|
||
try:
|
||
asyncio.run(_main())
|
||
except KeyboardInterrupt:
|
||
logger.info("[accounting_worker] 收到中断信号，退出")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|