Add comments, stats, weekly report and a scheduled workflow example

Author: dynabe

.github/workflows/substack-weekly.yml

@@ -0,0 +1,26 @@
+# Example: run Substack automations on a schedule with GitHub Actions.
+# 1. Add COTLET_API_KEY under Settings > Secrets and variables > Actions.
+# 2. Uncomment `schedule`. Reading stats and comments needs no confirmation.
+name: Substack weekly
+
+on:
+  workflow_dispatch:
+  # schedule:
+  #   - cron: "0 8 * * 1"
+
+jobs:
+  substack:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Weekly subscriber and revenue report
+        env:
+          COTLET_API_KEY: ${{ secrets.COTLET_API_KEY }}
+        run: python weekly_report.py
+      - name: Comments waiting for a reply
+        env:
+          COTLET_API_KEY: ${{ secrets.COTLET_API_KEY }}
+        run: python unanswered_comments.py --days 7

post_stats.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+"""Opens, open rate, views, likes, restacks and new subscribers for your recent Substack posts, in one table.
+
+    python post_stats.py --last 10
+    python post_stats.py --since 2026-09-01 --out post-stats.csv
+    python post_stats.py --post-id 148213977
+
+Numbers keep growing for days after a send, so a post sent yesterday is marked as early.
+"""
+import argparse
+from datetime import datetime, timedelta, timezone
+
+from cotlet_client import call, paginate, run, write_rows
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--post-id", help="Only this post")
+    parser.add_argument("--last", type=int, default=10, help="Number of recent published posts. Default: 10")
+    parser.add_argument("--since", help="Only posts published on or after this date (YYYY-MM-DD)")
+    parser.add_argument("--out", help="Save to .csv or .json instead of printing")
+    args = parser.parse_args()
+
+    if args.post_id:
+        posts = [call("get_post", {"post_id": args.post_id})]
+    else:
+        posts = []
+        for post in paginate("list_posts", {"status": "published", "limit": 50}, "posts"):
+            if args.since and post["published_at"][:10] < args.since:
+                break
+            posts.append(post)
+            if not args.since and len(posts) >= args.last:
+                break
+
+    early = datetime.now(timezone.utc) - timedelta(days=2)
+    rows = []
+    for post in posts:
+        stats = call("get_post_stats", {"post_id": post["post_id"]})
+        published = post.get("published_at")
+        rows.append({
+            "title": post["title"],
+            "published_at": published,
+            **{k: v for k, v in stats.items() if k != "post_id"},
+            "early": bool(published) and datetime.fromisoformat(published.replace("Z", "+00:00")) > early,
+            "url": post.get("url"),
+        })
+    write_rows(rows, args.out)
+
+
+if __name__ == "__main__":
+    run(main)

publication_stats.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+"""Total, paid, new and churned Substack subscribers and revenue for the last 7, 30 or 90 days.
+
+    python publication_stats.py --period 7d
+"""
+import argparse
+
+from cotlet_client import call, run
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--period", choices=["7d", "30d", "90d"], default="30d")
+    args = parser.parse_args()
+
+    s = call("get_publication_stats", {"period": args.period})
+    paid = s.get("subscribers_paid") or 0
+    churn = (s["subscribers_churned"] / paid * 100) if paid else 0
+    print(f"Substack publication ({s['period']})")
+    print(f"Subscribers:  {s['subscribers_total']:,}")
+    print(f"Paid:         {paid:,}")
+    print(f"New:          +{s['subscribers_new']:,}")
+    print(f"Churned:      -{s['subscribers_churned']:,}  ({churn:.1f}% of paid)")
+    print(f"Revenue:      ${s.get('revenue_usd', 0):,}")
+
+
+if __name__ == "__main__":
+    run(main)

reply_to_comment.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+"""Reply as you under a comment on one of your Substack posts.
+
+    python reply_to_comment.py --comment-id c_90417 --body "Thank you, Maya. Pricing is the whole of part two."
+
+The commenter gets Substack's usual reply notification. By default Cotlet asks you to confirm first.
+"""
+import argparse
+import json
+
+from cotlet_client import call, run
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--comment-id", required=True, help="Comment ID from unanswered_comments.py")
+    parser.add_argument("--body", required=True, help="Reply text")
+    args = parser.parse_args()
+
+    print(json.dumps(call("reply_to_comment", {"comment_id": args.comment_id, "body": args.body}), indent=2))
+
+
+if __name__ == "__main__":
+    run(main)

