zzhyyyyy
c97fbbad0e
mentor 要求把会话存档轮询从 3s 提到 0.5s。WX_FINANCE_POLL_INTERVAL_SEC 由 int 改 float (默认 0.5), poller 的可打断 sleep 从 range(int) 改为按 ≤0.5s 分片的 while 循环(原 range() 不吃 float、无法支持亚秒间隔), 启动日志 %d→%s。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
131 lines
4.8 KiB
Python
131 lines
4.8 KiB
Python
"""会话存档【独立进程】轮询 poller —— 单独 systemd 服务跑, 与 app-server 隔离。
|
|
|
|
为什么独立进程:WeWorkFinanceSdk 是 Go 编译的 c-shared 库, 自带 Go runtime。嵌进 app-server(uvicorn)
|
|
后台线程里跑会 segfault、把主进程带崩(2026-07-17 实测 core-dump 每 5s 循环)。独立进程让 Go runtime
|
|
独占进程 + 主线程, 崩了 systemd 只重启本 poller、不动 app-server。拉到美团卡片(weapp)/截图(image)
|
|
→ POST app-server 内部端点 /internal/wx-finance/pending 打比价信号(下游心跳→弹窗→比价 完全复用)。
|
|
|
|
手动跑: cd /opt/shaguabijia-app-server && .venv/bin/python scripts/wx_finance_poller.py
|
|
生产: deploy/wx-finance-poller.service 常驻(systemctl enable --now wx-finance-poller)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import signal
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
from app.core.logging import setup_logging
|
|
from app.integrations.wx_finance_sdk import WxFinanceError, WxFinanceSdk
|
|
|
|
logger = logging.getLogger("shagua.wx_finance_poller")
|
|
|
|
_stop = False
|
|
|
|
|
|
def _on_signal(_signum: int, _frame: object) -> None:
|
|
global _stop
|
|
_stop = True
|
|
|
|
|
|
def _load_seq() -> int:
|
|
try:
|
|
return int(Path(settings.WX_FINANCE_SEQ_FILE).read_text().strip() or "0")
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
def _save_seq(seq: int) -> None:
|
|
try:
|
|
p = Path(settings.WX_FINANCE_SEQ_FILE)
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(str(seq))
|
|
except Exception:
|
|
logger.exception("seq 持久化失败")
|
|
|
|
|
|
def _notify(client: httpx.Client, source: str, kind: str) -> None:
|
|
"""POST 内部端点让 app-server 打信号(set_pending 在 app-server 进程内存)。"""
|
|
url = f"{settings.WX_FINANCE_INTERNAL_URL.rstrip('/')}/internal/wx-finance/pending"
|
|
try:
|
|
resp = client.post(
|
|
url,
|
|
json={"source": source, "kind": kind},
|
|
headers={"X-Internal-Secret": settings.INTERNAL_API_SECRET},
|
|
timeout=10,
|
|
)
|
|
logger.info("notify app-server (%s) → %s", kind, resp.status_code)
|
|
except Exception:
|
|
logger.exception("notify app-server 失败")
|
|
|
|
|
|
def _handle(client: httpx.Client, msg: dict) -> None:
|
|
"""明文消息:外部用户(from≠接收成员)发来的 weapp/image → 通知 app-server 打信号。"""
|
|
msgtype = msg.get("msgtype") or ""
|
|
frm = msg.get("from") or ""
|
|
receiver = settings.WX_FINANCE_RECEIVER_USERID
|
|
if receiver and frm == receiver:
|
|
return # 成员自己发的, 跳过
|
|
if msgtype not in ("weapp", "image"):
|
|
return
|
|
logger.info("命中触发 from=%s type=%s", frm, msgtype)
|
|
_notify(client, "meituan", msgtype)
|
|
|
|
|
|
def main() -> None:
|
|
setup_logging(debug=settings.APP_DEBUG)
|
|
if not settings.wx_finance_configured:
|
|
logger.warning("WX_FINANCE 未配齐(ENABLED/corpid/存档Secret), poller 退出")
|
|
return
|
|
if not settings.INTERNAL_API_SECRET:
|
|
logger.error("INTERNAL_API_SECRET 未配置, poller 无法通知 app-server, 退出")
|
|
return
|
|
signal.signal(signal.SIGTERM, _on_signal)
|
|
signal.signal(signal.SIGINT, _on_signal)
|
|
|
|
sdk = WxFinanceSdk(
|
|
settings.WX_FINANCE_SDK_PATH,
|
|
settings.WX_KF_CORP_ID,
|
|
settings.WX_FINANCE_SECRET,
|
|
Path(settings.WX_FINANCE_PRIVATE_KEY_PATH).read_bytes(),
|
|
)
|
|
seq = _load_seq()
|
|
logger.info(
|
|
"wx_finance poller 启动, 从 seq=%d 轮询, 间隔 %ss", seq, settings.WX_FINANCE_POLL_INTERVAL_SEC
|
|
)
|
|
with httpx.Client() as client:
|
|
while not _stop:
|
|
try:
|
|
msgs = sdk.get_chat_data(seq, limit=1000)
|
|
if msgs:
|
|
logger.info("拉到 %d 条(seq>%d)", len(msgs), seq)
|
|
for item in msgs:
|
|
try:
|
|
plain = sdk.decrypt(item["encrypt_random_key"], item["encrypt_chat_msg"])
|
|
_handle(client, plain)
|
|
except Exception:
|
|
logger.exception("解密/处理单条失败 seq=%s", item.get("seq"))
|
|
seq = max(seq, int(item.get("seq", seq)))
|
|
if msgs:
|
|
_save_seq(seq)
|
|
except WxFinanceError:
|
|
logger.exception("GetChatData 出错")
|
|
except Exception:
|
|
logger.exception("轮询异常")
|
|
# 可打断的 sleep:按 ≤0.5s 分片, 支持亚秒间隔(如 0.5s), 收到 SIGTERM 尽快退
|
|
interval = max(0.1, settings.WX_FINANCE_POLL_INTERVAL_SEC)
|
|
step = min(0.5, interval)
|
|
slept = 0.0
|
|
while slept < interval and not _stop:
|
|
time.sleep(step)
|
|
slept += step
|
|
sdk.close()
|
|
logger.info("wx_finance poller 已停止")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|