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

```

Browse Source 
feat(lzwcai-agile-db): 更新版本至0.4.4并优化数据库管理技能文档

- 更新版本号从0.4.2到0.4.4
- 优化API密钥权限管理说明，明确grant_api_key_permissions仅支持追加不支持撤销
- 新增add_sql_tool_to_datasource工具，提供一键创建SQL工具功能
- 调整create_sql_tool说明，强调需技能已存在
- 强化数据写操作安全机制，插入/更新/删除前必须预览并等待用户确认
- 完善导入数据功能说明，详细解释confirm_import_data参数传递方式
- 补充技能与工具管理流程，提供更清晰的操作指引
- 新增数字员工平台数据库技能配置指南文档
```
This commit is contained in:
yuanzhipeng
2026-06-26 16:21:41 +08:00
parent ba5cd4bbe1
commit 635313a7ab
43 changed files with 3464 additions and 686 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
lzwcai_mcpskills_generate_reports/README.md
View File

@@ -13,26 +13,55 @@ pip install -e .
## Python API
### 渲染文档
### 核心入口 `generate` / `scan_template`
```python
from lzwcai_mcpskills_generate_reports import generate
from lzwcai_mcpskills_generate_reports import generate, scan_template
# 扫描模板需要哪些占位符 / for / if 块
result = scan_template("./模板.docx") # 本地路径或 http/https URL
# 渲染
out_path = generate(
data="data.json", # dict 或 JSON 文件路径
template="./模板.docx", # 用户自己的 docx 模板路径
template="./模板.docx", # 本地路径或 http/https URL（自动下载）
out_path="_out/报价方案.docx",
style_ref="./用户样式.docx", # 可选：把用户模板的 theme/字体套过来
)
```
### 扫描模板占位符
### 便捷封装 `main` 模块
`main` 模块在核心入口之上做了两点增强，适合程序化调用：
1. `data` 除 dict / 本地 JSON 路径外，还支持 **JSON 文件 URL**（自动下载、用完即删）。
2. `out` **可省略**；省略时落到当前目录 `_out/`，文件名按 `模板名_时间戳.docx` 自动生成。
```python
from lzwcai_mcpskills_generate_reports import scan_template
from lzwcai_mcpskills_generate_reports.main import generate_report, scan_report
result = scan_template("./模板.docx")
print(result)
# data 传 dict，out 省略 -> 默认 _out/ 下自动命名
generate_report(template="./模板.docx", data={...})
# data 传本地 JSON 路径
generate_report(template="./模板.docx", data="data.json", out="_out/a.docx")
# template / data 都传 URL
generate_report(
template="https://host/模板.docx",
data="https://host/data.json",
)
# 扫描占位符，支持本地路径或 URL
scan_report(template="https://host/模板.docx")
```
`generate_report` 返回 `{"output": 输出文件绝对路径}`。
### 扫描结果结构
```python
# scan_template / scan_report 返回：
# {
# "placeholders": ["project_title", "contact_person", "equipments", ...],
# "blocks": [
@@ -43,30 +72,16 @@ print(result)
# }
```
## 命令行
```powershell
# 渲染
generate-report generate --template ./模板.docx --data data.json --out _out/报价方案.docx
# 扫描占位符
generate-report scan --template ./模板.docx
# 样式迁移
generate-report generate --template ./模板.docx --data data.json --style-ref ./用户样式.docx --out _out/报价方案_定制.docx
```
## MCP Server
本包同时提供 MCP Server（stdio 模式），把渲染引擎暴露成 3 个 MCP 工具：
本包同时提供 MCP Server（stdio 模式），把渲染引擎暴露成 2 个 MCP 工具：
| 工具 | 说明 | 必填参数 |
|------|------|----------|
| `generate_report` | 模板 + 数据 → 渲染输出 docx，返回输出文件绝对路径 | `template`, `data`, `out`（可选 `style_ref`） |
| `scan_template` | 扫描模板占位符与 for/if 块结构 | `template` |
| `validate_report_data` | 校验数据契约（不渲染） | `data` |
| 工具 | 说明 | 参数 |
|------|------|------|
| `generate_report` | 模板 + 数据 → 渲染输出 docx，返回输出文件绝对路径 | 必填 `template`、`data`；可选 `out`（省略落到 `_out/` 自动命名）、`style_ref` |
| `scan_template` | 扫描模板占位符与 for/if 块结构 | 必填 `template` |
其中 `data` 既可以传 JSON 对象，也可以传指向 JSON 文件的路径字符串。
其中 `data` 既可以传 JSON 对象，也可以传指向 JSON 文件的路径或 URL 字符串；`template` 支持本地路径或 http/https URL。数据契约校验由 `generate_report` 内部自动完成（不合法会报错）。
### 启动
@@ -74,8 +89,8 @@ generate-report generate --template ./模板.docx --data data.json --style-ref .
# 安装后用 console script 启动
lzwcai-mcpskills-generate-reports
# 或直接运行入口模块
python main.py
# 或以模块方式运行
python -m lzwcai_mcpskills_generate_reports.server
```
### MCP 客户端配置示例
@@ -92,6 +107,13 @@ python main.py
> stdio 模式下 stdout 被 MCP 协议占用，所有日志写入 `logs/` 目录与 stderr。
## 环境变量
| 变量 | 默认 | 说明 |
|------|------|------|
| `LOG_LEVEL` | `INFO` | 日志级别（DEBUG/INFO/WARNING/ERROR/CRITICAL）。 |
| `LZWCAI_INSECURE_SSL` | 关闭 | 设为 `1`/`true`/`yes` 时，下载模板/数据/图片**关闭 SSL 证书校验**。仅用于内网自签名证书等可信场景，生产慎用。 |
## 数据契约（QuoteData）
```json
@@ -117,7 +139,11 @@ python main.py
}
```
图片字段约定：路径/URL → 真实图；`""` → 默认占位图；`None` → 不显示。
- 必填顶层字段：`project_title`、`contact_person`、`contact_phone`、`requirements`。
- `requirements` 传列表会自动拼成多行字符串。
- `equipments[].index` 省略时自动从"四"起按中文数字补全（前三章固定为公司简介 / 客户要求 / 布局图）。
- `params[].v` 允许为 `0`、空串等合法假值；仅当键缺失或值为 `null` 才算缺失。
- 图片字段约定：路径/URL → 真实图；`""` → 默认占位图；`None` → 不显示。
## 目录结构
@@ -125,20 +151,23 @@ python main.py
lzwcai_mcpskills_generate_reports/
├── pyproject.toml
├── README.md
├── templates/ # 用户模板（示例，不在包内）
├── main.py # 使用本包的示例脚本（仓库根，非包内）
├── templates/ # 用户模板（示例，不在包内）
│ └── standard/
│ ├── template.docx
│ └── meta.json
├── samples/ # 示例数据（不在包内）
├── samples/ # 示例数据（不在包内）
│ └── sample_data.json
└── lzwcai_mcpskills_generate_reports/ # Python 包
├── __init__.py # 公共 API 入口
├── cli.py # 命令行
├── pipeline.py # 总入口
├── schema.py # 数据契约 + 校验器
├── render_quote.py # 渲染引擎
├── style_transfer.py # 样式迁移
└── template_scanner.py # 模板占位符扫描
└── lzwcai_mcpskills_generate_reports/ # Python 包
├── __init__.py # 公共 API 入口
├── main.py # 程序化便捷封装（URL data / out 可省略）
├── server.py # MCP Server (stdio)
├── pipeline.py # 总入口
├── schema.py # 数据契约 + 校验器
├── render_quote.py # 渲染引擎
├── style_transfer.py # 样式迁移
├── template_scanner.py # 模板占位符扫描
└── utils/ # 下载 / 日志等工具
```
## 模板约定
@@ -156,4 +185,4 @@ lzwcai_mcpskills_generate_reports/
}
```
- BUSINESS 逻辑（内置模板维护、模板市场等）交给包外的上层系统处理。
- BUSINESS 逻辑（内置模板维护、模板市场等）交给包外的上层系统处理。

10
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/__init__.py
View File

@@ -5,9 +5,10 @@ lzwcai-mcpskills-generate-reports
纯渲染引擎，不内置模板。
对外暴露的公共 API：
- generate: 数据 + 模板路径 -> docx（核心入口）
- scan_template: 模板路径 -> 占位符 JSON
- scan_template: 返回 QuoteData 数据契约结构（该传哪些字段、嵌套结构）
- validate: 校验数据契约
- normalize: 归一化数据
- describe: 返回数据契约结构（scan_template 的底层实现）
- transplant_style: 将用户模板样式迁移到结果文档
"""
@@ -15,15 +16,20 @@ __version__ = "0.1.0"
from .pipeline import generate
from .template_scanner import scan_template
from .schema import validate, normalize, DEFAULTS
from .schema import validate, normalize, describe, DEFAULTS
from .style_transfer import transplant_style
from .utils.fetch import is_url, download_to_temp, local_file
__all__ = [
"generate",
"scan_template",
"validate",
"normalize",
"describe",
"transplant_style",
"is_url",
"download_to_temp",
"local_file",
"DEFAULTS",
"__version__",
]

BIN
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/__pycache__/pipeline.cpython-312.pyc
View File

Binary file not shown.

BIN
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/__pycache__/render_quote.cpython-312.pyc
View File

Binary file not shown.

BIN
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/__pycache__/schema.cpython-312.pyc
View File

Binary file not shown.

BIN
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/__pycache__/style_transfer.cpython-312.pyc
View File

Binary file not shown.

122
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/cli.py
View File

@@ -1,122 +0,0 @@
# -*- coding: utf-8 -*-
"""
cli.py：公共入口。
程序调用（推荐）:
from lzwcai_mcpskills_generate_reports.cli import generate_report, scan_report
generate_report(
template="./模板.docx",
data={"project_name": "x", ...},
out="_out/a.docx",
)
scan_report(template="./模板.docx")
命令行:
generate-report generate --template ./模板.docx --data data.json --out _out/a.docx
generate-report scan --template ./模板.docx
"""
import argparse
import json
import os
import sys
try:
sys.stdout.reconfigure(encoding="utf-8")
except (AttributeError, OSError):
pass
from lzwcai_mcpskills_generate_reports import generate, scan_template
def _load_data(data):
"""把 data 归一化为 dict：支持 dict 或 JSON 文件路径字符串。"""
if isinstance(data, str):
if not os.path.isfile(data):
raise FileNotFoundError(f"数据文件不存在: {data}")
with open(data, "r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise TypeError(f"data 必须是 dict 或 JSON 文件路径字符串，实际类型: {type(data).__name__}")
return data
def generate_report(template, data, out, style_ref=None):
"""生成 docx 报告。
参数:
template: 模板 docx 文件路径
data: dict 数据 或 JSON 文件路径字符串
out: 输出 docx 文件路径
style_ref: 用户样式参考 docx 路径（可选）
返回:
dict: {"output": 输出文件绝对路径}
"""
data = _load_data(data)
out = generate(data=data, template=template, out_path=out, style_ref=style_ref)
return {"output": out}
def scan_report(template):
"""扫描模板占位符。
参数:
template: 模板 docx 文件路径
返回:
dict: 占位符扫描结果
"""
return scan_template(template)
def _build_arg_parser():
parser = argparse.ArgumentParser(
description="docx 模板渲染与占位符扫描",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
generate-report generate --template ./模板.docx --data data.json --out _out/a.docx
generate-report generate --template ./模板.docx --data data.json --style-ref 用户样式.docx --out _out/b.docx
generate-report scan --template ./模板.docx
""",
)
sub = parser.add_subparsers(dest="command", required=True)
gen_parser = sub.add_parser("generate", help="渲染生成 docx")
gen_parser.add_argument("--template", required=True, help="模板 docx 文件路径")
gen_parser.add_argument("--data", required=True, help="数据 JSON 文件路径")
gen_parser.add_argument("--out", required=True, help="输出 docx 文件路径")
gen_parser.add_argument("--style-ref", default=None, help="用户上传的样式参考 docx（可选）")
scan_parser = sub.add_parser("scan", help="扫描模板占位符")
scan_parser.add_argument("--template", required=True, help="模板 docx 文件路径")
return parser
def main():
"""命令行入口（console_scripts 调用）。"""
parser = _build_arg_parser()
args = parser.parse_args()
try:
if args.command == "generate":
result = generate_report(
template=args.template,
data=args.data,
out=args.out,
style_ref=args.style_ref,
)
print(f"生成成功: {result['output']}")
elif args.command == "scan":
result = scan_report(template=args.template)
print(json.dumps(result, ensure_ascii=False, indent=2))
except Exception as e:
print(f"错误: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()

276
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/logs/lzwcai_mcpskills_generate_reports.log
View File

File diff suppressed because one or more lines are too long

78
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/logs/lzwcai_mcpskills_generate_reports_error.log
View File

@@ -0,0 +1,78 @@
2026-06-22 22:30:08 - lzwcai_mcpskills_generate_reports.server - ERROR - [server.py:166] - 工具执行失败: generate_report: data 看起来是 JSON 内容但解析失败: Expecting value: line 1 column 1019 (char 1018)
Traceback (most recent call last):
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\main.py", line 55, in _load_data
obj = json.loads(s)
^^^^^^^^^^^^^
File "D:\anaconda3\Lib\json\__init__.py", line 346, in loads
return _default_decoder.decode(s)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\anaconda3\Lib\json\decoder.py", line 337, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\anaconda3\Lib\json\decoder.py", line 355, in raw_decode
raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1019 (char 1018)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\server.py", line 163, in handle_call_tool
result = await anyio.to_thread.run_sync(handler, args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\anaconda3\Lib\site-packages\anyio\to_thread.py", line 63, in run_sync
return await get_async_backend().run_sync_in_worker_thread(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\anaconda3\Lib\site-packages\anyio\_backends\_asyncio.py", line 2502, in run_sync_in_worker_thread
return await future
^^^^^^^^^^^^
File "D:\anaconda3\Lib\site-packages\anyio\_backends\_asyncio.py", line 986, in run
result = context.run(func, *args)
^^^^^^^^^^^^^^^^^^^^^^^^
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\server.py", line 120, in _do_generate_report
return _generate_report(
^^^^^^^^^^^^^^^^^
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\main.py", line 99, in generate_report
data = _load_data(data)
^^^^^^^^^^^^^^^^
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\main.py", line 57, in _load_data
raise ValueError(f"data 看起来是 JSON 内容但解析失败: {e}")
ValueError: data 看起来是 JSON 内容但解析失败: Expecting value: line 1 column 1019 (char 1018)
2026-06-22 22:30:32 - lzwcai_mcpskills_generate_reports.server - ERROR - [server.py:166] - 工具执行失败: generate_report: data 看起来是 JSON 内容但解析失败: Expecting value: line 1 column 1012 (char 1011)
Traceback (most recent call last):
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\main.py", line 55, in _load_data
obj = json.loads(s)
^^^^^^^^^^^^^
File "D:\anaconda3\Lib\json\__init__.py", line 346, in loads
return _default_decoder.decode(s)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\anaconda3\Lib\json\decoder.py", line 337, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\anaconda3\Lib\json\decoder.py", line 355, in raw_decode
raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1012 (char 1011)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\server.py", line 163, in handle_call_tool
result = await anyio.to_thread.run_sync(handler, args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\anaconda3\Lib\site-packages\anyio\to_thread.py", line 63, in run_sync
return await get_async_backend().run_sync_in_worker_thread(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\anaconda3\Lib\site-packages\anyio\_backends\_asyncio.py", line 2502, in run_sync_in_worker_thread
return await future
^^^^^^^^^^^^
File "D:\anaconda3\Lib\site-packages\anyio\_backends\_asyncio.py", line 986, in run
result = context.run(func, *args)
^^^^^^^^^^^^^^^^^^^^^^^^
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\server.py", line 120, in _do_generate_report
return _generate_report(
^^^^^^^^^^^^^^^^^
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\main.py", line 99, in generate_report
data = _load_data(data)
^^^^^^^^^^^^^^^^
File "E:\yh-ai\project\lzwcai-szyg\lzwcai-mcp-server-package\lzwcai_mcpskills_generate_reports\lzwcai_mcpskills_generate_reports\main.py", line 57, in _load_data
raise ValueError(f"data 看起来是 JSON 内容但解析失败: {e}")
ValueError: data 看起来是 JSON 内容但解析失败: Expecting value: line 1 column 1012 (char 1011)

115
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/main.py Normal file
View File

@@ -0,0 +1,115 @@
# -*- coding: utf-8 -*-
"""
main.py：程序化调用入口（非 CLI、非 MCP server）。
两点特性：
1. template / data 既可传本地文件路径，也可传 http/https URL（自动下载）。
2. 输出路径 out 可省略；省略时落到当前目录下的 _out/，
文件名按 模板名 + 时间戳 自动生成。
用法:
from lzwcai_mcpskills_generate_reports.main import generate_report, scan_report
# data 传 dict
generate_report(template="./模板.docx", data={...})
# data 传本地 JSON 路径，out 省略
generate_report(template="./模板.docx", data="data.json")
# template / data 都传 URL
generate_report(
template="https://host/模板.docx",
data="https://host/data.json",
out="_out/a.docx",
)
# 扫描占位符，支持本地路径或 URL
scan_report(template="https://host/模板.docx")
"""
import json
import os
import sys
import time
try:
sys.stdout.reconfigure(encoding="utf-8")
except (AttributeError, OSError):
pass
from . import generate, scan_template
from .utils.fetch import is_url, local_file
def _load_data(data):
"""把 data 归一化为 dict：支持 dict、JSON 内容字符串、本地 JSON 文件路径、或 JSON 文件 URL。
字符串的判定顺序：
1. 以 '{' 开头 -> 当作 JSON 内容直接解析；
2. 否则尝试 JSON 解析（捕获失败则当作路径）；
3. 本地路径 / URL（URL 会下载到临时文件读取，用完即删）。
"""
if isinstance(data, dict):
return data
if isinstance(data, str):
s = data.strip()
# 1) 看起来就是 JSON 对象内容（路径/URL 不会以 '{' 开头）
if s.startswith("{"):
try:
obj = json.loads(s)
except json.JSONDecodeError as e:
raise ValueError(f"data 看起来是 JSON 内容但解析失败: {e}")
if not isinstance(obj, dict):
raise TypeError("data 为 JSON 内容时必须是对象（dict）")
return obj
# 2) 尝试当作 JSON 字符串解析（捕获失败说明是路径）
try:
obj = json.loads(s)
if isinstance(obj, dict):
return obj
except (json.JSONDecodeError, ValueError):
pass
# 3) 本地路径 / URL
with local_file(data, suffix=".json") as path:
if not is_url(data) and not os.path.isfile(path):
raise FileNotFoundError(f"数据文件不存在: {data}")
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
raise TypeError(
f"data 必须是 dict、JSON 内容/文件路径/URL 字符串，实际类型: {type(data).__name__}"
)
def _default_out_path(template):
"""out 省略时的默认输出路径：<cwd>/_out/<模板名>_<时间戳>.docx。"""
name = os.path.basename(template.split("?")[0]) # 去掉 URL 查询串
base = os.path.splitext(name)[0] or "report"
stamp = time.strftime("%Y%m%d_%H%M%S")
return os.path.join(os.getcwd(), "_out", f"{base}_{stamp}.docx")
def generate_report(template, data, out=None, style_ref=None):
"""生成 docx 报告。
参数:
template: 模板 docx 文件路径，或 http/https URL（自动下载）
data: dict、JSON 文件路径，或 JSON 文件 URL（自动下载）
out: 输出 docx 文件路径；省略则用默认路径（_out/ 下按模板名+时间戳命名）
style_ref: 用户样式参考 docx 路径或 URL（可选）
返回:
dict: {"output": 输出文件绝对路径}
"""
data = _load_data(data)
if not out:
out = _default_out_path(template)
out_path = generate(data=data, template=template, out_path=out, style_ref=style_ref)
return {"output": out_path}
def scan_report(template=None):
"""返回 QuoteData 数据契约结构（该传哪些字段、是否必填、嵌套结构）。
参数:
template: 已忽略，保留兼容旧签名。
返回:
dict: 数据契约，见 schema.describe()。
"""
return scan_template(template)

40
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/pipeline.py
View File

@@ -7,10 +7,18 @@ pipeline.py：总入口 — 串起 数据校验 -> 渲染 -> (可选)样式迁
"""
import os
import json
import contextlib
from .schema import validate, normalize
from .render_quote import render
from .style_transfer import transplant_style
from .utils.fetch import is_url, local_file
@contextlib.contextmanager
def _noop():
"""style_ref 为空时占位用的空上下文，产出 None。"""
yield None
def _load_data(data):
@@ -40,9 +48,9 @@ def generate(data, template, out_path, style_ref=None):
参数:
data: QuoteData dict（或 JSON 文件路径字符串）
template: 模板 docx 文件路径
template: 模板 docx 文件路径，或 http/https URL（自动下载）
out_path: 输出 docx 路径
style_ref: 用户上传的样式参考 docx 路径（可选）
style_ref: 用户上传的样式参考 docx 路径，或 URL（可选，自动下载）
返回:
生成的 docx 绝对路径
@@ -50,10 +58,7 @@ def generate(data, template, out_path, style_ref=None):
data = _load_data(data)
if not isinstance(template, str):
raise TypeError(f"template 必须是文件路径字符串，实际类型: {type(template).__name__}")
if not os.path.isfile(template):
raise FileNotFoundError(f"模板文件不存在: {template}")
template_path = os.path.abspath(template)
raise TypeError(f"template 必须是文件路径或 URL 字符串，实际类型: {type(template).__name__}")
# 归一化 + 校验
data = normalize(data)
@@ -66,14 +71,21 @@ def generate(data, template, out_path, style_ref=None):
if out_dir:
os.makedirs(out_dir, exist_ok=True)
# 渲染（优先读取同目录 meta.json 作为图片配置）
meta = _load_meta_for_template(template_path)
render(data, out_path, template_path, meta=meta)
# 把 template / style_ref 统一解析成本地路径（URL 会下载到临时文件，用完即删）
with local_file(template) as template_path, \
(local_file(style_ref) if style_ref else _noop()) as style_path:
if not is_url(template) and not os.path.isfile(template_path):
raise FileNotFoundError(f"模板文件不存在: {template}")
template_path = os.path.abspath(template_path)
# 可选样式迁移
if style_ref:
if not os.path.isfile(style_ref):
raise FileNotFoundError(f"样式参考文件不存在: {style_ref}")
transplant_style(out_path, style_ref, out_path)
# 渲染（优先读取同目录 meta.json 作为图片配置；URL 模板无同目录 meta，则为空）
meta = _load_meta_for_template(template_path)
render(data, out_path, template_path, meta=meta)
# 可选样式迁移
if style_ref:
if not is_url(style_ref) and not os.path.isfile(style_path):
raise FileNotFoundError(f"样式参考文件不存在: {style_ref}")
transplant_style(out_path, style_path, out_path)
return os.path.abspath(out_path)

7
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/render_quote.py
View File

@@ -8,7 +8,6 @@ render_quote.py：渲染引擎 — data + 模板路径 -> docx。
"""
import copy
import os
import ssl
import sys
import tempfile
import urllib.request
@@ -22,6 +21,8 @@ except (AttributeError, OSError):
from docxtpl import DocxTemplate, InlineImage
from docx.shared import Mm
from .utils.fetch import make_ssl_context
def _resolve_image_path(src, tmp_files):
"""把图片字段值解析为本地文件路径。
@@ -30,9 +31,7 @@ def _resolve_image_path(src, tmp_files):
"""
def _download(url):
try:
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
ctx = make_ssl_context()
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
data = urllib.request.urlopen(req, context=ctx, timeout=30).read()
ext = os.path.splitext(url.split("?")[0])[1] or ".png"

75
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/schema.py
View File

@@ -64,8 +64,11 @@ def validate(data):
if not isinstance(p, dict):
errors.append(f"equipments[{i}].params[{j}] 必须为对象")
continue
if not p.get("k") or not p.get("v"):
errors.append(f"equipments[{i}].params[{j}] 缺少 k 或 v")
# 用"键是否存在/为空"判断，避免把合法假值（0、False、空串当 v）误判为缺失
if not p.get("k"):
errors.append(f"equipments[{i}].params[{j}] 缺少 k")
if "v" not in p or p.get("v") is None:
errors.append(f"equipments[{i}].params[{j}] 缺少 v")
# quote_items
items = data.get("quote_items", [])
@@ -121,3 +124,71 @@ def normalize(data):
d["section_after_sales"] = _CN_DIGITS[after_sales_idx - 1] if after_sales_idx <= len(_CN_DIGITS) else str(after_sales_idx)
return d
def describe():
"""返回 QuoteData 数据契约结构（该传哪些字段、是否必填、嵌套结构）。
与 validate / normalize 保持一致，是「调用方该如何组织数据」的权威说明，
可直接对照 samples/sample_data.json。返回新副本，调用方可安全修改。
"""
return {
"type": "QuoteData",
"required": ["project_title", "contact_person", "contact_phone", "requirements"],
"fields": {
"project_title": {"type": "string", "required": True, "desc": "项目/方案标题"},
"contact_person": {"type": "string", "required": True, "desc": "联系人"},
"contact_phone": {"type": "string", "required": True, "desc": "联系电话"},
"contact_company": {"type": "string", "required": False, "desc": "客户公司名（模板用到才生效）"},
"requirements": {
"type": "string | list[string]",
"required": True,
"desc": "客户要求；传列表会自动拼成多行字符串",
},
"layout_image": {"type": "string", "required": False, "desc": "整线布局图，本地路径或 URL；空串=占位图，None=不显示"},
"layout_title": {"type": "string", "required": False, "default": DEFAULTS["layout_title"]},
"show_layout": {"type": "bool", "required": False, "default": DEFAULTS["show_layout"]},
"equipments": {
"type": "list",
"required": False,
"desc": "设备清单",
"item": {
"name": {"type": "string", "required": True, "desc": "设备名称"},
"index": {"type": "string", "required": False, "desc": "章节序号，缺省自动按中文数字（四、五…）补全"},
"images": {"type": "list[string]", "required": False, "desc": "设备图，路径或 URL 列表；缺省为 ['']"},
"features": {
"type": "list",
"required": False,
"item": {
"title": {"type": "string", "required": True, "desc": "特点标题"},
"lines": {"type": "list[string]", "required": True, "desc": "特点说明，多行"},
},
},
"params": {
"type": "list",
"required": False,
"item": {
"k": {"type": "string", "required": True, "desc": "参数名"},
"v": {"type": "string | number", "required": True, "desc": "参数值，允许 0/空串等合法假值"},
},
},
},
},
"quote_items": {
"type": "list",
"required": False,
"desc": "报价表条目",
"item": {
"name": {"type": "string", "required": True, "desc": "条目名称"},
"qty": {"type": "string", "required": False, "desc": "数量，如 '1套'"},
"image": {"type": "string", "required": False, "desc": "条目图，路径或 URL；缺省为空串"},
"desc": {"type": "string", "required": False, "desc": "条目说明"},
"price": {"type": "string", "required": False, "desc": "价格，如 '面议'"},
},
},
},
"auto_generated": {
"section_quote_table": "报价表章节序号，按设备数量自动计算，无需提供",
"section_after_sales": "售后服务章节序号，按设备数量自动计算，无需提供",
},
}

120
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/server.py
View File

@@ -2,14 +2,12 @@
"""
lzwcai-mcpskills-generate-reports MCP Server
把 docx 模板渲染引擎封装成 MCP 工具，提供三个工具：
- generate_report: 数据 + 模板路径 -> 渲染输出 docx
- scan_template: 扫描模板占位符 / for / if 块
- validate_report_data: 校验数据契约（不渲染）
把 docx 模板渲染引擎封装成 MCP 工具，提供两个工具：
- generate_report: 数据 + 模板路径 -> 渲染输出 docx
- scan_template: 返回 QuoteData 数据契约结构（该传哪些字段、嵌套结构）
stdio 模式运行；所有日志走 stderr，stdout 留给 MCP 协议。
"""
import os
import json
import logging
@@ -22,13 +20,15 @@ from mcp.server.stdio import stdio_server
try:
from .utils.logger_config import setup_system_logging, get_logger
from . import generate, scan_template, validate, normalize
from . import scan_template
from .main import generate_report as _generate_report
except ImportError:
from lzwcai_mcpskills_generate_reports.utils.logger_config import (
setup_system_logging, get_logger,
)
from lzwcai_mcpskills_generate_reports import (
generate, scan_template, validate, normalize,
from lzwcai_mcpskills_generate_reports import scan_template
from lzwcai_mcpskills_generate_reports.main import (
generate_report as _generate_report,
)
# 初始化日志系统
@@ -39,22 +39,8 @@ logger = get_logger(__name__)
server = Server("lzwcai_mcpskills_generate_reports")
def _load_data(data):
"""把 data 归一化为 dict：支持 dict 或 JSON 文件路径字符串。"""
if isinstance(data, str):
if not os.path.isfile(data):
raise FileNotFoundError(f"数据文件不存在: {data}")
with open(data, "r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise TypeError(
f"data 必须是对象或 JSON 文件路径字符串，实际类型: {type(data).__name__}"
)
return data
# ── 工具定义 ──────────────────────────────────────────────
_DATA_DESC = "报价数据：可以是 JSON 对象，或指向 JSON 文件的路径字符串"
_DATA_DESC = "报价数据：可以是 JSON 对象，或指向 JSON 文件的路径/URL 字符串"
TOOL_DEFS = [
types.Tool(
@@ -68,56 +54,54 @@ TOOL_DEFS = [
"properties": {
"template": {
"type": "string",
"description": "模板 docx 文件路径（必填）",
"description": "模板 docx 文件路径，或 http/https URL（会自动下载）（必填）",
},
"data": {
"type": ["object", "string"],
"type": "string",
"description": _DATA_DESC + "（必填）",
},
"out": {
"type": "string",
"description": "输出 docx 文件路径（必填）",
"description": "输出 docx 文件路径（可选）；省略则落到当前目录 _out/，按 模板名_时间戳.docx 自动命名",
},
"style_ref": {
"type": "string",
"description": "用户上传的样式参考 docx 路径（可选），会把其 theme/字体套到结果文档",
"description": "用户上传的样式参考 docx 路径或 URL（可选），会把其 theme/字体套到结果文档",
},
},
"required": ["template", "data", "out"],
"required": ["template", "data"],
},
# 输出契约：成功返回 {"output": 路径}；失败返回 {"error":..., "tool_name":...}。
# MCP 要求 outputSchema 顶层必须是 type:"object"，所以 oneOf 只用来约束
# 两种 required 组合，否则出错时 structuredContent 会过不了校验。
outputSchema={
"type": "object",
"properties": {
"output": {"type": "string", "description": "输出 docx 文件绝对路径"},
"error": {"type": "string", "description": "错误信息"},
"tool_name": {"type": "string"},
},
"oneOf": [
{"required": ["output"]},
{"required": ["error", "tool_name"]},
],
},
),
types.Tool(
name="scan_template",
description=(
"扫描 docx 模板中的 Jinja2 占位符，返回需要外部提供的顶层变量列表和 for/if 块结构。"
"用于在渲染前了解模板需要哪些数据字段。"
"返回 QuoteData 数据契约结构：该传哪些字段、是否必填、equipments/features/params/"
"quote_items 等嵌套结构，以及自动生成无需提供的字段。用于在渲染前了解数据该如何组织。"
),
inputSchema={
"type": "object",
"properties": {
"template": {
"type": "string",
"description": "模板 docx 文件路径（必填）",
"description": "（已忽略，保留兼容）模板路径或 URL；所有模板共用同一数据契约",
},
},
"required": ["template"],
},
),
types.Tool(
name="validate_report_data",
description=(
"校验报价数据是否符合契约（必填字段、equipments/quote_items/features/params 结构），"
"不渲染文档。返回校验结果和错误列表。"
),
inputSchema={
"type": "object",
"properties": {
"data": {
"type": ["object", "string"],
"description": _DATA_DESC + "（必填）",
},
},
"required": ["data"],
"required": [],
},
),
]
@@ -132,31 +116,22 @@ async def handle_list_tools() -> list[types.Tool]:
# ── 工具实现（同步函数，放线程池执行）──────────────────────
def _do_generate_report(arguments: dict) -> dict:
template = arguments["template"]
data = arguments["data"]
out = arguments["out"]
style_ref = arguments.get("style_ref")
data = _load_data(data)
out_path = generate(data=data, template=template, out_path=out, style_ref=style_ref)
return {"output": out_path}
# 复用 main.generate_report：data 支持 URL，out 可省略（落到 _out/ 自动命名）
return _generate_report(
template=arguments["template"],
data=arguments["data"],
out=arguments.get("out"),
style_ref=arguments.get("style_ref"),
)
def _do_scan_template(arguments: dict) -> dict:
return scan_template(arguments["template"])
def _do_validate(arguments: dict) -> dict:
data = _load_data(arguments["data"])
norm = normalize(data)
errors = validate(norm)
return {"valid": not errors, "errors": errors}
return scan_template(arguments.get("template"))
_HANDLERS = {
"generate_report": _do_generate_report,
"scan_template": _do_scan_template,
"validate_report_data": _do_validate,
}
@@ -164,8 +139,14 @@ _HANDLERS = {
async def handle_call_tool(
name: str,
arguments: dict | None,
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""调用工具"""
) -> tuple[list[types.TextContent], dict]:
"""调用工具。
返回 (content, structured) 二元组：
- structured: 结构化结果 dict，SDK 自动填入响应的 structuredContent，
调用方可直接取值，无需再对 content[].text 做 json.loads。
- content: 序列化后的 TextContent，保留对老客户端的向后兼容。
"""
logger.info(
f"收到 CallTool 请求: name={name}, "
f"arguments={json.dumps(arguments, ensure_ascii=False) if arguments else 'None'}"
@@ -185,12 +166,13 @@ async def handle_call_tool(
logger.error(f"工具执行失败: {name}: {e}", exc_info=True)
result = {"error": str(e), "tool_name": name}
return [
content = [
types.TextContent(
type="text",
text=json.dumps(result, ensure_ascii=False, indent=2),
)
]
return content, result
async def run_server():

136
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/template_scanner.py
View File

@@ -1,133 +1,23 @@
# -*- coding: utf-8 -*-
"""
template_scanner.py：扫描 docx 模板中的 Jinja2 占位符。
template_scanner.py：返回 QuoteData 数据契约结构。
只读模板，不渲染；返回模板里要求外部提供的数据字段清单。
历史上本模块扫描 docx 模板里的 Jinja2 占位符；现已改为直接返回
schema.describe() 的数据契约（该传哪些字段、是否必填、嵌套结构），
更直观、可直接对照 samples/sample_data.json。
保留 scan_template 这个名字与 template 参数，向后兼容既有调用方。
"""
import re
import zipfile
import xml.etree.ElementTree as ET
from jinja2 import Environment, meta
from jinja2 import nodes as jinja_nodes
from .schema import describe
def _iter_docx_text(docx_path):
"""遍历 docx 中所有 XML 文本节点，产出原始字符串片段。"""
with zipfile.ZipFile(docx_path, "r") as z:
for name in z.namelist():
# 只关心 word 主文档、页眉、页脚
if not (name.startswith("word/document") or name.startswith("word/header") or name.startswith("word/footer")):
continue
data = z.read(name)
try:
root = ET.fromstring(data)
except ET.ParseError:
continue
# w:t 节点存放文本
for elem in root.iter():
if elem.tag.endswith("}t"):
if elem.text:
yield elem.text
def scan_template(template_path=None):
"""返回 QuoteData 数据契约结构。
def _extract_source(docx_path):
"""把 docx 所有文本片段拼成一段连续的源文本，便于 Jinja2 解析。"""
return "\n".join(_iter_docx_text(docx_path))
def _normalize_docxtpl_tags(source):
"""把 docxtpl 段落/行/单元格级标签 {%p ... %} {%tr ... %} {%tc ... %} 还原为 {% ... %}。
docxtpl 用 {%p、{%tr、{%tc 控制块作用于段落、表格行、表格单元格；
扫描占位符时不需要这些粒度信息，统一成标准 Jinja2 标签即可解析。
"""
return re.sub(r"{%\s*(?:p|tr|tc)\s+", "{% ", source)
def _walk_blocks(node, result=None):
"""遍历 Jinja2 AST，收集 For / If 块信息。"""
if result is None:
result = []
if node is None:
return result
if isinstance(node, jinja_nodes.For):
iter_name = _expression_name(node.iter)
target_name = _expression_name(node.target)
result.append({
"type": "for",
"iterator": target_name,
"variable": iter_name,
})
for child in node.body:
_walk_blocks(child, result)
for child in node.else_ or []:
_walk_blocks(child, result)
return result
if isinstance(node, jinja_nodes.If):
test_name = _expression_name(node.test)
result.append({
"type": "if",
"condition": test_name,
})
for child in node.body:
_walk_blocks(child, result)
for child in node.else_ or []:
_walk_blocks(child, result)
return result
if hasattr(node, "body"):
for child in node.body:
_walk_blocks(child, result)
if hasattr(node, "else_") and node.else_:
for child in node.else_:
_walk_blocks(child, result)
return result
def _expression_name(expr):
"""把 Jinja2 表达式尽量还原为可读的字符串。"""
if expr is None:
return None
if isinstance(expr, jinja_nodes.Name):
return expr.name
if isinstance(expr, jinja_nodes.Const):
return str(expr.value)
if isinstance(expr, jinja_nodes.Getattr):
return f"{_expression_name(expr.node)}.{expr.attr}"
if isinstance(expr, jinja_nodes.Getitem):
return f"{_expression_name(expr.node)}[{_expression_name(expr.arg)}]"
return str(expr)
def scan_template(template_path):
"""扫描 docx 模板，返回占位符信息 JSON。
参数:
template_path: 兼容旧签名，已忽略（所有模板共用同一数据契约）。
返回:
{
"placeholders": ["project_title", "equipments", ...], # 需要外部提供的最顶层变量
"blocks": [
{"type": "for", "iterator": "eq", "variable": "equipments"},
{"type": "if", "condition": "show_layout"},
...
]
}
dict: 数据契约，见 schema.describe()。
"""
source = _normalize_docxtpl_tags(_extract_source(template_path))
if not source.strip():
return {"placeholders": [], "blocks": []}
env = Environment()
ast = env.parse(source)
variables = sorted(meta.find_undeclared_variables(ast))
blocks = _walk_blocks(ast)
return {
"placeholders": variables,
"blocks": blocks,
}
return describe()

77
lzwcai_mcpskills_generate_reports/lzwcai_mcpskills_generate_reports/utils/fetch.py Normal file
View File

@@ -0,0 +1,77 @@
# -*- coding: utf-8 -*-
"""
fetch.py：把远程文件地址（http/https）下载成本地临时文件。
用于让 generate / scan_template 既能接收本地路径，也能直接接收一个 URL，
下载后按本地文件处理，用完即删。
"""
import contextlib
import os
import ssl
import tempfile
import urllib.request
def is_url(s):
"""判断字符串是否是 http/https URL。"""
return isinstance(s, str) and s.lower().startswith(("http://", "https://"))
def make_ssl_context():
"""构造下载用的 SSL 上下文。
默认开启证书校验（安全）。仅当环境变量 LZWCAI_INSECURE_SSL 设为
1/true/yes 时才关闭校验，用于内网自签名证书等可信场景。
"""
ctx = ssl.create_default_context()
if os.environ.get("LZWCAI_INSECURE_SSL", "").lower() in ("1", "true", "yes"):
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
return ctx
def download_to_temp(url, suffix=".docx"):
"""下载 URL 内容到临时文件，返回本地路径。
参数:
url: 远程文件地址（http/https）
suffix: 当 URL 末尾无扩展名时使用的默认后缀
返回:
本地临时文件绝对路径（调用方负责删除）
"""
ctx = make_ssl_context()
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
data = urllib.request.urlopen(req, context=ctx, timeout=60).read()
ext = os.path.splitext(url.split("?")[0])[1] or suffix
fd, path = tempfile.mkstemp(suffix=ext)
try:
os.write(fd, data)
finally:
os.close(fd)
return path
@contextlib.contextmanager
def local_file(path_or_url, suffix=".docx"):
"""统一把"本地路径或 URL"解析成本地路径的上下文管理器。
- 传入本地路径：原样产出，退出时不删除。
- 传入 URL：下载到临时文件并产出其路径，退出时自动删除。
用法:
with local_file(template) as path:
... 用 path 读模板 ...
"""
if is_url(path_or_url):
tmp = download_to_temp(path_or_url, suffix=suffix)
try:
yield tmp
finally:
try:
os.remove(tmp)
except OSError:
pass
else:
yield path_or_url

12
lzwcai_mcpskills_generate_reports/main.py
View File

@@ -1,9 +1,15 @@
# -*- coding: utf-8 -*-
"""
Entry point for lzwcai-mcpskills-generate-reports
main.py：启动 lzwcai-mcpskills-generate-reports MCP Server (stdio 模式)。
Runs the MCP server (stdio mode) for docx report generation.
运行:
python main.py
等价于:
python -m lzwcai_mcpskills_generate_reports.server
stdio 模式下 stdout 被 MCP 协议占用，所有日志写入 logs/ 目录与 stderr。
"""
from lzwcai_mcpskills_generate_reports.server import main
if __name__ == "__main__":

4
lzwcai_mcpskills_generate_reports/pyproject.toml
View File

@@ -4,10 +4,11 @@ build-backend = "setuptools.build_meta"
[project]
name = "lzwcai-mcpskills-generate-reports"
version = "0.1.0"
version = "0.1.3"
description = "Render styled quotation documents from user-supplied docx templates and structured data"
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
keywords = ["docx", "quotation", "report", "template", "jinja2"]
authors = [
{ name = "LzwCai", email = "lzwcai@example.com" },
@@ -27,7 +28,6 @@ dependencies = [
]
[project.scripts]
generate-report = "lzwcai_mcpskills_generate_reports.cli:main"
lzwcai-mcpskills-generate-reports = "lzwcai_mcpskills_generate_reports.server:main"
[project.urls]