diff --git a/pipeline_bidding/bid_flow.py b/pipeline_bidding/bid_flow.py index d7c6c01..44b6a52 100644 --- a/pipeline_bidding/bid_flow.py +++ b/pipeline_bidding/bid_flow.py @@ -26,7 +26,7 @@ from .bid_common import ( HT_TENDER_FILE, HT_HUMAN_DOCS, HT_DELIVERY_CONFIRM, ) -logger = logging.getLogger("pipeline.bidding.flow") +logger = logging.getLogger("pipeline.bidding") R_ANALYST = "agent.tender_analyst" R_PREP = "agent.bid_prep" @@ -39,6 +39,9 @@ MAX_ANALYST_ATTEMPT = 2 # 解析阶段最多重派次数(防死循环) MAX_PREP_ATTEMPT = 2 POLL_INTERVAL = 15 +# poller 心跳(供 /pipeline-bidding/api/bid_probe.dspy 诊断;服务内可见) +POLLER_STATE = {"last_run": "", "last_error": "", "rounds": 0} + async def _count(sor, sql, p): recs = await sor.sqlExe(sql, p) @@ -338,17 +341,34 @@ def start_poller(): async def _once(): async with db.sqlorContext(dbname) as sor: res = await _reconcile_all_with(sor) - for pid, acts in (res or {}).items(): - for a in acts: - if a.startswith(("created", "escalated", "completed", "error")): - logger.info("bid_flow %s: %s", pid, a) + return res + + import datetime # watchdog:每轮最多 60 秒,防连接池/MDL 锁卡死导致 poller 永久停摆 - await asyncio.wait_for(_once(), timeout=60) + res = await asyncio.wait_for(_once(), timeout=60) + POLLER_STATE["rounds"] += 1 + POLLER_STATE["last_run"] = datetime.datetime.now().strftime("%H:%M:%S") + POLLER_STATE["last_error"] = "" + for pid, acts in (res or {}).items(): + for a in acts: + if a.startswith(("created", "escalated", "completed", "error")): + try: + from appPublic.log import debug as _dbg + _dbg("bid_flow %s: %s" % (pid, a)) + except Exception: + pass + logger.info("bid_flow %s: %s", pid, a) except Exception as e: + POLLER_STATE["last_error"] = "%s: %s" % (type(e).__name__, str(e)[:160]) logger.warning("bid_poller error/timeout: %s", str(e)[:200]) await asyncio.sleep(POLL_INTERVAL) asyncio.create_task(_loop()) + try: + from appPublic.log import debug as _dbg + _dbg("bid_flow poller started (interval=%ss)" % POLL_INTERVAL) + except Exception: + pass logger.info("bid_flow poller started (interval=%ss)", POLL_INTERVAL) add_startup(_bid_poller) diff --git a/scripts/load_path.py b/scripts/load_path.py index b0a6688..40b4dc9 100644 --- a/scripts/load_path.py +++ b/scripts/load_path.py @@ -34,9 +34,10 @@ for t in TABLES: "/%s/%s/update_%s.dspy" % (MOD, t, t), "/%s/%s/delete_%s.dspy" % (MOD, t, t), ] -# 产线业务 API(当前无自写 dspy;保留结构,后续新增在此登记) +# 产线业务 API PATHS_LOGINED += [ "/%s/api/" % MOD, + "/%s/api/bid_probe.dspy" % MOD, ] diff --git a/wwwroot/api/bid_probe.dspy b/wwwroot/api/bid_probe.dspy new file mode 100644 index 0000000..b2b609d --- /dev/null +++ b/wwwroot/api/bid_probe.dspy @@ -0,0 +1,36 @@ +# bid_probe.dspy — 投标产线服务内探针(诊断用) +# GET/POST: action=status | reconcile +# status → 模块装载状态 + poller 心跳 + 活跃投标项目数 +# reconcile → 在服务进程内跑一轮全量对账,返回动作明细 + +import json + +action = (params_kw or {}).get('action', 'status') +dbname = get_module_dbname('pipeline-bidding') + +try: + import pipeline_bidding + from pipeline_bidding import bid_flow + loaded = True + ver = getattr(pipeline_bidding, '__version__', '?') +except Exception as e: + loaded = False + ver = str(e)[:100] + +if action == 'reconcile': + try: + res = await bid_flow.reconcile_all() + return json.dumps({"success": True, "projects": res}, ensure_ascii=False, default=str) + except Exception as e: + return json.dumps({"success": False, "error": str(e)[:300]}, ensure_ascii=False) + +out = {"loaded": loaded, "version": ver, + "poller_state": getattr(bid_flow, 'POLLER_STATE', None) if loaded else None} +if loaded: + async with DBPools().sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT COUNT(*) AS c FROM sd_projects WHERE pipeline_id='bidding_general' " + "AND status NOT IN ('paused','archived','completed','cancelled')", {}) + await sor.sqlExe("COMMIT", {}) + out["active_projects"] = getattr(recs[0], 'c', 0) if recs else 0 +return json.dumps(out, ensure_ascii=False, default=str)