成稿自动投递：把早报和周报一键推到 IM 与知识库

flowchart LR
A["#13 产物:archive/科技早报-*.md"] --> B["① 发现 discover<br />glob 取最新一份<br />判定:只取成品,不碰 sources/"]
C["#15 产物:weekly/周报-*.md"] --> B
B --> D["② 解析 parse<br />读标题+前N条二级标题<br />判定:只读成品md,绝不读 diff"]
D --> E["③ 组装 assemble<br />拼成一条 markdown 摘要<br />判定:内容可复现"]
E --> F["④ 多端投递 deliver<br />企业微信/邮件/ima 适配器<br />判定:无报告 exit 1 且 errcode=0"]
F -.->|"可选"| G["接 #14 定时<br />每天跑完自动投<br />判定:失败有人喊"]

"""
deliver.py —— 成稿自动投递（示例代码，已实测可运行）
流程：发现报告文件 -> 解析摘要 -> 组装成一条投递消息 -> 通过适配器推送到 IM/邮件/知识库
依赖：仅 Python 标准库；推送目标（webhook）由参数或环境变量注入。
文中已用本地 mock 服务器实测「企业微信机器人 webhook」通路。
"""
import argparse
import datetime as dt
import glob
import json
import os
import re
import urllib.request
from pathlib import Path

# 报告发现规则：每类取"最新修改"的一份
REPORT_GLOBS = {
"早报": "archive/科技早报-*.md",
"周报": "weekly/周报-*.md",
}

def discover(pattern: str) -> list[Path]:
files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True)
return [Path(f) for f in files]

def parse_digest(path: Path, kind: str, limit: int = 5) -> dict:
"""读报告正文，取标题 + 前 limit 条二级标题作摘要。只读成品文件，绝不碰源数据/脚本。"""
text = path.read_text(encoding="utf-8")
title_m = re.search(r"^#\s+(.+)$", text, re.M)
title = title_m.group(1).strip() if title_m else path.name
items = re.findall(r"^##\s+(.+)$", text, re.M)[:limit]
return {"kind": kind, "file": path.name, "title": title, "items": items}

def build_message(reports: list[dict]) -> str:
lines = [f"# 今日投递 · {dt.date.today().isoformat()}", ""]
total = 0
for r in reports:
lines.append(f"## {r['title']}（{len(r['items'])} 条预览）")
for it in r["items"]:
lines.append(f"- {it}")
lines.append(f"  _来源文件：{r['file']}_")
lines.append("")
total += len(r["items"])
lines.append(f"> 共投递 {len(reports)} 份报告、{total} 条内容预览")
return "\n".join(lines)

def fit_bytes(content: str, limit: int = 4096) -> str:
    """按 UTF-8 字节截断，且不切坏多字节字符——企业微信按字节而非字符计算长度"""
    raw = content.encode("utf-8")
    if len(raw) <= limit:
        return content
    return raw[:limit].decode("utf-8", errors="ignore")

class WebhookAdapter:
"""企业微信机器人 webhook（markdown 类型）。其他 IM（钉钉/飞书）同理，只换 payload 结构。"""

def __init__(self, url: str, timeout: int = 15):
self.url = url
self.timeout = timeout

def send(self, content: str) -> tuple[int, int, str]:
# 企业微信 markdown 消息体上限 4096 字节，超长截断
payload = json.dumps({
"msgtype": "markdown",
"markdown": {"content": fit_bytes(content)},
}).encode("utf-8")
req = urllib.request.Request(
self.url,
data=payload,
headers={"Content-Type": "application/json"},
)
# 企业微信在内容超限、参数非法、频率超限时会返回 HTTP 200 但 errcode 非 0，只判状态码等于假成功
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
status = resp.status
body = json.loads(resp.read().decode("utf-8"))
return status, body.get("errcode", -1), body.get("errmsg", "")

