From e0dbac6f9d430073a80681c4c061ac4784c2a20e Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Fri, 4 Sep 2026 19:32:43 +0800
Subject: [PATCH] =?UTF-8?q?feat(accounting):=20=E7=8B=AC=E7=AB=8B=E5=BC=82?=
 =?UTF-8?q?=E6=AD=A5=E8=AE=B0=E8=B4=A6=E8=BF=9B=E7=A8=8B=E2=80=94=E2=80=94?=
 =?UTF-8?q?=E6=A8=A1=E5=9E=8B=E7=94=A8=E9=87=8F=E5=87=BA=E8=B4=A6(?=
 =?UTF-8?q?=E5=AE=A2=E6=88=B7=E4=BB=98/=E5=95=86=E6=88=B7=E8=90=A5?=
 =?UTF-8?q?=E6=94=B6/=E4=BE=9B=E5=BA=94=E5=95=86=E6=88=90=E6=9C=AC),?=
 =?UTF-8?q?=E5=90=8C=E6=AD=A5=E8=AE=B0=E8=B4=A6=E7=95=99web,=E8=BF=9B?=
 =?UTF-8?q?=E7=A8=8B=E7=94=B1start/stop.sh=E7=AE=A1=E7=90=86?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/pipeline_accounting_worker.py | 83 +++++++++++++++++++++++++++++++
 start.sh                          | 19 +++++++
 stop.sh                           | 11 ++++
 3 files changed, 113 insertions(+)
 create mode 100644 app/pipeline_accounting_worker.py

diff --git a/app/pipeline_accounting_worker.py b/app/pipeline_accounting_worker.py
new file mode 100644
index 0000000..c98c893
--- /dev/null
+++ b/app/pipeline_accounting_worker.py
@@ -0,0 +1,83 @@
+#!/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
+
+    # 记账链路必需模块（顺序同主应用：pricing 先于资源/产品模块）
+    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_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()
diff --git a/start.sh b/start.sh
index fc9f41a..530cc03 100755
--- a/start.sh
+++ b/start.sh
@@ -35,3 +35,22 @@ export PATH="$WORKDIR/bin:$PATH"
 $WORKDIR/py3/bin/python $WORKDIR/app/pipeline_app.py -p $PORT -w $WORKDIR >> $WORKDIR/logs/pipeline.log 2>&1 &
 echo $! > "$PIDFILE"
 echo "Started PID $(cat $PIDFILE)"
+
+# 独立异步记账进程（模型用量出账：客户付/商户营收/供应商成本）
+# 只随主进程/worker 启动；web 单独模式不启动（同步记账在 web 内）
+if [ "$MODE" != "web" ]; then
+    ACC_PIDFILE="pipeline-accounting.pid"
+    if [ -f "$ACC_PIDFILE" ]; then
+        acc_pid=$(cat "$ACC_PIDFILE")
+        if kill -0 "$acc_pid" 2>/dev/null; then
+            echo "Accounting worker already running (PID $acc_pid)"
+        else
+            rm -f "$ACC_PIDFILE"
+        fi
+    fi
+    if [ ! -f "$ACC_PIDFILE" ]; then
+        $WORKDIR/py3/bin/python $WORKDIR/app/pipeline_accounting_worker.py -w $WORKDIR >> $WORKDIR/logs/pipeline-accounting.log 2>&1 &
+        echo $! > "$ACC_PIDFILE"
+        echo "Started accounting worker PID $(cat $ACC_PIDFILE)"
+    fi
+fi
diff --git a/stop.sh b/stop.sh
index 78b1483..5c239c3 100755
--- a/stop.sh
+++ b/stop.sh
@@ -14,3 +14,14 @@ if [ -f pipeline.pid ]; then
 else
     echo "No pid file found"
 fi
+
+if [ -f pipeline-accounting.pid ]; then
+    pid=$(cat pipeline-accounting.pid)
+    if kill -0 "$pid" 2>/dev/null; then
+        kill "$pid"
+        echo "Stopped accounting worker PID $pid"
+    else
+        echo "Accounting worker $pid not running"
+    fi
+    rm -f pipeline-accounting.pid
+fi