Wiring WeChat Push into Python Automation with PushPlus

PushPlus is the shortest path from a Python script to your phone. One HTTP POST with a token, a title, and a markdown body, and the message lands in WeChat. No app to build, no bot framework, no SMTP relay that gets flagged as spam. For unattended jobs — data collection, model retraining, portfolio snapshots — that single channel is usually enough.

This post covers the setup end to end: getting the token, a send function that survives network hiccups, naming conventions that keep dozens of scripts distinguishable, and the templates I reuse for daily reports versus alerts.

Getting the token

Register at pushplus.plus (or the Chinese mirror, pushplus.plus routes there too) with WeChat scan login. The account is bound to your WeChat, which is what makes the push arrive as a service message.

After login, open Send Message → One-on-One Message. The page shows a 32-character token. Treat it like a password: it authorizes anyone who holds it to send messages to your WeChat. Store it in an environment variable, not in source code.

# .env or your shell profile
export PUSHPLUS_TOKEN="a1b2c3d4e5f6..."

import os

TOKEN = os.environ["PUSHPLUS_TOKEN"]

A free account allows a limited number of pushes per day. For a handful of jobs that run a few times daily, that ceiling is never a problem. If you push per-row records or log every heartbeat, you will hit it — batch instead.

A send function that does not lie to you

The naive version is requests.post(url, json=payload) and done. That fails in three ways that matter for unattended jobs:

1. No timeout. A hung connection blocks the job forever.
2. No status check. PushPlus returns HTTP 200 even when the token is invalid; the error is in the JSON body.
3. No retry. A transient DNS failure kills the notification for the one run that needed it.

Here is the version I actually use:

import logging
import time

import requests

PUSHPLUS_URL = "https://www.pushplus.plus/send"
TOKEN = os.environ["PUSHPLUS_TOKEN"]

logger = logging.getLogger(__name__)


def push(
    title: str,
    content: str,
    *,
    template: str = "markdown",
    token: str = TOKEN,
    timeout: float = 10.0,
    retries: int = 2,
) -> bool:
    """Send a WeChat push via PushPlus. Returns True on confirmed delivery."""
    payload = {
        "token": token,
        "title": title,
        "content": content,
        "template": template,
    }

    for attempt in range(retries + 1):
        try:
            resp = requests.post(PUSHPLUS_URL, json=payload, timeout=timeout)
            resp.raise_for_status()
            body = resp.json()
        except requests.RequestException as exc:
            logger.warning("push attempt %d failed: %s", attempt + 1, exc)
            if attempt < retries:
                time.sleep(2 ** attempt)
                continue
            return False

        if body.get("code") == 200:
            return True

        # Business-level error: bad token, quota exceeded, etc.
        logger.error("push rejected: %s", body.get("msg"))
        return False

    return False

Three details worth calling out.

Check code, not just the HTTP status. PushPlus wraps everything in {"code": 200, "msg": "请求成功", "data": "..."}. An expired token returns HTTP 200 with code: 401. If you only check raise_for_status(), you will believe a dead token is working.

Exponential backoff on network errors only. Retrying a business error (bad token, quota) just burns quota. Retrying a socket timeout is worth it. The code above separates the two paths.

Return a boolean. Callers need to know whether the push succeeded. A job that cannot report its own failure should at least be able to log that the reporting failed.

Title conventions: prefix everything

Once you have five or six scripts pushing to the same WeChat thread, the notification list becomes a wall of text. The fix is a fixed prefix per system, so a glance at the lock screen tells you what fired and whether it matters.

[QUANT] Daily NAV — 2024-06-14
[SCRAPE] Job failed — 3 retries exhausted
[CLEAN] Weekly dedup done — 12,340 rows
[MONITOR] Disk usage 91% on host-02

The pattern is [SYSTEM] Short description — key metric or timestamp. The bracket tag is scannable; the em dash separates the event from its payload. Keep the tag under eight characters and uppercase it.

I centralize this in a tiny wrapper so no script invents its own format:

def notify(system: str, subject: str, body: str) -> bool:
    title = f"[{system.upper()}] {subject}"
    return push(title, body)

Call sites stay clean:

notify("quant", "Daily NAV — 2024-06-14", report_md)
notify("scrape", "Job failed — 3 retries exhausted", traceback_md)

Two rules keep this useful. First, never put the variable part of the message in the prefix — the prefix is for filtering, the suffix is for reading. Second, keep the tag vocabulary small. If you have fifteen systems, group them into four or five tags (QUANT, DATA, OPS, ALERT) and put the specific name in the subject.

Markdown templates: reports versus alerts

PushPlus renders template: "markdown" natively in WeChat. That means headers, bold, tables, and code blocks all render. Use it — a wall of plain text is harder to parse on a phone than a structured card.

The mistake is using one template for everything. A daily report and a failure alert have opposite reading patterns. The report is scanned for the numbers; the alert is scanned for the cause and the action.

Daily report template

Reports are read on a schedule, often half-awake. Front-load the numbers, push detail below.

