import os, requests, time
NOTION = os.environ["NOTION_API_KEY"]
MV = os.environ["MAVERA_API_KEY"]
NB = "https://api.notion.com/v1"
MB = "https://app.mavera.io/api/v1"
NH = {
"Authorization": f"Bearer {NOTION}",
"Notion-Version": "2022-06-28",
"Content-Type": "application/json",
}
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
KB_DB_ID = "your-knowledge-base-database-id"
CATEGORIES = ["Product Docs", "Blog Drafts", "Style Guide", "Internal Comms"]
# 1. Query knowledge base pages
all_pages = []
for category in CATEGORIES:
r = requests.post(f"{NB}/databases/{KB_DB_ID}/query", headers=NH, json={
"filter": {"property": "Category", "select": {"equals": category}},
"sorts": [{"property": "Last edited time", "direction": "descending"}],
"page_size": 10,
})
if r.status_code == 429:
time.sleep(1)
r = requests.post(f"{NB}/databases/{KB_DB_ID}/query", headers=NH, json={
"filter": {"property": "Category", "select": {"equals": category}},
"page_size": 10,
})
data = r.json()
all_pages.extend(data.get("results", []))
time.sleep(0.4)
print(f"Found {len(all_pages)} knowledge base pages across {len(CATEGORIES)} categories")
# 2. Extract text from pages
def get_blocks_text(block_id):
texts = []
cursor = None
while True:
params = {"page_size": 100}
if cursor:
params["start_cursor"] = cursor
r = requests.get(f"{NB}/blocks/{block_id}/children", headers=NH, params=params)
if r.status_code == 429:
time.sleep(1); continue
r.raise_for_status()
data = r.json()
for block in data.get("results", []):
btype = block.get("type", "")
rt = block.get(btype, {}).get("rich_text", [])
text = "".join(t.get("plain_text", "") for t in rt)
if text.strip():
texts.append(text)
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.4)
return "\n".join(texts)
samples = []
for page in all_pages[:25]:
props = page.get("properties", {})
title_parts = props.get("Name", props.get("Title", {})).get("title", [])
title = "".join(t.get("plain_text", "") for t in title_parts)
text = get_blocks_text(page["id"])
if len(text) > 100:
samples.append(f"=== {title} ===\n{text[:1500]}")
combined = "\n\n---\n\n".join(samples)
print(f"Collected {len(samples)} pages ({len(combined)} chars)")
# 3. Create Brand Voice
bv = requests.post(f"{MB}/brand-voices", headers=MH, json={
"name": "Notion Knowledge Base Voice",
"samples": [combined[:50000]],
"description": (
f"Extracted from {len(samples)} internal knowledge base pages "
f"across categories: {', '.join(CATEGORIES)}. "
"Captures organizational writing patterns, technical vocabulary, and tone."
),
}).json()
print(f"Brand Voice: {bv.get('id', 'error')}")
# 4. Wait and inspect
time.sleep(5)
detail = requests.get(f"{MB}/brand-voices/{bv['id']}", headers=MH).json()
print(f"Status: {detail.get('status', 'unknown')}")
if detail.get("traits"):
print(f"Traits: {detail['traits']}")
# 5. Test with a generation
from openai import OpenAI
mavera = OpenAI(api_key=MV, base_url=MB)
test = mavera.responses.create(
model="mavera-1",
input=[{"role": "user", "content": "Write a 100-word product update announcement for a new API versioning feature."}],
extra_body={"brand_voice_id": bv["id"]},
)
print(f"\nTest generation:\n{test.output[0].content[0].text}")