105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
import os
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime
|
|
|
|
# Load env
|
|
def load_env():
|
|
paths = [".env.notion", "../.env.notion", "/app/.env.notion"]
|
|
for path in paths:
|
|
if os.path.exists(path):
|
|
with open(path) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line and not line.startswith('#') and '=' in line:
|
|
key, val = line.split('=', 1)
|
|
os.environ[key] = val.strip('"\'')
|
|
break
|
|
|
|
load_env()
|
|
TOKEN = os.environ.get("NOTION_API_KEY")
|
|
|
|
def request(method, url, data=None):
|
|
if not TOKEN:
|
|
print("ERROR: NOTION_API_KEY not found.")
|
|
return None
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {TOKEN}",
|
|
"Content-Type": "application/json",
|
|
"Notion-Version": "2022-06-28"
|
|
}
|
|
|
|
req = urllib.request.Request(url, headers=headers, method=method)
|
|
if data:
|
|
req.data = json.dumps(data).encode('utf-8')
|
|
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read().decode('utf-8'))
|
|
except urllib.error.HTTPError as e:
|
|
print(f"HTTP Error {e.code}: {e.read().decode('utf-8')}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Request Error: {e}")
|
|
return None
|
|
|
|
def find_db(title):
|
|
res = request("POST", "https://api.notion.com/v1/search", {
|
|
"query": title,
|
|
"filter": {"value": "database", "property": "object"}
|
|
})
|
|
if not res: return None
|
|
for item in res.get("results", []):
|
|
if item["title"][0]["plain_text"].lower() == title.lower():
|
|
return item["id"]
|
|
return None
|
|
|
|
def query_db(db_id, filter_payload=None):
|
|
payload = {}
|
|
if filter_payload: payload["filter"] = filter_payload
|
|
res = request("POST", f"https://api.notion.com/v1/databases/{db_id}/query", payload)
|
|
return res.get("results", []) if res else []
|
|
|
|
def get_title(page):
|
|
props = page.get("properties", {})
|
|
for key, val in props.items():
|
|
if val["type"] == "title" and val["title"]:
|
|
return val["title"][0]["plain_text"]
|
|
return "Untitled"
|
|
|
|
def get_status_options(db_id):
|
|
res = request("GET", f"https://api.notion.com/v1/databases/{db_id}")
|
|
if not res: return []
|
|
props = res.get("properties", {})
|
|
status = props.get("Status", {}).get("status", {})
|
|
return [opt["name"] for opt in status.get("options", [])]
|
|
|
|
def get_page_properties(page_id):
|
|
res = request("GET", f"https://api.notion.com/v1/pages/{page_id}")
|
|
return res.get("properties", {}) if res else {}
|
|
|
|
def get_property_value(page_id, prop_name):
|
|
props = get_page_properties(page_id)
|
|
if not props: return None
|
|
prop = props.get(prop_name)
|
|
if not prop: return None
|
|
|
|
if prop["type"] == "number":
|
|
return prop["number"]
|
|
# Add other types if needed
|
|
return None
|
|
|
|
def update_page(page_id, props):
|
|
return request("PATCH", f"https://api.notion.com/v1/pages/{page_id}", {"properties": props})
|
|
|
|
def append_blocks(page_id, blocks):
|
|
return request("PATCH", f"https://api.notion.com/v1/blocks/{page_id}/children", {"children": blocks})
|
|
|
|
def create_page(parent_db, props):
|
|
return request("POST", "https://api.notion.com/v1/pages", {
|
|
"parent": {"database_id": parent_db},
|
|
"properties": props
|
|
})
|