Add engagement metrics to article archives

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 <skill-directory>\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 <skill-directory>\scripts\wechat_agent.py history "https://mp.weixin.qq.com/s/..." --output ".\wechat-history" --days 7 --timeout 300
+py -3 <skill-directory>\scripts\wechat_agent.py history "https://mp.weixin.qq.com/s/..." --output ".\wechat-history" --days 7 --metrics --timeout 300
 
 py -3 <skill-directory>\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
 

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 个自然日")