2651 lines
109 KiB
Python
Executable File
2651 lines
109 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
import argparse
|
||
import asyncio
|
||
import base64
|
||
import datetime as dt
|
||
import hashlib
|
||
import hmac
|
||
import io
|
||
import json
|
||
import logging
|
||
import mimetypes
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import urllib.parse
|
||
import uuid
|
||
|
||
from dingtalk_stream import AckMessage, CallbackMessage, ChatbotHandler, ChatbotMessage
|
||
from dingtalk_stream import Credential, DingTalkStreamClient
|
||
import requests
|
||
|
||
try:
|
||
from PIL import Image
|
||
except Exception:
|
||
Image = None
|
||
|
||
try:
|
||
import boto3
|
||
from botocore.config import Config as BotoConfig
|
||
except Exception:
|
||
boto3 = None
|
||
BotoConfig = None
|
||
|
||
|
||
MARKET_HINT_KEYWORDS = (
|
||
"市场分析",
|
||
"市场数据分析",
|
||
"行业分析",
|
||
"类目分析",
|
||
"竞品分析",
|
||
"趋势分析",
|
||
)
|
||
|
||
SALES_HINT_KEYWORDS = (
|
||
"销售数据",
|
||
"销售报表",
|
||
"销售看板",
|
||
"公司销售",
|
||
"业绩报表",
|
||
"业绩数据",
|
||
"费比监控",
|
||
)
|
||
|
||
SENSITIVE_SALES_KEYWORDS = (
|
||
"销售",
|
||
"业绩",
|
||
"售后",
|
||
"退款",
|
||
"店铺",
|
||
"发货",
|
||
"工单",
|
||
"费比",
|
||
)
|
||
|
||
EXPLICIT_AMOUNT_KEYWORDS = (
|
||
"金额",
|
||
"费用",
|
||
"收入",
|
||
"利润",
|
||
"成本",
|
||
"客单价",
|
||
"gmv",
|
||
"销售额",
|
||
"退款额",
|
||
"毛利",
|
||
)
|
||
|
||
INSTALL_MEDIA_HINT_KEYWORDS = (
|
||
"安装图纸",
|
||
"安装视频",
|
||
"安装说明书",
|
||
"图纸地址",
|
||
"视频地址",
|
||
"图纸链接",
|
||
"视频链接",
|
||
"订单号查图纸",
|
||
"订单号查视频",
|
||
)
|
||
|
||
INSTALL_MEDIA_STITCH_HINT_KEYWORDS = (
|
||
"拼接",
|
||
"合并",
|
||
"一张图",
|
||
"整合图",
|
||
"汇总图",
|
||
)
|
||
|
||
VIDEO_ANALYSIS_HINT_KEYWORDS = (
|
||
"视频",
|
||
"video",
|
||
"短片",
|
||
"片段",
|
||
)
|
||
|
||
VIDEO_ANALYSIS_ACTION_KEYWORDS = (
|
||
"分析",
|
||
"解读",
|
||
"总结",
|
||
"提炼",
|
||
"复盘",
|
||
"讲解",
|
||
"评价",
|
||
"建议",
|
||
"看看",
|
||
"看下",
|
||
"说明",
|
||
)
|
||
|
||
VIDEO_SCRIPT_HINT_KEYWORDS = (
|
||
"脚本",
|
||
"分镜",
|
||
"复刻",
|
||
"仿拍",
|
||
"同款",
|
||
"类似视频",
|
||
"拍摄方案",
|
||
"镜头脚本",
|
||
)
|
||
|
||
DALONGXIA_ANALYTICS_HINT_KEYWORDS = (
|
||
"壹点需求",
|
||
"需求系统",
|
||
"任务进度总览",
|
||
"进度总览",
|
||
"数据分析总览",
|
||
"需求统计",
|
||
"设计师绩效",
|
||
"流程效率",
|
||
"预警",
|
||
"维度分析",
|
||
"主部门",
|
||
"designerperformance",
|
||
"dimensionanalysis",
|
||
"taskprogressoverview",
|
||
"designerdeduction",
|
||
"designerscore",
|
||
"primarydept",
|
||
"primarytask",
|
||
"analytics",
|
||
)
|
||
|
||
EXCEL_IMAGE_LINK_HINT_KEYWORDS = (
|
||
"图片链接",
|
||
"链接转图片",
|
||
"转成图片",
|
||
"回填",
|
||
"excel",
|
||
"xlsx",
|
||
"表格",
|
||
"文件",
|
||
)
|
||
|
||
|
||
URL_RE = re.compile(
|
||
r"https?://[A-Za-z0-9\-._~:/?#\[\]@!$&'()*+,;=%]+",
|
||
re.IGNORECASE,
|
||
)
|
||
IP_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}(?::\d+)?\b")
|
||
DOMAIN_RE = re.compile(r"\b(?:[a-zA-Z0-9-]+\.)+(?:com|cn|net|ai|org|io)\b", re.IGNORECASE)
|
||
IMAGE_EXT_RE = re.compile(r"\.(?:png|jpg|jpeg|gif|webp|bmp|svg)(?:$|[?#])", re.IGNORECASE)
|
||
VIDEO_EXT_RE = re.compile(r"\.(?:mp4|mov|m4v|webm|mkv|avi|flv|wmv|ts|m3u8)(?:$|[?#])", re.IGNORECASE)
|
||
ORDER_ID_RE = re.compile(r"\b\d{12,24}(?:-\d{2,4})?\b")
|
||
SMALLTALK_SET = {
|
||
"你好",
|
||
"您好",
|
||
"嗨",
|
||
"哈喽",
|
||
"在吗",
|
||
"在嘛",
|
||
"早上好",
|
||
"上午好",
|
||
"中午好",
|
||
"下午好",
|
||
"晚上好",
|
||
"hi",
|
||
"hello",
|
||
}
|
||
|
||
|
||
|
||
def bool_env(name: str, default: bool) -> bool:
|
||
raw = os.getenv(name)
|
||
if raw is None:
|
||
return default
|
||
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
||
|
||
|
||
class OpenClawDingTalkBot(ChatbotHandler):
|
||
def __init__(self, logger: logging.Logger):
|
||
super().__init__()
|
||
self.logger = logger
|
||
self.openclaw_bin = os.getenv("OPENCLAW_BIN", "/usr/bin/openclaw")
|
||
self.openclaw_timeout = int(os.getenv("OPENCLAW_TIMEOUT", "600"))
|
||
self.market_brief_bin = os.getenv("MARKET_BRIEF_BIN", "/usr/local/bin/market-brief-report")
|
||
self.market_brief_timeout = int(os.getenv("MARKET_BRIEF_TIMEOUT", "180"))
|
||
self.install_media_bin = os.getenv("INSTALL_MEDIA_BIN", "/usr/local/bin/install-media-lookup")
|
||
self.install_media_timeout = int(os.getenv("INSTALL_MEDIA_TIMEOUT", "120"))
|
||
self.prefix = os.getenv("BOT_PREFIX", "/oc").strip()
|
||
self.require_prefix = bool_env("REQUIRE_PREFIX", False)
|
||
self.reply_max_chars = int(os.getenv("REPLY_MAX_CHARS", "3500"))
|
||
self.session_prefix = os.getenv("OPENCLAW_SESSION_PREFIX", "dingtalk")
|
||
self.hide_source_urls = bool_env("HIDE_SOURCE_URLS", True)
|
||
self.send_processing_hint = bool_env("SEND_PROCESSING_HINT", True)
|
||
self.dedupe_ttl_seconds = int(os.getenv("DEDUPE_TTL_SECONDS", "300"))
|
||
self.session_max_lines = int(os.getenv("OPENCLAW_SESSION_MAX_LINES", "120"))
|
||
self.sessions_dir = os.getenv("OPENCLAW_SESSIONS_DIR", "/root/.openclaw/agents/main/sessions")
|
||
self.unmask_url_hosts = {
|
||
h.strip().lower()
|
||
for h in os.getenv(
|
||
"UNMASK_URL_HOSTS",
|
||
"img.alicdn.com,static.jiansun.vip,img.jiansuntuzhi.cn,v.youku.com",
|
||
).split(",")
|
||
if h.strip()
|
||
}
|
||
self._recent_event_ts = {}
|
||
self.processing_hint_text = os.getenv(
|
||
"PROCESSING_HINT_TEXT",
|
||
"已收到，正在处理你的需求，请稍等1-3分钟。",
|
||
).strip()
|
||
self.reply_style = os.getenv("REPLY_STYLE", "action_card").strip().lower()
|
||
self.action_card_title = os.getenv("ACTION_CARD_TITLE", "OpenClaw 助理").strip()
|
||
self.action_card_tips = os.getenv("ACTION_CARD_TIPS", "").strip()
|
||
self.action_card_logo = os.getenv("ACTION_CARD_LOGO", "").strip()
|
||
self.action_card_btn_text = os.getenv("ACTION_CARD_BTN_TEXT", "").strip()
|
||
self.action_card_btn_url = os.getenv("ACTION_CARD_BTN_URL", "").strip()
|
||
self.smalltalk_reply_text = os.getenv(
|
||
"SMALLTALK_REPLY_TEXT",
|
||
"你好，我在。请直接告诉我你的任务目标（例如：查询某店上周售后原因TOP5并给出改进建议）。",
|
||
).strip()
|
||
self.max_parallel_jobs = max(1, min(int(os.getenv("BOT_MAX_PARALLEL_JOBS", "4")), 16))
|
||
self._work_semaphore = asyncio.Semaphore(self.max_parallel_jobs)
|
||
self._conversation_locks = {}
|
||
self.card_disable_seconds = max(300, min(int(os.getenv("CARD_DISABLE_SECONDS", "43200")), 86400))
|
||
self.card_disabled_until = 0.0
|
||
self.install_media_stitch_enabled = bool_env("INSTALL_MEDIA_STITCH_ENABLED", True)
|
||
self.install_media_stitch_min_images = max(2, min(int(os.getenv("INSTALL_MEDIA_STITCH_MIN_IMAGES", "2")), 10))
|
||
self.install_media_stitch_max_images = max(
|
||
self.install_media_stitch_min_images,
|
||
min(int(os.getenv("INSTALL_MEDIA_STITCH_MAX_IMAGES", "20")), 50),
|
||
)
|
||
self.install_media_fetch_timeout = max(3, min(int(os.getenv("INSTALL_MEDIA_FETCH_TIMEOUT", "12")), 60))
|
||
self.install_media_max_width = max(720, min(int(os.getenv("INSTALL_MEDIA_MAX_WIDTH", "1600")), 4000))
|
||
self.s3_endpoint = (
|
||
os.getenv("INSTALL_MEDIA_S3_ENDPOINT", "").strip()
|
||
or os.getenv("S3_ENDPOINT", "").strip()
|
||
)
|
||
self.s3_bucket = (
|
||
os.getenv("INSTALL_MEDIA_S3_BUCKET", "").strip()
|
||
or os.getenv("S3_BUCKET", "").strip()
|
||
)
|
||
self.s3_region = (
|
||
os.getenv("INSTALL_MEDIA_S3_REGION", "").strip()
|
||
or os.getenv("S3_REGION", "us-east-1").strip()
|
||
)
|
||
self.s3_access_key = (
|
||
os.getenv("INSTALL_MEDIA_S3_ACCESS_KEY_ID", "").strip()
|
||
or os.getenv("S3_ACCESS_KEY_ID", "").strip()
|
||
)
|
||
self.s3_secret_key = (
|
||
os.getenv("INSTALL_MEDIA_S3_SECRET_ACCESS_KEY", "").strip()
|
||
or os.getenv("S3_SECRET_ACCESS_KEY", "").strip()
|
||
)
|
||
self.s3_public_domain = (
|
||
os.getenv("INSTALL_MEDIA_S3_PUBLIC_DOMAIN", "").strip()
|
||
or os.getenv("S3_PUBLIC_DOMAIN", "").strip()
|
||
)
|
||
self.s3_key_prefix = (
|
||
os.getenv("INSTALL_MEDIA_S3_FILE_PATH", "").strip()
|
||
or os.getenv("S3_FILE_PATH", "").strip()
|
||
)
|
||
self.s3_acl = (
|
||
os.getenv("INSTALL_MEDIA_S3_ACL", "").strip()
|
||
or os.getenv("S3_ACL", "public-read").strip()
|
||
)
|
||
self.s3_path_style = bool_env("INSTALL_MEDIA_S3_PATH_STYLE", bool_env("S3_PATH_STYLE", True))
|
||
if self.s3_key_prefix and not self.s3_key_prefix.endswith("/"):
|
||
self.s3_key_prefix += "/"
|
||
self.image_analysis_enabled = bool_env("IMAGE_ANALYSIS_ENABLED", True)
|
||
self.image_analysis_model = os.getenv("IMAGE_ANALYSIS_MODEL", "qwen2.5vl:3b").strip() or "qwen2.5vl:3b"
|
||
self.image_analysis_timeout = max(10, min(int(os.getenv("IMAGE_ANALYSIS_TIMEOUT", "90")), 300))
|
||
self.image_max_images = max(1, min(int(os.getenv("IMAGE_MAX_IMAGES", "3")), 8))
|
||
self.image_download_timeout = max(3, min(int(os.getenv("IMAGE_DOWNLOAD_TIMEOUT", "20")), 60))
|
||
self.image_max_bytes = max(262144, min(int(os.getenv("IMAGE_MAX_BYTES", "10485760")), 31457280))
|
||
self.image_upload_to_s3 = bool_env("IMAGE_UPLOAD_TO_S3", True)
|
||
self.image_vision_output_max_chars = max(300, min(int(os.getenv("IMAGE_VISION_OUTPUT_MAX_CHARS", "2200")), 6000))
|
||
self.video_analysis_enabled = bool_env("VIDEO_ANALYSIS_ENABLED", True)
|
||
self.video_analysis_model = os.getenv("VIDEO_ANALYSIS_MODEL", "qwen-vl-plus").strip() or "qwen-vl-plus"
|
||
self.video_analysis_fallback_model = (
|
||
os.getenv("VIDEO_ANALYSIS_FALLBACK_MODEL", "qwen-vl-plus").strip() or "qwen-vl-plus"
|
||
)
|
||
self.video_analysis_base_url = (
|
||
os.getenv("VIDEO_ANALYSIS_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions").strip()
|
||
or "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
|
||
)
|
||
self.video_analysis_timeout = max(15, min(int(os.getenv("VIDEO_ANALYSIS_TIMEOUT", "360")), 900))
|
||
self.video_analysis_max_tokens = max(256, min(int(os.getenv("VIDEO_ANALYSIS_MAX_TOKENS", "1800")), 8192))
|
||
self.video_analysis_max_urls = max(1, min(int(os.getenv("VIDEO_ANALYSIS_MAX_URLS", "2")), 3))
|
||
self.excel_link_feature_enabled = bool_env("EXCEL_IMAGE_LINK_FEATURE_ENABLED", True)
|
||
self.excel_converter_script = (
|
||
os.getenv(
|
||
"EXCEL_IMAGE_CONVERTER_SCRIPT",
|
||
"/opt/openclaw-dingtalk-bridge/skills/dingtalk-excel-image-return/convert_excel_image_links.py",
|
||
).strip()
|
||
or "/opt/openclaw-dingtalk-bridge/skills/dingtalk-excel-image-return/convert_excel_image_links.py"
|
||
)
|
||
self.excel_converter_timeout = max(60, min(int(os.getenv("EXCEL_IMAGE_CONVERTER_TIMEOUT", "480")), 3600))
|
||
self.excel_link_column = os.getenv("EXCEL_IMAGE_LINK_COLUMN", "图片链接").strip() or "图片链接"
|
||
self.excel_image_column_header = os.getenv("EXCEL_IMAGE_COLUMN_HEADER", "图片").strip() or "图片"
|
||
self.excel_image_max_width = max(80, min(int(os.getenv("EXCEL_IMAGE_MAX_WIDTH", "140")), 1200))
|
||
self.excel_image_max_height = max(80, min(int(os.getenv("EXCEL_IMAGE_MAX_HEIGHT", "140")), 1200))
|
||
self.excel_image_download_timeout = max(3, min(int(os.getenv("EXCEL_IMAGE_DOWNLOAD_TIMEOUT", "20")), 120))
|
||
self.excel_file_download_timeout = max(5, min(int(os.getenv("EXCEL_FILE_DOWNLOAD_TIMEOUT", "60")), 300))
|
||
self.excel_skip_tls_verify = bool_env("EXCEL_IMAGE_SKIP_TLS_VERIFY", False)
|
||
self.excel_text_prompt_cooldown_seconds = max(
|
||
10,
|
||
min(int(os.getenv("EXCEL_TEXT_PROMPT_COOLDOWN_SECONDS", "120")), 600),
|
||
)
|
||
self._recent_excel_success_ts = {}
|
||
|
||
def _should_handle(self, msg: ChatbotMessage, text: str) -> bool:
|
||
if text.startswith(self.prefix):
|
||
return True
|
||
if msg.is_in_at_list:
|
||
return True
|
||
return not self.require_prefix
|
||
|
||
def _normalize_prompt(self, text: str) -> str:
|
||
text = text.strip()
|
||
if self.prefix and text.startswith(self.prefix):
|
||
text = text[len(self.prefix):].strip()
|
||
text = re.sub(r"^@\S+\s*", "", text).strip()
|
||
return text
|
||
|
||
def _is_smalltalk_prompt(self, prompt: str) -> bool:
|
||
normalized = re.sub(r"[\s，,。.!！?？~～]+", "", (prompt or "").strip().lower())
|
||
if not normalized:
|
||
return False
|
||
if normalized in SMALLTALK_SET:
|
||
return True
|
||
if normalized.startswith(("你好", "您好", "hi", "hello")) and len(normalized) <= 6:
|
||
return True
|
||
return False
|
||
|
||
def _safe_session_component(self, raw: str) -> str:
|
||
original = (raw or "default").strip().replace(" ", "_")
|
||
safe = re.sub(r"[^A-Za-z0-9_-]+", "-", original)
|
||
safe = re.sub(r"-{2,}", "-", safe).strip("-")
|
||
if not safe:
|
||
safe = "default"
|
||
if safe != original:
|
||
digest = hashlib.sha1(original.encode("utf-8")).hexdigest()[:8]
|
||
safe = f"{safe}-{digest}"
|
||
return safe[:96]
|
||
|
||
def _session_id(self, msg: ChatbotMessage) -> str:
|
||
raw_conversation_id = str(msg.conversation_id or "default")
|
||
safe_conversation_id = self._safe_session_component(raw_conversation_id)
|
||
base_session = f"{self.session_prefix}-{safe_conversation_id}"
|
||
|
||
session_file = os.path.join(self.sessions_dir, f"{base_session}.jsonl")
|
||
if self.session_max_lines > 0 and os.path.exists(session_file):
|
||
try:
|
||
with open(session_file, "rb") as fh:
|
||
line_count = sum(1 for _ in fh)
|
||
if line_count >= self.session_max_lines:
|
||
rotate_suffix = time.strftime("%Y%m%d%H%M")
|
||
rotated = f"{base_session}-{rotate_suffix}"
|
||
self.logger.info(
|
||
"session auto-rotated: base=%s lines=%s rotated=%s",
|
||
base_session,
|
||
line_count,
|
||
rotated,
|
||
)
|
||
return rotated
|
||
except Exception:
|
||
self.logger.exception("failed to inspect session file: %s", session_file)
|
||
|
||
return base_session
|
||
|
||
def _event_key(self, callback: CallbackMessage, msg: ChatbotMessage, prompt: str) -> str:
|
||
if getattr(msg, "message_id", None):
|
||
return f"msg:{msg.message_id}"
|
||
|
||
data = callback.data if isinstance(callback.data, dict) else {}
|
||
for key in ("msgId", "messageId", "eventId", "processQueryKey"):
|
||
value = data.get(key)
|
||
if value:
|
||
return f"data:{key}:{value}"
|
||
|
||
headers = getattr(callback, "headers", {}) or {}
|
||
for key in ("messageId", "eventId", "processQueryKey"):
|
||
value = headers.get(key)
|
||
if value:
|
||
return f"hdr:{key}:{value}"
|
||
|
||
raw = "|".join(
|
||
[
|
||
str(getattr(msg, "conversation_id", "") or ""),
|
||
str(getattr(msg, "sender_id", "") or ""),
|
||
str(getattr(msg, "sender_staff_id", "") or ""),
|
||
str(getattr(msg, "create_at", "") or ""),
|
||
prompt[:200],
|
||
]
|
||
)
|
||
return "fp:" + hashlib.sha1(raw.encode("utf-8")).hexdigest()
|
||
|
||
def _seen_or_mark_event(self, key: str) -> bool:
|
||
now = time.time()
|
||
expire_before = now - max(self.dedupe_ttl_seconds, 30)
|
||
|
||
stale_keys = [k for k, ts in self._recent_event_ts.items() if ts < expire_before]
|
||
for stale_key in stale_keys:
|
||
self._recent_event_ts.pop(stale_key, None)
|
||
|
||
if key in self._recent_event_ts:
|
||
return True
|
||
|
||
self._recent_event_ts[key] = now
|
||
return False
|
||
|
||
def _is_market_prompt(self, prompt: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", prompt)
|
||
|
||
if any(k in normalized for k in SALES_HINT_KEYWORDS):
|
||
return False
|
||
|
||
if "电脑桌" in normalized:
|
||
return True
|
||
|
||
if any(k in normalized for k in MARKET_HINT_KEYWORDS):
|
||
return True
|
||
|
||
if "市场" in normalized and any(
|
||
k in normalized for k in ("搜索", "飙升词", "热词", "需求供给比", "价格带", "客单价", "洞察", "趋势")
|
||
):
|
||
return True
|
||
|
||
if "市场" in normalized and "分析" in normalized:
|
||
return True
|
||
|
||
return False
|
||
|
||
def _is_install_media_prompt(self, prompt: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", prompt)
|
||
if not normalized:
|
||
return False
|
||
if any(k in normalized for k in INSTALL_MEDIA_HINT_KEYWORDS):
|
||
return True
|
||
if "安装" in normalized and ("图纸" in normalized or "视频" in normalized):
|
||
return True
|
||
return False
|
||
|
||
def _is_install_media_stitch_prompt(self, prompt: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", prompt)
|
||
if not normalized:
|
||
return False
|
||
return any(k in normalized for k in INSTALL_MEDIA_STITCH_HINT_KEYWORDS)
|
||
|
||
def _is_video_url(self, raw_url: str) -> bool:
|
||
try:
|
||
parsed = urllib.parse.urlparse(raw_url)
|
||
except Exception:
|
||
return False
|
||
|
||
if parsed.scheme not in {"http", "https"}:
|
||
return False
|
||
|
||
host = (parsed.hostname or "").lower()
|
||
target = f"{parsed.path or ''}?{parsed.query or ''}"
|
||
if VIDEO_EXT_RE.search(target):
|
||
return True
|
||
|
||
video_hosts = (
|
||
"douyin.com",
|
||
"iesdouyin.com",
|
||
"bilibili.com",
|
||
"youku.com",
|
||
"qq.com",
|
||
"weixin.qq.com",
|
||
"ixigua.com",
|
||
"kuaishou.com",
|
||
"mgtv.com",
|
||
"1905.com",
|
||
"xiaohongshu.com",
|
||
"xhslink.com",
|
||
"tiktok.com",
|
||
"youtube.com",
|
||
"youtu.be",
|
||
)
|
||
return any(host == h or host.endswith(f".{h}") for h in video_hosts)
|
||
|
||
def _extract_video_urls(self, text: str, limit: int = 2) -> list:
|
||
urls = []
|
||
seen = set()
|
||
for match in URL_RE.finditer(text or ""):
|
||
raw = match.group(0).rstrip(".,;!?")
|
||
if not self._is_video_url(raw):
|
||
continue
|
||
if raw in seen:
|
||
continue
|
||
seen.add(raw)
|
||
urls.append(raw)
|
||
if len(urls) >= limit:
|
||
break
|
||
return urls
|
||
|
||
def _extract_any_urls(self, text: str, limit: int = 5) -> list:
|
||
urls = []
|
||
seen = set()
|
||
for match in URL_RE.finditer(text or ""):
|
||
raw = match.group(0).rstrip(".,;!?")
|
||
if raw in seen:
|
||
continue
|
||
seen.add(raw)
|
||
urls.append(raw)
|
||
if len(urls) >= limit:
|
||
break
|
||
return urls
|
||
|
||
def _is_video_analysis_prompt(self, prompt: str) -> bool:
|
||
if self._is_install_media_prompt(prompt):
|
||
return False
|
||
|
||
normalized = re.sub(r"\s+", "", (prompt or "")).lower()
|
||
if not normalized:
|
||
return False
|
||
|
||
has_video_word = ("视频" in normalized) or ("video" in normalized)
|
||
has_video_intent = any(k in normalized for k in VIDEO_ANALYSIS_HINT_KEYWORDS) or any(
|
||
k in normalized for k in VIDEO_ANALYSIS_ACTION_KEYWORDS
|
||
)
|
||
if has_video_word and has_video_intent:
|
||
return True
|
||
video_urls = self._extract_video_urls(prompt, limit=1)
|
||
if not video_urls:
|
||
return False
|
||
if has_video_intent:
|
||
return True
|
||
return len(normalized) <= 120
|
||
|
||
def _needs_video_script(self, prompt: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", (prompt or "")).lower()
|
||
if not normalized:
|
||
return False
|
||
return any(k in normalized for k in VIDEO_SCRIPT_HINT_KEYWORDS)
|
||
|
||
def _is_dalongxia_analytics_prompt(self, prompt: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", (prompt or "")).lower()
|
||
if not normalized:
|
||
return False
|
||
if any(k in normalized for k in DALONGXIA_ANALYTICS_HINT_KEYWORDS):
|
||
return True
|
||
if "任务进度" in normalized and ("总览" in normalized or "overview" in normalized):
|
||
return True
|
||
if "设计师绩效" in normalized or "流程效率" in normalized:
|
||
return True
|
||
if "primary" in normalized and ("dept" in normalized or "task" in normalized):
|
||
return True
|
||
return False
|
||
|
||
def _parse_analytics_mode(self, prompt: str) -> str:
|
||
normalized = re.sub(r"\s+", "", (prompt or "")).lower()
|
||
compact = re.sub(r"[^a-z0-9\u4e00-\u9fa5]+", "", normalized)
|
||
|
||
mode_aliases = (
|
||
("primarytaskbranddistribution", "primary_task_brand_distribution"),
|
||
("primarytaskflowefficiency", "primary_task_flow_efficiency"),
|
||
("primarydeptstatistics", "primary_dept_statistics"),
|
||
("primarydeptuserranking", "primary_dept_user_ranking"),
|
||
("primarydeptranking", "primary_dept_ranking"),
|
||
("primarytask", "primary_task_flow_efficiency"),
|
||
("primarydept", "primary_dept_ranking"),
|
||
("designerdeductiondetails", "designer_deduction"),
|
||
("designerscoredetails", "designer_score"),
|
||
("designerdeduction", "designer_deduction"),
|
||
("designerscore", "designer_score"),
|
||
("taskprogressoverview", "task_progress_overview"),
|
||
("taskprogress", "task_progress_overview"),
|
||
("progress", "task_progress_overview"),
|
||
("demandstatistics", "demand_statistics"),
|
||
("designerperformance", "designer_performance"),
|
||
("processefficiency", "process_efficiency"),
|
||
("dimensionanalysis", "dimension_analysis"),
|
||
("overview", "overview"),
|
||
("warnings", "warnings"),
|
||
)
|
||
for token, mode in mode_aliases:
|
||
if token in compact:
|
||
return mode
|
||
|
||
if "主部门" in normalized and "人员" in normalized and "排名" in normalized:
|
||
return "primary_dept_user_ranking"
|
||
if "主部门" in normalized and "品牌" in normalized and "分布" in normalized:
|
||
return "primary_task_brand_distribution"
|
||
if "主部门" in normalized and "任务流程" in normalized and "效率" in normalized:
|
||
return "primary_task_flow_efficiency"
|
||
if "主部门" in normalized and "统计" in normalized:
|
||
return "primary_dept_statistics"
|
||
if "主部门" in normalized and "排名" in normalized:
|
||
return "primary_dept_ranking"
|
||
if "设计师" in normalized and ("扣分明细" in normalized or "扣分详情" in normalized):
|
||
return "designer_deduction"
|
||
if "设计师" in normalized and ("评分明细" in normalized or "得分明细" in normalized):
|
||
return "designer_score"
|
||
if "维度" in normalized and "分析" in normalized:
|
||
return "dimension_analysis"
|
||
if "流程效率" in normalized:
|
||
return "process_efficiency"
|
||
if "设计师绩效" in normalized:
|
||
return "designer_performance"
|
||
if "需求统计" in normalized:
|
||
return "demand_statistics"
|
||
if "预警" in normalized:
|
||
return "warnings"
|
||
if "任务进度" in normalized and ("总览" in normalized or "overview" in normalized):
|
||
return "task_progress_overview"
|
||
if "数据分析" in normalized and "总览" in normalized:
|
||
return "overview"
|
||
return "overview"
|
||
|
||
def _parse_analytics_date_range(self, prompt: str) -> str:
|
||
normalized = re.sub(r"\s+", "", (prompt or "")).lower()
|
||
if any(k in normalized for k in ("all", "全部", "全量", "所有时间")):
|
||
return "all"
|
||
if any(k in normalized for k in ("quarter", "季度", "q1", "q2", "q3", "q4")):
|
||
return "quarter"
|
||
if any(k in normalized for k in ("week", "本周", "上周", "近7天", "最近7天", "7天")):
|
||
return "week"
|
||
return "month"
|
||
|
||
def _parse_analytics_month(self, prompt: str) -> str:
|
||
text = prompt or ""
|
||
m = re.search(r"(20\d{2})[年/-](\d{1,2})", text)
|
||
if not m:
|
||
m = re.search(r"(20\d{2})(\d{2})", text)
|
||
if m:
|
||
year = int(m.group(1))
|
||
month = int(m.group(2))
|
||
if 1 <= month <= 12:
|
||
return f"{year:04d}-{month:02d}"
|
||
now = time.localtime()
|
||
return f"{now.tm_year:04d}-{now.tm_mon:02d}"
|
||
|
||
def _parse_analytics_ids(self, prompt: str) -> tuple:
|
||
text = prompt or ""
|
||
dept_id = ""
|
||
user_id = ""
|
||
|
||
m = re.search(r"(?:部门ID|deptId|dept_id)[：:\s]*([0-9]{1,20})", text, re.IGNORECASE)
|
||
if m:
|
||
dept_id = m.group(1)
|
||
m = re.search(r"(?:用户ID|设计师ID|成员ID|userId|user_id)[：:\s]*([0-9]{1,20})", text, re.IGNORECASE)
|
||
if m:
|
||
user_id = m.group(1)
|
||
return dept_id, user_id
|
||
|
||
def _normalize_analytics_datetime(self, raw: str, end_of_day: bool = False) -> str:
|
||
text = (raw or "").strip()
|
||
if not text:
|
||
return ""
|
||
text = text.replace("/", "-").replace(" ", "T")
|
||
if re.match(r"^\d{4}-\d{1,2}-\d{1,2}$", text):
|
||
suffix = "23:59:59" if end_of_day else "00:00:00"
|
||
return f"{text}T{suffix}"
|
||
if re.match(r"^\d{4}-\d{1,2}-\d{1,2}T\d{1,2}$", text):
|
||
return f"{text}:00:00"
|
||
if re.match(r"^\d{4}-\d{1,2}-\d{1,2}T\d{1,2}:\d{1,2}$", text):
|
||
return f"{text}:00"
|
||
if re.match(r"^\d{4}-\d{1,2}-\d{1,2}T\d{1,2}:\d{1,2}:\d{1,2}$", text):
|
||
return text
|
||
return ""
|
||
|
||
def _parse_analytics_time_window(self, prompt: str, month: str) -> tuple:
|
||
text = prompt or ""
|
||
start_time = ""
|
||
end_time = ""
|
||
|
||
range_match = re.search(
|
||
r"(20\d{2}[/-]\d{1,2}[/-]\d{1,2})\s*(?:到|至|~|-)\s*(20\d{2}[/-]\d{1,2}[/-]\d{1,2})",
|
||
text,
|
||
)
|
||
if range_match:
|
||
start_time = self._normalize_analytics_datetime(range_match.group(1), end_of_day=False)
|
||
end_time = self._normalize_analytics_datetime(range_match.group(2), end_of_day=True)
|
||
|
||
if not start_time:
|
||
m = re.search(r"(?:startTime|start_time|开始时间)[：:=\s]*([0-9/:\- T]{8,25})", text, re.IGNORECASE)
|
||
if m:
|
||
start_time = self._normalize_analytics_datetime(m.group(1), end_of_day=False)
|
||
if not end_time:
|
||
m = re.search(r"(?:endTime|end_time|结束时间)[：:=\s]*([0-9/:\- T]{8,25})", text, re.IGNORECASE)
|
||
if m:
|
||
end_time = self._normalize_analytics_datetime(m.group(1), end_of_day=True)
|
||
|
||
if start_time and end_time:
|
||
return start_time, end_time
|
||
|
||
month_match = re.match(r"^(20\d{2})-(\d{2})$", month or "")
|
||
if month_match:
|
||
year = int(month_match.group(1))
|
||
mon = int(month_match.group(2))
|
||
first = dt.datetime(year, mon, 1, 0, 0, 0)
|
||
if mon == 12:
|
||
next_first = dt.datetime(year + 1, 1, 1, 0, 0, 0)
|
||
else:
|
||
next_first = dt.datetime(year, mon + 1, 1, 0, 0, 0)
|
||
last = next_first - dt.timedelta(seconds=1)
|
||
return first.strftime("%Y-%m-%dT%H:%M:%S"), last.strftime("%Y-%m-%dT%H:%M:%S")
|
||
|
||
now = dt.datetime.now()
|
||
first = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
||
last = now.replace(hour=23, minute=59, second=59, microsecond=0)
|
||
return first.strftime("%Y-%m-%dT%H:%M:%S"), last.strftime("%Y-%m-%dT%H:%M:%S")
|
||
|
||
def _parse_include_task_progress(self, prompt: str) -> bool:
|
||
return bool(
|
||
re.search(
|
||
r"(含|包含|带上|带|include)\s*任务进度|includeTaskProgress\s*[:=]\s*true",
|
||
prompt or "",
|
||
re.IGNORECASE,
|
||
)
|
||
)
|
||
|
||
def _build_dalongxia_analytics_request(self, mode: str, app_key: str, app_secret: str, params: dict) -> tuple:
|
||
timestamp = str(int(time.time() * 1000))
|
||
date_range = params.get("date_range", "month")
|
||
month = params.get("month", "")
|
||
dept_id = str(params.get("dept_id", "") or "")
|
||
user_id = str(params.get("user_id", "") or "")
|
||
include_task_progress = bool(params.get("include_task_progress", False))
|
||
include_str = str(include_task_progress).lower()
|
||
start_time = params.get("start_time", "")
|
||
end_time = params.get("end_time", "")
|
||
|
||
if mode == "task_progress_overview":
|
||
action = "analytics_task_progress_overview"
|
||
payload = "|".join([action, app_key, date_range, month, dept_id, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"dateRange": date_range,
|
||
"month": month,
|
||
}
|
||
if dept_id:
|
||
body["deptId"] = dept_id
|
||
return "/api/open/analytics/task-progress/overview", "壹点需求系统：任务进度总览", body
|
||
|
||
if mode == "demand_statistics":
|
||
action = "analytics_demand_statistics"
|
||
payload = "|".join([action, app_key, date_range, month, dept_id, user_id, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"dateRange": date_range,
|
||
"month": month,
|
||
}
|
||
if dept_id:
|
||
body["deptId"] = dept_id
|
||
if user_id:
|
||
body["userId"] = user_id
|
||
return "/api/open/analytics/demand-statistics", "壹点需求系统：需求统计", body
|
||
|
||
if mode == "designer_performance":
|
||
action = "analytics_designer_performance"
|
||
payload = "|".join([action, app_key, month, dept_id, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"month": month,
|
||
}
|
||
if dept_id:
|
||
body["deptId"] = dept_id
|
||
return "/api/open/analytics/designer-performance", "壹点需求系统：设计师绩效", body
|
||
|
||
if mode == "process_efficiency":
|
||
action = "analytics_process_efficiency"
|
||
payload = "|".join([action, app_key, month, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"month": month,
|
||
}
|
||
return "/api/open/analytics/process-efficiency", "壹点需求系统：流程效率", body
|
||
|
||
if mode == "warnings":
|
||
action = "analytics_warnings"
|
||
payload = "|".join([action, app_key, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
}
|
||
return "/api/open/analytics/warnings", "壹点需求系统：预警信息", body
|
||
|
||
if mode == "dimension_analysis":
|
||
action = "analytics_dimension_analysis"
|
||
payload = "|".join([action, app_key, date_range, month, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"dateRange": date_range,
|
||
"month": month,
|
||
}
|
||
return "/api/open/analytics/dimension-analysis", "壹点需求系统：维度分析", body
|
||
|
||
if mode == "designer_deduction":
|
||
action = "analytics_designer_deduction_details"
|
||
detail_user_id = user_id or os.getenv("ANALYTICS_DEFAULT_USER_ID", "38").strip()
|
||
payload = "|".join([action, app_key, str(detail_user_id), month, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"userId": str(detail_user_id),
|
||
"month": month,
|
||
}
|
||
return "/api/open/analytics/designer-deduction-details", "壹点需求系统：设计师扣分明细", body
|
||
|
||
if mode == "designer_score":
|
||
action = "analytics_designer_score_details"
|
||
detail_user_id = user_id or os.getenv("ANALYTICS_DEFAULT_USER_ID", "38").strip()
|
||
payload = "|".join([action, app_key, str(detail_user_id), month, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"userId": str(detail_user_id),
|
||
"month": month,
|
||
}
|
||
return "/api/open/analytics/designer-score-details", "壹点需求系统：设计师得分明细", body
|
||
|
||
if mode == "primary_dept_statistics":
|
||
action = "analytics_primary_dept_statistics"
|
||
payload = "|".join([action, app_key, start_time, end_time, dept_id, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"startTime": start_time,
|
||
"endTime": end_time,
|
||
}
|
||
if dept_id:
|
||
body["deptId"] = dept_id
|
||
return "/api/open/analytics/primary-dept/statistics", "壹点需求系统：主部门统计", body
|
||
|
||
if mode == "primary_dept_ranking":
|
||
action = "analytics_primary_dept_ranking"
|
||
payload = "|".join([action, app_key, start_time, end_time, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"startTime": start_time,
|
||
"endTime": end_time,
|
||
}
|
||
return "/api/open/analytics/primary-dept/ranking", "壹点需求系统：主部门排名", body
|
||
|
||
if mode == "primary_task_flow_efficiency":
|
||
action = "analytics_primary_task_flow_efficiency"
|
||
payload = "|".join([action, app_key, start_time, end_time, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"startTime": start_time,
|
||
"endTime": end_time,
|
||
}
|
||
return "/api/open/analytics/primary-task/flow-efficiency", "壹点需求系统：主部门任务流转效率", body
|
||
|
||
if mode == "primary_task_brand_distribution":
|
||
action = "analytics_primary_task_brand_distribution"
|
||
payload = "|".join([action, app_key, start_time, end_time, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"startTime": start_time,
|
||
"endTime": end_time,
|
||
}
|
||
return "/api/open/analytics/primary-task/brand-distribution", "壹点需求系统：主部门任务品牌分布", body
|
||
|
||
if mode == "primary_dept_user_ranking":
|
||
action = "analytics_primary_dept_user_ranking"
|
||
payload = "|".join([action, app_key, start_time, end_time, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"startTime": start_time,
|
||
"endTime": end_time,
|
||
}
|
||
return "/api/open/analytics/primary-dept/user-ranking", "壹点需求系统：主部门成员排名", body
|
||
|
||
action = "analytics_overview"
|
||
payload = "|".join([action, app_key, date_range, month, dept_id, user_id, include_str, timestamp])
|
||
body = {
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": self._build_open_api_sign(app_secret, payload),
|
||
"dateRange": date_range,
|
||
"month": month,
|
||
"includeTaskProgress": include_task_progress,
|
||
}
|
||
if dept_id:
|
||
body["deptId"] = dept_id
|
||
if user_id:
|
||
body["userId"] = user_id
|
||
return "/api/open/analytics/overview", "壹点需求系统：数据分析总览", body
|
||
|
||
def _format_dalongxia_analytics_result(self, mode: str, title: str, body: dict, data: dict) -> str:
|
||
def _f(v):
|
||
if v is None:
|
||
return "-"
|
||
if isinstance(v, float):
|
||
s = f"{v:.2f}"
|
||
return s.rstrip("0").rstrip(".")
|
||
return str(v)
|
||
|
||
def _is_num(v):
|
||
return isinstance(v, (int, float))
|
||
|
||
def _sum_num(items, key):
|
||
total = 0
|
||
found = False
|
||
for it in items:
|
||
v = it.get(key) if isinstance(it, dict) else None
|
||
if _is_num(v):
|
||
total += v
|
||
found = True
|
||
return total if found else None
|
||
|
||
query_params = []
|
||
for key in ("dateRange", "month", "deptId", "userId", "includeTaskProgress", "startTime", "endTime"):
|
||
if key in body and body[key] not in ("", None):
|
||
query_params.append(f"{key}={body[key]}")
|
||
|
||
lines = [title]
|
||
if query_params:
|
||
lines.append(f"- 查询参数: {', '.join(query_params)}")
|
||
if isinstance(data, dict):
|
||
if data.get("time"):
|
||
lines.append(f"- 统计时间: {data.get('time')}")
|
||
if data.get("traceId"):
|
||
lines.append(f"- traceId: {data.get('traceId')}")
|
||
|
||
root = (data or {}).get("data") if isinstance(data, dict) else None
|
||
if root is None:
|
||
root = data
|
||
|
||
if mode == "task_progress_overview" and isinstance(root, dict):
|
||
metrics = root.get("metrics") or {}
|
||
status = root.get("statusDistribution") or {}
|
||
trend = (root.get("trend") or [])[:5]
|
||
lines.append(
|
||
"- 核心指标: "
|
||
f"总任务={_f(metrics.get('totalTasks'))}，完成={_f(metrics.get('completedTasks'))}，"
|
||
f"处理中={_f(metrics.get('inProgressTasks'))}，逾期={_f(metrics.get('overdueTasks'))}，"
|
||
f"完成率={_f(metrics.get('completionRate'))}%"
|
||
)
|
||
if status:
|
||
parts = [f"{k}={_f(v)}" for k, v in status.items() if _is_num(v)]
|
||
if parts:
|
||
lines.append(f"- 状态分布: {'，'.join(parts[:8])}")
|
||
if trend:
|
||
lines.append("- 趋势(最近5个点):")
|
||
for t in trend:
|
||
lines.append(
|
||
f" - {_f(t.get('date'))}: 新建={_f(t.get('createdCount'))}，完成={_f(t.get('completedCount'))}"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
if mode == "designer_performance" and isinstance(root, dict):
|
||
items = root.get("designerStats") or []
|
||
if not isinstance(items, list):
|
||
items = []
|
||
items = sorted(
|
||
[it for it in items if isinstance(it, dict)],
|
||
key=lambda x: (x.get("completedCount") or 0, x.get("receivedCount") or 0),
|
||
reverse=True,
|
||
)
|
||
total_received = _sum_num(items, "receivedCount")
|
||
total_completed = _sum_num(items, "completedCount")
|
||
total_score = _sum_num(items, "totalScore")
|
||
total_net_score = _sum_num(items, "netScore")
|
||
lines.append(
|
||
"- 汇总: "
|
||
f"设计师数={len(items)}，接收任务={_f(total_received)}，完成任务={_f(total_completed)}，"
|
||
f"总积分={_f(total_score)}，净积分={_f(total_net_score)}"
|
||
)
|
||
lines.append("- TOP设计师(按完成量，含每人积分):")
|
||
for idx, it in enumerate(items[:10], 1):
|
||
total_score_item = it.get("totalScore")
|
||
net_score_item = it.get("netScore")
|
||
point_value = net_score_item if _is_num(net_score_item) else total_score_item
|
||
lines.append(
|
||
f" {idx}. {_f(it.get('realName') or it.get('userName'))}｜{_f(it.get('deptName'))}｜"
|
||
f"完成={_f(it.get('completedCount'))}｜积分={_f(point_value)}｜"
|
||
f"扣分={_f(it.get('deductionScore'))}｜目标={_f(it.get('targetScore'))}｜"
|
||
f"准时率={_f(it.get('onTimeRate'))}%｜完成率={_f(it.get('completionRate'))}%"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
if mode == "dimension_analysis" and isinstance(root, dict):
|
||
items = root.get("categoryAnalysis") or []
|
||
if not isinstance(items, list):
|
||
items = []
|
||
items = sorted(
|
||
[it for it in items if isinstance(it, dict)],
|
||
key=lambda x: (x.get("totalTasks") or 0),
|
||
reverse=True,
|
||
)
|
||
lines.append(f"- 类目数: {len(items)}")
|
||
lines.append("- TOP类目(按任务量):")
|
||
for idx, it in enumerate(items[:10], 1):
|
||
lines.append(
|
||
f" {idx}. {_f(it.get('categoryName'))}｜任务={_f(it.get('totalTasks'))}｜"
|
||
f"完成={_f(it.get('completedTasks'))}｜进行中={_f(it.get('inProgressTasks'))}｜完成率={_f(it.get('completionRate'))}%"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
if mode == "overview" and isinstance(root, dict):
|
||
module_keys = list(root.keys())
|
||
lines.append(f"- 模块: {', '.join(module_keys[:8]) if module_keys else '-'}")
|
||
demand = root.get("demandStatistics") or {}
|
||
user_stats = demand.get("userStats") if isinstance(demand, dict) else []
|
||
if isinstance(user_stats, list) and user_stats:
|
||
lines.append(f"- 需求统计: 用户数={len(user_stats)}")
|
||
top = sorted(
|
||
[it for it in user_stats if isinstance(it, dict)],
|
||
key=lambda x: (x.get("totalDemands") or 0),
|
||
reverse=True,
|
||
)[:5]
|
||
lines.append("- 需求量TOP5:")
|
||
for idx, it in enumerate(top, 1):
|
||
lines.append(
|
||
f" {idx}. {_f(it.get('realName') or it.get('userName'))}｜{_f(it.get('deptName'))}｜"
|
||
f"总需求={_f(it.get('totalDemands'))}｜完成={_f(it.get('completedDemands'))}"
|
||
)
|
||
task_progress = root.get("taskProgressOverview") or root.get("taskProgress") or {}
|
||
if isinstance(task_progress, dict) and task_progress.get("metrics"):
|
||
m = task_progress.get("metrics") or {}
|
||
lines.append(
|
||
"- 任务进度: "
|
||
f"总任务={_f(m.get('totalTasks'))}，完成={_f(m.get('completedTasks'))}，逾期={_f(m.get('overdueTasks'))}"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
if mode in {
|
||
"primary_dept_ranking",
|
||
"primary_dept_statistics",
|
||
"primary_task_flow_efficiency",
|
||
"primary_task_brand_distribution",
|
||
"primary_dept_user_ranking",
|
||
}:
|
||
items = root if isinstance(root, list) else []
|
||
if not items and isinstance(root, dict):
|
||
for key in ("items", "list", "rows", "data"):
|
||
if isinstance(root.get(key), list):
|
||
items = root.get(key)
|
||
break
|
||
items = [it for it in items if isinstance(it, dict)]
|
||
lines.append(f"- 条数: {len(items)}")
|
||
lines.append("- TOP10:")
|
||
for idx, it in enumerate(items[:10], 1):
|
||
name = it.get("deptName") or it.get("userName") or it.get("brandName") or it.get("name")
|
||
task_count = it.get("taskCount") or it.get("totalTasks") or it.get("count")
|
||
rate = it.get("approvalRate") or it.get("completionRate") or it.get("onTimeRate")
|
||
extra = it.get("avgResponseHours") or it.get("avgCycleHours")
|
||
part = f" {idx}. {_f(name)}｜任务={_f(task_count)}"
|
||
if rate is not None:
|
||
part += f"｜比率={_f(rate)}%"
|
||
if extra is not None:
|
||
part += f"｜时长={_f(extra)}h"
|
||
lines.append(part)
|
||
return "\n".join(lines)
|
||
|
||
if mode == "warnings":
|
||
items = root if isinstance(root, list) else []
|
||
if isinstance(root, dict):
|
||
for key in ("warnings", "items", "list", "rows"):
|
||
if isinstance(root.get(key), list):
|
||
items = root.get(key)
|
||
break
|
||
if isinstance(items, list):
|
||
lines.append(f"- 预警条数: {len(items)}")
|
||
for idx, it in enumerate(items[:10], 1):
|
||
if isinstance(it, dict):
|
||
msg = it.get("message") or it.get("warnMsg") or it.get("content") or it.get("title")
|
||
lines.append(f" {idx}. {_f(msg)}")
|
||
else:
|
||
lines.append(f" {idx}. {_f(it)}")
|
||
return "\n".join(lines)
|
||
|
||
if mode in {"designer_deduction", "designer_score"}:
|
||
items = root if isinstance(root, list) else []
|
||
if isinstance(root, dict):
|
||
for key in ("details", "items", "list", "rows", "data"):
|
||
if isinstance(root.get(key), list):
|
||
items = root.get(key)
|
||
break
|
||
items = [it for it in items if isinstance(it, dict)]
|
||
lines.append(f"- 记录数: {len(items)}")
|
||
lines.append("- 前10条:")
|
||
for idx, it in enumerate(items[:10], 1):
|
||
lines.append(
|
||
f" {idx}. 时间={_f(it.get('time') or it.get('createTime') or it.get('date'))}｜"
|
||
f"分值={_f(it.get('score') or it.get('deductionScore') or it.get('value'))}｜"
|
||
f"原因={_f(it.get('reason') or it.get('remark') or it.get('content'))}"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
# fallback
|
||
fallback = json.dumps(data, ensure_ascii=False, indent=2)
|
||
if len(fallback) > 1800:
|
||
fallback = fallback[:1800] + "\n...(已截断)"
|
||
lines.append("- 结果预览:")
|
||
lines.append(fallback)
|
||
return "\n".join(lines)
|
||
|
||
def _build_open_api_sign(self, secret: str, payload: str) -> str:
|
||
return hmac.new(secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest()
|
||
|
||
def _post_open_api_json(self, base_url: str, path: str, body: dict) -> dict:
|
||
url = f"{base_url.rstrip('/')}{path}"
|
||
timeout = max(10, min(int(os.getenv("OPEN_API_TIMEOUT", "60")), 300))
|
||
last_err = None
|
||
|
||
def _parse_response(resp):
|
||
content_type = (resp.headers.get("Content-Type") or "").lower()
|
||
if "json" not in content_type:
|
||
text = (resp.text or "").strip()
|
||
raise RuntimeError(f"HTTP {resp.status_code} 非JSON响应: {text[:300]}")
|
||
data = resp.json()
|
||
if not resp.ok:
|
||
code = data.get("code", "unknown")
|
||
msg = data.get("message", data)
|
||
raise RuntimeError(f"HTTP {resp.status_code} / code={code} / msg={msg}")
|
||
if data.get("code") != 200:
|
||
raise RuntimeError(f"code={data.get('code')} / msg={data.get('message')}")
|
||
return data.get("data")
|
||
|
||
try:
|
||
resp = requests.post(url, json=body, timeout=timeout)
|
||
return _parse_response(resp)
|
||
except Exception as exc:
|
||
last_err = exc
|
||
self.logger.warning("open api request attempt1 failed: %s", exc)
|
||
|
||
try:
|
||
with requests.Session() as session:
|
||
session.trust_env = False
|
||
resp = session.post(url, json=body, timeout=timeout)
|
||
return _parse_response(resp)
|
||
except Exception as exc:
|
||
last_err = exc
|
||
self.logger.warning("open api request attempt2 failed: %s", exc)
|
||
|
||
raise last_err or RuntimeError("open api request failed")
|
||
|
||
def _run_dalongxia_analytics(self, prompt: str) -> str:
|
||
base_url = (
|
||
os.getenv("OPEN_API_BASE_URL", "").strip()
|
||
or os.getenv("OPEN_API_BASE", "").strip()
|
||
or "https://qd.yydd.ai"
|
||
)
|
||
app_key = os.getenv("OPEN_API_APP_KEY", "").strip()
|
||
app_secret = os.getenv("OPEN_API_APP_SECRET", "").strip()
|
||
if not app_key or not app_secret:
|
||
return "壹点需求系统查询失败：缺少 OPEN_API_APP_KEY / OPEN_API_APP_SECRET。"
|
||
|
||
mode = self._parse_analytics_mode(prompt)
|
||
date_range = self._parse_analytics_date_range(prompt)
|
||
month = self._parse_analytics_month(prompt)
|
||
dept_id, user_id = self._parse_analytics_ids(prompt)
|
||
include_task_progress = self._parse_include_task_progress(prompt)
|
||
start_time, end_time = self._parse_analytics_time_window(prompt, month)
|
||
try:
|
||
path, title, body = self._build_dalongxia_analytics_request(
|
||
mode=mode,
|
||
app_key=app_key,
|
||
app_secret=app_secret,
|
||
params={
|
||
"date_range": date_range,
|
||
"month": month,
|
||
"dept_id": dept_id,
|
||
"user_id": user_id,
|
||
"include_task_progress": include_task_progress,
|
||
"start_time": start_time,
|
||
"end_time": end_time,
|
||
},
|
||
)
|
||
data = self._post_open_api_json(base_url, path, body)
|
||
wrapped = data if isinstance(data, dict) and ("data" in data or "traceId" in data or "time" in data) else {"data": data}
|
||
return self._format_dalongxia_analytics_result(mode=mode, title=title, body=body, data=wrapped)
|
||
except Exception as exc:
|
||
return f"壹点需求系统查询失败: {exc}"
|
||
|
||
def _is_sales_prompt(self, prompt: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", prompt).lower()
|
||
return any(k.lower() in normalized for k in SENSITIVE_SALES_KEYWORDS)
|
||
|
||
def _explicit_amount_requested(self, prompt: str) -> bool:
|
||
normalized = re.sub(r"\s+", "", prompt).lower()
|
||
return any(k.lower() in normalized for k in EXPLICIT_AMOUNT_KEYWORDS)
|
||
|
||
def _mask_sensitive_amounts(self, prompt: str, answer: str) -> str:
|
||
if not self._is_sales_prompt(prompt):
|
||
return answer
|
||
if self._explicit_amount_requested(prompt):
|
||
return answer
|
||
|
||
filtered_lines = []
|
||
removed = 0
|
||
for line in answer.splitlines():
|
||
if re.search(r"(金额|退款率|费比|收入|利润|成本|客单价|GMV|毛利|\d[\d,\.]*\s*(元|万元|万|亿))", line, re.IGNORECASE):
|
||
removed += 1
|
||
continue
|
||
filtered_lines.append(line)
|
||
|
||
cleaned = "\n".join(filtered_lines).strip()
|
||
if removed > 0:
|
||
note = "（已按配置自动隐藏金额类敏感指标，仅保留件数/订单维度。）"
|
||
if cleaned:
|
||
cleaned = f"{cleaned}\n\n{note}"
|
||
else:
|
||
cleaned = note
|
||
|
||
return cleaned or "已按配置隐藏金额类敏感指标，请改问件数/订单/原因结构维度。"
|
||
|
||
def _extract_market_subject(self, prompt: str) -> str:
|
||
normalized = prompt.strip()
|
||
|
||
def clean_subject(raw: str) -> str:
|
||
v = raw.strip().replace("\u201c", "").replace("\u201d", "").strip("\"'")
|
||
v = re.sub(r"\s+", "", v)
|
||
v = re.sub(r"^(帮我|请|麻烦|给我|我想|想要|想看|想做|分析|查看|查询|关于|整理一份|整理|输出|生成)+", "", v)
|
||
v = re.sub(r"^(近(?:[一二三四五六七八九十两半0-9]+)(?:天|周|个月|月|年)|过去(?:[一二三四五六七八九十两半0-9]+)(?:天|周|个月|月|年)|近30天|近7天|近15天|近一个月|近三个月|近半年|近一年)+", "", v)
|
||
v = re.sub(r"(的|情况|数据|分析)+$", "", v)
|
||
return v.strip()
|
||
|
||
def pick(raw: str):
|
||
v = clean_subject(raw)
|
||
if len(v) < 2:
|
||
return None
|
||
if v in {"市场", "行业", "类目", "竞品", "趋势", "市场数据", "关键词", "搜索"}:
|
||
return None
|
||
return v
|
||
|
||
if "电脑桌" in normalized:
|
||
return "电脑桌"
|
||
|
||
m = re.search(r"[\"\u201c\u201d']([^\"\u201c\u201d']{2,30})[\"\u201c\u201d']", normalized)
|
||
if m:
|
||
v = pick(m.group(1))
|
||
if v:
|
||
return v
|
||
|
||
m = re.search(
|
||
r"关于([\u4e00-\u9fa5A-Za-z0-9]{2,20})(?:的)?(?:市场数据分析|市场分析|行业分析|类目分析|竞品分析|趋势分析|市场数据|市场情况)",
|
||
normalized,
|
||
)
|
||
if m:
|
||
v = pick(m.group(1))
|
||
if v:
|
||
return v
|
||
|
||
m = re.search(
|
||
r"([\u4e00-\u9fa5A-Za-z0-9]{2,20})(?:的)?(?:市场数据分析|市场分析|行业分析|类目分析|竞品分析|趋势分析|市场数据)",
|
||
normalized,
|
||
)
|
||
if m:
|
||
v = pick(m.group(1))
|
||
if v:
|
||
return v
|
||
|
||
m = re.search(r"([\u4e00-\u9fa5A-Za-z0-9]{2,20})类目", normalized)
|
||
if m:
|
||
v = pick(m.group(1))
|
||
if v:
|
||
return v
|
||
|
||
m = re.search(r"([\u4e00-\u9fa5A-Za-z0-9]{2,20})(?:市场|市场搜索|关键词|飙升词|热词)", normalized)
|
||
if m:
|
||
v = pick(m.group(1))
|
||
if v:
|
||
return v
|
||
|
||
m = re.search(r"分析(?:一下|下)?([\u4e00-\u9fa5A-Za-z0-9]{2,20})", normalized)
|
||
if m:
|
||
v = pick(m.group(1))
|
||
if v:
|
||
return v
|
||
|
||
m = re.search(r"查(?:一下|下)?([\u4e00-\u9fa5A-Za-z0-9]{2,20})", normalized)
|
||
if m:
|
||
v = pick(m.group(1))
|
||
if v:
|
||
return v
|
||
|
||
return "市场主题"
|
||
|
||
def _extract_order_id(self, prompt: str) -> str:
|
||
text = prompt.strip()
|
||
for pattern in (
|
||
r"订单号[：:\s]*([A-Za-z0-9-]{8,40})",
|
||
r"订单编号[：:\s]*([A-Za-z0-9-]{8,40})",
|
||
r"订单[：:\s]*([A-Za-z0-9-]{8,40})",
|
||
):
|
||
m = re.search(pattern, text, re.IGNORECASE)
|
||
if m:
|
||
return m.group(1).strip()
|
||
|
||
m2 = ORDER_ID_RE.search(text)
|
||
if m2:
|
||
return m2.group(0).strip()
|
||
return ""
|
||
|
||
def _extract_product_no(self, prompt: str) -> str:
|
||
text = prompt.strip()
|
||
for pattern in (
|
||
r"(?:产品货号|货号|商品货号|SKU编码|SKUID|sku编码|skuid|sku)[：:\s]*([A-Za-z0-9_-]{5,40})",
|
||
):
|
||
m = re.search(pattern, text, re.IGNORECASE)
|
||
if m:
|
||
return m.group(1).strip()
|
||
m2 = re.search(r"\b[A-Za-z]{1,4}\d{4,}[A-Za-z0-9_-]{2,}\b", text)
|
||
if m2:
|
||
return m2.group(0).strip()
|
||
return ""
|
||
|
||
def _is_code_prompt(self, prompt: str) -> bool:
|
||
normalized = (prompt or "").lower()
|
||
hints = (
|
||
"代码",
|
||
"sql",
|
||
"python",
|
||
"java",
|
||
"javascript",
|
||
"脚本",
|
||
"bash",
|
||
"shell",
|
||
"命令",
|
||
"函数",
|
||
"class",
|
||
"yaml",
|
||
"json",
|
||
"正则",
|
||
)
|
||
return any(h in normalized for h in hints)
|
||
|
||
def _clean_dialog_noise(self, answer: str) -> str:
|
||
if not answer:
|
||
return answer
|
||
|
||
cleaned = []
|
||
for raw in answer.splitlines():
|
||
line = raw.strip()
|
||
lower = line.lower()
|
||
if not line:
|
||
cleaned.append(raw)
|
||
continue
|
||
|
||
if line.startswith("[plugins]"):
|
||
continue
|
||
if "registered" in lower and "tool" in lower:
|
||
continue
|
||
if "no toolbeencalled" in lower:
|
||
continue
|
||
if "without a preceding text-to-speech" in lower:
|
||
continue
|
||
if line.startswith("原始输出:"):
|
||
continue
|
||
if line.startswith("{") and "\"runId\"" in line:
|
||
continue
|
||
|
||
cleaned.append(raw)
|
||
|
||
out = "\n".join(cleaned).strip()
|
||
return out or answer
|
||
|
||
def _trim_noncode_blocks(self, prompt: str, answer: str) -> str:
|
||
if not answer or self._is_code_prompt(prompt):
|
||
return answer
|
||
|
||
pattern = re.compile(r"```[\s\S]*?```", re.MULTILINE)
|
||
|
||
def _repl(match):
|
||
block = match.group(0)
|
||
lines = [ln for ln in block.splitlines() if ln.strip()]
|
||
if len(lines) <= 12:
|
||
return block
|
||
return "```text\n[代码块已省略，如需完整代码请明确说明]\n```"
|
||
|
||
return pattern.sub(_repl, answer)
|
||
|
||
def _disable_card_temporarily(self, reason: str) -> None:
|
||
self.card_disabled_until = time.time() + self.card_disable_seconds
|
||
self.logger.warning("card reply disabled for %ss, reason=%s", self.card_disable_seconds, reason)
|
||
|
||
def _extract_incoming_download_codes(self, msg: ChatbotMessage) -> list:
|
||
try:
|
||
raw_codes = msg.get_image_list() or []
|
||
except Exception:
|
||
self.logger.exception("failed to read incoming image list")
|
||
return []
|
||
|
||
seen = set()
|
||
codes = []
|
||
for raw in raw_codes:
|
||
code = str(raw or "").strip()
|
||
if not code or code in seen:
|
||
continue
|
||
seen.add(code)
|
||
codes.append(code)
|
||
if len(codes) >= self.image_max_images:
|
||
break
|
||
return codes
|
||
|
||
def _normalize_file_name(self, raw_name: str) -> str:
|
||
name = str(raw_name or "").strip().replace("\\", "_").replace("/", "_")
|
||
name = re.sub(r"\s+", "_", name)
|
||
name = re.sub(r"[^0-9A-Za-z._-]+", "_", name).strip("._")
|
||
if not name:
|
||
return f"upload_{int(time.time())}.xlsx"
|
||
return name[:120]
|
||
|
||
def _is_supported_excel_file(self, file_name: str) -> bool:
|
||
return str(file_name or "").strip().lower().endswith(".xlsx")
|
||
|
||
def _is_excel_link_prompt(self, prompt: str) -> bool:
|
||
raw = str(prompt or "").strip().lower()
|
||
if not raw:
|
||
return False
|
||
return any(keyword in raw for keyword in EXCEL_IMAGE_LINK_HINT_KEYWORDS)
|
||
|
||
def _find_download_code_payload(self, value):
|
||
stack = [value]
|
||
while stack:
|
||
current = stack.pop()
|
||
if current is None:
|
||
continue
|
||
if isinstance(current, str):
|
||
content = current.strip()
|
||
if not content:
|
||
continue
|
||
if content.startswith("{") or content.startswith("["):
|
||
try:
|
||
stack.append(json.loads(content))
|
||
continue
|
||
except Exception:
|
||
continue
|
||
if isinstance(current, dict):
|
||
lowered = {str(k).lower(): k for k in current.keys()}
|
||
if "downloadcode" in lowered:
|
||
raw_code = current.get(lowered["downloadcode"])
|
||
code = str(raw_code or "").strip()
|
||
if code:
|
||
file_name = ""
|
||
for candidate in ("fileName", "filename", "file_name", "name", "title"):
|
||
if candidate in current and str(current.get(candidate) or "").strip():
|
||
file_name = str(current.get(candidate)).strip()
|
||
break
|
||
return {"download_code": code, "file_name": file_name}
|
||
for child in current.values():
|
||
stack.append(child)
|
||
continue
|
||
if isinstance(current, list):
|
||
for child in current:
|
||
stack.append(child)
|
||
return None
|
||
|
||
def _extract_incoming_file_payload(self, callback_data: dict, msg: ChatbotMessage) -> dict:
|
||
if str(getattr(msg, "message_type", "")).lower() != "file":
|
||
return {}
|
||
payload = self._find_download_code_payload(callback_data)
|
||
if not payload:
|
||
payload = self._find_download_code_payload(getattr(msg, "extensions", {}))
|
||
if not payload:
|
||
return {}
|
||
file_name = payload.get("file_name") or ""
|
||
if not file_name and isinstance(callback_data, dict):
|
||
for key in ("fileName", "filename", "name", "title"):
|
||
raw = callback_data.get(key)
|
||
if raw:
|
||
file_name = str(raw).strip()
|
||
if file_name:
|
||
break
|
||
payload["file_name"] = self._normalize_file_name(file_name or "upload.xlsx")
|
||
return payload
|
||
|
||
def _guess_image_ext(self, content_type: str, source_url: str) -> str:
|
||
ctype = (content_type or "").split(";")[0].strip().lower()
|
||
ext = ""
|
||
if ctype:
|
||
ext = mimetypes.guess_extension(ctype) or ""
|
||
if not ext:
|
||
path = urllib.parse.urlparse(source_url or "").path
|
||
_, path_ext = os.path.splitext(path)
|
||
ext = path_ext
|
||
ext = (ext or ".jpg").lower()
|
||
if ext.startswith("."):
|
||
ext = ext[1:]
|
||
if ext == "jpeg":
|
||
return "jpg"
|
||
if ext not in {"jpg", "png", "webp", "gif", "bmp"}:
|
||
return "jpg"
|
||
return ext
|
||
|
||
def _upload_chat_image_bytes(self, image_bytes: bytes, ext: str, conversation_id: str) -> str:
|
||
if not self.image_upload_to_s3:
|
||
return ""
|
||
if boto3 is None or BotoConfig is None:
|
||
return ""
|
||
if not (self.s3_bucket and self.s3_access_key and self.s3_secret_key):
|
||
return ""
|
||
|
||
safe_conv = self._safe_session_component(str(conversation_id or "chat"))
|
||
stamp = time.strftime("%Y%m%d%H%M%S")
|
||
filename = f"{safe_conv}_{stamp}_{uuid.uuid4().hex[:8]}.{ext}"
|
||
object_key = f"{self.s3_key_prefix}chat-images/{filename}"
|
||
content_type = f"image/{'jpeg' if ext == 'jpg' else ext}"
|
||
with tempfile.TemporaryDirectory(prefix="chat_image_") as tmpdir:
|
||
local_path = os.path.join(tmpdir, filename)
|
||
with open(local_path, "wb") as fh:
|
||
fh.write(image_bytes)
|
||
self._upload_to_s3(local_path, object_key, content_type)
|
||
return self._build_public_s3_url(object_key)
|
||
|
||
def _analyze_image_with_local_vlm(self, image_bytes: bytes, user_prompt: str, image_index: int) -> str:
|
||
if not self.image_analysis_enabled:
|
||
return ""
|
||
payload = {
|
||
"model": self.image_analysis_model,
|
||
"stream": False,
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": (
|
||
"你现在只做OCR和画面客观描述，不做任何分析和建议。\n"
|
||
"请输出格式：\n"
|
||
"OCR_TEXT_BEGIN\n"
|
||
"<逐行文字，尽量保留原样，包含中文/英文/数字/LOGO字母>\n"
|
||
"OCR_TEXT_END\n"
|
||
"SCENE_DESC_BEGIN\n"
|
||
"<画面客观描述：人物/物体/场景/颜色/布局；若无法判断写 UNKNOWN_SCENE>\n"
|
||
"SCENE_DESC_END\n"
|
||
"规则：\n"
|
||
"- 只能输出OCR和画面描述，不要解释；\n"
|
||
"- 不确定字符用 ? 标记；\n"
|
||
"- 完全无文字时输出 NO_TEXT；\n"
|
||
"- 画面描述不得给建议、结论或推断。\n"
|
||
f"图片序号：{image_index}\n"
|
||
f"用户问题：{user_prompt or '请识别并分析图片'}"
|
||
),
|
||
"images": [base64.b64encode(image_bytes).decode("ascii")],
|
||
}
|
||
],
|
||
}
|
||
try:
|
||
resp = requests.post(
|
||
"http://127.0.0.1:11434/api/chat",
|
||
json=payload,
|
||
timeout=self.image_analysis_timeout,
|
||
)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
text = str((data.get("message") or {}).get("content", "")).strip()
|
||
if not text:
|
||
return ""
|
||
return text[: self.image_vision_output_max_chars]
|
||
except Exception as exc:
|
||
self.logger.warning("image vlm analysis failed image=%s err=%s", image_index, exc)
|
||
return ""
|
||
|
||
def _build_image_context_from_dingtalk(self, msg: ChatbotMessage, prompt: str, download_codes: list) -> dict:
|
||
blocks = []
|
||
uploaded_urls = []
|
||
warnings = []
|
||
used_images = 0
|
||
|
||
for idx, code in enumerate(download_codes, 1):
|
||
try:
|
||
download_url = self.get_image_download_url(code)
|
||
if not download_url:
|
||
warnings.append(f"图片{idx}下载地址获取失败")
|
||
continue
|
||
|
||
resp = requests.get(download_url, timeout=self.image_download_timeout)
|
||
resp.raise_for_status()
|
||
image_bytes = resp.content or b""
|
||
if not image_bytes:
|
||
warnings.append(f"图片{idx}为空内容")
|
||
continue
|
||
if len(image_bytes) > self.image_max_bytes:
|
||
warnings.append(f"图片{idx}过大，已跳过")
|
||
continue
|
||
|
||
ext = self._guess_image_ext(resp.headers.get("Content-Type", ""), download_url)
|
||
s3_url = ""
|
||
try:
|
||
s3_url = self._upload_chat_image_bytes(image_bytes, ext, str(getattr(msg, "conversation_id", "")))
|
||
except Exception as exc:
|
||
self.logger.warning("upload incoming image failed idx=%s err=%s", idx, exc)
|
||
if s3_url:
|
||
uploaded_urls.append(s3_url)
|
||
|
||
vision = self._analyze_image_with_local_vlm(image_bytes, prompt, idx)
|
||
if not vision:
|
||
warnings.append(f"图片{idx}识别失败")
|
||
continue
|
||
|
||
item_lines = [f"图片{idx}感知结果（OCR+场景）:"]
|
||
item_lines.append(vision)
|
||
blocks.append("\n".join(item_lines))
|
||
used_images += 1
|
||
except Exception as exc:
|
||
warnings.append(f"图片{idx}处理异常: {exc}")
|
||
|
||
result = {
|
||
"used_images": used_images,
|
||
"context_text": "\n\n".join(blocks).strip(),
|
||
"uploaded_urls": uploaded_urls,
|
||
"warnings": warnings,
|
||
}
|
||
self.logger.info(
|
||
"image context built: incoming=%s used=%s uploaded=%s warnings=%s",
|
||
len(download_codes),
|
||
used_images,
|
||
len(uploaded_urls),
|
||
len(warnings),
|
||
)
|
||
return result
|
||
|
||
def _merge_prompt_with_image_context(self, prompt: str, image_context: str) -> str:
|
||
base = (prompt or "").strip() or "请识别并分析我上传的图片内容。"
|
||
return (
|
||
f"{base}\n\n"
|
||
"【图片OCR与场景结果（本地模型）】\n"
|
||
f"{image_context}\n\n"
|
||
"请仅基于以上OCR文本和场景描述做理解、分析和建议。\n"
|
||
"注意：图片已完成本地识别，不要再尝试访问图片URL，也不要调用browser/image工具。"
|
||
).strip()
|
||
|
||
def _sanitize_answer(self, prompt: str, answer: str) -> str:
|
||
if self._is_market_prompt(prompt) or self._is_install_media_prompt(prompt):
|
||
return answer
|
||
|
||
if not self.hide_source_urls:
|
||
return answer
|
||
|
||
keep_map = {}
|
||
|
||
def _keep_or_mask_url(match):
|
||
raw = match.group(0)
|
||
try:
|
||
host = (urllib.parse.urlparse(raw).hostname or "").lower()
|
||
except Exception:
|
||
host = ""
|
||
|
||
if host and (
|
||
host in self.unmask_url_hosts
|
||
or host == "feishu.cn"
|
||
or host.endswith(".feishu.cn")
|
||
):
|
||
token = f"__KEEP_URL_{len(keep_map)}__"
|
||
keep_map[token] = raw
|
||
return token
|
||
|
||
return "[内部数据源地址已脱敏]"
|
||
|
||
out = URL_RE.sub(_keep_or_mask_url, answer)
|
||
out = IP_RE.sub("[内部地址已脱敏]", out)
|
||
out = DOMAIN_RE.sub("[内部域名已脱敏]", out)
|
||
|
||
# 保底脱敏（防止某些格式未被正则覆盖）
|
||
for token in ("hq.yucekj.com", "yucekj.com", "218.93.9.141"):
|
||
out = out.replace(token, "[内部数据源已脱敏]")
|
||
|
||
for token, raw in keep_map.items():
|
||
out = out.replace(token, raw)
|
||
|
||
return out
|
||
|
||
def _run_market_brief(self, subject: str) -> str:
|
||
cmd = [self.market_brief_bin, subject]
|
||
self.logger.info("market-brief call: subject=%s", subject)
|
||
proc = subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=self.market_brief_timeout,
|
||
)
|
||
output = (proc.stdout or "").strip()
|
||
if proc.returncode != 0:
|
||
err = (proc.stderr or output or "market brief error").strip()
|
||
return f"市场分析执行失败: {err[:500]}"
|
||
if not output:
|
||
return "市场分析执行完成，但未返回文本结果。"
|
||
return output
|
||
|
||
def _parse_install_media_output(self, output: str) -> dict:
|
||
parsed = {"status": "", "mappings": [], "media": [], "warnings": []}
|
||
current_set = ""
|
||
in_warnings = False
|
||
for raw in (output or "").splitlines():
|
||
line = raw.strip()
|
||
if not line:
|
||
continue
|
||
if line.startswith("STATUS:"):
|
||
parsed["status"] = line.split(":", 1)[1].strip()
|
||
in_warnings = False
|
||
continue
|
||
if line.startswith("NAME:"):
|
||
current_set = line.split(":", 1)[1].strip()
|
||
in_warnings = False
|
||
continue
|
||
if line == "WARNINGS:":
|
||
in_warnings = True
|
||
current_set = ""
|
||
continue
|
||
if in_warnings and line.startswith("- "):
|
||
parsed["warnings"].append(line[2:].strip())
|
||
continue
|
||
if not line.startswith("{"):
|
||
continue
|
||
try:
|
||
obj = json.loads(line)
|
||
except Exception:
|
||
continue
|
||
if current_set == "order_to_product_mapping":
|
||
parsed["mappings"].append(obj)
|
||
elif current_set == "install_media_links":
|
||
parsed["media"].append(obj)
|
||
return parsed
|
||
|
||
def _run_install_media_lookup(self, prompt: str) -> str:
|
||
order_id = self._extract_order_id(prompt)
|
||
product_no = self._extract_product_no(prompt)
|
||
if not order_id and not product_no:
|
||
return "请提供订单号（或产品货号），我将按“订单号 -> 货号 -> 安装图纸/视频库”流程查询。"
|
||
|
||
cmd = [self.install_media_bin]
|
||
if order_id:
|
||
cmd += ["--order-id", order_id]
|
||
if product_no:
|
||
cmd += ["--sku-id", product_no]
|
||
cmd += ["--mysql-schema", "jiansunblueprint", "--limit", "200"]
|
||
|
||
self.logger.info("install-media call: order_id=%s product_no=%s", order_id or "-", product_no or "-")
|
||
try:
|
||
proc = subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=self.install_media_timeout,
|
||
)
|
||
except subprocess.TimeoutExpired:
|
||
return f"安装资源查询超时（>{self.install_media_timeout}秒），请重试。"
|
||
|
||
raw_output = (proc.stdout or "").strip()
|
||
if proc.returncode != 0 and not raw_output:
|
||
err = (proc.stderr or "install media lookup error").strip()
|
||
return f"安装资源查询失败: {err[:500]}"
|
||
|
||
parsed = self._parse_install_media_output(raw_output)
|
||
mappings = parsed.get("mappings", [])
|
||
media_rows = parsed.get("media", [])
|
||
warnings = parsed.get("warnings", [])
|
||
|
||
product_nos = sorted(
|
||
{
|
||
str(x.get("product_no", "")).strip()
|
||
for x in mappings
|
||
if str(x.get("product_no", "")).strip() and str(x.get("product_no", "")).strip() != "(none)"
|
||
}
|
||
)
|
||
|
||
drawing_rows = [x for x in media_rows if str(x.get("media_type", "")).lower() == "drawing"]
|
||
video_rows = [x for x in media_rows if str(x.get("media_type", "")).lower() == "video"]
|
||
stitch_required = self.install_media_stitch_enabled or self._is_install_media_stitch_prompt(prompt)
|
||
stitch_url = ""
|
||
stitch_note = ""
|
||
if stitch_required and drawing_rows:
|
||
drawing_urls = []
|
||
seen_urls = set()
|
||
for row in drawing_rows:
|
||
url = str(row.get("media_url", "")).strip()
|
||
if not url or url in seen_urls:
|
||
continue
|
||
seen_urls.add(url)
|
||
drawing_urls.append(url)
|
||
if len(drawing_urls) >= self.install_media_stitch_min_images:
|
||
product_hint = (
|
||
(product_nos[0] if product_nos else "")
|
||
or product_no
|
||
or order_id
|
||
or "unknown"
|
||
)
|
||
stitch_url, stitch_note = self._stitch_and_upload_install_media(
|
||
drawing_urls=drawing_urls[: self.install_media_stitch_max_images],
|
||
product_hint=product_hint,
|
||
)
|
||
|
||
lines = ["安装资源查询结果"]
|
||
if order_id:
|
||
lines.append(f"- 订单号: {order_id}")
|
||
if product_nos:
|
||
lines.append(f"- 命中货号: {', '.join(product_nos[:8])}")
|
||
elif product_no:
|
||
lines.append(f"- 输入货号: {product_no}")
|
||
|
||
lines.append(f"- 图纸: {len(drawing_rows)} 条，视频: {len(video_rows)} 条")
|
||
if stitch_url:
|
||
lines.append(f"- 拼接图: {stitch_url}")
|
||
elif stitch_note and self._is_install_media_stitch_prompt(prompt):
|
||
lines.append(f"- 拼接图: {stitch_note}")
|
||
|
||
display_rows = media_rows
|
||
if stitch_url:
|
||
# 用户要求仅返回拼接图，不再返回原始多条图纸链接；视频链接保留
|
||
display_rows = [x for x in media_rows if str(x.get("media_type", "")).lower() != "drawing"]
|
||
|
||
if display_rows:
|
||
lines.append("- 资源链接（前20条）:")
|
||
for i, row in enumerate(display_rows[:20], 1):
|
||
mtype = str(row.get("media_type", "")).lower()
|
||
tag = "图纸" if mtype == "drawing" else ("视频" if mtype == "video" else "资源")
|
||
url = str(row.get("media_url", "")).strip()
|
||
if url:
|
||
lines.append(f"{i}. [{tag}] {url}")
|
||
if len(display_rows) > 20:
|
||
lines.append(f"- 其余 {len(display_rows) - 20} 条已省略。")
|
||
else:
|
||
if stitch_url:
|
||
lines.append("- 原始图纸链接已按配置隐藏。")
|
||
else:
|
||
lines.append("- 未查到安装图纸/视频链接。")
|
||
|
||
if warnings:
|
||
lines.append("- 警告:")
|
||
for w in warnings[:3]:
|
||
lines.append(f" - {w}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
def _run_video_analysis(self, prompt: str) -> str:
|
||
if not self.video_analysis_enabled:
|
||
return "视频分析功能未启用。"
|
||
|
||
video_urls = self._extract_video_urls(prompt, limit=self.video_analysis_max_urls)
|
||
if not video_urls:
|
||
any_urls = self._extract_any_urls(prompt, limit=self.video_analysis_max_urls)
|
||
if any_urls:
|
||
video_urls = any_urls
|
||
self.logger.warning("video-analysis fallback to raw url(s): %s", ",".join(video_urls))
|
||
else:
|
||
return "请提供可访问的视频链接（http/https），我再开始分析。"
|
||
|
||
api_key = (
|
||
os.getenv("DASHSCOPE_API_KEY", "").strip()
|
||
or os.getenv("BAILIAN_API_KEY", "").strip()
|
||
or os.getenv("QWEN_API_KEY", "").strip()
|
||
)
|
||
if not api_key:
|
||
return "视频分析未配置API Key，请设置 DASHSCOPE_API_KEY 后重试。"
|
||
|
||
model_name = (self.video_analysis_model or "").strip()
|
||
lower_model = model_name.lower()
|
||
if (
|
||
("vl" not in lower_model and "omni" not in lower_model)
|
||
and self.video_analysis_fallback_model
|
||
and self.video_analysis_fallback_model != model_name
|
||
):
|
||
self.logger.warning(
|
||
"video-analysis model may not support video input, auto fallback: %s -> %s",
|
||
model_name,
|
||
self.video_analysis_fallback_model,
|
||
)
|
||
model_name = self.video_analysis_fallback_model
|
||
|
||
need_script = self._needs_video_script(prompt)
|
||
user_goal = self._remove_image_urls(prompt, video_urls).strip()
|
||
if not user_goal:
|
||
if need_script:
|
||
user_goal = "请深度拆解该视频，并输出可直接复刻的分镜脚本和拍摄执行方案。"
|
||
else:
|
||
user_goal = "请分析该视频的核心内容、关键信息和可执行建议。"
|
||
elif "视频" not in user_goal and "video" not in user_goal.lower():
|
||
user_goal = f"{user_goal}\n请围绕视频内容输出结论。"
|
||
|
||
content_items = [{"type": "video_url", "video_url": {"url": video_urls[0]}}]
|
||
content_items.append({"type": "text", "text": user_goal})
|
||
|
||
system_prompt = (
|
||
"你是视频内容分析助手。请基于视频画面和字幕内容输出结构化结论，避免臆测。"
|
||
)
|
||
if need_script:
|
||
system_prompt = (
|
||
"你是资深短视频导演与编剧。请基于视频内容做可执行复刻拆解。\n"
|
||
"输出必须包含：\n"
|
||
"1) 视频定位与核心卖点（1-2句）\n"
|
||
"2) 时间轴分镜脚本表（时间段/镜头景别/画面内容/人物动作/机位与运镜/字幕或口播/BGM与音效）\n"
|
||
"3) 拍摄执行清单（场景、道具、服装、灯光、设备）\n"
|
||
"4) 剪辑脚本（节奏点、转场、卡点、字幕样式）\n"
|
||
"5) 复刻风险与规避建议（至少3条）\n"
|
||
"6) 可直接给拍摄团队的最终脚本（按步骤）\n"
|
||
"要求：具体、可落地、少空话。"
|
||
)
|
||
|
||
payload = {
|
||
"model": model_name,
|
||
"messages": [
|
||
{
|
||
"role": "system",
|
||
"content": system_prompt,
|
||
},
|
||
{
|
||
"role": "user",
|
||
"content": content_items,
|
||
},
|
||
],
|
||
"temperature": 0.2,
|
||
"max_tokens": self.video_analysis_max_tokens,
|
||
}
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
self.logger.info("video-analysis call: model=%s url_count=%s urls=%s", model_name, len(video_urls), ",".join(video_urls))
|
||
try:
|
||
resp = requests.post(
|
||
self.video_analysis_base_url,
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=self.video_analysis_timeout,
|
||
)
|
||
except Exception as exc:
|
||
self.logger.error("video-analysis request failed: %s", exc)
|
||
return f"视频分析调用失败：{exc}"
|
||
|
||
text = ""
|
||
if resp.status_code >= 400:
|
||
try:
|
||
err_data = resp.json()
|
||
err_msg = (
|
||
err_data.get("error", {}).get("message")
|
||
or err_data.get("message")
|
||
or resp.text
|
||
)
|
||
except Exception:
|
||
err_msg = resp.text
|
||
self.logger.error(
|
||
"video-analysis http error: status=%s model=%s body=%s",
|
||
resp.status_code,
|
||
model_name,
|
||
(str(err_msg) or "")[:800],
|
||
)
|
||
hint = ""
|
||
em = (str(err_msg) or "").lower()
|
||
if (
|
||
"video" in em
|
||
and ("not support" in em or "invalid" in em or "unsupported" in em)
|
||
and "vl" not in model_name.lower()
|
||
):
|
||
hint = "（当前模型可能不支持视频输入，建议改为 qwen-vl-plus / qwen-vl-max）"
|
||
return f"视频分析失败（HTTP {resp.status_code}）：{str(err_msg).strip()[:300]}{hint}"
|
||
|
||
try:
|
||
data = resp.json()
|
||
choices = data.get("choices") or []
|
||
message = (choices[0] or {}).get("message", {}) if choices else {}
|
||
content = message.get("content", "")
|
||
if isinstance(content, list):
|
||
chunks = []
|
||
for item in content:
|
||
if isinstance(item, dict):
|
||
t = str(item.get("text", "")).strip()
|
||
if t:
|
||
chunks.append(t)
|
||
text = "\n".join(chunks).strip()
|
||
else:
|
||
text = str(content or "").strip()
|
||
except Exception as exc:
|
||
self.logger.exception("video-analysis parse failed")
|
||
self.logger.error("video-analysis raw body: %s", (resp.text or "")[:1000])
|
||
return f"视频分析返回解析失败：{exc}"
|
||
|
||
if not text:
|
||
return "视频分析完成，但未返回有效文本结果。"
|
||
|
||
if len(video_urls) > 1:
|
||
return f"{text}\n\n（已按配置先分析第1个视频链接；如需逐条分析请分别发送。）"
|
||
return text
|
||
|
||
def _stitch_and_upload_install_media(self, drawing_urls: list, product_hint: str) -> tuple:
|
||
if Image is None:
|
||
return "", "未执行（缺少 Pillow 依赖）"
|
||
if boto3 is None or BotoConfig is None:
|
||
return "", "未执行（缺少 boto3 依赖）"
|
||
if not (self.s3_bucket and self.s3_access_key and self.s3_secret_key):
|
||
return "", "未执行（S3 配置不完整）"
|
||
|
||
images = []
|
||
try:
|
||
for url in drawing_urls:
|
||
try:
|
||
resp = requests.get(url, timeout=self.install_media_fetch_timeout)
|
||
resp.raise_for_status()
|
||
with Image.open(io.BytesIO(resp.content)) as img:
|
||
images.append(img.convert("RGB"))
|
||
except Exception as exc:
|
||
self.logger.warning("install-media image download failed: %s %s", url, exc)
|
||
|
||
if len(images) < self.install_media_stitch_min_images:
|
||
return "", "未执行（可下载图纸不足）"
|
||
|
||
target_width = min(max(img.width for img in images), self.install_media_max_width)
|
||
resized_images = []
|
||
total_height = 0
|
||
for img in images:
|
||
if img.width != target_width:
|
||
new_height = max(1, int(img.height * target_width / img.width))
|
||
if hasattr(Image, "Resampling"):
|
||
resized = img.resize((target_width, new_height), Image.Resampling.LANCZOS)
|
||
else:
|
||
resized = img.resize((target_width, new_height), Image.LANCZOS)
|
||
else:
|
||
resized = img.copy()
|
||
resized_images.append(resized)
|
||
total_height += resized.height
|
||
|
||
stitched = Image.new("RGB", (target_width, total_height), (255, 255, 255))
|
||
y = 0
|
||
for img in resized_images:
|
||
stitched.paste(img, (0, y))
|
||
y += img.height
|
||
|
||
safe_product = re.sub(r"[^A-Za-z0-9_-]+", "-", (product_hint or "unknown")).strip("-")[:40] or "unknown"
|
||
stamp = time.strftime("%Y%m%d%H%M%S")
|
||
filename = f"{safe_product}_{stamp}_{uuid.uuid4().hex[:8]}.jpg"
|
||
object_key = f"{self.s3_key_prefix}install-media/{filename}"
|
||
|
||
with tempfile.TemporaryDirectory(prefix="install_media_") as tmpdir:
|
||
local_path = os.path.join(tmpdir, filename)
|
||
stitched.save(local_path, format="JPEG", quality=88, optimize=True)
|
||
self._upload_to_s3(local_path, object_key, "image/jpeg")
|
||
|
||
return self._build_public_s3_url(object_key), ""
|
||
except Exception as exc:
|
||
self.logger.exception("install-media stitch/upload failed")
|
||
return "", f"未执行（{exc}）"
|
||
finally:
|
||
for img in images:
|
||
try:
|
||
img.close()
|
||
except Exception:
|
||
pass
|
||
|
||
def _upload_to_s3(self, file_path: str, object_key: str, content_type: str) -> None:
|
||
endpoint = self.s3_endpoint or None
|
||
config = BotoConfig(s3={"addressing_style": "path" if self.s3_path_style else "virtual"})
|
||
client = boto3.client(
|
||
"s3",
|
||
endpoint_url=endpoint,
|
||
region_name=self.s3_region,
|
||
aws_access_key_id=self.s3_access_key,
|
||
aws_secret_access_key=self.s3_secret_key,
|
||
config=config,
|
||
)
|
||
with open(file_path, "rb") as fh:
|
||
kwargs = {
|
||
"Bucket": self.s3_bucket,
|
||
"Key": object_key,
|
||
"Body": fh,
|
||
"ContentType": content_type,
|
||
}
|
||
if self.s3_acl:
|
||
kwargs["ACL"] = self.s3_acl
|
||
client.put_object(**kwargs)
|
||
|
||
def _build_public_s3_url(self, object_key: str) -> str:
|
||
safe_key = urllib.parse.quote(object_key, safe="/-_.~")
|
||
if self.s3_public_domain:
|
||
return f"{self.s3_public_domain.rstrip('/')}/{safe_key}"
|
||
if self.s3_endpoint:
|
||
base = self.s3_endpoint.rstrip("/")
|
||
if self.s3_path_style:
|
||
return f"{base}/{self.s3_bucket}/{safe_key}"
|
||
return f"{base}/{safe_key}"
|
||
return safe_key
|
||
|
||
def _run_openclaw(self, session_id: str, prompt: str) -> str:
|
||
cmd = [
|
||
self.openclaw_bin,
|
||
"agent",
|
||
"--session-id",
|
||
session_id,
|
||
"--message",
|
||
prompt,
|
||
"--json",
|
||
"--timeout",
|
||
str(self.openclaw_timeout),
|
||
]
|
||
self.logger.info("openclaw call: session=%s", session_id)
|
||
try:
|
||
proc = subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=self.openclaw_timeout + 20,
|
||
)
|
||
except subprocess.TimeoutExpired:
|
||
self.logger.error("openclaw call timeout: session=%s timeout=%ss", session_id, self.openclaw_timeout)
|
||
return f"OpenClaw 调用超时（>{self.openclaw_timeout}秒），请重试。"
|
||
if proc.returncode != 0:
|
||
err = (proc.stderr or proc.stdout or "openclaw error").strip()
|
||
return f"OpenClaw 调用失败: {err[:500]}"
|
||
|
||
try:
|
||
raw_stdout = (proc.stdout or "").strip()
|
||
decoder = json.JSONDecoder()
|
||
data = None
|
||
parse_error = None
|
||
try:
|
||
data, _ = decoder.raw_decode(raw_stdout)
|
||
except Exception as exc:
|
||
parse_error = exc
|
||
idx = raw_stdout.find("{")
|
||
while idx != -1:
|
||
candidate = raw_stdout[idx:]
|
||
try:
|
||
data, _ = decoder.raw_decode(candidate)
|
||
break
|
||
except Exception:
|
||
idx = raw_stdout.find("{", idx + 1)
|
||
if data is None:
|
||
raise parse_error
|
||
|
||
payloads = data.get("result", {}).get("payloads", [])
|
||
text_items = [p.get("text", "") for p in payloads if isinstance(p, dict)]
|
||
answer = "\n".join(t for t in text_items if t).strip()
|
||
if not answer:
|
||
answer = "任务已执行，但未返回文本结果。"
|
||
|
||
market_fallback_hints = (
|
||
"无法直接从页面抓取",
|
||
"需要登录后的操作",
|
||
"工具版本不支持点击交互",
|
||
"无法正常访问该系统",
|
||
"您没有权限访问此页面",
|
||
"尚未登录",
|
||
"请先前往登录页登录",
|
||
"device identity required",
|
||
"控制用户界面需要HTTPS",
|
||
)
|
||
if self._is_market_prompt(prompt) and any(h in answer for h in market_fallback_hints):
|
||
subject = self._extract_market_subject(prompt)
|
||
fallback = self._run_market_brief(subject)
|
||
return f"已切换为市场分析直连模式（{subject}）：\n\n{fallback}"
|
||
|
||
return answer
|
||
except Exception as exc:
|
||
self.logger.exception("parse openclaw output failed")
|
||
raw = (proc.stdout or "").strip()
|
||
return f"结果解析失败: {exc}.\n原始输出: {raw[:500]}"
|
||
|
||
def _reply_file_by_media_id(self, msg: ChatbotMessage, media_id: str) -> bool:
|
||
if not media_id:
|
||
return False
|
||
request_headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "*/*",
|
||
}
|
||
values = {
|
||
"msgtype": "file",
|
||
"file": {
|
||
"media_id": media_id,
|
||
},
|
||
"at": {
|
||
"atUserIds": [msg.sender_staff_id] if msg.sender_staff_id else [],
|
||
},
|
||
}
|
||
try:
|
||
response_text = ""
|
||
response = requests.post(
|
||
msg.session_webhook,
|
||
headers=request_headers,
|
||
data=json.dumps(values),
|
||
timeout=20,
|
||
)
|
||
response_text = response.text
|
||
response.raise_for_status()
|
||
except Exception as exc:
|
||
self.logger.error("reply file failed, error=%s, response.text=%s", exc, response_text)
|
||
return False
|
||
return True
|
||
|
||
def _run_excel_image_link_job(self, msg: ChatbotMessage, file_payload: dict) -> tuple[str, bool]:
|
||
if not self.excel_link_feature_enabled:
|
||
return "Excel 图片链接转图片功能未启用。", False
|
||
script_path = self.excel_converter_script
|
||
if not os.path.exists(script_path):
|
||
return f"转换脚本不存在：{script_path}", False
|
||
|
||
download_code = str(file_payload.get("download_code") or "").strip()
|
||
file_name = self._normalize_file_name(file_payload.get("file_name") or "upload.xlsx")
|
||
if not download_code:
|
||
return "未获取到文件下载码，请重试上传。", False
|
||
if not self._is_supported_excel_file(file_name):
|
||
return "当前仅支持 .xlsx 文件，请上传Excel后重试。", False
|
||
|
||
download_url = self.get_image_download_url(download_code)
|
||
if not download_url:
|
||
return "文件下载地址获取失败，请重试。", False
|
||
|
||
with tempfile.TemporaryDirectory(prefix="dingtalk_excel_link_") as tmpdir:
|
||
input_path = os.path.join(tmpdir, file_name)
|
||
output_name = f"{os.path.splitext(file_name)[0]}_图片回填.xlsx"
|
||
output_path = os.path.join(tmpdir, output_name)
|
||
try:
|
||
response = requests.get(download_url, timeout=self.excel_file_download_timeout)
|
||
response.raise_for_status()
|
||
with open(input_path, "wb") as fh:
|
||
fh.write(response.content)
|
||
except Exception as exc:
|
||
self.logger.exception("download incoming excel failed")
|
||
return f"下载源文件失败：{exc}", False
|
||
|
||
cmd = [
|
||
sys.executable,
|
||
script_path,
|
||
"--input",
|
||
input_path,
|
||
"--output",
|
||
output_path,
|
||
"--link-column",
|
||
self.excel_link_column,
|
||
"--image-column-header",
|
||
self.excel_image_column_header,
|
||
"--max-width",
|
||
str(self.excel_image_max_width),
|
||
"--max-height",
|
||
str(self.excel_image_max_height),
|
||
"--timeout",
|
||
str(self.excel_image_download_timeout),
|
||
]
|
||
if self.excel_skip_tls_verify:
|
||
cmd.append("--skip-tls-verify")
|
||
try:
|
||
proc = subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=self.excel_converter_timeout,
|
||
)
|
||
except subprocess.TimeoutExpired:
|
||
return f"Excel 转换超时（>{self.excel_converter_timeout}秒），请稍后重试。", False
|
||
|
||
stdout_text = (proc.stdout or "").strip()
|
||
stderr_text = (proc.stderr or "").strip()
|
||
if proc.returncode != 0 or not os.path.exists(output_path):
|
||
detail = stderr_text or stdout_text or "未知错误"
|
||
self.logger.error("excel converter failed code=%s detail=%s", proc.returncode, detail[:800])
|
||
return f"Excel 转换失败：{detail[:500]}", False
|
||
|
||
summary_line = ""
|
||
for line in stdout_text.splitlines():
|
||
if line.startswith("SUMMARY:"):
|
||
summary_line = line.replace("SUMMARY:", "").strip()
|
||
break
|
||
|
||
try:
|
||
with open(output_path, "rb") as fh:
|
||
media_content = fh.read()
|
||
media_id = self.dingtalk_client.upload_to_dingtalk(
|
||
media_content,
|
||
filetype="file",
|
||
filename=output_name,
|
||
mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
)
|
||
except Exception as exc:
|
||
self.logger.exception("upload converted excel failed")
|
||
return f"处理完成但上传失败：{exc}", False
|
||
|
||
if not media_id:
|
||
return "处理完成但上传媒体失败，请稍后重试。", False
|
||
|
||
sent = self._reply_file_by_media_id(msg, media_id)
|
||
if not sent:
|
||
return "处理完成但回传文件失败，请稍后重试。", False
|
||
|
||
summary_text = summary_line or "已完成图片回填。"
|
||
return f"文件已回传：{output_name}\n处理结果：{summary_text}", True
|
||
|
||
def _is_image_url(self, raw_url: str) -> bool:
|
||
try:
|
||
parsed = urllib.parse.urlparse(raw_url)
|
||
except Exception:
|
||
return False
|
||
|
||
host = (parsed.hostname or "").lower()
|
||
path = (parsed.path or "").lower()
|
||
if IMAGE_EXT_RE.search(path):
|
||
return True
|
||
if host.startswith("img.") or "image" in host:
|
||
return True
|
||
return False
|
||
|
||
def _extract_image_urls(self, text: str, limit: int = 5) -> list:
|
||
urls = []
|
||
seen = set()
|
||
for match in URL_RE.finditer(text or ""):
|
||
raw = match.group(0).rstrip(".,;!?")
|
||
if not self._is_image_url(raw):
|
||
continue
|
||
if raw in seen:
|
||
continue
|
||
seen.add(raw)
|
||
urls.append(raw)
|
||
if len(urls) >= limit:
|
||
break
|
||
return urls
|
||
|
||
def _remove_image_urls(self, text: str, urls: list) -> str:
|
||
out = text
|
||
for u in urls:
|
||
out = out.replace(u, "")
|
||
out = re.sub(r"\n{3,}", "\n\n", out)
|
||
return out.strip()
|
||
|
||
def _build_action_buttons(self) -> list:
|
||
if not self.action_card_btn_url:
|
||
return []
|
||
return [
|
||
{
|
||
"text": self.action_card_btn_text or "打开链接",
|
||
"url": self.action_card_btn_url,
|
||
"iosUrl": self.action_card_btn_url,
|
||
"color": "gray",
|
||
}
|
||
]
|
||
|
||
def _is_card_reply_ok(self, reply_obj) -> bool:
|
||
if reply_obj is None:
|
||
return False
|
||
card_instance_id = getattr(reply_obj, "card_instance_id", None)
|
||
if isinstance(card_instance_id, str):
|
||
return bool(card_instance_id.strip())
|
||
return bool(card_instance_id)
|
||
|
||
def _reply_message(self, msg: ChatbotMessage, text: str, prefer_card: bool = True) -> bool:
|
||
payload = (text or "").strip() or "任务已执行，但未返回文本结果。"
|
||
|
||
if self.reply_style != "action_card":
|
||
prefer_card = False
|
||
if self.card_disabled_until > time.time():
|
||
prefer_card = False
|
||
|
||
if prefer_card:
|
||
image_urls = self._extract_image_urls(payload, limit=5)
|
||
if image_urls:
|
||
slider = [(f"图片{i + 1}", u) for i, u in enumerate(image_urls)]
|
||
markdown = self._remove_image_urls(payload, image_urls)
|
||
if not markdown:
|
||
markdown = "已整理相关图片如下："
|
||
try:
|
||
card = self.reply_carousel_card(
|
||
incoming_message=msg,
|
||
markdown=markdown,
|
||
image_slider=slider,
|
||
button_text=self.action_card_btn_text or "查看",
|
||
title=self.action_card_title,
|
||
logo=self.action_card_logo,
|
||
)
|
||
if self._is_card_reply_ok(card):
|
||
return True
|
||
self.logger.warning("carousel card reply returned empty id, fallback to action card/text")
|
||
self._disable_card_temporarily("carousel_card_empty_id")
|
||
except Exception as exc:
|
||
self.logger.exception("carousel card reply failed, fallback to action card")
|
||
if "Card.Instance.Write" in str(exc) or "AccessDenied" in str(exc):
|
||
self._disable_card_temporarily("card_permission_denied")
|
||
|
||
markdown = payload
|
||
if self.action_card_title:
|
||
markdown = f"### {self.action_card_title}\n\n{payload}"
|
||
try:
|
||
card = self.reply_markdown_button(
|
||
incoming_message=msg,
|
||
markdown=markdown,
|
||
button_list=self._build_action_buttons(),
|
||
tips=self.action_card_tips,
|
||
title=self.action_card_title,
|
||
logo=self.action_card_logo,
|
||
)
|
||
if self._is_card_reply_ok(card):
|
||
return True
|
||
self.logger.warning("action card reply returned empty id, fallback to text")
|
||
self._disable_card_temporarily("action_card_empty_id")
|
||
except Exception as exc:
|
||
self.logger.exception("action card reply failed, fallback to text")
|
||
if "Card.Instance.Write" in str(exc) or "AccessDenied" in str(exc):
|
||
self._disable_card_temporarily("card_permission_denied")
|
||
|
||
resp = self.reply_text(payload, msg)
|
||
if resp is None:
|
||
self.logger.error("text reply failed")
|
||
return False
|
||
return True
|
||
|
||
async def process(self, callback: CallbackMessage):
|
||
try:
|
||
msg = ChatbotMessage.from_dict(callback.data)
|
||
except Exception:
|
||
self.logger.exception("invalid callback payload")
|
||
return AckMessage.STATUS_BAD_REQUEST, "bad payload"
|
||
|
||
text = ""
|
||
if msg.text and getattr(msg.text, "content", None):
|
||
text = msg.text.content.strip()
|
||
else:
|
||
text_list = msg.get_text_list() or []
|
||
text = text_list[0].strip() if text_list else ""
|
||
|
||
incoming_image_codes = self._extract_incoming_download_codes(msg)
|
||
incoming_file_payload = self._extract_incoming_file_payload(callback.data, msg)
|
||
has_excel_file = bool(incoming_file_payload)
|
||
conversation_key = re.sub(r"[^A-Za-z0-9_-]+", "-", str(getattr(msg, "conversation_id", "") or "default"))
|
||
|
||
if not text and not incoming_image_codes and not has_excel_file:
|
||
return AckMessage.STATUS_OK, "ignored"
|
||
|
||
if not has_excel_file and not self._should_handle(msg, text):
|
||
return AckMessage.STATUS_OK, "ignored"
|
||
|
||
prompt = self._normalize_prompt(text)
|
||
if not prompt:
|
||
if has_excel_file:
|
||
prompt = "请处理我上传的Excel文件，把图片链接转成图片并回传。"
|
||
elif incoming_image_codes:
|
||
prompt = "请识别并分析我上传的图片内容，输出结论和建议。"
|
||
else:
|
||
self._reply_message(
|
||
msg,
|
||
f"用法: {self.prefix} 你的问题\n例如: {self.prefix} 查询亿家达旗舰店上周业绩",
|
||
prefer_card=False,
|
||
)
|
||
return AckMessage.STATUS_OK, "ok"
|
||
elif not has_excel_file and self.excel_link_feature_enabled and self._is_excel_link_prompt(prompt):
|
||
now_ts = time.time()
|
||
last_ts = self._recent_excel_success_ts.get(conversation_key, 0)
|
||
if last_ts and now_ts - last_ts <= self.excel_text_prompt_cooldown_seconds:
|
||
self._reply_message(msg, "文件刚处理完成，请查看上一条回传文件。", prefer_card=False)
|
||
else:
|
||
self._reply_message(msg, "请直接上传 .xlsx 文件（包含“图片链接”列），我会自动回填并回传。", prefer_card=False)
|
||
return AckMessage.STATUS_OK, "excel_text_prompt_hint"
|
||
|
||
event_key = self._event_key(callback, msg, prompt)
|
||
if self._seen_or_mark_event(event_key):
|
||
self.logger.info(
|
||
"duplicate callback ignored: key=%s conversation=%s sender=%s",
|
||
event_key,
|
||
getattr(msg, "conversation_id", ""),
|
||
getattr(msg, "sender_id", ""),
|
||
)
|
||
return AckMessage.STATUS_OK, "duplicate_ignored"
|
||
|
||
if self._is_smalltalk_prompt(prompt):
|
||
try:
|
||
self._reply_message(msg, self.smalltalk_reply_text, prefer_card=False)
|
||
except Exception:
|
||
self.logger.exception("failed to send smalltalk reply")
|
||
return AckMessage.STATUS_OK, "smalltalk_ok"
|
||
|
||
if self.send_processing_hint:
|
||
hint = self.processing_hint_text or "已收到，正在处理，请稍等。"
|
||
try:
|
||
self.reply_text(hint, msg)
|
||
except Exception:
|
||
self.logger.exception("failed to send processing hint")
|
||
|
||
conv_lock = self._conversation_locks.setdefault(conversation_key, asyncio.Lock())
|
||
|
||
async with conv_lock:
|
||
session_id = self._session_id(msg)
|
||
image_context_result = {"used_images": 0, "context_text": "", "uploaded_urls": [], "warnings": []}
|
||
prompt_for_exec = prompt
|
||
try:
|
||
async with self._work_semaphore:
|
||
self.logger.info(
|
||
"worker acquired: session=%s conversation=%s max_parallel=%s",
|
||
session_id,
|
||
conversation_key,
|
||
self.max_parallel_jobs,
|
||
)
|
||
if has_excel_file:
|
||
file_answer, file_sent = await asyncio.to_thread(
|
||
self._run_excel_image_link_job,
|
||
msg,
|
||
incoming_file_payload,
|
||
)
|
||
if file_sent:
|
||
self._recent_excel_success_ts[conversation_key] = time.time()
|
||
self._reply_message(msg, file_answer, prefer_card=False)
|
||
self.logger.info(
|
||
"excel file processed: session=%s file=%s",
|
||
session_id,
|
||
incoming_file_payload.get("file_name", ""),
|
||
)
|
||
return AckMessage.STATUS_OK, "excel_file_ok"
|
||
answer = file_answer
|
||
elif incoming_image_codes and not self._is_install_media_prompt(prompt):
|
||
image_context_result = await asyncio.to_thread(
|
||
self._build_image_context_from_dingtalk,
|
||
msg,
|
||
prompt,
|
||
incoming_image_codes,
|
||
)
|
||
context_text = image_context_result.get("context_text", "")
|
||
if context_text:
|
||
prompt_for_exec = self._merge_prompt_with_image_context(prompt, context_text)
|
||
if self._is_install_media_prompt(prompt):
|
||
answer = await asyncio.to_thread(self._run_install_media_lookup, prompt)
|
||
elif self._is_dalongxia_analytics_prompt(prompt):
|
||
answer = await asyncio.to_thread(self._run_dalongxia_analytics, prompt)
|
||
elif self._is_video_analysis_prompt(prompt):
|
||
answer = await asyncio.to_thread(self._run_video_analysis, prompt)
|
||
elif self._is_market_prompt(prompt):
|
||
subject = self._extract_market_subject(prompt)
|
||
answer = await asyncio.to_thread(self._run_market_brief, subject)
|
||
else:
|
||
answer = await asyncio.to_thread(self._run_openclaw, session_id, prompt_for_exec)
|
||
except Exception:
|
||
self.logger.exception("worker execution failed: session=%s", session_id)
|
||
answer = "系统执行异常，请稍后重试。"
|
||
|
||
if incoming_image_codes and image_context_result.get("used_images", 0) == 0:
|
||
warnings = image_context_result.get("warnings", [])
|
||
warn_text = "；".join(warnings[:2]) if warnings else "图片下载或识别失败"
|
||
answer = f"已收到图片，但未完成识别：{warn_text}。请重发清晰图片后再试。"
|
||
elif image_context_result.get("warnings"):
|
||
answer = f"{answer}\n\n（图片处理提示：{'；'.join(image_context_result['warnings'][:2])}）"
|
||
|
||
answer = self._clean_dialog_noise(answer)
|
||
answer = self._trim_noncode_blocks(prompt, answer)
|
||
answer = self._sanitize_answer(prompt, answer)
|
||
answer = self._mask_sensitive_amounts(prompt, answer)
|
||
if len(answer) > self.reply_max_chars:
|
||
answer = answer[: self.reply_max_chars] + "\n...(已截断)"
|
||
|
||
try:
|
||
prefer_card = not self._is_install_media_prompt(prompt)
|
||
sent = self._reply_message(msg, answer, prefer_card=prefer_card)
|
||
if sent:
|
||
self.logger.info("reply sent: session=%s chars=%s", session_id, len(answer))
|
||
else:
|
||
self.logger.error("reply failed: session=%s chars=%s", session_id, len(answer))
|
||
except Exception:
|
||
self.logger.exception("failed to send final reply: session=%s", session_id)
|
||
return AckMessage.STATUS_OK, "ok"
|
||
|
||
|
||
def build_client(logger: logging.Logger) -> DingTalkStreamClient:
|
||
client_id = os.getenv("DINGTALK_CLIENT_ID", "").strip()
|
||
client_secret = os.getenv("DINGTALK_CLIENT_SECRET", "").strip()
|
||
|
||
if not client_id or not client_secret:
|
||
raise RuntimeError("DINGTALK_CLIENT_ID / DINGTALK_CLIENT_SECRET 未配置")
|
||
|
||
credential = Credential(client_id, client_secret)
|
||
client = DingTalkStreamClient(credential, logger=logger)
|
||
bot = OpenClawDingTalkBot(logger)
|
||
client.register_callback_handler(ChatbotMessage.TOPIC, bot)
|
||
if bool_env("REGISTER_DELEGATE_TOPIC", False):
|
||
client.register_callback_handler(ChatbotMessage.DELEGATE_TOPIC, bot)
|
||
logger.info("delegate topic enabled")
|
||
else:
|
||
logger.info("delegate topic disabled (default)")
|
||
return client
|
||
|
||
|
||
def self_check(logger: logging.Logger) -> int:
|
||
required = ["DINGTALK_CLIENT_ID", "DINGTALK_CLIENT_SECRET"]
|
||
missing = [k for k in required if not os.getenv(k)]
|
||
if missing:
|
||
logger.error("missing env: %s", ", ".join(missing))
|
||
return 2
|
||
|
||
openclaw_bin = os.getenv("OPENCLAW_BIN", "/usr/bin/openclaw")
|
||
proc = subprocess.run([openclaw_bin, "--version"], capture_output=True, text=True)
|
||
if proc.returncode != 0:
|
||
logger.error("openclaw check failed: %s", (proc.stderr or proc.stdout).strip())
|
||
return 3
|
||
|
||
market_brief_bin = os.getenv("MARKET_BRIEF_BIN", "/usr/local/bin/market-brief-report")
|
||
if not os.path.exists(market_brief_bin):
|
||
logger.error("market brief bin not found: %s", market_brief_bin)
|
||
return 4
|
||
|
||
install_media_bin = os.getenv("INSTALL_MEDIA_BIN", "/usr/local/bin/install-media-lookup")
|
||
if not os.path.exists(install_media_bin):
|
||
logger.error("install media bin not found: %s", install_media_bin)
|
||
return 5
|
||
|
||
logger.info("self-check ok, openclaw=%s", (proc.stdout or "").strip())
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="OpenClaw DingTalk Bridge")
|
||
parser.add_argument("--check", action="store_true", help="run checks and exit")
|
||
args = parser.parse_args()
|
||
|
||
logging.basicConfig(
|
||
level=os.getenv("LOG_LEVEL", "INFO").upper(),
|
||
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
||
)
|
||
logger = logging.getLogger("openclaw-dingtalk-bridge")
|
||
|
||
if args.check:
|
||
return self_check(logger)
|
||
|
||
client = build_client(logger)
|
||
logger.info("bridge started, listening DingTalk stream")
|
||
client.start_forever()
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|