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

feat: 添加资产变动确认卡片和Excel图片解析功能

Browse Source 
- 新增 `send_asset_confirmation_card` 工具，用于发送资产变动确认卡片
- 新增 `upload_image_by_excel` 工具，支持从Excel中根据image_key定位并上传图片
- 在file-tools中添加 `excel_image_key_to_temp_file` 和 `upload_file_to_minio` 工具
- 新增配置文件管理和MinIO集成支持
- 更新项目依赖版本，添加openpyxl和minio库
This commit is contained in:
Sucan
2026-02-13 19:57:02 +08:00
parent 0a308726a6
commit 135c8e379e
8 changed files with 1035 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

479
lzwcai_lark_mcp/lzwcai_lark_mcp/tools.py
View File

@@ -28,6 +28,28 @@ tools = [
"required": ["url"]
}
),
Tool(
name="upload_image_by_excel",
description="从Excel中根据image_key定位图片并上传返回image_key",
inputSchema={
"type": "object",
"properties": {
"excel_path": {
"type": "string",
"description": "Excel文件路径"
},
"image_key": {
"type": "string",
"description": "Excel中的image_key"
},
"image_type": {
"type": "string",
"description": "图片类型，例如 message"
}
},
"required": ["excel_path", "image_key"]
}
),
Tool(
name="send_card_message",
description="发送消息卡片，入参为receiver_ids、person_id和image_key",
@@ -52,6 +74,38 @@ tools = [
},
"required": ["image_key", "person_id"]
}
),
Tool(
name="send_asset_confirmation_card",
description="发送资产变动确认卡片，入参为user_id、time、inputs和outputs",
inputSchema={
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": "消息接收者ID，同时用于卡片内person组件"
},
"time": {
"type": "string",
"description": "资产变动发生时间"
},
"inputs": {
"type": "array",
"description": "入库资产列表",
"items": {
"type": "string"
}
},
"outputs": {
"type": "array",
"description": "出库资产列表",
"items": {
"type": "string"
}
}
},
"required": ["user_id", "time"]
}
)
]
@@ -94,6 +148,400 @@ def upload_image_by_url(token: str, url: str, image_type: str) -> str:
return image_key
def upload_image_by_file(token: str, file_path: str, image_type: str) -> str:
if not os.path.isfile(file_path):
raise FileNotFoundError(f"image file not found: {file_path}")
filename = os.path.basename(file_path) or "image"
content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
with open(file_path, "rb") as handle:
content = handle.read()
headers = {"Authorization": f"Bearer {token}"}
data = {"image_type": image_type}
files = {"image": (filename, content, content_type)}
response = requests.post(
"https://open.feishu.cn/open-apis/im/v1/images",
headers=headers,
data=data,
files=files,
timeout=30
)
response.raise_for_status()
payload = response.json()
if payload.get("code") not in (0, None):
raise RuntimeError(f"lark image upload failed: {payload}")
image_key = payload.get("data", {}).get("image_key")
if not image_key:
raise RuntimeError(f"lark image upload missing image_key: {payload}")
return image_key
def _is_url(value: str) -> bool:
parsed = urlparse(value)
return parsed.scheme in ("http", "https")
def _normalize_header(value: object) -> str:
if value is None:
return ""
return str(value).strip().lower()
def _resolve_image_source_from_excel(excel_path: str, image_key: str) -> str:
from openpyxl import load_workbook
workbook = load_workbook(excel_path, read_only=True, data_only=True)
worksheet = workbook.worksheets[0]
header_row = next(worksheet.iter_rows(min_row=1, max_row=1, values_only=True), None)
if not header_row:
raise ValueError("excel header row is empty")
header_map = {}
for index, header in enumerate(header_row):
name = _normalize_header(header)
if name:
header_map[name] = index
key_col_index = None
for key_name in ("image_key", "imagekey", "key"):
if key_name in header_map:
key_col_index = header_map[key_name]
break
if key_col_index is None:
raise ValueError("missing image_key column in excel")
candidate_columns = ["image_path", "image_url", "image", "url", "file_path", "path", "image_file"]
candidate_indices = [header_map[name] for name in candidate_columns if name in header_map]
matched_source = ""
for row in worksheet.iter_rows(min_row=2, values_only=True):
if key_col_index >= len(row):
continue
cell_value = row[key_col_index]
if cell_value is None:
continue
if str(cell_value).strip() != image_key:
continue
for index in candidate_indices:
if index < len(row):
candidate_value = row[index]
if candidate_value is not None and str(candidate_value).strip():
matched_source = str(candidate_value).strip()
break
if not matched_source:
matched_source = str(cell_value).strip()
break
if not matched_source:
raise ValueError(f"missing image source for image_key: {image_key}")
if not _is_url(matched_source):
if not os.path.isabs(matched_source):
matched_source = os.path.join(os.path.dirname(excel_path), matched_source)
return matched_source
def _normalize_asset_items(value: object) -> List[str]:
if value is None:
return []
if isinstance(value, str):
segments = [segment.strip() for segment in re.split(r"[，,、;；]+", value)]
return [segment for segment in segments if segment]
if isinstance(value, list):
items: List[str] = []
for item in value:
items.extend(_normalize_asset_items(item))
return items
if isinstance(value, dict):
if "inputs" in value:
return _normalize_asset_items(value.get("inputs"))
if "outputs" in value:
return _normalize_asset_items(value.get("outputs"))
if "name" in value:
return _normalize_asset_items(value.get("name"))
items: List[str] = []
for item in value.values():
items.extend(_normalize_asset_items(item))
return items
text = str(value).strip()
return [text] if text else []
def _build_asset_options(inputs: List[str], outputs: List[str]) -> tuple[list[Dict[str, object]], str]:
labels = []
options = []
for item in outputs:
label = f"{item}(出)"
labels.append(label)
options.append({
"text": {"tag": "plain_text", "content": label},
"value": f"{item}|出"
})
for item in inputs:
label = f"{item}(入)"
labels.append(label)
options.append({
"text": {"tag": "plain_text", "content": label},
"value": f"{item}|入"
})
return options, "、".join(labels)
def send_asset_confirmation_card(token: str, user_id: str, change_time: str, inputs: object, outputs: object) -> str:
normalized_user_id = str(user_id).strip()
if not normalized_user_id:
raise ValueError("missing user_id")
normalized_time = str(change_time or "").strip()
if not normalized_time:
normalized_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
input_items = _normalize_asset_items(inputs)
output_items = _normalize_asset_items(outputs)
options, summary = _build_asset_options(input_items, output_items)
content = {
"schema": "2.0",
"config": {
"update_multi": True,
"style": {
"text_size": {
"normal_v2": {
"default": "normal",
"pc": "normal",
"mobile": "heading"
}
}
}
},
"body": {
"direction": "vertical",
"elements": [
{
"tag": "column_set",
"flex_mode": "stretch",
"horizontal_spacing": "12px",
"horizontal_align": "left",
"columns": [
{
"tag": "column",
"width": "weighted",
"elements": [
{
"tag": "markdown",
"content": f"**<font color='blue-600'>用户:</font>** <person id={normalized_user_id} show_name=true show_avatar=true style='normal'></person>",
"text_align": "left",
"text_size": "normal"
}
],
"vertical_spacing": "8px",
"horizontal_align": "left",
"vertical_align": "top",
"weight": 1
},
{
"tag": "column",
"width": "weighted",
"elements": [
{
"tag": "markdown",
"content": f"**<font color='blue-600'>发生时间:</font>** <font color='grey'>{normalized_time}</font>",
"text_align": "left",
"text_size": "normal"
}
],
"vertical_spacing": "8px",
"horizontal_align": "left",
"vertical_align": "top",
"weight": 1
}
],
"margin": "0px 0px 0px 0px"
},
{
"tag": "hr",
"margin": "0px 0px 0px 0px"
},
{
"tag": "form",
"elements": [
{
"tag": "markdown",
"content": "**<font color='blue-600'>\\*请准确选择属于您的变动项:</font>**",
"text_align": "left",
"text_size": "normal",
"margin": "0px 0px 8px 0px"
},
{
"tag": "multi_select_static",
"placeholder": {
"tag": "plain_text",
"content": "请选择资产变动项"
},
"options": options,
"type": "default",
"width": "fill",
"required": False,
"name": "asset_changes",
"margin": "0px 0px 16px 0px",
"element_id": "cIiptD7Z4hCtAeR5Rb0b"
},
{
"tag": "markdown",
"content": "**<font color='blue-600'>其他说明:</font>**",
"text_align": "left",
"text_size": "normal_v2",
"margin": "0px 0px 0px 0px"
},
{
"tag": "input",
"placeholder": {
"tag": "plain_text",
"content": "请输入"
},
"default_value": "",
"width": "fill",
"name": "Input_3h27n7woqci",
"margin": "0px 0px 0px 0px"
},
{
"tag": "column_set",
"flex_mode": "flow",
"horizontal_spacing": "8px",
"horizontal_align": "right",
"columns": [
{
"tag": "column",
"width": "auto",
"elements": [
{
"tag": "button",
"text": {
"tag": "plain_text",
"content": "确认"
},
"type": "primary_filled",
"width": "default",
"size": "medium",
"behaviors": [
{
"type": "callback",
"value": {
"action": "confirm_asset_changes"
}
}
],
"form_action_type": "submit",
"name": "confirm_button",
"margin": "4px 0px 4px 0px"
}
],
"vertical_spacing": "8px",
"horizontal_align": "left",
"vertical_align": "top"
},
{
"tag": "column",
"width": "auto",
"elements": [
{
"tag": "button",
"text": {
"tag": "plain_text",
"content": "反馈问题"
},
"type": "default",
"width": "default",
"confirm": {
"title": {
"tag": "plain_text",
"content": "反馈并废除该确认单吗？"
},
"text": {
"tag": "plain_text",
"content": summary
}
},
"behaviors": [
{
"type": "callback",
"value": {
"": ""
}
}
],
"form_action_type": "submit",
"name": "feedback_button",
"margin": "4px 0px 4px 0px"
}
],
"vertical_spacing": "8px",
"horizontal_align": "left",
"vertical_align": "top"
}
]
}
],
"direction": "vertical",
"horizontal_align": "left",
"vertical_align": "top",
"padding": "12px 12px 12px 12px",
"margin": "0px 0px 0px 0px",
"name": "asset_confirmation_form"
},
{
"tag": "hr",
"margin": "0px 0px 0px 0px"
}
]
},
"header": {
"title": {
"tag": "plain_text",
"content": "资产变动单"
},
"subtitle": {
"tag": "plain_text",
"content": ""
},
"text_tag_list": [
{
"tag": "text_tag",
"text": {
"tag": "plain_text",
"content": "待确认"
},
"color": "orange"
}
],
"template": "blue",
"icon": {
"tag": "standard_icon",
"token": "googledrive_outlined"
},
"padding": "12px 8px 12px 8px"
}
}
payload = {
"receive_id": normalized_user_id,
"msg_type": "interactive",
"content": json.dumps(content, ensure_ascii=False)
}
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
response = requests.post(
"https://open.feishu.cn/open-apis/im/v1/messages",
params={"receive_id_type": "user_id"},
headers=headers,
json=payload,
timeout=30
)
try:
data = response.json()
except ValueError:
data = {"raw": response.text}
if not response.ok:
raise RuntimeError(f"lark send message http error: {data}")
if data.get("code") not in (0, None):
raise RuntimeError(f"lark send message failed: {data}")
message_id = data.get("data", {}).get("message_id")
if not message_id:
raise RuntimeError(f"lark send message missing message_id: {data}")
return message_id
def send_card_message(token: str, receiver_id: str, person_id: str, image_key: str) -> str:
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
content = {
@@ -261,6 +709,29 @@ async def handle_call_tool(name: str, arguments: Dict[str, object], token: str)
raise ValueError("missing url")
result = upload_image_by_url(token, url, image_type)
elif name == "upload_image_by_excel":
image_type = str(arguments.get("image_type") or "message").strip()
excel_path = str(arguments.get("excel_path", "")).strip()
image_key = str(arguments.get("image_key", "")).strip()
if not image_type:
raise ValueError("missing image_type")
if not excel_path:
raise ValueError("missing excel_path")
if not image_key:
raise ValueError("missing image_key")
if not os.path.exists(excel_path):
raise FileNotFoundError(f"excel_path not found: {excel_path}")
excel_extension = os.path.splitext(excel_path)[1].lower()
if excel_extension in (".xls",):
raise ValueError("xls is not supported, please convert to xlsx")
if excel_extension in (".xlsx", ".xlsm", ".xltx", ".xltm"):
image_source = _resolve_image_source_from_excel(excel_path, image_key)
else:
image_source = excel_path
if _is_url(image_source):
result = upload_image_by_url(token, image_source, image_type)
else:
result = upload_image_by_file(token, image_source, image_type)
elif name == "send_card_message":
image_key = str(arguments.get("image_key", "")).strip()
receiver_ids = arguments.get("receiver_ids")
@@ -273,6 +744,14 @@ async def handle_call_tool(name: str, arguments: Dict[str, object], token: str)
if not isinstance(receiver_ids, list):
raise ValueError("receiver_ids must be a list")
result = send_card_messages(token, receiver_ids, person_id, image_key)
elif name == "send_asset_confirmation_card":
user_id = str(arguments.get("user_id", "")).strip()
change_time = str(arguments.get("time", "")).strip()
inputs = arguments.get("inputs")
outputs = arguments.get("outputs")
if not user_id:
raise ValueError("missing user_id")
result = send_asset_confirmation_card(token, user_id, change_time, inputs, outputs)
else:
raise ValueError(f"unknown tool name: {name}")
return [TextContent(type="text", text=result)]