新增工具

mcp-for-crm/src/server.py

@@ -343,6 +343,93 @@ async def update_opportunity(
     }, ensure_ascii=False, indent=2)
 
 
+# ============================================================
+# 工具 6: get_inquiry_info — 询价单回读（只读）
+# ============================================================
+
+@app.tool()
+async def get_inquiry_info(inquiry_id: str) -> str:
+    """按询价单号回读询价单整行（只读）。用于存档后验证：状态、图号、零件号、年用量、目标价、创建时间。"""
+    pool = await get_pool()
+    row = await pool.fetchrow("SELECT * FROM inquiry WHERE inquiry_id = $1", inquiry_id)
+    if not row:
+        return json.dumps({"success": False, "message": f"未找到询价单 {inquiry_id}"}, ensure_ascii=False, indent=2)
+    return json.dumps({"success": True, "inquiry": dict(row)}, ensure_ascii=False, indent=2, default=str)
+
+
+# ============================================================
+# 工具 7: get_opportunity_info — 商机回读（只读）
+# ============================================================
+
+@app.tool()
+async def get_opportunity_info(opportunity_id: str = None, customer_code: str = None) -> str:
+    """按商机号或客户编码回读商机（只读）。用于存档后验证：stage/probability/expected_amount/更新时间。"""
+    pool = await get_pool()
+    if opportunity_id:
+        row = await pool.fetchrow("SELECT * FROM opportunity WHERE opportunity_id = $1", opportunity_id)
+        rows = [row] if row else []
+    elif customer_code:
+        rows = await pool.fetch("SELECT * FROM opportunity WHERE customer_code = $1 ORDER BY created_at DESC", customer_code)
+    else:
+        return json.dumps({"success": False, "message": "需提供 opportunity_id 或 customer_code"}, ensure_ascii=False, indent=2)
+    return json.dumps({"total": len(rows), "opportunities": [dict(r) for r in rows]}, ensure_ascii=False, indent=2, default=str)
+
+
+# ============================================================
+# 工具 8: get_record_counts — 行数统计（只读）
+# ============================================================
+
+@app.tool()
+async def get_record_counts() -> str:
+    """返回 inquiry/quotation/opportunity 三表行数（只读）。用于数据卫生检查。"""
+    pool = await get_pool()
+    counts = {t: await pool.fetchval(f"SELECT COUNT(*) FROM {t}") for t in ("inquiry", "quotation", "opportunity")}
+    return json.dumps({"success": True, "counts": counts}, ensure_ascii=False, indent=2)
+
+
+# ============================================================
+# 工具 9: purge_test_records — 受限测试数据清理（管理）
+# ============================================================
+
+@app.tool()
+async def purge_test_records(
+    quotation_ids: list = None,
+    inquiry_ids: list = None,
+    opportunity_ids: list = None
+) -> str:
+    """清理 TC-03 回归测试数据（受限管理工具）：仅删除 remarks/notes 含「TC-03回归测试」或 title 以「TC-03」开头的记录，其余一律拒绝，防止误删演示/生产数据。"""
+    pool = await get_pool()
+    deleted = {"quotations": 0, "inquiries": 0, "opportunities": 0, "rejected": []}
+
+    for qid in quotation_ids or []:
+        row = await pool.fetchrow("SELECT remarks FROM quotation WHERE quotation_id = $1", qid)
+        if row and "TC-03回归测试" in (row["remarks"] or ""):
+            await pool.execute("DELETE FROM quotation_cost_detail WHERE quotation_id = $1", qid)
+            await pool.execute("DELETE FROM quotation WHERE quotation_id = $1", qid)
+            deleted["quotations"] += 1
+        else:
+            deleted["rejected"].append(qid)
+
+    for iid in inquiry_ids or []:
+        row = await pool.fetchrow("SELECT notes FROM inquiry WHERE inquiry_id = $1", iid)
+        if row and "TC-03回归测试" in (row["notes"] or ""):
+            await pool.execute("DELETE FROM inquiry_item WHERE inquiry_id = $1", iid)
+            await pool.execute("DELETE FROM inquiry WHERE inquiry_id = $1", iid)
+            deleted["inquiries"] += 1
+        else:
+            deleted["rejected"].append(iid)
+
+    for oid in opportunity_ids or []:
+        row = await pool.fetchrow("SELECT title FROM opportunity WHERE opportunity_id = $1", oid)
+        if row and (row["title"] or "").startswith("TC-03"):
+            await pool.execute("DELETE FROM opportunity WHERE opportunity_id = $1", oid)
+            deleted["opportunities"] += 1
+        else:
+            deleted["rejected"].append(oid)
+
+    return json.dumps({"success": True, "deleted": deleted}, ensure_ascii=False, indent=2)
+
+
 # ============================================================
 # 启动 HTTP Server
 # ============================================================