How to Automate Notion Content Publishing using the API

1import os
2import hmac
3import hashlib
4import time
5import threading
6import requests
7from flask import Flask, request, jsonify
8
9app = Flask(__name__)
10
11NOTION_TOKEN = os.environ['NOTION_TOKEN']
12WEBHOOK_SECRET = os.environ['NOTION_WEBHOOK_SECRET']
13WP_BASE = os.environ['WP_BASE_URL']  # e.g. https://yoursite.com/wp-json/wp/v2
14WP_AUTH = (os.environ['WP_USERNAME'], os.environ['WP_APP_PASSWORD'])
15
16H = {'Authorization': f'Bearer {NOTION_TOKEN}', 'Notion-Version': '2025-09-03'}
17
18def verify_sig(body, sig):
19    d = hmac.new(WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest()
20    return hmac.compare_digest(f'sha256={d}', sig)
21
22def notion_get(path):
23    for _ in range(5):
24        r = requests.get(f'https://api.notion.com{path}', headers=H)
25        if r.status_code == 429:
26            time.sleep(int(r.headers.get('Retry-After', 5)))
27            continue
28        r.raise_for_status()
29        time.sleep(0.35)
30        return r.json()
31    raise Exception(f'Max retries for {path}')
32
33def fetch_children(block_id):
34    blocks, cursor = [], None
35    while True:
36        params = {'page_size': 100}
37        if cursor: params['start_cursor'] = cursor
38        r = requests.get(f'https://api.notion.com/v1/blocks/{block_id}/children', headers=H, params=params)
39        if r.status_code == 429:
40            time.sleep(int(r.headers.get('Retry-After', 5)))
41            continue
42        r.raise_for_status()
43        data = r.json()
44        blocks.extend(data['results'])
45        if not data['has_more']: break
46        cursor = data['next_cursor']
47        time.sleep(0.35)
48    return blocks
49
50def fetch_all(block_id, depth=0):
51    if depth > 5: return []
52    result = []
53    for block in fetch_children(block_id):
54        block['_depth'] = depth
55        result.append(block)
56        if block.get('has_children') and block['type'] not in ['child_page', 'child_database']:
57            result.extend(fetch_all(block['id'], depth + 1))
58    return result
59
60def rich_to_html(arr):
61    out = ''
62    for rt in arr:
63        txt = rt.get('plain_text', '')
64        ann = rt.get('annotations', {})
65        if ann.get('bold'): txt = f'<strong>{txt}</strong>'
66        if ann.get('italic'): txt = f'<em>{txt}</em>'
67        if ann.get('code'): txt = f'<code>{txt}</code>'
68        if rt.get('href'): txt = f'<a href="{rt["href"]}">{txt}</a>'
69        out += txt
70    return out
71
72def to_html(blocks):
73    parts = []
74    for b in blocks:
75        t = b['type']
76        d = b.get(t, {})
77        rt = d.get('rich_text', [])
78        if t == 'paragraph': parts.append(f'<p>{rich_to_html(rt)}</p>')
79        elif t in ('heading_1','heading_2','heading_3'):
80            n = t[-1]
81            parts.append(f'<h{n}>{rich_to_html(rt)}</h{n}>')
82        elif t in ('bulleted_list_item','numbered_list_item'):
83            parts.append(f'<li>{rich_to_html(rt)}</li>')
84        elif t == 'code':
85            parts.append(f'<pre><code>{rich_to_html(rt)}</code></pre>')
86        elif t == 'image':
87            src = b.get('_img_url', '#')
88            parts.append(f'<img src="{src}" alt="">')
89        elif t == 'quote':
90            parts.append(f'<blockquote>{rich_to_html(rt)}</blockquote>')
91        elif t == 'divider': parts.append('<hr>')
92    return '\n'.join(parts)
93
94def publish_page(page):
95    page_id = page['id']
96    title = page['properties'].get('Title', {}).get('title', [])
97    title_text = title[0]['plain_text'] if title else 'Untitled'
98    blocks = fetch_all(page_id)
99    # Download images
100    for b in blocks:
101        if b['type'] == 'image':
102            img = b['image']
103            url = img['file']['url'] if img['type'] == 'file' else img['external']['url']
104            try:
105                r = requests.get(url, timeout=30)
106                b['_img_url'] = url  # In prod: upload to S3, use permanent URL
107            except: b['_img_url'] = '#'
108    html = to_html(blocks)
109    # Push to WordPress
110    wp_resp = requests.post(f'{WP_BASE}/posts', auth=WP_AUTH, json={
111        'title': title_text, 'content': html, 'status': 'publish'
112    })
113    wp_resp.raise_for_status()
114    print(f'Published to WordPress: {wp_resp.json()["link"]}')
115    # Mark Notion page as Published
116    requests.patch(f'https://api.notion.com/v1/pages/{page_id}', headers={**H, 'Content-Type': 'application/json'},
117                   json={'properties': {'Status': {'select': {'name': 'Published'}}}})
118
119@app.route('/webhooks/notion', methods=['POST'])
120def webhook():
121    raw = request.get_data()
122    if not verify_sig(raw, request.headers.get('X-Notion-Signature', '')):
123        return jsonify({'error': 'bad sig'}), 401
124    p = request.json
125    page_id = p.get('entity', {}).get('id')
126    if page_id:
127        page = notion_get(f'/v1/pages/{page_id}')
128        if page.get('properties', {}).get('Status', {}).get('select', {}).get('name') == 'Ready to Publish':
129            threading.Thread(target=publish_page, args=(page,), daemon=True).start()
130    return jsonify({'ok': True})
131
132if __name__ == '__main__':
133    app.run(port=3000)