unanswered_comments.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+"""Comments on your recent Substack posts that you have not answered yet.
+
+    python unanswered_comments.py --last 3
+    python unanswered_comments.py --days 14 --out unanswered.csv
+
+Reply with reply_to_comment.py, or let your AI agent draft the replies over MCP.
+"""
+import argparse
+from datetime import date, timedelta
+
+from cotlet_client import call, paginate, run, write_rows
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    window = parser.add_mutually_exclusive_group()
+    window.add_argument("--last", type=int, default=3, help="Check the last N published posts. Default: 3")
+    window.add_argument("--days", type=int, help="Check every post published in the last N days")
+    parser.add_argument("--out", help="Save to .csv or .json instead of printing")
+    args = parser.parse_args()
+
+    since = (date.today() - timedelta(days=args.days)).isoformat() if args.days else None
+    rows = []
+    checked = 0
+    for post in paginate("list_posts", {"status": "published", "limit": 50}, "posts"):
+        if since and post["published_at"][:10] < since:
+            break
+        if not since and checked >= args.last:
+            break
+        checked += 1
+        for c in call("list_comments", {"post_id": post["post_id"], "unanswered_only": True}).get("comments", []):
+            rows.append({"post": post["title"], "comment_id": c["comment_id"], "author": c["author"],
+                         "body": c["body"], "created_at": c["created_at"], "post_url": post.get("url")})
+    print(f"Checked {checked} posts, {len(rows)} comments waiting for you")
+    write_rows(rows, args.out)
+
+
+if __name__ == "__main__":
+    run(main)

weekly_report.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""Weekly Substack report: new, paid and churned subscribers, revenue, and how each post of the week did.
+
+    python weekly_report.py
+    python weekly_report.py --save last-week.json
+
+With --save the totals are stored, and the next run with the same file adds a
+week-on-week line. Revenue is Substack's gross annualized figure, not cash received.
+"""
+import argparse
+import json
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from cotlet_client import call, paginate, run
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--save", help="JSON file to compare with and update, e.g. last-week.json")
+    args = parser.parse_args()
+
+    week = call("get_publication_stats", {"period": "7d"})
+    month = call("get_publication_stats", {"period": "30d"})
+    weekly_churn = month["subscribers_churned"] / 30 * 7
+
+    print(f"Subscribers: {week['subscribers_total']:,} total (+{week['subscribers_new']} new)")
+    flag = "  (flagged: above the 30-day weekly average)" if week["subscribers_churned"] > weekly_churn else ""
+    print(f"Paid: {week['subscribers_paid']:,}, churned {week['subscribers_churned']} "
+          f"(30-day weekly average: {weekly_churn:.0f}){flag}")
+    print(f"Gross annualized revenue: ${week.get('revenue_usd', 0):,}")
+
+    now = datetime.now(timezone.utc)
+    print("Posts this week:")
+    for post in paginate("list_posts", {"status": "published", "limit": 50}, "posts"):
+        published = datetime.fromisoformat(post["published_at"].replace("Z", "+00:00"))
+        if published < now - timedelta(days=7):
+            break
+        s = call("get_post_stats", {"post_id": post["post_id"]})
+        early = " (early, under 2 days of data)" if published > now - timedelta(days=2) else ""
+        print(f'  "{post["title"]}": open rate {s["open_rate"]:.1%}, {s["subscribers_gained"]} new subscribers{early}')
+
+    if args.save:
+        path = Path(args.save)
+        if path.exists():
+            prev = json.loads(path.read_text())
+            print(f"Week on week: total {week['subscribers_total'] - prev['subscribers_total']:+,}, "
+                  f"paid {week['subscribers_paid'] - prev['subscribers_paid']:+,}")
+        else:
+            print("Week on week: no stored copy yet, saved this week for next time")
+        path.write_text(json.dumps(week, indent=2))
+
+
+if __name__ == "__main__":
+    run(main)