def daily_report(date: str, metrics: dict, notes: str = "") -> str:
    rows = "\n".join(
        f"| {k} | {v} |" for k, v in metrics.items()
    )
    return f"""## Daily Report — {date}

| Metric | Value |
| --- | --- |
{rows}

**Notes**

{notes or "_none_"}
"""

The ## header renders as a section title in WeChat. The table aligns values. The notes section is where you put a one-liner about anything unusual — an empty notes field is a signal in itself.

Alert template

Alerts are read under stress. The first line must answer "what broke," the second "why," the third "what I should do." No preamble.

def alert_body(
    what: str,
    detail: str,
    action: str,
    traceback_text: str | None = None,
) -> str:
    body = f"""**What:** {what}

**Detail:** {detail}

**Action:** {action}
"""
    if traceback_text:
        body += f"\n```
{% endraw %}
\n{traceback_text[-1500:]}\n
{% raw %}
```\n"
    return body

The traceback is truncated to the last 1500 characters. The tail is where the actual exception lives; the head is usually framework noise. Truncating keeps the push readable and avoids hitting message size limits.

Use alerts sparingly. If every alert fires daily, none of them are alerts. Reserve the alert template for conditions that require a human decision — a failed job that will not self-heal, a threshold breach, a resource exhaustion.

Every unattended job ends with a push

This is the rule that matters most. Silence is not success.

A cron job that runs, does nothing visible, and exits is indistinguishable from a cron job that never started, crashed before logging, or was silently disabled by the scheduler. The only way to know is to make the job announce itself — every run, both outcomes.

import sys
import traceback


def main() -> None:
    try:
        result = run_pipeline()
    except Exception:
        notify(
            "quant",
            "Pipeline failed",
            alert_body(
                what="run_pipeline raised",
                detail="unhandled exception",
                action="check logs on host-02",
                traceback_text=traceback.format_exc(),
            ),
        )
        sys.exit(1)
    else:
        notify(
            "quant",
            f"Pipeline done — {result['rows']} rows",
            daily_report(result["date"], result["metrics"], result["notes"]),
        )


if __name__ == "__main__":
    main()

The try/except/else structure guarantees exactly one push per run. Success and failure both produce a message. If the process is killed by the OOM killer or the machine reboots, you get nothing — and that absence is itself the signal. A missing daily report at 09:00 is a problem worth investigating.

Two refinements on top of this:

Heartbeat for critical jobs. If a job must run daily and its absence is not obvious, add a separate dead-man's-switch. A tiny script that checks whether today's success push was recorded, and alerts if not. This catches the case where the job host is down entirely.

Deduplicate repeated failures. A job that fails every five minutes will send 288 alerts a day. Add a simple state file: if the same failure signature fired in the last hour, suppress the push and log instead.

import hashlib
import json
from pathlib import Path

STATE = Path("/var/lib/jobs/last_alert.json")


def should_alert(signature: str, cooldown_s: int = 3600) -> bool:
    now = time.time()
    try:
        state = json.loads(STATE.read_text())
    except (FileNotFoundError, json.JSONDecodeError):
        state = {}

    key = hashlib.sha1(signature.encode()).hexdigest()
    last = state.get(key, 0)
    if now - last < cooldown_s:
        return False

    state[key] = now
    STATE.write_text(json.dumps(state))
    return True

Call should_alert before the failure push. The signature can be the exception type plus the failing function name — enough to group identical failures without collapsing distinct ones.

Operational notes

A few things I learned the hard way:

• Token rotation. If a token leaks, anyone can push to your WeChat. Rotate it from the console and update the environment variable. Because the token lives in an env var, rotation is a one-line change and a restart, not a code edit.
• Timezone. PushPlus timestamps use the server's timezone. If your job runs in UTC and you read the push in local time, put the local date in the title explicitly, as the templates above do.
• Message size. Very large markdown bodies get truncated or rejected. Cap content at a few thousand characters and link to a full report if you need more.
• Quota awareness. Log the data field from successful responses — it contains the message ID, useful for debugging delivery. If you approach the daily quota, batch multiple metrics into one push rather than sending several.

The whole integration is maybe sixty lines of Python: a token in the environment, a send function with timeout and retry, a naming convention, two templates, and a hard rule that every job reports its own outcome. That is enough to make an unattended pipeline observable from your pocket.

More notes like this ship every week on this site.

---

Daily Picks

The following pairs are selected from the multi-timeframe trend scanner (Gate.io futures) and are for technical-analysis study only — not investment advice.
Data updated: 2026-10-10 08:46:32

Long

Pair	Signal	Price	Take Profit	Stop Loss	R/R
NEAR	高质量标的做多	$4.9102	$5.0821	$4.812	1:1.8

Short

Pair	Signal	Price	Take Profit	Stop Loss	R/R
MAGIC	热点跟涨	$0.1247	$0.1155	$0.1291	1:2.1

2 picks selected. Scanner runs every 15 minutes.

Comments

[PixQuill]

Python + PushPlus WeChat notification best‑practices: env‑var token, check business code not just HTTP status, always notify on both success and failure, add heartbeat and debounce for unattended jobs.