#!/usr/bin/env python3 from __future__ import annotations import argparse import base64 import hashlib import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request from collections import deque from datetime import datetime from pathlib import Path from typing import Any, Iterable DEFAULT_WCPP_BASE_URL = os.environ.get("WCPP_BASE_URL", "http://127.0.0.1:18238") DEFAULT_XIBAO_BASE_URL = os.environ.get("XIBAO_BASE_URL", "http://127.0.0.1:8787") DEFAULT_SESSION_FILE = Path( os.environ.get("WCPP_SESSION_FILE", "/root/WeChatPadPro_test_20260227/webui/.session.json") ) BASE_DIR = Path(__file__).resolve().parent DEFAULT_STATE_FILE = BASE_DIR / "data" / "wechat_bridge_state.json" DEFAULT_META_FILE = BASE_DIR / "data" / "wechat_bridge_meta.json" DEFAULT_DAILY_CLEANUP_TIME = os.environ.get("XIBAO_DAILY_CLEANUP_TIME", "00:10") RELAY_SERIAL_LINE_RE = re.compile(r"^\s*\d+\s*[、，,.．:：)\）]\s*") def log(msg: str) -> None: stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{stamp}] {msg}", flush=True) def read_json_file(path: Path, fallback: Any) -> Any: try: if not path.exists(): return fallback return json.loads(path.read_text(encoding="utf-8")) except Exception: return fallback def write_json_file(path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(path) def parse_hhmm(text: str) -> tuple[int, int]: val = str(text or "").strip() m = re.match(r"^(\d{1,2}):(\d{1,2})$", val) if not m: return (0, 10) hour = int(m.group(1)) minute = int(m.group(2)) if hour < 0 or hour > 23 or minute < 0 or minute > 59: return (0, 10) return (hour, minute) def flatten_wrapped_value(value: Any) -> Any: if isinstance(value, dict): for name in ("str", "string", "String", "value", "Value"): if name in value and value[name] not in (None, ""): return value[name] return value def find_first_value(data: Any, keys: Iterable[str]) -> Any: targets = {k.lower() for k in keys} queue: list[Any] = [data] while queue: current = queue.pop(0) if isinstance(current, dict): for key, val in current.items(): if key.lower() in targets: found = flatten_wrapped_value(val) if found not in (None, "", [], {}): return found queue.append(val) elif isinstance(current, list): queue.extend(current) return None def pick_primary_message(item: dict[str, Any]) -> dict[str, Any]: add_msgs = item.get("AddMsgs") if isinstance(add_msgs, list): for msg in add_msgs: if isinstance(msg, dict): return msg return item def summarize_message(item: Any) -> dict[str, Any]: if not isinstance(item, dict): text = str(item) return { "id": "-", "type": "-", "from": "-", "to": "-", "time": "-", "content": text, "raw": item, } primary = pick_primary_message(item) msg_id = find_first_value(primary, ["msg_id", "new_msg_id", "MsgId", "NewMsgId", "ClientMsgId", "id"]) msg_type = find_first_value(primary, ["msg_type", "MsgType", "ContentType", "Type"]) from_user = find_first_value( primary, ["from_user_name", "FromUserName", "FromWxid", "Sender", "Talker", "wxid", "userName"], ) to_user = find_first_value(primary, ["to_user_name", "ToUserName", "ToWxid", "Receiver"]) create_time = find_first_value(primary, ["create_time", "CreateTime", "Timestamp", "Time", "MsgTime"]) content = find_first_value(primary, ["text_content", "TextContent", "content", "Content", "Message", "Text"]) if content in (None, "", {}, []): content = find_first_value(primary, ["msg_source", "MsgSource"]) def _to_text(value: Any) -> str: value = flatten_wrapped_value(value) if isinstance(value, (dict, list)): return json.dumps(value, ensure_ascii=False) if value is None: return "-" return str(value) return { "id": _to_text(msg_id), "type": _to_text(msg_type), "from": _to_text(from_user), "to": _to_text(to_user), "time": _to_text(create_time), "content": _to_text(content), "raw": item, } def build_message_dedupe_key(msg: dict[str, Any]) -> str: msg_id = str(msg.get("id") or "-") msg_type = str(msg.get("type") or "-") if msg_id not in {"", "-"}: return f"id:{msg_id}|type:{msg_type}" seed = "|".join( [ str(msg.get("from") or "-"), str(msg.get("to") or "-"), str(msg.get("time") or "-"), str(msg.get("content") or "")[:160], ] ) digest = hashlib.sha1(seed.encode("utf-8")).hexdigest()[:20] return f"hash:{digest}|type:{msg_type}" class SeenStore: def __init__(self, path: Path, max_size: int = 8000) -> None: self.path = path self.max_size = max(100, int(max_size)) self._queue: deque[str] = deque() self._set: set[str] = set() self._dirty = False self.load() def load(self) -> None: raw = read_json_file(self.path, {}) seen = raw.get("seen") if isinstance(raw, dict) else [] if not isinstance(seen, list): seen = [] for item in seen: key = str(item).strip() if not key or key in self._set: continue self._queue.append(key) self._set.add(key) while len(self._queue) > self.max_size: old = self._queue.popleft() self._set.discard(old) self._dirty = False def contains(self, key: str) -> bool: return key in self._set def add(self, key: str) -> None: if not key or key in self._set: return self._queue.append(key) self._set.add(key) while len(self._queue) > self.max_size: old = self._queue.popleft() self._set.discard(old) self._dirty = True def flush(self, force: bool = False) -> None: if not force and not self._dirty: return data = { "updated_at": datetime.now().isoformat(timespec="seconds"), "seen": list(self._queue), } write_json_file(self.path, data) self._dirty = False class WechatXibaoBridge: def __init__(self, args: argparse.Namespace) -> None: self.wechat_base_url = str(args.wechat_base_url).rstrip("/") self.xibao_base_url = str(args.xibao_base_url).rstrip("/") self.auth_key = str(args.wechat_auth_key or "").strip() self.session_file = Path(args.wechat_session_file).resolve() self.sync_count = max(1, int(args.sync_count)) self.poll_interval = max(0.5, float(args.poll_interval)) self.max_images = max(0, int(args.max_images)) self.once = bool(args.once) self.dry_run = bool(args.dry_run) self.allow_from = {x.strip() for x in str(args.allow_from or "").split(",") if x.strip()} self.seen = SeenStore(Path(args.state_file).resolve()) self.meta_file = Path(args.meta_file).resolve() self.cleanup_hour, self.cleanup_minute = parse_hhmm(str(args.daily_cleanup_time)) self.explicit_auth = bool(self.auth_key) self.self_wxid = "" self.meta = self._load_meta() self.meta_dirty = False self._last_state_flush = time.time() self._last_meta_flush = time.time() def _load_meta(self) -> dict[str, Any]: raw = read_json_file(self.meta_file, {}) if not isinstance(raw, dict): raw = {} users = raw.get("users") if not isinstance(users, dict): users = {} else: normalized_users: dict[str, Any] = {} for wxid, item in users.items(): if not isinstance(item, dict): continue line_map = item.get("line_map") if not isinstance(line_map, dict): line_map = {} line_order = item.get("line_order") if not isinstance(line_order, list): line_order = [] normalized_users[str(wxid)] = { "updated_at": str(item.get("updated_at", "")), "line_map": {str(k): v for k, v in line_map.items() if isinstance(v, dict)}, "line_order": [str(x) for x in line_order if str(x).strip()], } users = normalized_users daily_cleanup = raw.get("daily_cleanup") if not isinstance(daily_cleanup, dict): daily_cleanup = {} daily_skip = raw.get("daily_skip") if not isinstance(daily_skip, dict): daily_skip = {} else: normalized_daily_skip: dict[str, Any] = {} for wxid, item in daily_skip.items(): if not isinstance(item, dict): continue count_raw = item.get("count") try: count_val = int(count_raw or 0) except Exception: count_val = 0 if count_val <= 0: continue normalized_daily_skip[str(wxid)] = { "date": str(item.get("date", "")).strip(), "count": max(1, min(count_val, 500)), "updated_at": str(item.get("updated_at", "")), } daily_skip = normalized_daily_skip return { "updated_at": str(raw.get("updated_at", "")), "users": users, "daily_cleanup": { "last_date": str(daily_cleanup.get("last_date", "")).strip(), }, "daily_skip": daily_skip, } def _mark_meta_dirty(self) -> None: self.meta_dirty = True self.meta["updated_at"] = datetime.now().isoformat(timespec="seconds") def _flush_meta(self, force: bool = False) -> None: if not force and not self.meta_dirty: return write_json_file(self.meta_file, self.meta) self.meta_dirty = False def _flush_state(self, force: bool = False) -> None: self.seen.flush(force=force) self._flush_meta(force=force) def _request_json( self, *, base_url: str, path: str, method: str, payload: dict[str, Any] | None = None, key: str | None = None, timeout: int = 30, accept_error: bool = False, ) -> tuple[int, dict[str, Any]]: url = f"{base_url.rstrip('/')}{path}" if key: joiner = "&" if "?" in url else "?" url = f"{url}{joiner}key={urllib.parse.quote(key)}" body = None headers: dict[str, str] = {} if payload is not None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=body, method=method.upper(), headers=headers) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode("utf-8") data = json.loads(raw) if raw else {} if not isinstance(data, dict): data = {"raw": data} return int(resp.status), data except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") try: data = json.loads(raw) if raw else {} except Exception: data = {"ok": False, "error": raw or f"HTTP {exc.code}"} if not isinstance(data, dict): data = {"raw": data} if accept_error: return int(exc.code), data raise RuntimeError(f"HTTP {exc.code}: {raw[:400]}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"请求失败: {exc}") from exc def _request_bytes(self, url: str, timeout: int = 45) -> tuple[bytes, str]: req = urllib.request.Request(url, method="GET") with urllib.request.urlopen(req, timeout=timeout) as resp: data = resp.read() content_type = resp.headers.get("Content-Type", "application/octet-stream") content_type = content_type.split(";", 1)[0].strip() or "application/octet-stream" return data, content_type def _load_auth_from_session(self) -> str: raw = read_json_file(self.session_file, {}) if isinstance(raw, dict): key = str(raw.get("authKey", "")).strip() if key: return key return "" def is_key_online(self, auth_key: str) -> bool: key = str(auth_key or "").strip() if not key: return False try: _, resp = self._request_json( base_url=self.wechat_base_url, path="/login/GetLoginStatus", method="GET", key=key, timeout=12, accept_error=True, ) return int(resp.get("Code") or 0) == 200 except Exception: return False def refresh_auth_key(self, force: bool = False) -> None: if self.explicit_auth and not force: return latest = self._load_auth_from_session() if not latest: return if latest != self.auth_key: if not self.auth_key: self.auth_key = latest self.self_wxid = "" log(f"已更新 authKey: {self.auth_key}") return latest_online = self.is_key_online(latest) current_online = self.is_key_online(self.auth_key) if latest_online or not current_online: self.auth_key = latest self.self_wxid = "" log(f"已更新 authKey: {self.auth_key}") else: log("检测到新的 authKey 但未在线，继续使用当前在线会话。") def ensure_self_wxid(self) -> None: if not self.auth_key or self.self_wxid: return try: _, resp = self._request_json( base_url=self.wechat_base_url, path="/user/GetProfile", method="GET", key=self.auth_key, ) except Exception as exc: log(f"获取自身 wxid 失败: {exc}") return if resp.get("Code") != 200: return data = resp.get("Data") self_id = find_first_value(data, ["wxid", "Wxid", "userName", "UserName", "to_user_name"]) if self_id: self.self_wxid = str(self_id).strip() if self.self_wxid: log(f"已识别当前账号 wxid: {self.self_wxid}") def _extract_feedback_records(self, data: dict[str, Any]) -> list[dict[str, Any]]: if not isinstance(data, dict): return [] result = data.get("result") if not isinstance(result, dict): return [] records = result.get("records") if isinstance(records, list) and records: return [x for x in records if isinstance(x, dict)] records = result.get("new_records") if isinstance(records, list) and records: return [x for x in records if isinstance(x, dict)] return [] def save_feedback_context(self, from_user: str, data: dict[str, Any]) -> None: from_user = str(from_user or "").strip() if not from_user: return records = self._extract_feedback_records(data) if not records: return line_map: dict[str, dict[str, Any]] = {} line_order: list[str] = [] fallback_index = 0 for item in records: source_line = str(item.get("source_line") or item.get("raw_text") or "").strip() if not source_line: continue line_serial = str(item.get("line_serial") or "").strip() if not line_serial: fallback_index += 1 line_serial = str(fallback_index) if line_serial in line_map: continue line_order.append(line_serial) line_map[line_serial] = { "source_line": source_line, "record": { "source_line": source_line, "raw_text": str(item.get("raw_text") or ""), "branch": str(item.get("branch") or ""), "amount": str(item.get("amount") or ""), "type": str(item.get("type") or ""), "page": str(item.get("page") or ""), "status": str(item.get("status") or ""), "line_serial": line_serial, "line_product_index": str(item.get("line_product_index") or ""), "dedup_key": str(item.get("dedup_key") or ""), "signature_key": str(item.get("signature_key") or ""), }, } if not line_map: return users = self.meta.setdefault("users", {}) users[from_user] = { "updated_at": datetime.now().isoformat(timespec="seconds"), "line_map": line_map, "line_order": line_order, } if len(users) > 400: keys = sorted(users.keys(), key=lambda k: str(users[k].get("updated_at", ""))) for key in keys[:-300]: users.pop(key, None) self._mark_meta_dirty() def resolve_feedback_item(self, from_user: str, serial: str) -> dict[str, Any] | None: users = self.meta.get("users") if not isinstance(users, dict): return None context = users.get(str(from_user or "").strip()) if not isinstance(context, dict): return None line_map = context.get("line_map") if not isinstance(line_map, dict): return None serial_text = str(serial or "").strip() if not serial_text: return None item = line_map.get(serial_text) if isinstance(item, dict): return item line_order = context.get("line_order") if isinstance(line_order, list) and serial_text.isdigit(): idx = int(serial_text) - 1 if 0 <= idx < len(line_order): mapped_serial = str(line_order[idx]).strip() item = line_map.get(mapped_serial) if isinstance(item, dict): return item return None def parse_feedback_command(self, text: str) -> tuple[str, str] | None: content = str(text or "").strip() if not content: return None m = re.match(r"^/?反馈\s*[+：:\-]?\s*(\d{1,4})(?:[\s，,;；:：\-]+(.*))?$", content) if not m: return None serial = str(m.group(1)).strip() note = str(m.group(2) or "").strip() return serial, note def parse_skip_command(self, text: str) -> int | None: content = str(text or "").strip() if not content: return None m = re.match(r"^/?跳过\s*[（(]?\s*(\d{1,4})\s*[)）]?\s*$", content) if not m: return None return max(0, min(int(m.group(1)), 500)) def _today_text(self) -> str: return datetime.now().strftime("%Y-%m-%d") def set_daily_skip(self, from_user: str, count: int) -> int: user = str(from_user or "").strip() if not user: return 0 daily_skip = self.meta.setdefault("daily_skip", {}) if not isinstance(daily_skip, dict): daily_skip = {} self.meta["daily_skip"] = daily_skip normalized_count = max(0, min(int(count or 0), 500)) if normalized_count <= 0: if user in daily_skip: daily_skip.pop(user, None) self._mark_meta_dirty() return 0 daily_skip[user] = { "date": self._today_text(), "count": normalized_count, "updated_at": datetime.now().isoformat(timespec="seconds"), } if len(daily_skip) > 1000: keys = sorted(daily_skip.keys(), key=lambda k: str(daily_skip[k].get("updated_at", ""))) for key in keys[:-800]: daily_skip.pop(key, None) self._mark_meta_dirty() return normalized_count def get_daily_skip(self, from_user: str) -> int: user = str(from_user or "").strip() if not user: return 0 daily_skip = self.meta.get("daily_skip") if not isinstance(daily_skip, dict): return 0 item = daily_skip.get(user) if not isinstance(item, dict): return 0 if str(item.get("date") or "").strip() != self._today_text(): daily_skip.pop(user, None) self._mark_meta_dirty() return 0 try: count = int(item.get("count") or 0) except Exception: count = 0 if count <= 0: daily_skip.pop(user, None) self._mark_meta_dirty() return 0 return max(1, min(count, 500)) def apply_daily_skip_to_raw_text(self, from_user: str, raw_text: str) -> tuple[str, int, int]: requested = self.get_daily_skip(from_user) if requested <= 0: return raw_text, 0, 0 lines = str(raw_text or "").splitlines() if not lines: return raw_text, 0, requested removed = 0 keep_lines: list[str] = [] for line in lines: source = str(line) if removed < requested and RELAY_SERIAL_LINE_RE.match(source.strip()): removed += 1 continue keep_lines.append(source) normalized = "\n".join(x for x in keep_lines if str(x).strip()) if not normalized.strip(): normalized = "#接龙" return normalized, removed, requested def handle_feedback(self, from_user: str, serial: str, note: str) -> None: item = self.resolve_feedback_item(from_user=from_user, serial=serial) if not item: self.send_text(from_user, "未找到可反馈的序号，请先发送 #接龙 生成后再反馈。") return source_line = str(item.get("source_line") or "").strip() record = item.get("record") if isinstance(item.get("record"), dict) else {} note_text = str(note or "").strip() or f"用户反馈序号{serial}" mark_type = "recognition_error" if "识别" in note_text else "generation_error" if self.dry_run: log( f"[dry-run] feedback from={from_user} serial={serial} mark_type={mark_type} " f"source_line={source_line} note={note_text}" ) self.send_text(from_user, f"反馈已接收（演练模式）：序号{serial}") return status, resp = self._request_json( base_url=self.xibao_base_url, path="/api/log/mark", method="POST", payload={ "mark_type": mark_type, "source_line": source_line, "note": f"{note_text} | from={from_user}", "record": record, }, accept_error=True, timeout=40, ) if not resp.get("ok"): err = str(resp.get("error") or f"HTTP {status}").strip() self.send_text(from_user, f"反馈记录失败: {err}") return issue = resp.get("issue") if isinstance(resp.get("issue"), dict) else {} issue_id = str(issue.get("id") or "").strip() suffix = f"，编号 {issue_id}" if issue_id else "" self.send_text(from_user, f"反馈已记录：序号{serial}{suffix}") def _clear_history_keep_feedback(self) -> bool: history_path = "" try: _, cfg = self._request_json( base_url=self.xibao_base_url, path="/api/config", method="GET", accept_error=False, timeout=20, ) history_path = str(cfg.get("resolved_history_file") or "").strip() except Exception as exc: log(f"读取配置失败（history路径）: {exc}") if history_path: p = Path(history_path) if p.exists() or p.parent.exists(): try: write_json_file(p, []) log(f"已清空历史文件: {p}") return True except Exception as exc: log(f"清空历史文件失败: {exc}") try: _, resp = self._request_json( base_url=self.xibao_base_url, path="/api/history/clear", method="POST", payload={}, accept_error=True, timeout=40, ) if resp.get("ok"): if resp.get("skipped_suppressed_cleared"): log("已清空历史（附带清空了屏蔽列表）。") else: log("已清空历史。") return True log(f"清空历史失败: {resp}") return False except Exception as exc: log(f"清空历史请求异常: {exc}") return False def maybe_run_daily_cleanup(self) -> None: now = datetime.now() today = now.strftime("%Y-%m-%d") daily = self.meta.setdefault("daily_cleanup", {}) last_date = str(daily.get("last_date") or "").strip() if last_date == today: return if (now.hour, now.minute) < (self.cleanup_hour, self.cleanup_minute): return log("开始执行每日清理：图片输出 + 历史记录（保留反馈记录）") output_ok = False history_ok = False try: _, resp = self._request_json( base_url=self.xibao_base_url, path="/api/output/clear", method="POST", payload={}, accept_error=True, timeout=120, ) output_ok = bool(resp.get("ok")) if output_ok: log("已清空输出图片。") else: log(f"清空输出图片失败: {resp}") except Exception as exc: log(f"清空输出图片异常: {exc}") history_ok = self._clear_history_keep_feedback() if output_ok and history_ok: daily["last_date"] = today self._mark_meta_dirty() log(f"每日清理完成: {today}") def send_text(self, to_user: str, text: str) -> None: if self.dry_run: log(f"[dry-run] send text to={to_user}: {text}") return body = { "MsgItem": [ { "MsgType": 1, "ToUserName": to_user, "TextContent": text, } ] } _, resp = self._request_json( base_url=self.wechat_base_url, path="/message/SendTextMessage", method="POST", payload=body, key=self.auth_key, timeout=30, ) if resp.get("Code") != 200: raise RuntimeError(f"SendTextMessage失败: {json.dumps(resp, ensure_ascii=False)}") def send_image_data_uri(self, to_user: str, data_uri: str) -> None: if self.dry_run: log(f"[dry-run] send image to={to_user}, bytes={len(data_uri)}") return body = { "MsgItem": [ { "MsgType": 2, "ToUserName": to_user, "ImageContent": data_uri, } ] } _, resp = self._request_json( base_url=self.wechat_base_url, path="/message/SendImageNewMessage", method="POST", payload=body, key=self.auth_key, timeout=60, ) if resp.get("Code") != 200: raise RuntimeError(f"SendImageNewMessage失败: {json.dumps(resp, ensure_ascii=False)}") def parse_command(self, text: str) -> dict[str, str] | None: content = str(text or "").strip() if not content: return None feedback = self.parse_feedback_command(content) if feedback is not None: serial, note = feedback return {"action": "feedback", "serial": serial, "payload": note} skip_count = self.parse_skip_command(content) if skip_count is not None: return {"action": "set_skip", "count": str(skip_count), "payload": ""} lower = content.lower() if lower in {"/喜报", "喜报", "#喜报", "/喜报 help", "/喜报 帮助", "喜报帮助"}: return {"action": "help", "payload": ""} prefixes = ["/喜报", "喜报", "#喜报"] for prefix in prefixes: if content.startswith(prefix): rest = content[len(prefix) :].strip(" ：:\n\t") if not rest: return {"action": "help", "payload": ""} rest_lower = rest.lower() if rest_lower in {"help", "帮助"}: return {"action": "help", "payload": ""} if rest.startswith("解析"): payload = rest[2:].strip(" ：:\n\t") return {"action": "parse", "payload": payload} if rest.startswith("生成"): payload = rest[2:].strip(" ：:\n\t") return {"action": "generate", "payload": payload} return {"action": "generate", "payload": rest} if "#接龙" in content: return {"action": "generate", "payload": content} return None def normalize_raw_text(self, payload: str) -> str: text = str(payload or "").strip() if not text: return "" if "#接龙" in text: return text return f"#接龙\n{text}" def format_parse_result(self, data: dict[str, Any]) -> str: result = data.get("result") if isinstance(data, dict) else {} if not isinstance(result, dict): return "解析完成。" summary = result.get("summary") if isinstance(result.get("summary"), dict) else {} parsed = int(summary.get("parsed") or 0) new_count = int(summary.get("new") or 0) duplicate = int(summary.get("duplicate") or 0) pending = int(summary.get("insurance_pending") or 0) return ( "解析完成\n" f"- 识别条数: {parsed}\n" f"- 可生成: {new_count}\n" f"- 重复: {duplicate}\n" f"- 待选保险年限: {pending}" ) def format_generate_result(self, data: dict[str, Any]) -> str: result = data.get("result") if isinstance(data, dict) else {} summary = result.get("summary") if isinstance(result, dict) and isinstance(result.get("summary"), dict) else {} generated = int(data.get("generated_count") or 0) duplicate = int(summary.get("duplicate") or 0) pending = int(summary.get("insurance_pending") or 0) if generated <= 0: return ( "本次没有可生成的新记录\n" f"- 重复记录: {duplicate}\n" f"- 待选保险年限: {pending}" ) return f"生成完成，本次生成 {generated} 张。若有问题请发送：反馈+序号+说明" def compose_help_text(self) -> str: return ( "喜报机器人用法:\n" "1) /喜报 生成 + 接龙文本\n" "2) /喜报 解析 + 接龙文本\n" "3) 直接发送包含 #接龙 的文本\n" "4) 生成有误时：反馈+序号+说明\n" "5) 发送 跳过3（或 跳过(3)），当天自动跳过前3条；跳过0 取消\n" "示例:\n" "/喜报 生成\\n#接龙\\n1、营江路揽收现金10万存一年\n" "反馈2 网点写错了\n" "跳过2" ) def handle_command(self, msg: dict[str, Any], command: dict[str, str]) -> None: from_user = str(msg.get("from") or "").strip() if not from_user: return action = command.get("action", "") payload = command.get("payload", "") if action == "feedback": serial = str(command.get("serial") or "").strip() if not serial: self.send_text(from_user, "反馈格式错误，请发送：反馈+序号+说明") return self.handle_feedback(from_user=from_user, serial=serial, note=payload) return if action == "set_skip": try: count = int(str(command.get("count") or "0").strip() or "0") except Exception: self.send_text(from_user, "跳过格式错误，请发送：跳过3 或 跳过(3)") return effective = self.set_daily_skip(from_user=from_user, count=count) if effective <= 0: self.send_text(from_user, "已取消今日自动跳过。") else: self.send_text(from_user, f"已设置今日自动跳过前 {effective} 条（仅当天生效）。") return if action == "help": self.send_text(from_user, self.compose_help_text()) return raw_text = self.normalize_raw_text(payload) if not raw_text: self.send_text(from_user, "未检测到有效文本，请发送 /喜报 帮助 查看用法。") return raw_text, skipped_count, requested_skip = self.apply_daily_skip_to_raw_text(from_user=from_user, raw_text=raw_text) endpoint = "/api/parse" if action == "parse" else "/api/generate" start_hint = "收到，正在解析..." if action == "parse" else "收到，正在生成喜报，请稍候..." if requested_skip > 0: start_hint = f"{start_hint}（已跳过 {skipped_count}/{requested_skip} 条）" self.send_text(from_user, start_hint) status, resp = self._request_json( base_url=self.xibao_base_url, path=endpoint, method="POST", payload={"raw_text": raw_text, "save_history": True} if action == "generate" else {"raw_text": raw_text}, accept_error=True, timeout=300, ) if not resp.get("ok"): err = str(resp.get("error") or resp.get("message") or f"HTTP {status}").strip() if resp.get("error_code") == "insurance_year_required": err = "包含保险记录但未指定 3年交/5年交。请在文本里写明年限后重试。" self.send_text(from_user, f"处理失败: {err}") return self.save_feedback_context(from_user=from_user, data=resp) if action == "parse": self.send_text(from_user, self.format_parse_result(resp)) return self.send_text(from_user, self.format_generate_result(resp)) images = resp.get("download_images") if not isinstance(images, list) or not images: return send_list = images[: self.max_images] if self.max_images > 0 else [] for idx, item in enumerate(send_list, start=1): download_url = str(item.get("download_url") or "").strip() if not download_url: continue full_url = urllib.parse.urljoin(f"{self.xibao_base_url}/", download_url.lstrip("/")) try: raw, content_type = self._request_bytes(full_url, timeout=90) b64 = base64.b64encode(raw).decode("ascii") data_uri = f"data:{content_type};base64,{b64}" self.send_image_data_uri(from_user, data_uri) log(f"已回发图片 {idx}/{len(send_list)} -> {from_user}") except Exception as exc: self.send_text(from_user, f"第{idx}张图片回发失败: {exc}") remain = len(images) - len(send_list) if remain > 0: self.send_text(from_user, f"还有 {remain} 张图片未回发（已达到单次上限 {self.max_images}）。") def should_ignore_message(self, msg: dict[str, Any]) -> bool: msg_type = str(msg.get("type") or "").strip() from_user = str(msg.get("from") or "").strip() to_user = str(msg.get("to") or "").strip() content = str(msg.get("content") or "").strip() if msg_type not in {"1", "01"}: return True if not from_user or not to_user or not content: return True if from_user == "filehelper": return True if self.self_wxid and from_user == self.self_wxid: return True if from_user.endswith("@chatroom") or to_user.endswith("@chatroom"): return True if self.allow_from and from_user not in self.allow_from: return True return False def handle_polled_message(self, raw: Any) -> None: msg = summarize_message(raw) if self.should_ignore_message(msg): return dedupe_key = build_message_dedupe_key(msg) if self.seen.contains(dedupe_key): return self.seen.add(dedupe_key) cmd = self.parse_command(str(msg.get("content") or "")) if not cmd: return log( f"收到指令 from={msg.get('from')} id={msg.get('id')} type={msg.get('type')} " f"content={str(msg.get('content') or '')[:120]}" ) try: self.handle_command(msg, cmd) except Exception as exc: log(f"处理指令失败: {exc}") try: self.send_text(str(msg.get("from") or ""), f"处理失败: {exc}") except Exception: pass def poll_once(self) -> None: self.refresh_auth_key(force=False) if not self.auth_key: log(f"未获取到 authKey（session: {self.session_file}），等待中...") return self.ensure_self_wxid() _, resp = self._request_json( base_url=self.wechat_base_url, path="/message/HttpSyncMsg", method="POST", payload={"Count": self.sync_count}, key=self.auth_key, timeout=30, accept_error=False, ) code = int(resp.get("Code") or 0) text = str(resp.get("Text") or "").strip() if code != 200: if "重新登录" in text or "不存在" in text: self.refresh_auth_key(force=True) log(f"消息轮询返回 code={code} text={text or '-'}") return data = resp.get("Data") if not isinstance(data, list) or not data: return for item in data: self.handle_polled_message(item) def run(self) -> int: log("微信桥接机器人启动") log(f"WeChat API: {self.wechat_base_url}") log(f"喜报 API: {self.xibao_base_url}") log(f"状态文件: {self.seen.path}") log(f"元数据文件: {self.meta_file}") log(f"每日清理时间: {self.cleanup_hour:02d}:{self.cleanup_minute:02d}") if self.allow_from: log(f"仅处理白名单发送者: {', '.join(sorted(self.allow_from))}") try: while True: try: self.maybe_run_daily_cleanup() self.poll_once() except Exception as exc: log(f"轮询异常: {exc}") now = time.time() if now - self._last_state_flush >= 6: self.seen.flush(force=False) self._last_state_flush = now if now - self._last_meta_flush >= 6: self._flush_meta(force=False) self._last_meta_flush = now if self.once: break time.sleep(self.poll_interval) except KeyboardInterrupt: log("收到中断信号，准备退出") finally: self._flush_state(force=True) return 0 def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Xibao <-> WeChatPadPro bridge") parser.add_argument("--wechat-base-url", default=DEFAULT_WCPP_BASE_URL, help="WeChatPadPro API 地址") parser.add_argument("--wechat-auth-key", default=os.environ.get("WCPP_AUTH_KEY", ""), help="WeChat authKey") parser.add_argument("--wechat-session-file", default=str(DEFAULT_SESSION_FILE), help="webui 会话文件路径") parser.add_argument("--xibao-base-url", default=DEFAULT_XIBAO_BASE_URL, help="喜报服务地址") parser.add_argument("--poll-interval", type=float, default=2.0, help="消息轮询间隔（秒）") parser.add_argument("--sync-count", type=int, default=30, help="每次轮询拉取条数") parser.add_argument("--max-images", type=int, default=3, help="每次最多回发图片数量") parser.add_argument("--state-file", default=str(DEFAULT_STATE_FILE), help="消息去重状态文件") parser.add_argument("--meta-file", default=str(DEFAULT_META_FILE), help="反馈上下文与清理状态文件") parser.add_argument("--daily-cleanup-time", default=DEFAULT_DAILY_CLEANUP_TIME, help="每日清理时间 HH:MM") parser.add_argument("--allow-from", default="", help="只处理这些发送者wxid（逗号分隔）") parser.add_argument("--once", action="store_true", help="只执行一次轮询") parser.add_argument("--dry-run", action="store_true", help="仅打印不真正回消息") return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv or sys.argv[1:]) bridge = WechatXibaoBridge(args) return bridge.run() if __name__ == "__main__": raise SystemExit(main())