651 lines
24 KiB
Python
Executable File
651 lines
24 KiB
Python
Executable File
#!/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
|
||
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 _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 = "任务已执行，但未返回文本结果。"
|
||
|
||
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())
|