97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
"""推送早报至企业微信群 webhook（超长自动分多条）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import certifi
|
||
import httpx
|
||
|
||
from daily.config import OUTPUT_DIR, ROOT, env_int, wecom_chunk_bytes
|
||
from daily.wecom_split import split_wecom_messages
|
||
|
||
_PUSH_GAP_MS = 300
|
||
|
||
|
||
def _load_webhook_key() -> str:
|
||
from daily.config import env
|
||
|
||
key = (env("WECOM_WEBHOOK_KEY") or "").strip()
|
||
if not key:
|
||
raise RuntimeError("请设置 WECOM_WEBHOOK_KEY（项目根 .env）")
|
||
return key
|
||
|
||
|
||
def _resolve_report_path(arg: str | None) -> Path:
|
||
if arg:
|
||
path = Path(arg)
|
||
if not path.is_absolute():
|
||
path = ROOT / path
|
||
return path
|
||
candidates = sorted(
|
||
OUTPUT_DIR.glob("*.wecom.md"),
|
||
key=lambda p: p.stat().st_mtime,
|
||
reverse=True,
|
||
)
|
||
if candidates:
|
||
return candidates[0]
|
||
legacy = sorted(ROOT.glob("*.wecom.md"), key=lambda p: p.stat().st_mtime, reverse=True)
|
||
if legacy:
|
||
return legacy[0]
|
||
raise RuntimeError("未找到 .wecom.md 报告，请先运行 python -m daily")
|
||
|
||
|
||
def _post_markdown(client: httpx.Client, url: str, content: str) -> None:
|
||
payload = {"msgtype": "markdown", "markdown": {"content": content}}
|
||
resp = client.post(url, json=payload)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
if data.get("errcode", 0) != 0:
|
||
raise RuntimeError(f"推送失败: errcode={data.get('errcode')} errmsg={data.get('errmsg')}")
|
||
|
||
|
||
def send_report(report_path: Path | None = None) -> None:
|
||
path = _resolve_report_path(str(report_path) if report_path else None)
|
||
if not path.exists():
|
||
raise RuntimeError(f"报告文件不存在: {path}")
|
||
|
||
content = path.read_text(encoding="utf-8").strip()
|
||
if not content:
|
||
raise RuntimeError(f"报告内容为空: {path}")
|
||
|
||
chunk_limit = wecom_chunk_bytes()
|
||
max_parts = env_int("DAILY_WECOM_MAX_PARTS", 5)
|
||
parts = split_wecom_messages(content, chunk_limit)
|
||
if len(parts) > max_parts:
|
||
raise RuntimeError(
|
||
f"早报需 {len(parts)} 条消息，超过 DAILY_WECOM_MAX_PARTS={max_parts}，请调小各区块条数"
|
||
)
|
||
|
||
key = _load_webhook_key()
|
||
url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={key}"
|
||
|
||
with httpx.Client(timeout=20.0, verify=certifi.where()) as client:
|
||
for i, part in enumerate(parts, 1):
|
||
if i > 1:
|
||
time.sleep(_PUSH_GAP_MS / 1000.0)
|
||
_post_markdown(client, url, part)
|
||
|
||
total_bytes = len(content.encode("utf-8"))
|
||
if len(parts) == 1:
|
||
print(f"已推送至企业微信: {path.name} ({total_bytes} bytes)")
|
||
else:
|
||
sizes = ", ".join(str(len(p.encode("utf-8"))) for p in parts)
|
||
print(f"已推送至企业微信: {path.name} ({total_bytes} bytes → {len(parts)} 条: {sizes})")
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
args = argv if argv is not None else sys.argv[1:]
|
||
try:
|
||
send_report(Path(args[0]) if args else None)
|
||
return 0
|
||
except Exception as exc:
|
||
print(f"ERROR: {exc}", file=sys.stderr)
|
||
return 1
|