From 65ef75f1efa96cd8ec3f0d94b30737603254cb48 Mon Sep 17 00:00:00 2001 From: jillion <439912596@qq.com> Date: Fri, 18 Sep 2026 17:25:38 +0800 Subject: [PATCH] =?UTF-8?q?=E8=90=BD=E5=9C=B0=E7=A7=81=E8=81=8A=E9=99=84?= =?UTF-8?q?=E4=BB=B6=E8=AF=A2=E4=BB=B7=EF=BC=9A=E5=8F=91=E5=8D=A1=E8=B5=B0?= =?UTF-8?q?=20H5=20=E6=94=B6=20Excel/PDF=EF=BC=8C=E8=AE=A4=E6=88=90?= =?UTF-8?q?=E5=8F=A3=E6=92=AD=E5=86=8D=E5=A4=8D=E7=94=A8=E4=B8=89=E7=BA=BF?= =?UTF-8?q?=E6=96=87=E5=AD=97=E6=B5=81=E7=A8=8B=EF=BC=8C=E9=81=BF=E5=85=8D?= =?UTF-8?q?=E9=99=84=E4=BB=B6=E5=8D=A1=E5=9C=A8=E8=AF=86=E5=88=AB=E6=88=96?= =?UTF-8?q?=E9=80=80=E6=88=90=E7=BA=AF=E6=96=87=E6=9C=AC=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- inquiry-agent/agent/channel/h5/__init__.py | 47 +- inquiry-agent/agent/channel/h5/copy_store.py | 17 + inquiry-agent/agent/channel/h5/upload_page.py | 319 +++++++++ inquiry-agent/agent/handlers/__init__.py | 4 +- .../agent/handlers/attachment_inquiry.py | 661 ++++++++++++++++++ inquiry-agent/agent/jobs/recognition.py | 6 +- .../agent/llm/mode_attachment_text.py | 84 ++- .../agent/policy/attachment_extract.py | 110 +++ inquiry-agent/agent/policy/attachment_wave.py | 413 +++++++++++ inquiry-agent/agent/policy/inquiry_copy.py | 36 + inquiry-agent/agent/redis_coord/keys.py | 4 + inquiry-agent/agent/routing/dispatch.py | 21 +- inquiry-agent/requirements.txt | 4 + .../tests/test_attachment_inquiry.py | 339 +++++++++ .../tests/test_attachment_upload_h5.py | 96 +++ inquiry-agent/tests/test_attachment_wave.py | 116 +++ .../tests/test_extract_attachment.py | 61 ++ inquiry-agent/tests/test_land_copy.py | 20 + .../tests/test_mode_attachment_text.py | 70 ++ 19 files changed, 2420 insertions(+), 8 deletions(-) create mode 100644 inquiry-agent/agent/channel/h5/upload_page.py create mode 100644 inquiry-agent/agent/handlers/attachment_inquiry.py create mode 100644 inquiry-agent/agent/policy/attachment_extract.py create mode 100644 inquiry-agent/agent/policy/attachment_wave.py create mode 100644 inquiry-agent/tests/test_attachment_inquiry.py create mode 100644 inquiry-agent/tests/test_attachment_upload_h5.py create mode 100644 inquiry-agent/tests/test_attachment_wave.py create mode 100644 inquiry-agent/tests/test_extract_attachment.py create mode 100644 inquiry-agent/tests/test_mode_attachment_text.py diff --git a/inquiry-agent/agent/channel/h5/__init__.py b/inquiry-agent/agent/channel/h5/__init__.py index d62bf81..e51b512 100644 --- a/inquiry-agent/agent/channel/h5/__init__.py +++ b/inquiry-agent/agent/channel/h5/__init__.py @@ -10,14 +10,20 @@ from __future__ import annotations import logging from typing import Any, Optional -from fastapi import APIRouter, HTTPException, Query +from fastapi import APIRouter, File, HTTPException, Query, UploadFile from fastapi.responses import HTMLResponse from pydantic import BaseModel, Field from agent.channel.clarification import validate_token from agent.channel.h5.copy_page import expired_copy_page, render_copy_page from agent.channel.h5.copy_store import lookup_copy_token +from agent.channel.h5.upload_page import ( + expired_upload_page, + render_upload_message, + render_upload_page, +) from agent.handlers.card_action import handle_card_action +from agent.policy import inquiry_copy as copy logger = logging.getLogger(__name__) @@ -67,6 +73,45 @@ def h5_jsapi_signature( raise HTTPException(status_code=503, detail="jsapi_unavailable") from exc +@router.get("/upload/{token}", response_class=HTMLResponse) +def h5_upload_open(token: str) -> HTMLResponse: + """ + 打开附件上传页。身份只认 token；过期给 410，不吐内部栈。 + """ + from agent.handlers.attachment_inquiry import lookup_upload_sender + + if lookup_upload_sender(token) is None: + return HTMLResponse(expired_upload_page(), status_code=410) + return HTMLResponse(render_upload_page()) + + +@router.post("/upload/{token}", response_class=HTMLResponse) +async def h5_upload_submit( + token: str, + files: Optional[list[UploadFile]] = File(default=None), +) -> HTMLResponse: + """ + 接收多个 Excel/PDF。请求线程只验 token、读字节、入队，不等千问。 + """ + from agent.handlers.attachment_inquiry import accept_h5_uploads, lookup_upload_sender + + sender_id = lookup_upload_sender(token) + if sender_id is None: + return HTMLResponse(expired_upload_page(), status_code=410) + incoming: list[tuple[str, bytes]] = [] + for item in files or []: + name = item.filename or "" + raw = await item.read() + incoming.append((name, raw)) + phase, hint = accept_h5_uploads(sender_id=sender_id, files=incoming, process_now=False) + logger.info("h5.upload phase=%s sender=%s n=%s", phase, sender_id, len(incoming)) + if phase == "expired": + return HTMLResponse(expired_upload_page(), status_code=410) + if phase in {"reject_type", "need_file"}: + return HTMLResponse(render_upload_message("无法上传", hint), status_code=400) + return HTMLResponse(render_upload_message("上传成功", copy.ATTACH_H5_OK)) + + @router.get("/{token}") def h5_open(token: str) -> dict[str, Any]: """ diff --git a/inquiry-agent/agent/channel/h5/copy_store.py b/inquiry-agent/agent/channel/h5/copy_store.py index aee2801..946edda 100644 --- a/inquiry-agent/agent/channel/h5/copy_store.py +++ b/inquiry-agent/agent/channel/h5/copy_store.py @@ -46,6 +46,23 @@ def issue_copy_token(work_order_no: str, *, ttl_seconds: int = COPY_TTL_SECONDS) return token +def inquiry_form_base() -> str: + """H5 公网根路径（含 /inquiry-form），不含 userid。""" + return _form_base() + + +def build_upload_url(token: str) -> str: + """ + 拼附件上传页：{inquiry-form}/upload/{token}。 + + token 为空则返回空串，调用方不要出跳转按钮。 + """ + raw = (token or "").strip() + if not raw: + return "" + return f"{_form_base()}/upload/{raw}" + + def lookup_copy_token(token: str) -> Optional[str]: """校验 token 未过期；返回工单号或 None。不消费，可重复打开。""" raw = (token or "").strip() diff --git a/inquiry-agent/agent/channel/h5/upload_page.py b/inquiry-agent/agent/channel/h5/upload_page.py new file mode 100644 index 0000000..cee6508 --- /dev/null +++ b/inquiry-agent/agent/channel/h5/upload_page.py @@ -0,0 +1,319 @@ +""" +附件询价 H5 上传页。 + +本文件职责：渲染选文件页与过期/成功/拒收提示。URL 不含 userid。 +样式按企微内嵌、手机竖屏：卡片、虚线选区、已选文件列表；不改上传规则。 +""" + +from __future__ import annotations + +from html import escape +import json + +from agent.policy import inquiry_copy as copy + +# 三页共用：浅底、白卡片、企微蓝主按钮。禁止在页面里写 userid。 +_PAGE_CSS = """ +:root { + --bg: #eef3f9; + --card: #ffffff; + --text: #17233d; + --muted: #5b6b7c; + --line: #c5d4ea; + --brand: #1677ff; + --brand-press: #0f62d9; + --danger: #d83931; + --ok: #0f8a5f; + --shadow: 0 10px 28px rgba(23, 51, 91, 0.10); +} +* { box-sizing: border-box; } +html, body { margin: 0; min-height: 100%; } +body { + font-family: -apple-system, BlinkMacSystemFont, "PingFang SC", "Microsoft YaHei", sans-serif; + background: var(--bg); + color: var(--text); + -webkit-font-smoothing: antialiased; +} +.wrap { + max-width: 480px; + margin: 0 auto; + padding: 20px 16px calc(24px + env(safe-area-inset-bottom)); +} +.card { + background: var(--card); + border-radius: 16px; + padding: 22px 20px 24px; + box-shadow: var(--shadow); +} +.hero { + display: flex; + align-items: flex-start; + gap: 12px; + margin-bottom: 12px; +} +.badge { + width: 44px; + height: 44px; + border-radius: 12px; + background: #e8f1ff; + color: var(--brand); + display: flex; + align-items: center; + justify-content: center; + flex-shrink: 0; +} +.badge svg { width: 22px; height: 22px; } +h1 { font-size: 20px; line-height: 1.3; margin: 0 0 4px; font-weight: 600; } +.lead { margin: 0; color: var(--muted); font-size: 14px; line-height: 1.6; } +.hint { margin: 12px 0 0; color: var(--muted); font-size: 13px; line-height: 1.7; } +.pills { display: flex; gap: 8px; margin: 14px 0 0; } +.pill { + font-size: 12px; + color: #2b5aa8; + background: #eef4ff; + border-radius: 999px; + padding: 4px 10px; +} +.drop { + position: relative; + display: block; + margin-top: 18px; + padding: 22px 16px; + border: 1.5px dashed var(--line); + border-radius: 12px; + background: #f7faff; + text-align: center; + cursor: pointer; + -webkit-tap-highlight-color: transparent; +} +.drop:focus-within { border-color: var(--brand); box-shadow: 0 0 0 3px rgba(22, 119, 255, 0.16); } +.drop input { + position: absolute; + top: 0; + right: 0; + bottom: 0; + left: 0; + opacity: 0; + width: 100%; + height: 100%; + cursor: pointer; +} +.drop-title { display: block; font-size: 16px; font-weight: 600; } +.drop-sub { display: block; margin-top: 6px; font-size: 13px; color: var(--muted); } +.picked { list-style: none; margin: 14px 0 0; padding: 0; } +.picked li { + display: flex; + align-items: center; + gap: 10px; + padding: 10px 12px; + margin-top: 8px; + background: #f6f8fb; + border-radius: 10px; + font-size: 14px; +} +.idx { + width: 22px; + height: 22px; + border-radius: 50%; + background: #e8f1ff; + color: var(--brand); + font-size: 12px; + font-weight: 700; + display: flex; + align-items: center; + justify-content: center; + flex-shrink: 0; +} +.name { word-break: break-all; } +button[type="submit"], .btn { + width: 100%; + min-height: 48px; + margin-top: 20px; + border: 0; + border-radius: 10px; + background: var(--brand); + color: #fff; + font-size: 16px; + font-weight: 700; + cursor: pointer; + touch-action: manipulation; +} +button[type="submit"]:active, .btn:active { background: var(--brand-press); } +button[disabled] { background: #94a3b8; cursor: not-allowed; } +.err { + min-height: 1.4em; + margin: 12px 0 0; + color: var(--danger); + font-size: 14px; + line-height: 1.5; +} +.rules { margin: 16px 0 0; padding-left: 18px; color: var(--muted); font-size: 13px; line-height: 1.8; } +.tone-ok .badge { background: #e7f7ef; color: var(--ok); } +.tone-warn .badge { background: #fdecea; color: var(--danger); } +.tone-expired .badge { background: #eef1f5; color: var(--muted); } +.body { margin: 8px 0 0; font-size: 15px; line-height: 1.7; } +.back { margin: 10px 0 0; color: var(--muted); font-size: 14px; line-height: 1.6; } +@media (prefers-reduced-motion: reduce) { + * { transition: none !important; } +} +""" + +_ICON_FILE = ( + '' +) +_ICON_OK = ( + '' +) +_ICON_WARN = ( + '' +) +_ICON_EXPIRED = ( + '' +) + + +def _shell(*, title: str, body: str, tone: str = "ok") -> str: + """成功 / 拒绝 / 过期共用卡片壳，避免三页三种简陋白底。""" + icons = {"ok": _ICON_OK, "warn": _ICON_WARN, "expired": _ICON_EXPIRED} + icon = icons.get(tone, _ICON_WARN) + extra = '

请回企微查看识别结果。

' if tone == "ok" else "" + return f""" + + + + + {title} + + + +
+
+
+
{icon}
+
+

{title}

+
+
+

{body}

+ {extra} +
+
+ + +""" + + +def render_upload_page() -> str: + """上传页：一次可选多个 Excel/PDF；图片请发回企微。""" + reject = json.dumps(copy.ATTACH_H5_REJECT_TYPE, ensure_ascii=False) + need = json.dumps(copy.ATTACH_H5_NEED_FILE, ensure_ascii=False) + return f""" + + + + + 上传询价附件 + + + +
+
+
+
{_ICON_FILE}
+
+

上传询价附件

+

这里只传 Excel、PDF。图片请直接发到当前企微会话。

+
+
+
ExcelPDF
+

一次可选多个文件，顺序按你选中的顺序。提交后请回企微查看识别结果。

+
+ + + +
+ +
    +
  • 只接受 .xls / .xlsx / .pdf
    • +
  • 图片请发到当前企微会话，不要在本页上传
    • +
+
+
+ + + +""" + + +def render_upload_message(title: str, body: str, *, expired: bool = False) -> str: + """成功 / 拒绝 / 过期页。""" + tone = "expired" if expired else ("ok" if title == "上传成功" else "warn") + return _shell(title=escape(title), body=escape(body), tone=tone) + + +def expired_upload_page() -> str: + return render_upload_message("链接已过期", copy.ATTACH_H5_EXPIRED, expired=True) diff --git a/inquiry-agent/agent/handlers/__init__.py b/inquiry-agent/agent/handlers/__init__.py index 7e5bd3d..b519f32 100644 --- a/inquiry-agent/agent/handlers/__init__.py +++ b/inquiry-agent/agent/handlers/__init__.py @@ -2,15 +2,17 @@ Handler 包：一类意图 / 一类卡片一个模块。 禁止：单文件处理全部卡片；禁止把 Schema 校验复制粘贴进每个 handler（放 schema/policy）。 -当前：文字询价空运/海运/陆运闭环 + 私聊图片询价 + 海运应用群协同 + 空运 BOT 群文字/附件 + echo 联调残留。 +当前：文字询价空运/海运/陆运闭环 + 私聊图片询价 + 私聊附件询价 + 海运应用群协同 + 空运 BOT 群文字/附件 + echo 联调残留。 """ from agent.handlers.echo_text import handle_echo_text from agent.handlers.image_inquiry import handle_image_inquiry +from agent.handlers.attachment_inquiry import handle_attachment_inquiry from agent.handlers.text_inquiry import handle_text_inquiry, handle_text_other __all__ = [ "handle_echo_text", + "handle_attachment_inquiry", "handle_image_inquiry", "handle_text_inquiry", "handle_text_other", diff --git a/inquiry-agent/agent/handlers/attachment_inquiry.py b/inquiry-agent/agent/handlers/attachment_inquiry.py new file mode 100644 index 0000000..e265fa4 --- /dev/null +++ b/inquiry-agent/agent/handlers/attachment_inquiry.py @@ -0,0 +1,661 @@ +""" +意图：私聊附件询价。 + +本文件职责：发卡等材料 → H5 收 Excel/PDF、企微收图 → 抽原文/认图 → 拼口播 → 新开后复用文字询价。 +禁止：千问填最终缺项；另写海运/空运/陆运附件专线；回调/H5 线程同步等千问；群附件走本 Handler。 +""" + +from __future__ import annotations + +import logging +import re +import threading +import time +from typing import Any, Callable, Optional + +from agent.channel.clarification import issue_token, validate_token +from agent.channel.h5.copy_store import build_upload_url +from agent.channel.queue import get_message_store +from agent.channel.wecom.models import InboundMessage +from agent.policy import inquiry_copy as copy +from agent.policy.attachment_extract import allowed_attachment, extract_attachment_text +from agent.policy.attachment_wave import WAIT_TTL_SECONDS, get_attach_wave_book +from agent.policy.image_wave import get_image_wave_book + +logger = logging.getLogger(__name__) + +ExtractFn = Callable[[str, bytes], dict[str, Any]] +AttachTextFn = Callable[..., dict[str, Any]] +VisionFn = Callable[..., dict[str, Any]] +DownloadFn = Callable[[str], tuple[bytes, str]] +PutFn = Callable[..., dict[str, Any]] + +_STASH: dict[str, bytes] = {} +_STASH_LOCK = threading.Lock() +_PRECEDE_NOISE = ( + r"附件询价", + r"上传询价附件", + r"上传附件", + r"帮我传个\s*excel\s*询价", + r"我有个\s*excel\s*要询价", + r"传个\s*(excel|pdf)", + r"传份\s*(excel|pdf)", +) + + +def extra_precede(text: str) -> str: + """开口里去掉「要传附件」这类话，剩下的才带进口播。""" + body = (text or "").strip() + if not body: + return "" + cleaned = body + for pat in _PRECEDE_NOISE: + cleaned = re.sub(pat, " ", cleaned, flags=re.IGNORECASE) + return " ".join(cleaned.split()) + + +def _stash_put(key: str, data: bytes) -> None: + """同进程单测/回退：对象存储 dry_run 时仍能把字节交给识别。""" + if not key: + return + with _STASH_LOCK: + _STASH[key] = data or b"" + + +def _stash_get(key: str) -> bytes: + with _STASH_LOCK: + return _STASH.get(key, b"") + + +def drop_all_private_bookmarks(sender_id: str) -> None: + """三线当前书签都丢掉；已出号仍能按工单号找回。""" + from agent.handlers.image_inquiry import drop_all_private_bookmarks as _drop + + _drop(sender_id) + + +def is_active_attachment_wait(sender_id: str, *, now: Optional[float] = None) -> bool: + """业务结果尚未发出、链接也未过期的附件等待票。""" + return get_attach_wave_book().active(sender_id, now=now) is not None + + +def append_attach_text(sender_id: str, text: str) -> bool: + """认完之前的字并进本票，不走文字询价。""" + return get_attach_wave_book().append_follow_text(sender_id, text) + + +def handle_attachment_inquiry( + message: InboundMessage, + *, + store=None, + now: Optional[float] = None, +) -> str: + """ + 听成要传附件：新开一票、发上传卡。不建单、不等 H5。 + + 须已有 sender_id。群消息不要调用。 + """ + if not message.has_sender(): + logger.warning("attachment_inquiry 拒绝：无 sender_id") + return "reject_no_sender" + if (message.chat_type or "single") == "group": + logger.info("attachment_inquiry 忽略群 chat=%s", message.chat_id) + return "ignore_group_attachment" + + stamp = time.time() if now is None else float(now) + img_book = get_image_wave_book() + if img_book.active(message.sender_id) is not None: + img_book.mark_business_done(message.sender_id) + + drop_all_private_bookmarks(message.sender_id) + token = issue_token( + thread_id=f"attach:{message.sender_id}", + wait_version=0, + sender_id=message.sender_id, + allowed_actions=["upload_attachment"], + ttl_seconds=WAIT_TTL_SECONDS, + ) + wave = get_attach_wave_book().open_wait( + message.sender_id, + precede_text=extra_precede(message.content or ""), + token=token, + now=stamp, + ) + url = build_upload_url(token) + payload = copy.attachment_upload_card(url) + msg_store = store or get_message_store() + msg_store.enqueue_outbound( + touser=message.sender_id, + content="上传询价附件", + dedupe_key=f"attach-card:{wave.wave_id}", + payload=payload, + ) + logger.info("attachment_inquiry 发卡 wave=%s sender=%s", wave.wave_id, message.sender_id) + return "wait_upload" + + +def join_attachment_image( + message: InboundMessage, + *, + store=None, + vision_fn: Optional[VisionFn] = None, + download_fn: Optional[DownloadFn] = None, + put_fn: Optional[PutFn] = None, + process_now: bool = True, + now: Optional[float] = None, +) -> str: + """等材料期间的私聊图：并入本票，必要时 ack「正在识别图片」。""" + if not message.has_sender(): + return "reject_no_sender" + stamp = time.time() if now is None else float(now) + book = get_attach_wave_book() + media_id = str((message.media or {}).get("media_id") or "") + filename = str((message.media or {}).get("filename") or "image.jpg") + wave, how = book.join_image( + message.sender_id, media_id, filename=filename, now=stamp + ) + if how != "joined" or wave is None: + return "inactive" + msg_store = store or get_message_store() + index = len(wave.materials) - 1 + if not wave.ack_sent: + msg_store.enqueue_outbound( + touser=message.sender_id, + content=copy.IMAGE_RECOGNIZING, + dedupe_key=f"attach-ack:{wave.wave_id}", + ) + book.mark_ack(message.sender_id, "image") + if process_now and vision_fn is not None: + _recognize_image( + sender_id=message.sender_id, + index=index, + media_id=media_id, + filename=filename, + vision_fn=vision_fn, + download_fn=download_fn, + put_fn=put_fn, + ) + wave = book.active(message.sender_id) or wave + if book.pending(wave) <= 0: + return finish_attach_wave(message.sender_id, store=msg_store, seed_message=message) + return "recognizing" + _enqueue_attach( + sender_id=message.sender_id, + index=index, + role="image", + filename=filename, + media_id=media_id, + message_id=message.message_id, + wave_id=wave.wave_id, + ) + return "recognizing" + + +def accept_h5_uploads( + *, + sender_id: str, + files: list[tuple[str, bytes]], + store=None, + extract_fn: Optional[ExtractFn] = None, + attach_text_fn: Optional[AttachTextFn] = None, + put_fn: Optional[PutFn] = None, + process_now: bool = True, + now: Optional[float] = None, +) -> tuple[str, str]: + """ + H5 一次提交：校验后缀、按顺序入材料、开始认。 + + 返回 (phase, hint)。hint 给网页看；phase 给调用方。 + """ + sid = (sender_id or "").strip() + stamp = time.time() if now is None else float(now) + book = get_attach_wave_book() + wave = book.active(sid, now=stamp) + if wave is None: + return "expired", copy.ATTACH_H5_EXPIRED + incoming = [(str(n or "").strip(), raw or b"") for n, raw in files if str(n or "").strip()] + if not incoming: + return "need_file", copy.ATTACH_H5_NEED_FILE + for name, _raw in incoming: + if not allowed_attachment(name): + return "reject_type", copy.ATTACH_H5_REJECT_TYPE + start_index = len(wave.materials) + wave = book.add_files(sid, [n for n, _r in incoming], now=stamp) + if wave is None: + return "expired", copy.ATTACH_H5_EXPIRED + msg_store = store or get_message_store() + if not wave.ack_sent: + msg_store.enqueue_outbound( + touser=sid, + content=copy.ATTACH_RECOGNIZING, + dedupe_key=f"attach-ack:{wave.wave_id}", + ) + book.mark_ack(sid, "file") + for offset, (name, raw) in enumerate(incoming): + index = start_index + offset + object_key = f"inquiry-attach/{sid}/{wave.wave_id}/{index}/{name}" + _store_bytes(object_key, raw, put_fn=put_fn) + if process_now: + _recognize_file( + sender_id=sid, + index=index, + filename=name, + raw=raw, + object_key=object_key, + extract_fn=extract_fn, + attach_text_fn=attach_text_fn, + ) + else: + _enqueue_attach( + sender_id=sid, + index=index, + role="file", + filename=name, + object_key=object_key, + wave_id=wave.wave_id, + ) + _stash_put(object_key, raw) + wave = book.active(sid, now=stamp) + if process_now and wave is not None and book.pending(wave) <= 0: + finish_attach_wave(sid, store=msg_store) + return "done", copy.ATTACH_H5_OK + return "recognizing", copy.ATTACH_H5_OK + + +def finish_queued_attachment(payload: dict[str, Any]) -> dict[str, Any]: + """Worker 识别槽：认完一份材料，pending 为 0 则出业务结果。""" + sender_id = str(payload.get("sender_id") or "") + index = int(payload.get("index") or 0) + role = str(payload.get("role") or "file") + filename = str(payload.get("filename") or "") + if role == "image": + raw_hook = payload.get("__bytes") + vision_hook = payload.get("__vision") + + def _vision(**_kwargs: Any) -> dict[str, Any]: + if isinstance(vision_hook, dict): + return dict(vision_hook) + from agent.config import get_settings + from agent.llm.mode_vision import invoke_vision + + settings = get_settings() + return invoke_vision( + payload=_kwargs.get("payload") or {}, + allow_network=bool(settings.llm_allow_network and settings.llm_data_usage_confirmed), + ) + + def _download(mid: str) -> tuple[bytes, str]: + if isinstance(raw_hook, (bytes, bytearray)): + return bytes(raw_hook), "" + from agent.channel.outbox.sender import WeComAppClient + from agent.config import get_settings + + settings = get_settings() + client = WeComAppClient( + corp_id=settings.wecom_corp_id, + secret=settings.wecom_secret, + agent_id=int(settings.wecom_agent_id or "0") or 0, + ) + return client.download_media(mid) + + _recognize_image( + sender_id=sender_id, + index=index, + media_id=str(payload.get("media_id") or ""), + filename=filename or "image.jpg", + vision_fn=_vision, + download_fn=_download, + put_fn=None, + ) + else: + raw = payload.get("__bytes") + object_key = str(payload.get("object_key") or "") + blob = bytes(raw) if isinstance(raw, (bytes, bytearray)) else _load_bytes(object_key) + extract_hook = payload.get("__extract") + text_hook = payload.get("__attach_text") + _recognize_file( + sender_id=sender_id, + index=index, + filename=filename, + raw=blob, + object_key=object_key, + extract_fn=extract_hook if callable(extract_hook) else None, + attach_text_fn=text_hook if callable(text_hook) else None, + ) + book = get_attach_wave_book() + wave = book.active(sender_id) + if wave is None: + return {"ok": False, "phase": "no_active_wave", "pending": 0} + if book.pending(wave) > 0: + return {"ok": True, "phase": "recognizing", "pending": book.pending(wave)} + phase = finish_attach_wave( + sender_id, + seed_message=InboundMessage( + sender_id=sender_id, + message_id=str(payload.get("message_id") or f"attach:{payload.get('wave_id') or sender_id}:{index}"), + content="", + msg_type="file", + ), + ) + return {"ok": True, "phase": phase} + + +def finish_attach_wave( + sender_id: str, + *, + store=None, + seed_message: Optional[InboundMessage] = None, +) -> str: + """本票材料都认完：拼口播、当文字询价；看不清则说明+补问同一条。""" + from agent.handlers.text_inquiry import handle_text_inquiry + from agent.policy.air_text_flow import get_air_text_flow + from agent.policy.land_text_flow import get_land_text_flow + from agent.policy.sea_text_flow import get_sea_text_flow + + book = get_attach_wave_book() + wave = book.active(sender_id) + oral = book.compose_oral(sender_id) + reason = "" + if not oral.strip(): + reason = copy.ATTACH_TECH_FAIL if book.has_errors(sender_id) else copy.ATTACH_UNREADABLE + + msg_store = store or get_message_store() + prefix_used = {"v": False} + + class _StoreProxy: + def enqueue_outbound(self, **kwargs): + body = str(kwargs.get("content") or "") + extra = dict(kwargs.get("payload") or {}) + if reason and not prefix_used["v"] and body: + body = copy.image_fail_then_clarify(reason, body) + kwargs["content"] = body + prefix_used["v"] = True + return msg_store.enqueue_outbound( + touser=kwargs.get("touser") or sender_id, + content=body, + dedupe_key=str(kwargs.get("dedupe_key") or ""), + payload=extra, + ) + + inbound = InboundMessage( + sender_id=sender_id, + message_id=( + seed_message.message_id + if seed_message and seed_message.message_id + else f"attach:{(wave.wave_id if wave else sender_id)}:{int(time.time())}" + ), + content=oral, + msg_type="text", + media=dict(seed_message.media) if seed_message else {}, + raw=dict(seed_message.raw) if seed_message else {}, + ) + phase = handle_text_inquiry(inbound, store=_StoreProxy()) + if reason and not prefix_used["v"]: + msg_store.enqueue_outbound( + touser=sender_id, + content=copy.image_fail_then_clarify(reason, copy.ASK_TRANSPORT), + dedupe_key=f"attach-empty:{(wave.wave_id if wave else inbound.message_id)}", + ) + phase = "need_mode" + book.mark_business_done(sender_id) + sess = ( + get_sea_text_flow().session_of(sender_id) + or get_air_text_flow().session_of(sender_id) + or get_land_text_flow().session_of(sender_id) + ) + wo = (sess.work_order_no if sess else "") or "" + if wo and wave: + ledger = None + for engine in (get_sea_text_flow(), get_air_text_flow(), get_land_text_flow()): + hit = engine.session_of(sender_id) + if hit and hit.work_order_no == wo: + ledger = engine.ledger + break + if ledger is not None: + for row in wave.materials: + key = row.object_key + if not key: + continue + name = row.filename or ("image.jpg" if row.kind == "image" else "attachment.bin") + ctype = "image/jpeg" if row.kind == "image" else "application/octet-stream" + try: + ledger.attach_file( + work_order_no=wo, + file_name=name, + object_key=key, + content_type=ctype, + sender_id=sender_id, + ) + except Exception: + logger.exception("挂附件失败 wo=%s name=%s", wo, name) + logger.info("attachment_inquiry 波次完成 phase=%s oral_len=%s", phase, len(oral)) + return phase + + +def lookup_upload_sender(token: str) -> Optional[str]: + """ + H5 打开/提交：token 有效且对应附件票仍活跃才返回 sender_id。 + 不消费 token，业务结果发出前可再传。 + """ + rec = validate_token(token, action="upload_attachment", consume=False) + if rec is None: + return None + book = get_attach_wave_book() + wave = book.active(rec.sender_id) + if wave is None: + return None + if wave.token and wave.token != token: + return None + return rec.sender_id + + +def _enqueue_attach(**payload: Any) -> None: + try: + from agent.jobs.recognition import enqueue_recognition + + enqueue_recognition( + kind="attachment", + payload=dict(payload), + idempotency_key=str( + payload.get("message_id") + or f"{payload.get('wave_id')}:{payload.get('index')}" + ), + ) + except ModuleNotFoundError: + logger.info("attachment 识别队跳过（本环境无 jobs 依赖） sender=%s", payload.get("sender_id")) + except Exception: + logger.exception("attachment 入识别队失败 sender=%s", payload.get("sender_id")) + + +def _blob_put(object_key: str, raw: bytes) -> None: + """ + 把附件字节交给 Redis 短交接，供 Worker 识别槽读取。 + + HTTP 与 Worker 不是同一进程，内存 stash 跨不过去；对象存储 dry_run 也不落盘。 + TTL 与等待票一致。失败只打日志，不在请求线程抛给销售。 + """ + key = (object_key or "").strip() + if not key: + return + try: + import base64 + + from agent.redis_coord.keys import ATTACH_BLOB + from agent.redis_coord.runtime import get_redis_runtime + + rt = get_redis_runtime() + packed = base64.b64encode(raw or b"").decode("ascii") + rt.client.raw.set(rt.client.key(ATTACH_BLOB, key), packed, ex=WAIT_TTL_SECONDS) + except Exception: + logger.exception("附件字节 Redis 交接失败 key=%s", key) + + +def _blob_get(object_key: str) -> bytes: + """Worker 读 HTTP 放下的附件字节。没有则空。""" + key = (object_key or "").strip() + if not key: + return b"" + try: + import base64 + + from agent.redis_coord.keys import ATTACH_BLOB + from agent.redis_coord.runtime import get_redis_runtime + + rt = get_redis_runtime() + raw = rt.client.raw.get(rt.client.key(ATTACH_BLOB, key)) + if not raw: + return b"" + if isinstance(raw, bytes): + raw = raw.decode("ascii") + return base64.b64decode(str(raw)) + except Exception: + logger.exception("读附件 Redis 交接失败 key=%s", key) + return b"" + + +def _store_bytes(object_key: str, raw: bytes, *, put_fn: Optional[PutFn]) -> None: + _stash_put(object_key, raw) + _blob_put(object_key, raw) + if put_fn is not None: + try: + put_fn(key=object_key, data=raw, content_type="application/octet-stream") + except Exception: + logger.exception("附件落存储失败 key=%s", object_key) + return + try: + from agent.config import get_settings + from agent.storage.s3_client import S3ObjectStorage + + S3ObjectStorage.from_settings(get_settings(), allow_network=False).put_bytes( + key=object_key, data=raw, content_type="application/octet-stream" + ) + except Exception: + logger.info("附件对象存储跳过 key=%s", object_key) + + +def _load_bytes(object_key: str) -> bytes: + hit = _stash_get(object_key) + if hit: + return hit + blob = _blob_get(object_key) + if blob: + return blob + try: + from agent.config import get_settings + from agent.storage.s3_client import S3ObjectStorage + + settings = get_settings() + got = S3ObjectStorage.from_settings( + settings, + allow_network=bool(settings.llm_allow_network), + ).get_bytes(key=object_key) + data = bytes(got.get("data") or b"") + logger.info("读附件对象存储 key=%s bytes=%s", object_key, len(data)) + return data + except Exception: + logger.exception("读附件字节失败 key=%s", object_key) + return b"" + + +def _recognize_file( + *, + sender_id: str, + index: int, + filename: str, + raw: bytes, + object_key: str, + extract_fn: Optional[ExtractFn], + attach_text_fn: Optional[AttachTextFn], +) -> None: + book = get_attach_wave_book() + blob = raw or b"" + logger.info( + "attachment 认文件 sender=%s idx=%s name=%s bytes=%s", + sender_id, + index, + filename, + len(blob), + ) + if not blob: + book.complete_material(sender_id, index, object_key=object_key, error="empty_bytes") + return + extracted = (extract_fn or extract_attachment_text)(filename, blob) + raw_text = str(extracted.get("text") or "") + err = str(extracted.get("error") or "") + organized = raw_text + org_err = err + if raw_text.strip(): + if attach_text_fn is not None: + packed = attach_text_fn(payload={"raw_text": raw_text}) or {} + else: + from agent.config import get_settings + from agent.llm.mode_attachment_text import invoke_attachment_text + + settings = get_settings() + packed = invoke_attachment_text( + payload={"raw_text": raw_text}, + allow_network=bool(settings.llm_allow_network and settings.llm_data_usage_confirmed), + ) + if packed.get("ok") and str(packed.get("text") or "").strip(): + organized = str(packed.get("text") or "") + elif not packed.get("ok"): + org_err = str(packed.get("error") or org_err or "attachment_text_failed") + book.complete_material( + sender_id, + index, + text=organized, + object_key=object_key, + error=org_err if not (organized or "").strip() else "", + ) + + +def _recognize_image( + *, + sender_id: str, + index: int, + media_id: str, + filename: str, + vision_fn: Optional[VisionFn], + download_fn: Optional[DownloadFn], + put_fn: Optional[PutFn], +) -> None: + import base64 + + from agent.llm.mode_vision import sniff_image_mime + + book = get_attach_wave_book() + err = "" + raw = b"" + if download_fn is None: + raw, err = b"", "no_download" + else: + raw, err = download_fn(media_id) + if not raw: + book.complete_material(sender_id, index, error=err or "media_empty") + return + mime = sniff_image_mime(raw) or "image/jpeg" + object_key = f"inquiry-attach/{sender_id}/{media_id or filename}" + _store_bytes(object_key, raw, put_fn=put_fn) + vision: dict[str, Any] = {"ok": False, "text": "", "empty": True, "error": err or "no_vision"} + if vision_fn is not None: + b64 = base64.b64encode(raw).decode("ascii") + vision = vision_fn(payload={"image_b64": b64, "mime": mime, "media_id": media_id}) or vision + else: + from agent.config import get_settings + from agent.llm.mode_vision import invoke_vision + + settings = get_settings() + vision = invoke_vision( + payload={"image_b64": base64.b64encode(raw).decode("ascii"), "mime": mime}, + allow_network=bool(settings.llm_allow_network and settings.llm_data_usage_confirmed), + ) + text = str(vision.get("text") or "") + if not vision.get("ok"): + err = str(vision.get("error") or err or "vision_failed") + book.complete_material( + sender_id, + index, + text=text, + object_key=object_key, + error=err if not text.strip() else "", + ) diff --git a/inquiry-agent/agent/jobs/recognition.py b/inquiry-agent/agent/jobs/recognition.py index 2047daa..bd4a64f 100644 --- a/inquiry-agent/agent/jobs/recognition.py +++ b/inquiry-agent/agent/jobs/recognition.py @@ -1,7 +1,7 @@ """ 识别任务壳：入队 Redis Stream，Worker HTTP 槽认领后跑占位处理。 -本文件职责：识别入队；Worker 槽消费。kind=vision 跑私聊图片波次，其它 kind 仍占位。 +本文件职责：识别入队；Worker 槽消费。kind=vision 跑私聊图片波次，kind=attachment 跑附件材料，其它 kind 仍占位。 禁止：回调线程同步等识别完成；禁止本模块写主账六态。 """ @@ -74,6 +74,10 @@ def process_recognition(data: dict[str, Any]) -> dict[str, Any]: from agent.handlers.image_inquiry import finish_queued_image return finish_queued_image(payload) + if kind == "attachment": + from agent.handlers.attachment_inquiry import finish_queued_attachment + + return finish_queued_attachment(payload) return process_recognition_stub(data) diff --git a/inquiry-agent/agent/llm/mode_attachment_text.py b/inquiry-agent/agent/llm/mode_attachment_text.py index 0ff7e12..47fc5ec 100644 --- a/inquiry-agent/agent/llm/mode_attachment_text.py +++ b/inquiry-agent/agent/llm/mode_attachment_text.py @@ -1,30 +1,106 @@ """ -千问：附件文本 mode 壳。 +千问：整理 Excel/PDF 抽出的原文。 + +本文件职责：把抽文本结果交给 qwen-plus，只整理看得见的字，不填询价字段。 +属于 llm 包。单测注入 chat_fn。禁止算最终缺项、禁止发明港口/箱型。 """ from __future__ import annotations -from typing import Any +import logging +from typing import Any, Optional from agent.llm.modes import LlmMode +logger = logging.getLogger(__name__) + MODE = LlmMode.ATTACHMENT_TEXT +ATTACH_SYSTEM = ( + "下面是从 Excel 或 PDF 抽出来的原文。请整理成一段可读的询价口播：" + "只保留材料里已经出现的文字和数字，原样保留港口、机场三字码、箱型、重量。" + "看不清或没出现的不要猜，不要补询价缺项，不要发明港口、重量或日期。" + "材料完全没有询价信息时输出空。" +) + def mode_meta() -> dict[str, Any]: return { "mode": MODE.value, "provider": "qwen", "purpose": "attachment_text", - "implemented": False, + "implemented": True, + } + + +def _pack_text(text: str, *, stub: bool = False) -> dict[str, Any]: + body = (text or "").strip() + return { + "ok": True, + "stub": stub, + "mode": MODE.value, + "text": body, + "empty": not bool(body), + "error": "", } def invoke_shell(*, payload: dict[str, Any]) -> dict[str, Any]: + """ + 无网/单测入口：优先 assistant_text；否则把 raw_text 当整理结果。 + """ + data = dict(payload or {}) + if "assistant_text" in data: + return _pack_text(str(data.get("assistant_text") or ""), stub=True) + if "raw_text" in data: + return _pack_text(str(data.get("raw_text") or ""), stub=True) return { "ok": False, "stub": True, "mode": MODE.value, + "text": "", + "empty": True, "error": "attachment_text_shell_not_implemented", - "echo_keys": sorted(payload.keys()), + "echo_keys": sorted(data.keys()), } + + +def invoke_attachment_text( + *, + payload: dict[str, Any], + chat_fn: Optional[Any] = None, + allow_network: bool = False, +) -> dict[str, Any]: + """ + 整理一份抽出原文。chat_fn 单测注入；无网走 invoke_shell。 + """ + data = dict(payload or {}) + raw = str(data.get("raw_text") or "") + if chat_fn is None and not allow_network: + return invoke_shell(payload=data) + if not raw.strip(): + return _pack_text("", stub=chat_fn is None) + if chat_fn is None: + from agent.llm.http_client import LlmHttpClient + + client = LlmHttpClient.from_settings() + chat_fn = client.chat + messages = [ + {"role": "system", "content": ATTACH_SYSTEM}, + {"role": "user", "content": raw.strip()}, + ] + result = chat_fn(mode=MODE, messages=messages) + if not getattr(result, "ok", False): + err = str(getattr(result, "error", "") or "attachment_text_failed") + logger.info("attachment_text 失败 err=%s", err) + return { + "ok": False, + "stub": False, + "mode": MODE.value, + "text": "", + "empty": True, + "error": err, + } + packed = _pack_text(str(getattr(result, "content", "") or "")) + packed["stub"] = False + return packed diff --git a/inquiry-agent/agent/policy/attachment_extract.py b/inquiry-agent/agent/policy/attachment_extract.py new file mode 100644 index 0000000..8e247f4 --- /dev/null +++ b/inquiry-agent/agent/policy/attachment_extract.py @@ -0,0 +1,110 @@ +""" +Excel / PDF 抽出纯文本。 + +本文件职责：把允许的附件字节变成可见原文，供千问整理、再走文字询价。 +属于 policy 包。禁止在这里猜港口/箱型/缺项；禁止写六态。 +识别槽与 H5 都只调用本模块的纯函数。 +""" + +from __future__ import annotations + +import io +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +_ALLOWED_EXT = {".xls", ".xlsx", ".pdf"} +_REJECT_HINT = "only_xlsx_xls_pdf" + + +def attachment_ext(filename: str) -> str: + """小写后缀，含点。无后缀返回空。""" + name = (filename or "").strip().rsplit("/", 1)[-1].rsplit("\\", 1)[-1] + if "." not in name: + return "" + return "." + name.rsplit(".", 1)[-1].lower() + + +def allowed_attachment(filename: str) -> bool: + """H5 只收 xls / xlsx / pdf。""" + return attachment_ext(filename) in _ALLOWED_EXT + + +def extract_attachment_text(filename: str, raw: bytes) -> dict[str, Any]: + """ + 抽出单元格或 PDF 可见字。 + + 返回 ok/text/error/empty。失败时 text 为空，由上层走「没看清 / 暂时没认出」。 + """ + ext = attachment_ext(filename) + if ext not in _ALLOWED_EXT: + return {"ok": False, "text": "", "empty": True, "error": _REJECT_HINT} + blob = raw or b"" + if not blob: + return {"ok": False, "text": "", "empty": True, "error": "empty_bytes"} + try: + if ext == ".pdf": + text = _pdf_text(blob) + elif ext == ".xlsx": + text = _xlsx_text(blob) + else: + text = _xls_text(blob) + except Exception as exc: # noqa: BLE001 + logger.exception("抽附件文本失败 name=%s", filename) + return {"ok": False, "text": "", "empty": True, "error": f"extract_fail:{type(exc).__name__}"} + body = (text or "").strip() + return {"ok": True, "text": body, "empty": not bool(body), "error": ""} + + +def _xlsx_text(raw: bytes) -> str: + from openpyxl import load_workbook + + wb = load_workbook(io.BytesIO(raw), read_only=True, data_only=True) + parts: list[str] = [] + try: + for sheet in wb.worksheets: + title = str(sheet.title or "").strip() + if title: + parts.append(f"# {title}") + for row in sheet.iter_rows(values_only=True): + cells = [str(c).strip() for c in row if c is not None and str(c).strip()] + if cells: + parts.append("\t".join(cells)) + finally: + wb.close() + return "\n".join(parts) + + +def _xls_text(raw: bytes) -> str: + try: + import xlrd + except ImportError: + return "" + book = xlrd.open_workbook(file_contents=raw) + parts: list[str] = [] + for sheet in book.sheets(): + title = str(sheet.name or "").strip() + if title: + parts.append(f"# {title}") + for r in range(sheet.nrows): + cells = [ + str(sheet.cell_value(r, c)).strip() + for c in range(sheet.ncols) + if str(sheet.cell_value(r, c)).strip() + ] + if cells: + parts.append("\t".join(cells)) + return "\n".join(parts) + + +def _pdf_text(raw: bytes) -> str: + from pypdf import PdfReader + + reader = PdfReader(io.BytesIO(raw)) + parts: list[str] = [] + for page in reader.pages: + piece = page.extract_text() or "" + if piece.strip(): + parts.append(piece.strip()) + return "\n".join(parts) diff --git a/inquiry-agent/agent/policy/attachment_wave.py b/inquiry-agent/agent/policy/attachment_wave.py new file mode 100644 index 0000000..481719d --- /dev/null +++ b/inquiry-agent/agent/policy/attachment_wave.py @@ -0,0 +1,413 @@ +""" +私聊附件等材料波次账。 + +本文件职责：记住「这一票」Excel/PDF、企微图和开口/期间的字，供附件询价拼口播。 +属于 policy 包。生产波次落 Redis（HTTP 发卡/收 H5、Worker 识别共用）；单测可纯内存。 +禁止：调模型、建单、写六态；禁止一把大锁串全部销售。 + +未开始认时 30 分钟过期。一旦发图或 H5 提交，只看业务结果出没出。 +""" + +from __future__ import annotations + +import json +import logging +import threading +import uuid +from dataclasses import asdict, dataclass, field +from typing import Any, Optional + +from agent.policy.image_wave import WaveKV + +# 规格：上传链接与未开始认的等待同一时效。 +WAIT_TTL_SECONDS = 30 * 60 +# 已开始认的波次短协调：超时后当新开，不当六态。 +_WAVE_TTL_SEC = 2 * 3600 +logger = logging.getLogger(__name__) + + +@dataclass +class AttachMaterial: + """一份按到达/上传顺序记下的材料。text 在识别完成后才填。""" + + kind: str + filename: str = "" + media_id: str = "" + text: str = "" + object_key: str = "" + error: str = "" + done: bool = False + + +@dataclass +class AttachWave: + """ + 一张新询价对应的附件等材料波次。 + + started：已发图或已 H5 提交。pending 用 materials 里未 done 的份数。 + """ + + sender_id: str + wave_id: str + token: str = "" + opened_at: float = 0.0 + started: bool = False + ack_sent: bool = False + ack_kind: str = "" + business_done: bool = False + precede_text: str = "" + follow_text: str = "" + materials: list[AttachMaterial] = field(default_factory=list) + + +class AttachWaveBook: + """ + 附件波次账。按 sender 一把锁，不同销售互不等。 + + 生产必须挂 Redis：发卡在 HTTP、识别在 Worker，两边不能各记一份内存。 + """ + + def __init__(self, kv: Optional[WaveKV] = None) -> None: + self._lock = threading.Lock() + self._kv = kv + self._waves: dict[str, AttachWave] = {} + + def _hydrate(self, sender_id: str) -> None: + sid = (sender_id or "").strip() + if not sid or self._kv is None: + return + try: + raw = self._kv.get_json("wave", sid) + if raw: + self._waves[sid] = _wave_from_dict(raw) + else: + self._waves.pop(sid, None) + except Exception: + logger.exception("读附件波次失败 sender=%s", sid) + + def _persist(self, sender_id: str) -> None: + sid = (sender_id or "").strip() + if not sid or self._kv is None: + return + try: + wave = self._waves.get(sid) + if wave is None: + self._kv.delete("wave", sid) + else: + self._kv.set_json("wave", sid, _wave_to_dict(wave)) + except Exception: + logger.exception("写附件波次失败 sender=%s", sid) + + def pending(self, wave: Optional[AttachWave]) -> int: + """还没认完的材料份数。""" + if wave is None: + return 0 + return sum(1 for m in wave.materials if not m.done) + + def open_wait( + self, + sender_id: str, + *, + precede_text: str, + token: str, + now: float, + ) -> AttachWave: + """ + 听成附件询价：开等材料票。覆盖上一票附件波次。 + + 副作用：写账。不发送企微、不建单。 + """ + sid = (sender_id or "").strip() + wave = AttachWave( + sender_id=sid, + wave_id=uuid.uuid4().hex, + token=(token or "").strip(), + opened_at=float(now), + precede_text=(precede_text or "").strip(), + ) + with self._lock: + self._waves[sid] = wave + self._persist(sid) + return wave + + def active(self, sender_id: str, now: Optional[float] = None) -> Optional[AttachWave]: + """ + 未出业务结果的当前票。 + + 尚未开始认且超过 30 分钟：视为过期，不当活跃。 + """ + sid = (sender_id or "").strip() + stamp = float(now) if now is not None else 0.0 + with self._lock: + self._hydrate(sid) + wave = self._waves.get(sid) + if wave is None or wave.business_done: + return None + if not wave.started: + base = stamp if now is not None else wave.opened_at + if now is not None and (base - wave.opened_at) > WAIT_TTL_SECONDS: + return None + return wave + + def join_image( + self, + sender_id: str, + media_id: str, + *, + filename: str, + now: float, + ) -> tuple[Optional[AttachWave], str]: + """ + 等材料期间发来的图并入。无活跃票返回 inactive。 + + 副作用：materials 追加、started=True。 + """ + sid = (sender_id or "").strip() + wave = self.active(sid, now=now) + if wave is None: + return None, "inactive" + with self._lock: + self._hydrate(sid) + wave = self._waves.get(sid) + if wave is None or wave.business_done: + return None, "inactive" + wave.started = True + wave.materials.append( + AttachMaterial( + kind="image", + filename=(filename or "").strip() or "image.jpg", + media_id=(media_id or "").strip(), + ) + ) + self._persist(sid) + return wave, "joined" + + def add_files( + self, + sender_id: str, + filenames: list[str], + *, + now: float, + ) -> Optional[AttachWave]: + """H5 一次提交的多个文件按选中顺序追加。""" + sid = (sender_id or "").strip() + names = [str(x).strip() for x in (filenames or []) if str(x).strip()] + if not names: + return self.active(sid, now=now) + with self._lock: + self._hydrate(sid) + wave = self._waves.get(sid) + if wave is None or wave.business_done: + return None + if not wave.started and (float(now) - wave.opened_at) > WAIT_TTL_SECONDS: + return None + wave.started = True + for name in names: + wave.materials.append(AttachMaterial(kind="file", filename=name)) + self._persist(sid) + return wave + + def complete_material( + self, + sender_id: str, + index: int, + *, + text: str = "", + object_key: str = "", + error: str = "", + ) -> Optional[AttachWave]: + """ + 第 index 份材料认完。按上传顺序填原文，不按完成先后重排。 + + 调用：Worker 识别槽。不得在本方法出发业务回复。 + """ + sid = (sender_id or "").strip() + with self._lock: + self._hydrate(sid) + wave = self._waves.get(sid) + if wave is None: + return None + if index < 0 or index >= len(wave.materials): + logger.warning("附件材料下标越界 sender=%s idx=%s", sid, index) + return wave + row = wave.materials[index] + row.text = (text or "").strip() + if object_key: + row.object_key = object_key + if error: + row.error = error + row.done = True + self._persist(sid) + return wave + + def append_follow_text(self, sender_id: str, text: str) -> bool: + """业务结果发出前的字并进本票。发出后或无票返回 False。""" + body = (text or "").strip() + sid = (sender_id or "").strip() + if not sid or not body: + return False + with self._lock: + self._hydrate(sid) + wave = self._waves.get(sid) + if wave is None or wave.business_done: + return False + if wave.follow_text: + wave.follow_text = f"{wave.follow_text}\n{body}" + else: + wave.follow_text = body + self._persist(sid) + return True + + def mark_ack(self, sender_id: str, kind: str) -> None: + """已经回过「正在识别」，同一次不得再 ack。""" + sid = (sender_id or "").strip() + with self._lock: + self._hydrate(sid) + wave = self._waves.get(sid) + if wave is not None: + wave.ack_sent = True + wave.ack_kind = (kind or "").strip() + self._persist(sid) + + def compose_oral(self, sender_id: str) -> str: + """开口字 + 按材料顺序的原文 + 后到的字。""" + sid = (sender_id or "").strip() + with self._lock: + self._hydrate(sid) + wave = self._waves.get(sid) + if wave is None: + return "" + parts: list[str] = [] + if wave.precede_text.strip(): + parts.append(wave.precede_text.strip()) + for row in wave.materials: + if row.text.strip(): + parts.append(row.text.strip()) + if wave.follow_text.strip(): + parts.append(wave.follow_text.strip()) + return "\n".join(parts) + + def mark_business_done(self, sender_id: str) -> None: + """业务补问/卡片已发出。之后旧 H5 与并入图都必须拒绝。""" + sid = (sender_id or "").strip() + with self._lock: + self._hydrate(sid) + wave = self._waves.get(sid) + if wave is not None: + wave.business_done = True + self._persist(sid) + + def has_errors(self, sender_id: str) -> bool: + """本票是否有下载/识别技术失败（用于选没看清还是暂时没认出）。""" + sid = (sender_id or "").strip() + with self._lock: + self._hydrate(sid) + wave = self._waves.get(sid) + if wave is None: + return False + return any(bool(m.error) for m in wave.materials) + + +def _wave_to_dict(wave: AttachWave) -> dict[str, Any]: + data = asdict(wave) + return data + + +def _wave_from_dict(raw: dict[str, Any]) -> AttachWave: + materials = [] + for item in raw.get("materials") or []: + if not isinstance(item, dict): + continue + materials.append( + AttachMaterial( + kind=str(item.get("kind") or "file"), + filename=str(item.get("filename") or ""), + media_id=str(item.get("media_id") or ""), + text=str(item.get("text") or ""), + object_key=str(item.get("object_key") or ""), + error=str(item.get("error") or ""), + done=bool(item.get("done")), + ) + ) + return AttachWave( + sender_id=str(raw.get("sender_id") or ""), + wave_id=str(raw.get("wave_id") or uuid.uuid4().hex), + token=str(raw.get("token") or ""), + opened_at=float(raw.get("opened_at") or 0.0), + started=bool(raw.get("started")), + ack_sent=bool(raw.get("ack_sent")), + ack_kind=str(raw.get("ack_kind") or ""), + business_done=bool(raw.get("business_done")), + precede_text=str(raw.get("precede_text") or ""), + follow_text=str(raw.get("follow_text") or ""), + materials=materials, + ) + + +class RedisAttachKV: + """附件波次落到本前缀 Redis。HTTP 与 Worker 必须读同一把键。""" + + def __init__(self, client: Any) -> None: + self._client = client + + def get_json(self, kind: str, sender_id: str) -> Optional[dict[str, Any]]: + from agent.redis_coord.keys import ATTACH_WAVE + + _ = kind + raw = self._client.raw.get(self._client.key(ATTACH_WAVE, sender_id)) + if not raw: + return None + if isinstance(raw, bytes): + raw = raw.decode("utf-8") + data = json.loads(raw) + return dict(data) if isinstance(data, dict) else None + + def set_json(self, kind: str, sender_id: str, data: dict[str, Any]) -> None: + from agent.redis_coord.keys import ATTACH_WAVE + + _ = kind + self._client.raw.set( + self._client.key(ATTACH_WAVE, sender_id), + json.dumps(data, ensure_ascii=False), + ex=_WAVE_TTL_SEC, + ) + + def delete(self, kind: str, sender_id: str) -> None: + from agent.redis_coord.keys import ATTACH_WAVE + + _ = kind + self._client.raw.delete(self._client.key(ATTACH_WAVE, sender_id)) + + +def _production_kv() -> Optional[WaveKV]: + try: + from agent.redis_coord.runtime import get_redis_runtime + + rt = get_redis_runtime() + if rt is None or rt.client is None: + return None + return RedisAttachKV(rt.client) + except Exception: + logger.exception("附件波次 Redis 不可用，本进程退回内存账") + return None + + +_BOOK: Optional[AttachWaveBook] = None +_BOOK_GUARD = threading.Lock() + + +def get_attach_wave_book() -> AttachWaveBook: + """进程内一份句柄；生产内容在 Redis。""" + global _BOOK + with _BOOK_GUARD: + if _BOOK is None: + _BOOK = AttachWaveBook(kv=_production_kv()) + return _BOOK + + +def reset_attach_wave_book_for_test(kv: Optional[WaveKV] = None) -> AttachWaveBook: + """单测重置。传入同一 kv 可模拟 HTTP/Worker 两份句柄。""" + global _BOOK + with _BOOK_GUARD: + _BOOK = AttachWaveBook(kv=kv) + return _BOOK diff --git a/inquiry-agent/agent/policy/inquiry_copy.py b/inquiry-agent/agent/policy/inquiry_copy.py index 615d72b..6e62153 100644 --- a/inquiry-agent/agent/policy/inquiry_copy.py +++ b/inquiry-agent/agent/policy/inquiry_copy.py @@ -104,6 +104,42 @@ ASK_TRANSPORT = "请补充运输方式：海运 / 空运 / 陆运。" IMAGE_RECOGNIZING = "正在识别图片" IMAGE_UNREADABLE = "图片没看清询价信息，请换一张更清楚的，或直接打字。" IMAGE_TECH_FAIL = "图片暂时没认出来，请再发一张或直接打字。" +# 私聊附件询价：H5 或发图开始认；看不清/失败与补问拼成一条。 +ATTACH_RECOGNIZING = "正在识别附件" +ATTACH_UNREADABLE = "附件没看清询价信息，请换一份更清楚的 Excel / PDF（或把图片发到会话），或直接打字。" +ATTACH_TECH_FAIL = "附件暂时没认出来，请再传一次或直接打字。" +ATTACH_H5_REJECT_TYPE = "只接受 Excel、PDF；图片请发到当前企微会话。" +ATTACH_H5_NEED_FILE = "请选择要上传的 Excel 或 PDF。" +ATTACH_H5_EXPIRED = "已过期，请回企微重新说附件询价。" +ATTACH_H5_OK = "已收到，请回企微查看识别结果。" + + +def attachment_upload_card(url: str) -> dict[str, Any]: + """ + 附件上传入口模板卡（text_notice 跳转 H5）。 + + 必须跟现网同一卡型：企微 news_notice 强制要 card_image.url，缺了会 42044， + 出站再回退成一句「上传询价附件」，销售侧就看不到卡片。 + 文案对齐规格：Excel/PDF 走 H5，图片发当前会话。url 必须已含 token、不含 userid。 + """ + jump = (url or "").strip() + fallback = "https://ai.ytd-scm.com/inquiry-form" + action_url = jump or fallback + card: dict[str, Any] = { + "card_type": "text_notice", + "source": {"desc": "附件询价", "desc_color": 0}, + "main_title": { + "title": "上传询价附件", + "desc": "附件入口支持 Excel、PDF；图片请直接在企微聊天窗口发送", + }, + "sub_title_text": ( + "Excel、PDF 请点击下方入口上传；图片请直接在当前企微会话发送。" + "字段完整时自动进入建单与查价；缺失字段会在企微中补问" + ), + "jump_list": [{"type": 1, "title": "上传询价附件", "url": action_url}], + "card_action": {"type": 1, "url": action_url}, + } + return {"msgtype": "template_card", "template_card": card} def image_fail_then_clarify(reason: str, clarify: str) -> str: diff --git a/inquiry-agent/agent/redis_coord/keys.py b/inquiry-agent/agent/redis_coord/keys.py index 26a4816..8cd9b25 100644 --- a/inquiry-agent/agent/redis_coord/keys.py +++ b/inquiry-agent/agent/redis_coord/keys.py @@ -62,6 +62,10 @@ FLOW_BOOKMARK = "flow_bookmark" # 私聊图片波次：HTTP 与 Worker 必须共用，不当六态 IMAGE_WAVE = "image_wave" IMAGE_WAVE_TEXT = "image_wave_text" +# 私聊附件等材料波次：HTTP 发卡/H5 与 Worker 识别共用，不当六态 +ATTACH_WAVE = "attach_wave" +# 附件字节短交接：HTTP 收文件、Worker 识别读，TTL 与等待票一致，不当六态 +ATTACH_BLOB = "attach_blob" # 空运群同一句：BOT 与存档补听去重，不当六态 AIR_INBOUND_MSG = "air_inbound_msg" AIR_INBOUND_TEXT = "air_inbound_text" diff --git a/inquiry-agent/agent/routing/dispatch.py b/inquiry-agent/agent/routing/dispatch.py index a00821c..b37bcec 100644 --- a/inquiry-agent/agent/routing/dispatch.py +++ b/inquiry-agent/agent/routing/dispatch.py @@ -15,6 +15,7 @@ from agent.handlers import abnormal, card_action, close_deal, continue_thread, n from agent.handlers.echo_text import handle_echo_text from agent.handlers.group_collab import handle_group_collab from agent.handlers.image_inquiry import handle_image_inquiry +from agent.handlers.attachment_inquiry import handle_attachment_inquiry from agent.handlers.text_inquiry import handle_text_inquiry, handle_text_other, is_land_confirm_action from agent.routing.decision import RouteDecision @@ -57,6 +58,11 @@ def _wrap_new(message: InboundMessage) -> None: new_inquiry.handle_new_inquiry(message) +def _wrap_attachment(message: InboundMessage) -> None: + """私聊附件询价：发卡等材料，认完当口播新开，复用文字流程。""" + handle_attachment_inquiry(message) + + def _wrap_image(message: InboundMessage) -> None: """私聊图片询价：认图后当口播新开，复用文字流程。""" handle_image_inquiry(message) @@ -117,7 +123,7 @@ HANDLER_REGISTRY: dict[str, HandlerFn] = { "echo_text": _wrap_echo, "ordinary_text_inquiry": _wrap_text_inquiry, "ordinary_text_other": _wrap_text_other, - "attachment_inquiry": _wrap_new, + "attachment_inquiry": _wrap_attachment, "image_inquiry": _wrap_image, "multi_segment_transport": _wrap_new, "tms_read_only_quote": _wrap_continue, @@ -211,11 +217,24 @@ def dispatch_inbound(message: InboundMessage, decision: RouteDecision) -> str: kind = (message.msg_type or "text").lower() if kind == "image": + from agent.handlers.attachment_inquiry import is_active_attachment_wait, join_attachment_image + + if is_active_attachment_wait(message.sender_id): + join_attachment_image(message) + return "attachment_inquiry" handle_image_inquiry(message) return "image_inquiry" if kind in {"text", ""}: + from agent.handlers.attachment_inquiry import append_attach_text, is_active_attachment_wait from agent.handlers.image_inquiry import append_wave_text, is_active_image_wave + if is_active_attachment_wait(message.sender_id) and (message.content or "").strip(): + if decision.intent == "attachment_inquiry": + handle_attachment_inquiry(message) + return "attachment_inquiry" + append_attach_text(message.sender_id, message.content or "") + logger.info("dispatch 附件波次收字 sender=%s", message.sender_id) + return "attach_wave_follow_text" if is_active_image_wave(message.sender_id) and (message.content or "").strip(): append_wave_text(message.sender_id, message.content or "") logger.info("dispatch 图片波次收字 sender=%s", message.sender_id) diff --git a/inquiry-agent/requirements.txt b/inquiry-agent/requirements.txt index 3a48ddd..f799b89 100644 --- a/inquiry-agent/requirements.txt +++ b/inquiry-agent/requirements.txt @@ -16,3 +16,7 @@ psycopg-pool>=3.2,<4.0 redis>=5.0,<6.0 # 报价模板填表（按 mappingJson 写 xlsx） openpyxl>=3.1,<4.0 +# 附件询价：PDF 抽文本；旧版 .xls；H5 multipart 上传 +pypdf>=4.0,<6.0 +xlrd>=1.2,<3.0 +python-multipart>=0.0.9,<1.0 diff --git a/inquiry-agent/tests/test_attachment_inquiry.py b/inquiry-agent/tests/test_attachment_inquiry.py new file mode 100644 index 0000000..76870b3 --- /dev/null +++ b/inquiry-agent/tests/test_attachment_inquiry.py @@ -0,0 +1,339 @@ +""" +私聊附件询价：发卡、H5/图并入、口播复用文字流程。 +""" + +from __future__ import annotations + +import os +import sys +import unittest + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +os.environ.setdefault("YTD_ENV", "test") +os.environ.setdefault("REDIS_BACKEND", "memory") +os.environ.setdefault("CHECKPOINT_BACKEND", "memory") +os.environ.setdefault("MESSAGE_STORE_BACKEND", "memory") +os.environ.setdefault("LLM_DATA_USAGE_CONFIRMED", "false") +os.environ.setdefault("LLM_ALLOW_NETWORK", "false") +os.environ.setdefault("LEDGER_BACKEND", "memory") + +from agent.channel.queue import MemoryMessageStore +from agent.channel.wecom.models import InboundMessage +from agent.handlers import attachment_inquiry as attach_mod +from agent.handlers.attachment_inquiry import ( + accept_h5_uploads, + finish_queued_attachment, + handle_attachment_inquiry, + join_attachment_image, +) +from agent.handlers.text_inquiry import handle_text_inquiry +from agent.policy import inquiry_copy as copy +from agent.policy.air_text_flow import reset_air_text_flow_for_test +from agent.policy.attachment_wave import get_attach_wave_book, reset_attach_wave_book_for_test +from agent.policy.image_wave import reset_image_wave_book_for_test +from agent.policy.land_text_flow import reset_land_text_flow_for_test +from agent.policy.sea_text_flow import get_sea_text_flow, reset_sea_text_flow_for_test +from agent.routing.decision import RouteDecision +from agent.routing.dispatch import dispatch_inbound +from tests.test_sea_text_inquiry import COMPLETE_SEA + +SEA_ORAL = """海运 +起运港：上海 +目的港：洛杉矶 +品名：普货 +货量：20吨 +整柜或拼柜：整柜 +箱型箱量：1x40HQ +""" + + +def _extract(text: str): + def _fn(_name: str, _raw: bytes): + body = (text or "").strip() + return {"ok": True, "text": body, "empty": not bool(body), "error": ""} + + return _fn + + +def _vision(text: str, *, ok: bool = True, error: str = ""): + def _fn(**_kwargs): + body = (text or "").strip() + return {"ok": ok, "text": body, "empty": not bool(body), "error": error} + + return _fn + + +def _img(sender: str, mid: str) -> InboundMessage: + return InboundMessage( + sender_id=sender, + message_id=mid, + content="", + msg_type="image", + media={"media_id": mid, "filename": f"{mid}.jpg"}, + ) + + +class AttachmentInquiryTests(unittest.TestCase): + def setUp(self) -> None: + reset_sea_text_flow_for_test() + reset_land_text_flow_for_test() + reset_air_text_flow_for_test() + reset_attach_wave_book_for_test() + reset_image_wave_book_for_test() + self.store = MemoryMessageStore() + + def _bodies(self) -> list[str]: + return [i.content for i in self.store._outbox.values()] + + def _extras(self) -> list[dict]: + return [i.payload for i in self.store._outbox.values()] + + def test_intent_sends_upload_card_not_ticket(self) -> None: + phase = handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="帮我传个Excel询价，海运上海到纽约"), + store=self.store, + ) + self.assertEqual(phase, "wait_upload") + self.assertIsNone(get_sea_text_flow().session_of("s1")) + extras = self._extras() + self.assertTrue(extras) + card = extras[0].get("template_card") or {} + self.assertEqual(card.get("card_type"), "text_notice") + self.assertEqual(card.get("main_title", {}).get("title"), "上传询价附件") + self.assertIn("上传询价附件", (card.get("jump_list") or [{}])[0].get("title") or "") + url = (card.get("card_action") or {}).get("url") or "" + self.assertIn("/inquiry-form/upload/", url) + + def test_xlsx_complete_creates_sea_ticket(self) -> None: + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价"), + store=self.store, + ) + phase, hint = accept_h5_uploads( + sender_id="s1", + files=[("sea.xlsx", b"fake")], + store=self.store, + extract_fn=_extract(SEA_ORAL), + attach_text_fn=lambda **k: { + "ok": True, + "text": str((k.get("payload") or {}).get("raw_text") or ""), + "empty": False, + "error": "", + }, + put_fn=lambda **kw: {"ok": True, "key": kw.get("key") or "k"}, + ) + self.assertEqual(hint, copy.ATTACH_H5_OK) + self.assertIn(copy.ATTACH_RECOGNIZING, self._bodies()) + sess = get_sea_text_flow().session_of("s1") + self.assertIsNotNone(sess) + self.assertTrue((sess.work_order_no or "").strip()) + ticket = get_sea_text_flow().ledger.get_ticket(work_order_no=sess.work_order_no) + self.assertTrue(ticket.attachments) + + def test_empty_file_asks_mode(self) -> None: + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价"), + store=self.store, + ) + accept_h5_uploads( + sender_id="s1", + files=[("empty.xlsx", b"x")], + store=self.store, + extract_fn=_extract(""), + attach_text_fn=lambda **_k: {"ok": True, "text": "", "empty": True, "error": ""}, + ) + bodies = self._bodies() + self.assertTrue(any("没看清" in b for b in bodies)) + self.assertTrue(any("海运" in b and "空运" in b for b in bodies)) + + def test_image_during_wait_joins_same_ticket(self) -> None: + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价"), + store=self.store, + ) + join_attachment_image( + _img("s1", "mid-1"), + store=self.store, + vision_fn=_vision("截图"), + download_fn=lambda _m: (b"x", ""), + process_now=False, + ) + accept_h5_uploads( + sender_id="s1", + files=[("sea.xlsx", b"x")], + store=self.store, + extract_fn=_extract(SEA_ORAL), + attach_text_fn=lambda **k: { + "ok": True, + "text": str((k.get("payload") or {}).get("raw_text") or ""), + "empty": False, + }, + process_now=True, + ) + from agent.handlers.attachment_inquiry import finish_attach_wave + from agent.policy.attachment_wave import get_attach_wave_book + + book = get_attach_wave_book() + book.complete_material("s1", 0, text="截图") + finish_attach_wave("s1", store=self.store) + acks = [c for c in self._bodies() if c in {copy.IMAGE_RECOGNIZING, copy.ATTACH_RECOGNIZING}] + self.assertEqual(len(acks), 1) + self.assertEqual(acks[0], copy.IMAGE_RECOGNIZING) + sess = get_sea_text_flow().session_of("s1") + self.assertTrue(sess and sess.work_order_no) + + def test_mid_clarify_then_attach_is_new_ticket(self) -> None: + handle_text_inquiry( + InboundMessage(sender_id="s1", message_id="old", content="海运出号"), + store=self.store, + injected_facts=COMPLETE_SEA, + injected_mode="SEA", + ) + wo_old = get_sea_text_flow().session_of("s1").work_order_no + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m2", content="我有个excel要询价"), + store=self.store, + ) + accept_h5_uploads( + sender_id="s1", + files=[("sea.xlsx", b"x")], + store=self.store, + extract_fn=_extract(SEA_ORAL), + attach_text_fn=lambda **k: { + "ok": True, + "text": str((k.get("payload") or {}).get("raw_text") or ""), + "empty": False, + }, + ) + wo_new = get_sea_text_flow().session_of("s1").work_order_no + self.assertNotEqual(wo_old, wo_new) + self.assertIsNotNone(get_sea_text_flow().ledger.get_ticket(work_order_no=wo_old)) + + def test_reject_png_on_h5(self) -> None: + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价"), + store=self.store, + ) + phase, hint = accept_h5_uploads( + sender_id="s1", + files=[("shot.png", b"\x89PNG")], + store=self.store, + ) + self.assertEqual(phase, "reject_type") + self.assertIn("Excel", hint) + self.assertIsNone(get_sea_text_flow().session_of("s1")) + + def test_dispatch_image_joins_wait_not_new_image_inquiry(self) -> None: + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价"), + store=self.store, + ) + name = dispatch_inbound( + _img("s1", "mid-x"), + RouteDecision(intent="image_inquiry"), + ) + self.assertEqual(name, "attachment_inquiry") + + def test_follow_text_not_create_ticket_before_upload(self) -> None: + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价"), + store=self.store, + ) + name = dispatch_inbound( + InboundMessage(sender_id="s1", message_id="m2", content="海运上海到纽约 2个40尺"), + RouteDecision(intent="ordinary_text_inquiry"), + ) + self.assertEqual(name, "attach_wave_follow_text") + self.assertIsNone(get_sea_text_flow().session_of("s1")) + + def test_after_business_old_h5_rejected(self) -> None: + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价"), + store=self.store, + ) + accept_h5_uploads( + sender_id="s1", + files=[("sea.xlsx", b"x")], + store=self.store, + extract_fn=_extract(SEA_ORAL), + attach_text_fn=lambda **k: { + "ok": True, + "text": str((k.get("payload") or {}).get("raw_text") or ""), + "empty": False, + }, + ) + phase, _hint = accept_h5_uploads( + sender_id="s1", + files=[("again.xlsx", b"x")], + store=self.store, + extract_fn=_extract(SEA_ORAL), + ) + self.assertEqual(phase, "expired") + + def test_worker_reads_bytes_after_http_stash_cleared(self) -> None: + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价"), + store=self.store, + ) + accept_h5_uploads( + sender_id="s1", + files=[("sea.xlsx", b"fake-xlsx-bytes")], + store=self.store, + process_now=False, + ) + attach_mod._STASH.clear() + wave = get_attach_wave_book().active("s1") + self.assertIsNotNone(wave) + key = f"inquiry-attach/s1/{wave.wave_id}/0/sea.xlsx" + finish_queued_attachment( + { + "sender_id": "s1", + "index": 0, + "role": "file", + "filename": "sea.xlsx", + "object_key": key, + "wave_id": wave.wave_id, + "__extract": _extract(SEA_ORAL), + "__attach_text": lambda **k: { + "ok": True, + "text": str((k.get("payload") or {}).get("raw_text") or ""), + "empty": False, + "error": "", + }, + } + ) + self.assertIsNotNone(get_sea_text_flow().session_of("s1")) + + def test_second_unreadable_wave_still_replies(self) -> None: + def _empty(_name: str, _raw: bytes): + return {"ok": True, "text": "", "empty": True, "error": ""} + + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价"), + store=self.store, + ) + accept_h5_uploads( + sender_id="s1", + files=[("blank.xlsx", b"xx")], + store=self.store, + extract_fn=_empty, + ) + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m2", content="附件询价"), + store=self.store, + ) + accept_h5_uploads( + sender_id="s1", + files=[("blank2.xlsx", b"yy")], + store=self.store, + extract_fn=_empty, + ) + fails = [b for b in self._bodies() if "没看清" in b or "没认出来" in b] + self.assertGreaterEqual(len(fails), 2, fails) + + +if __name__ == "__main__": + unittest.main() diff --git a/inquiry-agent/tests/test_attachment_upload_h5.py b/inquiry-agent/tests/test_attachment_upload_h5.py new file mode 100644 index 0000000..8d1e8cf --- /dev/null +++ b/inquiry-agent/tests/test_attachment_upload_h5.py @@ -0,0 +1,96 @@ +""" +附件上传 H5：token 不含账号，过期 410，拒收图片。 +""" + +from __future__ import annotations + +import os +import sys +import unittest + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +os.environ.setdefault("YTD_ENV", "test") +os.environ.setdefault("REDIS_BACKEND", "memory") +os.environ.setdefault("CHECKPOINT_BACKEND", "memory") +os.environ.setdefault("MESSAGE_STORE_BACKEND", "memory") +os.environ.setdefault("LEDGER_BACKEND", "memory") + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from agent.channel.h5 import router as h5_router +from agent.channel.h5.copy_store import build_upload_url +from agent.channel.wecom.models import InboundMessage +from agent.handlers.attachment_inquiry import handle_attachment_inquiry, lookup_upload_sender +from agent.policy.attachment_wave import reset_attach_wave_book_for_test +from agent.policy.image_wave import reset_image_wave_book_for_test +from agent.policy.air_text_flow import reset_air_text_flow_for_test +from agent.policy.land_text_flow import reset_land_text_flow_for_test +from agent.policy.sea_text_flow import reset_sea_text_flow_for_test + + +class AttachmentUploadH5Tests(unittest.TestCase): + def setUp(self) -> None: + reset_attach_wave_book_for_test() + reset_image_wave_book_for_test() + reset_sea_text_flow_for_test() + reset_air_text_flow_for_test() + reset_land_text_flow_for_test() + app = FastAPI() + app.include_router(h5_router) + self.client = TestClient(app) + + def _open_card(self) -> str: + handle_attachment_inquiry( + InboundMessage(sender_id="s1", message_id="m1", content="附件询价") + ) + from agent.policy.attachment_wave import get_attach_wave_book + + wave = get_attach_wave_book().active("s1") + self.assertIsNotNone(wave) + url = build_upload_url(wave.token) + self.assertIn("/upload/", url) + return wave.token + + def test_open_page_ok_and_expired(self) -> None: + token = self._open_card() + self.assertEqual(lookup_upload_sender(token), "s1") + hit = self.client.get(f"/inquiry-form/upload/{token}") + self.assertEqual(hit.status_code, 200) + self.assertIn("Excel", hit.text) + self.assertNotIn("s1", hit.text) + miss = self.client.get("/inquiry-form/upload/not-a-real-token") + self.assertEqual(miss.status_code, 410) + self.assertIn("过期", miss.text) + + def test_upload_page_uses_card_chrome(self) -> None: + from agent.channel.h5.upload_page import expired_upload_page, render_upload_message, render_upload_page + + page = render_upload_page() + self.assertIn('class="card"', page) + self.assertIn('class="drop"', page) + self.assertIn('id="picked"', page) + self.assertIn("一次可选多个", page) + self.assertIn("Excel", page) + self.assertNotIn("s1", page) + ok = render_upload_message("上传成功", "已收到，请回企微查看识别结果。") + dead = expired_upload_page() + self.assertIn('class="card"', ok) + self.assertIn('class="card"', dead) + self.assertIn("过期", dead) + + def test_post_png_rejected(self) -> None: + token = self._open_card() + hit = self.client.post( + f"/inquiry-form/upload/{token}", + files=[("files", ("shot.png", b"\x89PNG", "image/png"))], + ) + self.assertEqual(hit.status_code, 400) + self.assertIn("Excel", hit.text) + + +if __name__ == "__main__": + unittest.main() diff --git a/inquiry-agent/tests/test_attachment_wave.py b/inquiry-agent/tests/test_attachment_wave.py new file mode 100644 index 0000000..b1f8380 --- /dev/null +++ b/inquiry-agent/tests/test_attachment_wave.py @@ -0,0 +1,116 @@ +""" +私聊附件等材料波次：发卡等待、发图/H5 并入、30 分钟时效。 +""" + +from __future__ import annotations + +import json +import os +import sys +import unittest + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +from agent.policy.attachment_wave import ( + WAIT_TTL_SECONDS, + AttachWaveBook, + reset_attach_wave_book_for_test, +) + + +class AttachWaveTests(unittest.TestCase): + def setUp(self) -> None: + self.book = reset_attach_wave_book_for_test() + + def test_wait_ttl_is_30_minutes(self) -> None: + self.assertEqual(WAIT_TTL_SECONDS, 1800) + + def test_open_wait_is_active_without_files(self) -> None: + wave = self.book.open_wait("s1", precede_text="海运上海到纽约", token="tok1", now=100.0) + self.assertTrue(wave.wave_id) + self.assertEqual(wave.precede_text, "海运上海到纽约") + self.assertFalse(wave.started) + self.assertIsNotNone(self.book.active("s1", now=100.0)) + + def test_wait_expires_after_30_minutes_if_not_started(self) -> None: + self.book.open_wait("s1", precede_text="海运", token="tok1", now=100.0) + self.assertIsNone(self.book.active("s1", now=100.0 + 1800.1)) + + def test_follow_text_while_waiting(self) -> None: + self.book.open_wait("s1", precede_text="", token="tok1", now=1.0) + self.assertTrue(self.book.append_follow_text("s1", "只要整柜")) + self.assertFalse(self.book.append_follow_text("s1", "")) + + def test_join_image_starts_recognition(self) -> None: + self.book.open_wait("s1", precede_text="海运", token="tok1", now=1.0) + wave, how = self.book.join_image("s1", "mid-a", filename="a.jpg", now=2.0) + self.assertEqual(how, "joined") + self.assertTrue(wave.started) + self.assertEqual(self.book.pending(wave), 1) + + def test_add_files_keeps_upload_order(self) -> None: + self.book.open_wait("s1", precede_text="", token="tok1", now=1.0) + wave = self.book.add_files("s1", ["sea.xlsx", "air.pdf"], now=2.0) + self.assertTrue(wave.started) + self.assertEqual([m.filename for m in wave.materials], ["sea.xlsx", "air.pdf"]) + self.assertEqual(self.book.pending(wave), 2) + + def test_compose_oral_uses_material_order(self) -> None: + self.book.open_wait("s1", precede_text="开口字", token="tok1", now=1.0) + self.book.add_files("s1", ["a.xlsx", "b.pdf"], now=2.0) + self.book.append_follow_text("s1", "只要40尺") + wave = self.book.active("s1", now=2.0) + self.assertIsNotNone(wave) + self.book.complete_material("s1", 1, text="空运 上海到纽约") + self.book.complete_material("s1", 0, text="海运 宁波到汉堡") + oral = self.book.compose_oral("s1") + self.assertEqual(oral, "开口字\n海运 宁波到汉堡\n空运 上海到纽约\n只要40尺") + + def test_after_business_done_not_active(self) -> None: + self.book.open_wait("s1", precede_text="", token="tok1", now=1.0) + self.book.add_files("s1", ["a.xlsx"], now=2.0) + self.book.complete_material("s1", 0, text="海运") + self.book.mark_business_done("s1") + self.assertIsNone(self.book.active("s1", now=3.0)) + self.assertFalse(self.book.append_follow_text("s1", "补一句")) + + def test_image_after_business_rejected(self) -> None: + self.book.open_wait("s1", precede_text="", token="tok1", now=1.0) + self.book.mark_business_done("s1") + wave, how = self.book.join_image("s1", "mid-b", filename="b.jpg", now=2.0) + self.assertEqual(how, "inactive") + self.assertIsNone(wave) + + def test_http_and_worker_share_kv(self) -> None: + kv = _MemoryWaveKV() + http = reset_attach_wave_book_for_test(kv=kv) + worker = AttachWaveBook(kv=kv) + http.open_wait("s1", precede_text="海运", token="tok", now=1.0) + self.assertIsNotNone(worker.active("s1", now=1.0)) + worker.add_files("s1", ["a.xlsx"], now=2.0) + worker.complete_material("s1", 0, text="宁波到汉堡") + worker.mark_business_done("s1") + self.assertIsNone(http.active("s1", now=3.0)) + + +class _MemoryWaveKV: + """单测假 Redis：两份账本共用同一字典。""" + + def __init__(self) -> None: + self._d: dict[str, dict] = {} + + def get_json(self, kind: str, sender_id: str): + row = self._d.get(f"{kind}:{sender_id}") + return json.loads(json.dumps(row)) if row is not None else None + + def set_json(self, kind: str, sender_id: str, data: dict) -> None: + self._d[f"{kind}:{sender_id}"] = json.loads(json.dumps(data)) + + def delete(self, kind: str, sender_id: str) -> None: + self._d.pop(f"{kind}:{sender_id}", None) + + +if __name__ == "__main__": + unittest.main() diff --git a/inquiry-agent/tests/test_extract_attachment.py b/inquiry-agent/tests/test_extract_attachment.py new file mode 100644 index 0000000..51e641e --- /dev/null +++ b/inquiry-agent/tests/test_extract_attachment.py @@ -0,0 +1,61 @@ +""" +Excel / PDF 抽纯文本：不猜字段。 +""" + +from __future__ import annotations + +import io +import os +import sys +import unittest + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +from agent.policy.attachment_extract import allowed_attachment, extract_attachment_text + + +def _xlsx_bytes() -> bytes: + from openpyxl import Workbook + + wb = Workbook() + ws = wb.active + ws.title = "询价" + ws["A1"] = "海运" + ws["B1"] = "上海到洛杉矶" + ws["A2"] = "1x40HQ" + buf = io.BytesIO() + wb.save(buf) + return buf.getvalue() + + +class ExtractAttachmentTests(unittest.TestCase): + def test_allow_only_excel_pdf(self) -> None: + self.assertTrue(allowed_attachment("a.XLSX")) + self.assertTrue(allowed_attachment("b.xls")) + self.assertTrue(allowed_attachment("c.pdf")) + self.assertFalse(allowed_attachment("d.png")) + self.assertFalse(allowed_attachment("e.docx")) + self.assertFalse(allowed_attachment("f.zip")) + + def test_reject_image_bytes(self) -> None: + got = extract_attachment_text("shot.png", b"\x89PNG") + self.assertFalse(got["ok"]) + self.assertEqual(got["error"], "only_xlsx_xls_pdf") + + def test_xlsx_cells_become_text(self) -> None: + got = extract_attachment_text("sea.xlsx", _xlsx_bytes()) + self.assertTrue(got["ok"]) + self.assertIn("海运", got["text"]) + self.assertIn("上海到洛杉矶", got["text"]) + self.assertIn("1x40HQ", got["text"]) + + def test_empty_bytes_fail(self) -> None: + got = extract_attachment_text("a.xlsx", b"") + self.assertFalse(got["ok"]) + self.assertTrue(got["empty"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/inquiry-agent/tests/test_land_copy.py b/inquiry-agent/tests/test_land_copy.py index 4dd207d..443b39d 100644 --- a/inquiry-agent/tests/test_land_copy.py +++ b/inquiry-agent/tests/test_land_copy.py @@ -320,6 +320,26 @@ class LandCopyTest(unittest.TestCase): self.assertTrue(merged.startswith(copy.IMAGE_UNREADABLE)) self.assertIn(copy.ASK_TRANSPORT, merged) + def test_attachment_copy_and_card(self): + self.assertEqual(copy.ATTACH_RECOGNIZING, "正在识别附件") + self.assertIn("没看清", copy.ATTACH_UNREADABLE) + self.assertIn("暂时没认出来", copy.ATTACH_TECH_FAIL) + payload = copy.attachment_upload_card("https://ai.ytd-scm.com/inquiry-form/upload/tok") + card = payload["template_card"] + self.assertEqual(payload["msgtype"], "template_card") + # 现网上传卡是 text_notice。news_notice 必须带 card_image，企微会 42044 后退成纯文本。 + self.assertEqual(card["card_type"], "text_notice") + self.assertNotIn("card_image", card) + self.assertEqual(card["source"]["desc"], "附件询价") + self.assertEqual(card["main_title"]["title"], "上传询价附件") + self.assertIn("Excel", card["main_title"]["desc"]) + self.assertIn("图片请直接在企微聊天窗口发送", card["main_title"]["desc"]) + self.assertIn("图片请直接在当前企微会话发送", card["sub_title_text"]) + self.assertEqual(card["jump_list"][0]["title"], "上传询价附件") + self.assertIn("/upload/tok", card["card_action"]["url"]) + self.assertEqual(card["card_action"]["type"], 1) + self.assertNotIn("userid", card["card_action"]["url"]) + if __name__ == "__main__": unittest.main() diff --git a/inquiry-agent/tests/test_mode_attachment_text.py b/inquiry-agent/tests/test_mode_attachment_text.py new file mode 100644 index 0000000..02ae1c2 --- /dev/null +++ b/inquiry-agent/tests/test_mode_attachment_text.py @@ -0,0 +1,70 @@ +""" +千问附件文本：只整理抽出原文，不算缺项。 +""" + +from __future__ import annotations + +import os +import sys +import unittest + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +from agent.llm.mode_attachment_text import invoke_attachment_text, invoke_shell +from agent.llm.modes import LlmMode + + +class ModeAttachmentTextTests(unittest.TestCase): + def test_shell_uses_raw_text(self) -> None: + got = invoke_shell(payload={"raw_text": "海运 宁波到汉堡 1x40HQ"}) + self.assertTrue(got["ok"]) + self.assertIn("宁波到汉堡", got["text"]) + self.assertFalse(got["empty"]) + + def test_blank_is_empty(self) -> None: + got = invoke_shell(payload={"assistant_text": " "}) + self.assertTrue(got["ok"]) + self.assertTrue(got["empty"]) + + def test_injected_chat_gets_raw_only(self) -> None: + seen: dict = {} + + class _R: + ok = True + content = "海运 上海到洛杉矶" + error = "" + + def chat(**kwargs): + seen.update(kwargs) + return _R() + + got = invoke_attachment_text( + payload={"raw_text": "# 询价\n海运\t上海到洛杉矶"}, + chat_fn=chat, + allow_network=False, + ) + self.assertTrue(got["ok"]) + self.assertEqual(got["text"], "海运 上海到洛杉矶") + self.assertEqual(seen["mode"], LlmMode.ATTACHMENT_TEXT) + sys_msg = seen["messages"][0]["content"] + self.assertIn("不要猜", sys_msg) + self.assertNotIn("缺项", seen["messages"][1]["content"] or "x") + + def test_chat_fail(self) -> None: + class _R: + ok = False + content = "" + error = "timeout" + + got = invoke_attachment_text( + payload={"raw_text": "海运"}, + chat_fn=lambda **_k: _R(), + ) + self.assertFalse(got["ok"]) + self.assertEqual(got["error"], "timeout") + + +if __name__ == "__main__": + unittest.main()