def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--root", default=".", help="报告根目录（含 archive/ 与 weekly/）")
ap.add_argument("--limit", type=int, default=5, help="每份报告取前几条作预览")
ap.add_argument("--webhook", default=os.environ.get("DELIVER_WEBHOOK", ""),
help="企业微信机器人 webhook URL（或设环境变量 DELIVER_WEBHOOK）")
ap.add_argument("--dry-run", action="store_true", help="只打印将要投递的内容，不真正发送")
args = ap.parse_args()

reports = []
for kind, pat in REPORT_GLOBS.items():
hits = discover(os.path.join(args.root, pat))[:1]  # 每类取最新一份
for p in hits:
reports.append(parse_digest(p, kind, args.limit))

if not reports:
# 守卫：没报告还"成功投递"就是静默失败，必须喊
print("[error] 未发现任何报告文件（期望 archive/科技早报-*.md 或 weekly/周报-*.md）")
return 1

msg = build_message(reports)
print("===== 将要投递的内容 =====")
print(msg)
print("==========================")

if args.dry_run:
print("[dry-run] 未发送，仅验证内容组装")
return 0

if not args.webhook:
print("[error] 未配置 --webhook（或 DELIVER_WEBHOOK 环境变量），无法投递；加 --dry-run 仅预览")
return 1

adapter = WebhookAdapter(args.webhook)
status, errcode, errmsg = adapter.send(msg)
print(f"[done] 投递完成，HTTP {status}，errcode={errcode} {errmsg}")
return 0 if (status == 200 and errcode == 0) else 1

if __name__ == "__main__":
raise SystemExit(main())

$ python deliver.py --root . --dry-run
===== 将要投递的内容 =====
# 今日投递 · 2026-10-25

## 科技早报 2026-10-24（3 条预览）
- 示例：某开源框架发布 2.0 版本，启动速度提升明显
- 示例：某云厂商上线 Agent 编排服务，支持可视化拖拽
- 示例：研究称小模型在垂直领域已逼近大模型水平
  _来源文件：科技早报-2026-10-24.md_

## 周报 · repo（窗口：7 days ago）（4 条预览）
- 新功能（1）
- 缺陷修复（1）
- 文档（1）
- 性能优化（1）
  _来源文件：周报-repo-7_days_ago.md_

> 共投递 2 份报告、7 条内容预览
==========================
[dry-run] 未发送，仅验证内容组装

$ python deliver.py --root . --webhook "http://127.0.0.1:8899/webhook"
...（内容与上同，略）...
[done] 投递完成，HTTP 200，errcode=0

$ python deliver.py --root empty_root --webhook "http://127.0.0.1:8899/webhook"
[error] 未发现任何报告文件（期望 archive/科技早报-*.md 或 weekly/周报-*.md）

# ---- 扩展适配器（接线骨架，需你的凭据；网络部分未在此环境实跑，填后自测） ----
import smtplib
from email.header import Header
from email.mime.text import MIMEText

class EmailAdapter:
    """SMTP 邮件投递：适合"留档"。正文用纯文本，避免把格式带进邮件。
    465 走 SMTP_SSL 直连，587 走 SMTP + starttls。"""
    def __init__(self, smtp_host: str, smtp_user: str, smtp_pass: str, to: str, port: int = 587, use_ssl: bool = False):
        self.host, self.user, self.passwd, self.to = smtp_host, smtp_user, smtp_pass, to
        self.port, self.use_ssl = port, use_ssl

    def send(self, content: str) -> int:
        msg = MIMEText(content, "plain", "utf-8")
        # 中文主题必须用 Header 编码，否则容易出现头部编码异常
        msg["Subject"] = Header("今日投递", "utf-8").encode()
        msg["From"], msg["To"] = self.user, self.to
        if self.use_ssl:
            with smtplib.SMTP_SSL(self.host, self.port) as s:
                s.login(self.user, self.passwd)
                s.send_message(msg)
        else:
            with smtplib.SMTP(self.host, self.port) as s:
                s.starttls()
                s.login(self.user, self.passwd)
                s.send_message(msg)
        return 0

