Add fetch_evening_tasks.py script and shared notion_helpers module
Refactors weekly-template's Notion client setup into a shared helper (with .env/dotenv support and full pagination), and adds a dedicated evening-prep script so both commands can fall back to script-based fetching instead of the MCP search/fetch pattern. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
46
scripts/notion_helpers.py
Normal file
46
scripts/notion_helpers.py
Normal file
@@ -0,0 +1,46 @@
|
||||
"""Shared Notion client setup and query helpers for the FullFocus scripts."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from notion_client import Client
|
||||
|
||||
load_dotenv(Path(__file__).resolve().parent.parent / ".env")
|
||||
|
||||
_notion_token = os.getenv('NOTION_TOKEN')
|
||||
if not _notion_token:
|
||||
print(json.dumps({"error": "NOTION_TOKEN environment variable not set. See scripts/README.md for setup instructions."}), file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
notion = Client(auth=_notion_token)
|
||||
|
||||
# Database IDs
|
||||
PROJECTS_DB = "2c0abd6c450a8090aca3e0b2b0373c17"
|
||||
TASKS_DB = "2c0abd6c450a805098d3cc0e7d3dfccf"
|
||||
READING_LIST_DB = "2c4abd6c450a80cbae55c440dd9e2427"
|
||||
|
||||
_data_source_id_cache = {}
|
||||
|
||||
def get_data_source_id(database_id):
|
||||
"""Resolve a database ID to its first data source ID (notion-client 3.x API)."""
|
||||
if database_id not in _data_source_id_cache:
|
||||
db = notion.databases.retrieve(database_id=database_id)
|
||||
_data_source_id_cache[database_id] = db["data_sources"][0]["id"]
|
||||
return _data_source_id_cache[database_id]
|
||||
|
||||
def query_all(data_source_id, **kwargs):
|
||||
"""Query a data source, following pagination until all matching pages are collected."""
|
||||
pages = []
|
||||
start_cursor = None
|
||||
while True:
|
||||
if start_cursor:
|
||||
kwargs["start_cursor"] = start_cursor
|
||||
results = notion.data_sources.query(data_source_id=data_source_id, **kwargs)
|
||||
pages.extend(results["results"])
|
||||
if not results.get("has_more"):
|
||||
break
|
||||
start_cursor = results["next_cursor"]
|
||||
return pages
|
||||
Reference in New Issue
Block a user