diff --git a/plugins/wechat-article-agent/skills/wechat-article-agent/SKILL.md b/plugins/wechat-article-agent/skills/wechat-article-agent/SKILL.md index dd482cd..3aa5946 100644 --- a/plugins/wechat-article-agent/skills/wechat-article-agent/SKILL.md +++ b/plugins/wechat-article-agent/skills/wechat-article-agent/SKILL.md @@ -1,6 +1,6 @@ --- name: wechat-article-agent -description: Transform publicly accessible WeChat Official Account content into portable local knowledge assets through conversation. Produces offline HTML with local media, Markdown, structured metadata, batch indexes, and user-authorized account-history archives. +description: Download publicly accessible WeChat Official Account articles into offline HTML, Markdown, structured metadata, batch indexes, user-authorized account-history archives, and best-effort public engagement counters. Use for acquisition and factual data handoff, not content analysis. --- # WeChat Archive Agent @@ -9,6 +9,14 @@ Operate the bundled archival engine through natural-language requests. Resolve t The default deliverable for each article is a portable folder containing article.html, article.md, meta.json, and locally downloaded assets. +## Scope boundary + +Act only as an acquisition layer: + +- Download and verify public articles, media assets, metadata, and requested public engagement counters. +- Return factual counts, timestamps, file paths, failures, and completeness status in manifests. +- Do not rank articles, infer performance causes, recommend samples, or analyze topics, titles, or structures. Hand the corpus and raw metrics to a separate analysis workflow. + ## Safety and privacy - Download only publicly accessible official-account articles requested by the user. @@ -69,10 +77,10 @@ py -3 \scripts\wechat_agent.py auth-url "https://mp.weixin.qq.c Ask the user to send the emitted URL to desktop WeChat's File Transfer Assistant and open it inside WeChat's built-in browser. The user must not send credentials. -Then run history with the chosen scope: +Then run history with the chosen scope. Add `--metrics` only when the user asks for performance data. It best-effort reads counters publicly displayed to the authorized WeChat client; unavailable counters must remain unavailable rather than guessed, and credentials must never be written to disk: ~~~powershell -py -3 \scripts\wechat_agent.py history "https://mp.weixin.qq.com/s/..." --output ".\wechat-history" --days 7 --timeout 300 +py -3 \scripts\wechat_agent.py history "https://mp.weixin.qq.com/s/..." --output ".\wechat-history" --days 7 --metrics --timeout 300 py -3 \scripts\wechat_agent.py history "https://mp.weixin.qq.com/s/..." --output ".\wechat-history" --from-date 2026-07-01 --to-date 2026-07-15 --timeout 300 @@ -97,6 +105,7 @@ The history command refuses to run without days, a custom start date, a positive - Confirm the HTML opens with the network disconnected and that resource attributes do not retain remote URLs. Sanitized ordinary hyperlinks may remain. - Search the output for temporary authorization parameter names and report any failure instead of claiming success. - Report warnings, skipped resources, and failed articles accurately. +- When `--metrics` is used, report the count of articles with available metrics separately from archive success. Do not replace unavailable counters with zero. ## Failure handling diff --git a/plugins/wechat-article-agent/skills/wechat-article-agent/scripts/wechat_agent.py b/plugins/wechat-article-agent/skills/wechat-article-agent/scripts/wechat_agent.py index 5bb6361..9682a1d 100644 --- a/plugins/wechat-article-agent/skills/wechat-article-agent/scripts/wechat_agent.py +++ b/plugins/wechat-article-agent/skills/wechat-article-agent/scripts/wechat_agent.py @@ -17,7 +17,7 @@ from datetime import datetime, timedelta from pathlib import Path from urllib.error import HTTPError, URLError -from urllib.parse import parse_qsl, quote, urlencode, urljoin, urlparse, urlunparse +from urllib.parse import parse_qs, parse_qsl, quote, urlencode, urljoin, urlparse, urlunparse from urllib.request import HTTPCookieProcessor, Request, build_opener from bs4 import BeautifulSoup, NavigableString, Tag @@ -200,6 +200,42 @@ def get( raise ArchiveError(f"网络连接失败：{exc.reason}") from None + def post( + self, + url: str, + data: dict[str, object], + timeout: float = 45, + headers: dict[str, str] | None = None, + ) -> HttpResponse: + request_headers = dict(self.headers) + request_headers["Content-Type"] = "application/x-www-form-urlencoded; charset=UTF-8" + if headers: + request_headers.update(headers) + request = Request( + url, + data=urlencode([(key, str(value)) for key, value in data.items()]).encode("utf-8"), + headers=request_headers, + method="POST", + ) + try: + with self.opener.open(request, timeout=timeout) as response: + return HttpResponse( + status_code=getattr(response, "status", 200), + url=response.geturl(), + headers=response.headers, + content=response.read(), + ) + except HTTPError as exc: + return HttpResponse( + status_code=exc.code, + url=exc.geturl(), + headers=exc.headers, + content=exc.read(), + ) + except URLError as exc: + raise ArchiveError(f"network connection failed: {exc.reason}") from None + + @dataclass class ArchiveResult: title: str @@ -722,6 +758,7 @@ def write_index(output_root: Path, results: list[ArchiveResult], errors: list[di PROFILE_ENDPOINT = "https://mp.weixin.qq.com/mp/profile_ext" +ARTICLE_METRICS_ENDPOINT = "https://mp.weixin.qq.com/mp/getappmsgext" SKIPPED_CACHE_DIRECTORIES = { "video", "filestorage", @@ -1148,6 +1185,88 @@ def history_items(page: dict[str, object]) -> list[dict[str, object]]: return items +def fetch_article_metrics( + session: HttpSession, + credential: Credentials, + item: dict[str, object], +) -> dict[str, object] | None: + """Best-effort counters exposed to the authorized WeChat client.""" + article_url = str(item.get("fetch_url") or item.get("source_url") or "") + query = parse_qs(urlparse(article_url).query) + biz = clean_text((query.get("__biz") or [""])[0]) + mid = clean_text((query.get("mid") or [""])[0]) + idx = clean_text((query.get("idx") or [""])[0]) + sn = clean_text((query.get("sn") or [""])[0]) + if not all((biz, mid, idx, sn)): + return None + params = { + **credential_params(biz, credential), + "appmsg_type": 9, + "mid": mid, + "idx": idx, + "sn": sn, + "is_only_read": 1, + "f": "json", + } + try: + response = session.post( + ARTICLE_METRICS_ENDPOINT, + data=params, + timeout=30, + headers={"Referer": article_url, "X-Requested-With": "XMLHttpRequest"}, + ) + if response.status_code == 429: + raise RiskControlError("WeChat metrics endpoint returned rate limiting") + if response.status_code >= 400: + return None + data = json.loads(response.text) + if not isinstance(data, dict): + return None + risk_marker = detect_risk_control(response.text) + if risk_marker: + raise RiskControlError(f"WeChat metrics endpoint risk-control response: {risk_marker}") + stat = data.get("appmsgstat") + if not isinstance(stat, dict): + return None + metrics: dict[str, object] = { + "retrieved_at": datetime.now().astimezone().isoformat(timespec="seconds"), + "source": "wechat_authorized_client", + } + aliases = { + "read_count": ("read_num", "read_count", "real_read_num"), + "like_count": ("like_num", "like_count", "old_like_num"), + "share_count": ("share_num", "share_count"), + "comment_count": ("comment_count", "comment_num"), + } + for name, keys in aliases.items(): + for key in keys: + value = stat.get(key) + if isinstance(value, (int, float)): + metrics[name] = int(value) + break + if isinstance(value, str) and value.isdigit(): + metrics[name] = int(value) + break + return metrics if len(metrics) > 2 else None + except RiskControlError: + raise + except Exception: + return None + + +def save_article_metrics(result: ArchiveResult, metrics: dict[str, object]) -> None: + """Add metrics to local metadata without ever writing credentials.""" + metadata_file = Path(result.metadata_file) + try: + metadata = json.loads(metadata_file.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return + if not isinstance(metadata, dict): + return + metadata["metrics"] = metrics + metadata_file.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8") + + def history_message_count(page: dict[str, object]) -> int: raw_messages = page.get("general_msg_list", {}) if isinstance(raw_messages, str): @@ -1312,6 +1431,8 @@ def archive_history( items: list[dict[str, object]], output_root: Path, pacing: PacingPolicy, + credential: Credentials | None = None, + collect_metrics: bool = False, ) -> tuple[list[ArchiveResult], list[dict[str, str]], Path, str]: output_root.mkdir(parents=True, exist_ok=True) results: list[ArchiveResult] = [] @@ -1322,6 +1443,13 @@ def archive_history( try: result = archive_article(str(item["fetch_url"]), output_root, session) results.append(result) + if collect_metrics and credential is not None: + metrics = fetch_article_metrics(session, credential, item) + if metrics is not None: + save_article_metrics(result, metrics) + emit("article_metrics", current=position, total=total, title=result.title, metrics=metrics) + else: + emit("article_metrics", current=position, total=total, title=result.title, status="unavailable") emit( "history_download", current=position, @@ -1460,7 +1588,25 @@ def command_history(args: argparse.Namespace) -> int: items, output_root, pacing, + credential=credential, + collect_metrics=bool(args.metrics), ) + metric_records: list[dict[str, object]] = [] + if args.metrics: + for result in results: + try: + metadata = json.loads(Path(result.metadata_file).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + continue + metrics = metadata.get("metrics") if isinstance(metadata, dict) else None + if isinstance(metrics, dict): + metric_records.append({ + "title": result.title, + "published_at": result.published_at, + "source_url": result.source_url, + "metadata_file": result.metadata_file, + "metrics": metrics, + }) manifest = { "account_biz": biz, "completed_at": datetime.now().astimezone().isoformat(timespec="seconds"), @@ -1473,6 +1619,9 @@ def command_history(args: argparse.Namespace) -> int: "errors": errors, "index_file": str(index_file.resolve()), "credential_storage": "memory_only", + "metrics_requested": bool(args.metrics), + "metrics_available_count": len(metric_records), + "metrics": metric_records, "stopped_for_risk_control": bool(risk_stop), } manifest_file = output_root / "history-result.json" @@ -1623,6 +1772,11 @@ def build_parser() -> argparse.ArgumentParser: history.add_argument("source", help="任意一篇文章链接或公众号 biz") history.add_argument("--output", default="wechat-history") history.add_argument("--timeout", type=int, default=300) + history.add_argument( + "--metrics", + action="store_true", + help="Best-effort public reading/like/share counters from the authorized WeChat client", + ) history.add_argument("--limit", type=int, default=0, help="最多下载篇数") history.add_argument("--max-pages", type=int, default=200) history.add_argument("--days", type=int, help="下载最近 N 个自然日")