class ImaAdapter:
    """写入 IMA 知识库笔记：适合"可检索问答"。按 #12 方法打标签，攒一个月能问。
    需你的 IMA 凭据；路径与鉴权以官方文档为准；此为骨架，未实测。"""
    def __init__(self, api_base: str, client_id: str, api_key: str, knowledge_base_id: str):
        self.base, self.cid, self.key, self.kb = api_base, client_id, api_key, knowledge_base_id

    def send(self, content: str) -> int:
        title = "今日投递"
        # 第一步：import_doc 创建可编辑笔记（note/v1，注意是单数；content_format=1 表示 markdown）
        body = json.dumps({"title": title, "content": content, "content_format": 1}).encode("utf-8")
        req = urllib.request.Request(
            f"{self.base}/openapi/note/v1/import_doc",
            data=body,
            headers={
                "Content-Type": "application/json",
                "ima-openapi-clientid": self.cid,
                "ima-openapi-apikey": self.key,
            },
        )
        with urllib.request.urlopen(req, timeout=30) as resp:
            result = json.loads(resp.read().decode("utf-8"))
        if result.get("code") != 0:
            return result.get("code", -1)
        note_id = result["data"]["note_id"]

        # 第二步：add_knowledge 关联进知识库（wiki/v1；media_type=11 表示笔记）
        # 注意：笔记走 note_info.content_id，不要传 media_id，否则报 220001
        add_body = json.dumps({
            "knowledge_base_id": self.kb,
            "media_type": 11,
            "note_info": {"content_id": note_id},
            "title": title,
            "folder_id": self.kb,  # 根目录时传 knowledge_base_id
        }).encode("utf-8")
        add_req = urllib.request.Request(
            f"{self.base}/openapi/wiki/v1/add_knowledge",
            data=add_body,
            headers={
                "Content-Type": "application/json",
                "ima-openapi-clientid": self.cid,
                "ima-openapi-apikey": self.key,
            },
        )
        with urllib.request.urlopen(add_req, timeout=30) as add_resp:
            add_result = json.loads(add_resp.read().decode("utf-8"))
        return add_result.get("code", -1)

flowchart TD
C["组装好的摘要 content"] --> R{"选投递渠道"}
R -->|"即时"| W["WebhookAdapter<br />企业微信机器人<br />判定:msgtype=markdown,≤4096字节(UTF-8)"]
R -->|"留档"| M["EmailAdapter<br />SMTP 邮件<br />判定:正文纯文本,不泄密"]
R -->|"可检索"| I["ImaAdapter<br />写入知识库笔记<br />判定:仅成品,打标签"]

flowchart TD
E["投递不对劲"] --> Q1{"推送返回非200/errcode!=0?"}
Q1 -->|"是"| C1["webhook 地址/权限错<br />查 DELIVER_WEBHOOK 环境变量<br />判定:地址勿入库,用环境变量"]
Q1 -->|"否"| Q2{"报告没生成却'成功'?"}
Q2 -->|"是"| C2["缺守卫→静默失败<br />接 #14 失败推送+exit1<br />判定:无报告必须 exit 1"]
Q2 -->|"否"| Q3{"群消息太长被截断?"}
Q3 -->|"是"| C3["超 4096 字节<br />减摘要条数/只发预览<br />判定:按UTF-8字节截断,不按字符数"]
Q3 -->|"否"| C4["内容含敏感/源码?<br />确认只读成品md<br />判定:不碰 sources/diff"]

flowchart LR
    L1["第1级 手动脚本<br />#13/#15 你敲命令它跑<br />判定:代码实测跑通"] --> L2["第2级 定时自动<br />#14 Agent 到点执行<br />判定:测试运行通过"]
    L2 --> L3["第2.5级 失败告警<br />空跑/坏源有人喊<br />判定:连续异常第1天知道"]
    L3 --> L4["第3级 自动投递闭环<br />本文:成稿推 IM/邮件/知识库<br />判定:产物有人收,不静默"]
    L4 --> L5["第4级 多源汇总月报<br />早报+周报合并复盘<br />判定:每步仍可单独验证"]