1
0
forked from erp-dev/erp

feat: wecomm support image link

This commit is contained in:
2026-01-23 21:11:13 +08:00
parent 81640cb84b
commit 4d33bc7f68
4 changed files with 99 additions and 8 deletions

2
.env
View File

@@ -7,3 +7,5 @@ DB_NAME=flower
DB_USER=postgres DB_USER=postgres
DB_PASSWORD=postgres DB_PASSWORD=postgres
PRINTING_SALES_ITEM_SOURCE_STATE_ID=2 PRINTING_SALES_ITEM_SOURCE_STATE_ID=2
WECOM_WEBHOOK_KEY=
PRINTING_JOB_STATE_ADVANCED_FOLLOWUP_URL_TEMPLATE=https://app.yuwen.cloud/workstation/production/batch-advance?orderId={order_id}

View File

@@ -10,16 +10,17 @@
- `api_v1/utils/wecom_webhook.py` - `api_v1/utils/wecom_webhook.py`
- 提供 `send_wecom_webhook_message(content=..., msgtype=..., key=..., timeout_seconds=...)` - 提供 `send_wecom_webhook_message(content=..., msgtype=..., key=..., timeout_seconds=...)`
- `msgtype` 支持:`text`(默认)、`markdown` - `msgtype` 支持:`text`(默认)、`markdown`
- command - command(通用测试发送)
- `api_v1/management/commands/wecom_webhook_send_text.py` - `api_v1/management/commands/wecom_webhook_send_text.py`
- 用法示例 - 用法:
- text默认
- `uv run python manage.py wecom_webhook_send_text "hello world"` - `uv run python manage.py wecom_webhook_send_text "hello world"`
- 发送 markdown`uv run python manage.py wecom_webhook_send_text "**bold**" --msgtype markdown` - markdown
- 可选覆盖 key`uv run python manage.py wecom_webhook_send_text "hello" --key xxx` - `uv run python manage.py wecom_webhook_send_text "**bold**\n> hello" --msgtype markdown`
- 参数: - 参数:
- `text`:必填,消息内容 - `text`:必填,消息内容
- `--msgtype`:可选,`text`(默认)或 `markdown` - `--msgtype`:可选,`text`(默认)或 `markdown`
- `--key`:可选,临时覆盖 `settings.WECOM_WEBHOOK_KEY` - `--key`:可选,临时覆盖 `.env``WECOM_WEBHOOK_KEY`
- `--timeout`可选HTTP 超时秒数(默认 10 - `--timeout`可选HTTP 超时秒数(默认 10
#### 2) 新增PrintingJob 每次状态推进state_advanced发送企业微信 markdown 通知 #### 2) 新增PrintingJob 每次状态推进state_advanced发送企业微信 markdown 通知
@@ -44,6 +45,7 @@
- 模板占位符: - 模板占位符:
- `{followup_line}`(由 service 渲染:有 url 才会输出“跟进地址”行) - `{followup_line}`(由 service 渲染:有 url 才会输出“跟进地址”行)
- 模板已增加字段:`{process_params_markdown}`(工序参数,来自 `state_log.get_all_parameters_summary()` - 模板已增加字段:`{process_params_markdown}`(工序参数,来自 `state_log.get_all_parameters_summary()`
- 工序参数渲染增强:当参数值是 http/https URL如图片链接自动渲染为 markdown Link`[url](url)`,便于点击查看
#### 3) 重构:提取“发送最新状态到企业微信”逻辑为可复用 service便于后续 API 化) #### 3) 重构:提取“发送最新状态到企业微信”逻辑为可复用 service便于后续 API 化)
@@ -52,3 +54,40 @@
- `render_printing_job_state_markdown(...)` - `render_printing_job_state_markdown(...)`
- command`printing/management/commands/wecom_notify_printing_job_latest_state.py` - command`printing/management/commands/wecom_notify_printing_job_latest_state.py`
- 发送者强制为“系统自动发送”(后续 API 化可传入基于 `request.user` 的 sender_label - 发送者强制为“系统自动发送”(后续 API 化可传入基于 `request.user` 的 sender_label
##### 3.1) command发送某个 PrintingJob 的“最新状态”通知(可 dry-run
- 文件:`printing/management/commands/wecom_notify_printing_job_latest_state.py`
- 用法:
- 预览(不发请求,适合只有 http 出网的机器):
- `uv run python manage.py wecom_notify_printing_job_latest_state <printing_job_id> --dry-run`
- 实际发送(需要 https 出网 + 配置 `.env``WECOM_WEBHOOK_KEY`
- `uv run python manage.py wecom_notify_printing_job_latest_state <printing_job_id>`
- 可选覆盖 key / timeout
- `uv run python manage.py wecom_notify_printing_job_latest_state <printing_job_id> --key xxx --timeout 5`
- 参数:
- `printing_job_id`必填PrintingJob.id
- `--dry-run`:仅输出 markdown不调用企业微信 webhook
- `--key`:可选,临时覆盖 `.env``WECOM_WEBHOOK_KEY`
- `--timeout`可选HTTP 超时秒数(默认 10
#### 4) 排查PrintingJob 查询 APIlist / retrieve均返回 “No PrintingJob matches the given query”
现象:
- `api_v1` 的 PrintingJob 查询接口(列表/单体)提示 “No PrintingJob matches the given query”但数据库中确有数据。
排查路径与结论(仅排查,未改动代码):
- `PrintingJobViewSet``api_v1/views/printing/views.py::PrintingJobViewSet`
- 继承 `CustomerVisibilityFilterMixin`,其 `get_queryset()` 会先做 **merchant 隔离**,再做 **客户可见性过滤**
- `CustomerVisibilityFilterMixin.filter_by_merchant()``api_v1/views/printing/mixins.py`
- 当前逻辑对非 superuser 强制 `queryset.filter(merchant=emp.merchant)`
- 当历史数据存在 `PrintingJob.merchant = NULL`(或不等于当前员工 merchant会被直接过滤掉导致 list/retrieve 都“查不到”。
- 佐证:
- `api_v1/views/printing/test_printing_job_api.py` 中明确把测试用户设为 `superuser`,并注释说明:
“PrintingJobViewSet 默认做 merchant 隔离;本测试集不关注该隔离逻辑,设为 superuser 以避免因测试数据未设置 merchant 导致的 404/列表为空。”
- 这与线上“明明有数据但 API 查不到”的现象一致(典型是 merchant 字段为空/不一致导致隔离过滤)。
建议方案(未实施):
- 保持多租户隔离前提下,对 `merchant=NULL` 的历史数据做兼容:
- 若模型自身 `merchant` 为空,则可回退使用关联 customer 的 merchant对 PrintingJob 可通过 `printing_order__customer__merchant`)判断归属。
- 并补充非 superuser 的回归测试覆盖该历史数据场景。

View File

@@ -10,6 +10,7 @@ This module is intended for reusable business logic that may be called by:
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from urllib.parse import urlparse
from django.conf import settings from django.conf import settings
from django.utils import timezone from django.utils import timezone
@@ -48,6 +49,26 @@ def _truncate_text(s: str, max_len: int) -> str:
return s[: max(0, int(max_len) - 6)] + "...(截断)" return s[: max(0, int(max_len) - 6)] + "...(截断)"
def _is_http_url(s: str) -> bool:
"""
Minimal URL check for markdown linkify.
Only treat http/https absolute URLs as linkable.
"""
s = (s or "").strip()
if not s:
return False
try:
p = urlparse(s)
except Exception:
return False
return p.scheme in ("http", "https") and bool(p.netloc)
def _escape_markdown_link_text(s: str) -> str:
# avoid breaking markdown link text
return (s or "").replace("[", r"\[").replace("]", r"\]")
def render_process_params_markdown(*, params: dict, max_items: int = 60, max_value_len: int = 200) -> str: def render_process_params_markdown(*, params: dict, max_items: int = 60, max_value_len: int = 200) -> str:
""" """
Render process params as markdown lines. Render process params as markdown lines.
@@ -61,7 +82,13 @@ def render_process_params_markdown(*, params: dict, max_items: int = 60, max_val
for k in sorted(params.keys(), key=lambda x: str(x)): for k in sorted(params.keys(), key=lambda x: str(x)):
v = params.get(k) v = params.get(k)
kk = _truncate_text(str(k), 80) kk = _truncate_text(str(k), 80)
vv = _truncate_text(str(v), int(max_value_len)) v_str = "" if v is None else str(v)
v_href = v_str.strip()
if _is_http_url(v_href):
v_disp = _escape_markdown_link_text(_truncate_text(v_href, int(max_value_len)))
vv = f"[{v_disp}]({v_href})"
else:
vv = _truncate_text(v_str, int(max_value_len))
items.append(f"- **{kk}**{vv}") items.append(f"- **{kk}**{vv}")
if len(items) > int(max_items): if len(items) > int(max_items):

View File

@@ -0,0 +1,23 @@
from django.test import SimpleTestCase
class WeComMarkdownRenderTest(SimpleTestCase):
def test_render_process_params_markdown_linkify_http_url(self):
from printing.services import render_process_params_markdown
md = render_process_params_markdown(
params={
"图片": "https://example.com/a.png",
"备注": "不是链接",
}
)
self.assertIn("- **图片**[https://example.com/a.png](https://example.com/a.png)", md)
self.assertIn("- **备注**:不是链接", md)
def test_render_process_params_markdown_linkify_strips_whitespace(self):
from printing.services import render_process_params_markdown
md = render_process_params_markdown(params={"图片": " https://example.com/a.png "})
self.assertIn("[https://example.com/a.png](https://example.com/a.png)", md)