full-focus/scripts/fetch_weekly_data.py

#!/usr/bin/env python3
"""
Fetch weekly planning data from Notion databases.
Returns JSON with projects, tasks, and reading suggestions.
"""

import json
import os
import sys
from datetime import datetime, timedelta
from notion_client import Client

# Initialize Notion client - will use NOTION_TOKEN env var
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"

def get_focus_projects():
    """Get projects with 'Focus this week' = true and Status = Active."""
    results = notion.databases.query(
        database_id=PROJECTS_DB,
        filter={
            "and": [
                {
                    "property": "Focus this week",
                    "checkbox": {
                        "equals": True
                    }
                },
                {
                    "property": "Status",
                    "select": {
                        "equals": "Active"
                    }
                }
            ]
        }
    )

    projects = []
    for page in results["results"]:
        props = page["properties"]
        projects.append({
            "id": page["id"],
            "url": page["url"],
            "name": props["Name"]["title"][0]["plain_text"] if props["Name"]["title"] else "",
            "weekly_goal": props["Weekly goal"]["rich_text"][0]["plain_text"] if props["Weekly goal"]["rich_text"] else "",
            "icon": page.get("icon", {}).get("emoji", "")
        })

    return projects

def get_weekly_tasks():
    """Get tasks due this week that are not Done."""
    # Calculate date range for this week
    today = datetime.now().date()
    week_end = today + timedelta(days=7)

    results = notion.databases.query(
        database_id=TASKS_DB,
        filter={
            "and": [
                {
                    "property": "Status",
                    "status": {
                        "does_not_equal": "Done"
                    }
                },
                {
                    "property": "Due date",
                    "date": {
                        "on_or_before": week_end.isoformat()
                    }
                }
            ]
        },
        sorts=[
            {
                "property": "Due date",
                "direction": "ascending"
            }
        ]
    )

    tasks = []
    for page in results["results"]:
        props = page["properties"]

        # Get due date
        due_date = None
        if props["Due date"]["date"]:
            due_date = props["Due date"]["date"]["start"]

        # Get project name if linked
        project_names = []
        if props["Project"]["relation"]:
            for rel in props["Project"]["relation"]:
                try:
                    project = notion.pages.retrieve(page_id=rel["id"])
                    project_name = project["properties"]["Name"]["title"][0]["plain_text"]
                    project_names.append(project_name)
                except:
                    pass

        # Get blocking status
        has_blocking = len(props["Blocking"]["relation"]) > 0 if props["Blocking"]["relation"] else False

        tasks.append({
            "id": page["id"],
            "url": page["url"],
            "name": props["Task name"]["title"][0]["plain_text"] if props["Task name"]["title"] else "",
            "due_date": due_date,
            "status": props["Status"]["status"]["name"] if props["Status"]["status"] else "",
            "priority": props["Priority"]["select"]["name"] if props["Priority"]["select"] else "",
            "effort_level": props["Effort level"]["select"]["name"] if props["Effort level"]["select"] else "",
            "energy_type": props["Energy type"]["select"]["name"] if props["Energy type"]["select"] else "",
            "task_type": [t["name"] for t in props["Task type"]["multi_select"]],
            "project": project_names[0] if project_names else "",
            "has_blocking": has_blocking
        })

    return tasks

def get_reading_suggestions():
    """Get articles with Status = 'To Read' or 'In Progress'."""
    results = notion.databases.query(
        database_id=READING_LIST_DB,
        filter={
            "or": [
                {
                    "property": "Status",
                    "status": {
                        "equals": "To Read"
                    }
                },
                {
                    "property": "Status",
                    "status": {
                        "equals": "In Progress"
                    }
                }
            ]
        }
    )

    articles = []
    for page in results["results"]:
        props = page["properties"]
        articles.append({
            "id": page["id"],
            "url": props["URL"]["url"] if props["URL"]["url"] else "",
            "title": props["Title"]["title"][0]["plain_text"] if props["Title"]["title"] else "",
            "topic": [t["name"] for t in props["Topic"]["multi_select"]],
            "read_time": props["Read time"]["select"]["name"] if props["Read time"]["select"] else "",
            "status": props["Status"]["status"]["name"] if props["Status"]["status"] else ""
        })

    return articles

def main():
    try:
        data = {
            "projects": get_focus_projects(),
            "tasks": get_weekly_tasks(),
            "reading": get_reading_suggestions(),
            "generated_at": datetime.now().isoformat()
        }

        print(json.dumps(data, indent=2))
        return 0

    except Exception as e:
        print(json.dumps({"error": str(e)}), file=sys.stderr)
        return 1

if __name__ == "__main__":
    sys.exit(main())