dongsk/mcp-auth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

聚焦场景（空串/纯空白 → None） 归化处理

Browse Source
This commit is contained in:
dongsk
2026-08-09 19:31:44 +08:00
parent 09487ee766
commit a77e44c68b
2 changed files with 92 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp-for-crm/src/server.py
+55 -3
View File
@@ -10,6 +10,19 @@ from datetime import datetime, timedelta
from mcp.server import MCPServer
from db import get_pool
def _s(value):
"""查询语句处的 str 归一化：None / 空串 / 纯空白 → 统一为 None，其他原值返回。
只处理 str 类型，不碰 int/float/list/bool。
用法：在每个工具函数开头对所有 str 参数执行 `x = _s(x)`，
确保后续 `if x:` / `if x is not None:` / LIKE 拼接 行为一致。
"""
if isinstance(value, str):
return value if value.strip() else None
return value
app = MCPServer(
name="mcp-for-crm-auto",
description="汽车零部件智能报价 CRM 数据服务",
@@ -28,6 +41,10 @@ async def get_customer_info(
oem_tier: str = None
) -> str:
"""查询 CRM 客户信息。支持按客户编码精确查询或按名称、OEM 层级模糊搜索。返回客户信用等级、折扣率、历史订单等。"""
# 查询语句处 str 归一化：空串/纯空白同 None
customer_code = _s(customer_code)
name = _s(name)
oem_tier = _s(oem_tier)
pool = await get_pool()
query = "SELECT * FROM customer WHERE 1=1"
params = []
@@ -62,6 +79,11 @@ async def create_inquiry(
notes: str = None
) -> str:
"""在 CRM 中创建新的询价记录。关联客户、图纸号、零件号、年用量。客户提交询价后建档使用。"""
# 查询语句处 str 归一化：空串/纯空白同 None
customer_code = _s(customer_code)
drawing_number = _s(drawing_number)
part_number = _s(part_number)
notes = _s(notes)
pool = await get_pool()
customer = await pool.fetchrow("SELECT name FROM customer WHERE customer_code = $1", customer_code)
@@ -110,6 +132,12 @@ async def save_quotation(
remarks: str = None
) -> str:
"""保存报价单到 CRM。包含模具费、单价、年用量、成本明细。报价生成后存档使用。"""
# 查询语句处 str 归一化：空串/纯空白同 None
inquiry_id = _s(inquiry_id)
customer_code = _s(customer_code)
payment_terms = _s(payment_terms)
delivery_terms = _s(delivery_terms)
remarks = _s(remarks)
pool = await get_pool()
customer = await pool.fetchrow("SELECT name FROM customer WHERE customer_code = $1", customer_code)
@@ -213,6 +241,12 @@ async def get_quotation_history(
latest_only: bool = False
) -> str:
"""查询历史报价记录。用于核价参考和价格一致性校验。支持按客户、询价编号、日期范围筛选。latest_only=true 时每个询价单仅返回最新版本。"""
# 查询语句处 str 归一化：空串/纯空白同 None
customer_code = _s(customer_code)
inquiry_id = _s(inquiry_id)
start_date = _s(start_date)
end_date = _s(end_date)
status = _s(status)
pool = await get_pool()
select_clause = (
"SELECT DISTINCT ON (q.inquiry_id) q.*, c.name as customer_name"
@@ -280,6 +314,15 @@ async def update_opportunity(
notes: str = None
) -> str:
"""更新 CRM 中的商机状态。支持创建新商机或更新已有商机的阶段、金额、概率。报价后跟进使用。"""
# 查询语句处 str 归一化：空串/纯空白同 None。str 变为 None 后，
# 后面 if value is not None 会自然跳过，避免空串被当作有效值写入。
opportunity_id = _s(opportunity_id)
customer_code = _s(customer_code)
title = _s(title)
stage = _s(stage)
oem_program = _s(oem_program)
expected_close_date = _s(expected_close_date)
notes = _s(notes)
pool = await get_pool()
close_date = None
@@ -350,6 +393,8 @@ async def update_opportunity(
@app.tool()
async def get_inquiry_info(inquiry_id: str) -> str:
"""按询价单号回读询价单整行（只读）。用于存档后验证：状态、图号、零件号、年用量、目标价、创建时间。"""
# 查询语句处 str 归一化：空串/纯空白同 None
inquiry_id = _s(inquiry_id)
pool = await get_pool()
row = await pool.fetchrow("SELECT * FROM inquiry WHERE inquiry_id = $1", inquiry_id)
if not row:
@@ -364,6 +409,9 @@ async def get_inquiry_info(inquiry_id: str) -> str:
@app.tool()
async def get_opportunity_info(opportunity_id: str = None, customer_code: str = None) -> str:
"""按商机号或客户编码回读商机（只读）。用于存档后验证：stage/probability/expected_amount/更新时间。"""
# 查询语句处 str 归一化：空串/纯空白同 None
opportunity_id = _s(opportunity_id)
customer_code = _s(customer_code)
pool = await get_pool()
if opportunity_id:
row = await pool.fetchrow("SELECT * FROM opportunity WHERE opportunity_id = $1", opportunity_id)
@@ -400,8 +448,12 @@ async def purge_test_records(
"""清理 TC-03 回归测试数据（受限管理工具）：仅删除 remarks/notes 含「TC-03回归测试」或 title 以「TC-03」开头的记录，其余一律拒绝，防止误删演示/生产数据。"""
pool = await get_pool()
deleted = {"quotations": 0, "inquiries": 0, "opportunities": 0, "rejected": []}
# 查询语句处 str 归一化：过滤掉 list 里的空串/纯空白 ID（同 None，跳过）
qids = [qid for qid in (quotation_ids or []) if _s(qid)]
iids = [iid for iid in (inquiry_ids or []) if _s(iid)]
oids = [oid for oid in (opportunity_ids or []) if _s(oid)]
for qid in quotation_ids or []:
for qid in qids:
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)
@@ -410,7 +462,7 @@ async def purge_test_records(
else:
deleted["rejected"].append(qid)
for iid in inquiry_ids or []:
for iid in iids:
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)
@@ -419,7 +471,7 @@ async def purge_test_records(
else:
deleted["rejected"].append(iid)
for oid in opportunity_ids or []:
for oid in oids:
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)
mcp-for-erp/src/server.py
+37 -6
View File
@@ -9,6 +9,19 @@ import json
from mcp.server import MCPServer
from db import get_pool
def _s(value):
"""查询语句处的 str 归一化：None / 空串 / 纯空白 → 统一为 None，其他原值返回。
只处理 str 类型，不碰 int/float/list/bool。
用法：在每个工具函数开头对所有 str 参数执行 `x = _s(x)`，
确保后续 `if x:` / `if x is not None:` / LIKE 拼接 行为一致。
"""
if isinstance(value, str):
return value if value.strip() else None
return value
app = MCPServer(
name="mcp-for-erp-auto",
description="汽车零部件智能报价 ERP 数据服务",
@@ -28,6 +41,10 @@ async def query_material_master(
die_casting_ton: int = None
) -> str:
"""查询 ERP 物料主数据。支持按物料编码精确查询，或按零件类别、材质、吨位等参数组合查询。返回物料列表，包含体积、重量、吨位等关键参数。"""
# 查询语句处 str 归一化：空串/纯空白同 None
material_code = _s(material_code)
part_category = _s(part_category)
material_grade = _s(material_grade)
pool = await get_pool()
query = "SELECT * FROM material_master WHERE 1=1"
params = []
@@ -58,6 +75,11 @@ async def query_material_master(
@app.tool()
async def match_material_grade(customer_grade: str, standard: str = None) -> str:
"""将客户指定材质牌号自动对照为企业内部等效牌号。支持 GB/ASTM/JIS/EN/ISO 标准体系，返回等效牌号、密度、铝锭参考价。"""
# 查询语句处 str 归一化：空串/纯空白同 None
customer_grade = _s(customer_grade)
standard = _s(standard)
if customer_grade is None:
return json.dumps({"matched": False, "message": "需提供 customer_grade（材质牌号）"}, ensure_ascii=False, indent=2)
pool = await get_pool()
query = """
@@ -97,6 +119,8 @@ async def match_material_grade(customer_grade: str, standard: str = None) -> str
@app.tool()
async def get_die_casting_params(material_code: str, volume_cc: float = None) -> str:
"""根据物料编码获取压铸参数。返回压铸机吨位、模次节拍、铝水重量、压铸成本。铝水重量 = 体积 × 密度。"""
# 查询语句处 str 归一化：空串/纯空白同 None
material_code = _s(material_code)
pool = await get_pool()
material = await pool.fetchrow("SELECT * FROM material_master WHERE material_code = $1", material_code)
@@ -136,6 +160,8 @@ async def get_machining_estimate(
mill_faces: int = 0
) -> str:
"""根据孔清单估算机加工工时。钻孔、攻丝、铰孔、铣面均有固化节拍，逐项累加得到总工时和成本。"""
# 查询语句处 str 归一化：空串/纯空白同 None
material_code = _s(material_code)
pool = await get_pool()
drill_holes = drill_holes or []
tap_holes = tap_holes or []
@@ -146,7 +172,7 @@ async def get_machining_estimate(
details = []
for hole in drill_holes:
spec = hole.get("spec", "Φ8")
spec = _s(hole.get("spec", "Φ8")) or "Φ8"
count = hole.get("count", 1)
row = await pool.fetchrow("SELECT * FROM machining_cycle WHERE operation_type='drill' AND spec=$1", spec)
if row:
@@ -157,7 +183,7 @@ async def get_machining_estimate(
details.append({"operation": "drill", "spec": spec, "count": count, "total_time_sec": time_sec})
for hole in tap_holes:
spec = hole.get("spec", "M8×1.25")
spec = _s(hole.get("spec", "M8×1.25")) or "M8×1.25"
count = hole.get("count", 1)
row = await pool.fetchrow("SELECT * FROM machining_cycle WHERE operation_type='tap' AND spec=$1", spec)
if row:
@@ -168,7 +194,7 @@ async def get_machining_estimate(
details.append({"operation": "tap", "spec": spec, "count": count, "total_time_sec": time_sec})
for hole in ream_holes:
spec = hole.get("spec", "Φ8H7")
spec = _s(hole.get("spec", "Φ8H7")) or "Φ8H7"
count = hole.get("count", 1)
row = await pool.fetchrow("SELECT * FROM machining_cycle WHERE operation_type='ream' AND spec=$1", spec)
if row:
@@ -209,6 +235,8 @@ async def calculate_part_cost(
profit_rate: float = 0.15
) -> str:
"""汇总核价：材料成本（铝水重量×单价）+ 压铸成本 + 机加工成本 + 后处理成本 + 管理费 + 利润。返回完整成本明细和最终报价。"""
# 查询语句处 str 归一化：空串/纯空白同 None
material_code = _s(material_code)
pool = await get_pool()
drill_holes = drill_holes or []
tap_holes = tap_holes or []
@@ -263,17 +291,20 @@ async def calculate_part_cost(
total_machining_time_sec = 0
for hole in drill_holes:
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='drill' AND spec=$1", hole.get("spec", "Φ8"))
spec = _s(hole.get("spec", "Φ8")) or "Φ8"
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='drill' AND spec=$1", spec)
if row:
total_machining_time_sec += float(row["cycle_time_sec"]) * hole.get("count", 1)
for hole in tap_holes:
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='tap' AND spec=$1", hole.get("spec", "M8×1.25"))
spec = _s(hole.get("spec", "M8×1.25")) or "M8×1.25"
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='tap' AND spec=$1", spec)
if row:
total_machining_time_sec += float(row["cycle_time_sec"]) * hole.get("count", 1)
for hole in ream_holes:
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='ream' AND spec=$1", hole.get("spec", "Φ8H7"))
spec = _s(hole.get("spec", "Φ8H7")) or "Φ8H7"
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='ream' AND spec=$1", spec)
if row:
total_machining_time_sec += float(row["cycle_time_sec"]) * hole.get("count", 1)