#!/usr/bin/env python3 import argparse import hashlib import json import logging import os import re import subprocess import time import urllib.parse from dingtalk_stream import AckMessage, CallbackMessage, ChatbotHandler, ChatbotMessage from dingtalk_stream import Credential, DingTalkStreamClient MARKET_HINT_KEYWORDS = ( "市场分析", "市场数据分析", "行业分析", "类目分析", "竞品分析", "趋势分析", ) SALES_HINT_KEYWORDS = ( "销售数据", "销售报表", "销售看板", "公司销售", "业绩报表", "业绩数据", "费比监控", ) SENSITIVE_SALES_KEYWORDS = ( "销售", "业绩", "售后", "退款", "店铺", "发货", "工单", "费比", ) EXPLICIT_AMOUNT_KEYWORDS = ( "金额", "费用", "收入", "利润", "成本", "客单价", "gmv", "销售额", "退款额", "毛利", ) URL_RE = re.compile(r"https?://[^\s)\]>]+", 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) 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.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").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() 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 _session_id(self, msg: ChatbotMessage) -> str: conversation_id = (msg.conversation_id or "default").replace(" ", "_") base_session = f"{self.session_prefix}-{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 "分析" in normalized: return True return False 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"^(帮我|请|麻烦|给我|我想|想要|想看|想做|分析|查看|查询|关于)+", "", 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 return "电脑桌" def _sanitize_answer(self, answer: str) -> str: 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: 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 _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: data = json.loads(proc.stdout) 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 = "任务已执行，但未返回文本结果。" if self._is_market_prompt(prompt) and ( "无法直接从页面抓取" in answer or "需要登录后的操作" in answer or "工具版本不支持点击交互" in answer ): 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 _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 prefer_card and self.reply_style == "action_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") except Exception: self.logger.exception("carousel card reply failed, fallback to action card") 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") except Exception: self.logger.exception("action card reply failed, fallback to text") 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 "" if not text: return AckMessage.STATUS_OK, "ignored" if not self._should_handle(msg, text): return AckMessage.STATUS_OK, "ignored" prompt = self._normalize_prompt(text) if not prompt: self._reply_message( msg, f"用法: {self.prefix} 你的问题\\n例如: {self.prefix} 查询亿家达旗舰店上周业绩", prefer_card=False, ) return AckMessage.STATUS_OK, "ok" 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.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") session_id = self._session_id(msg) if self._is_market_prompt(prompt): subject = self._extract_market_subject(prompt) answer = self._run_market_brief(subject) else: answer = self._run_openclaw(session_id, prompt) answer = self._sanitize_answer(answer) answer = self._mask_sensitive_amounts(prompt, answer) if len(answer) > self.reply_max_chars: answer = answer[: self.reply_max_chars] + "\n...(已截断)" try: sent = self._reply_message(msg, answer, prefer_card=True) 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 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())