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 sys
from datetime import datetime, timedelta

from notion_helpers import notion, get_data_source_id, query_all, PROJECTS_DB, TASKS_DB, READING_LIST_DB

def get_active_projects():
    """Get non-personal projects with Status = Active (candidates for weekly focus selection)."""
    pages = query_all(
        get_data_source_id(PROJECTS_DB),
        filter={
            "and": [
                {
                    "property": "Status",
                    "select": {
                        "equals": "Active"
                    }
                },
                {
                    "property": "Type",
                    "select": {
                        "does_not_equal": "Personal"
                    }
                }
            ]
        }
    )

    projects = []
    for page in pages:
        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 "",
            "focus_this_week": props["Focus this week"]["checkbox"],
            "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)

    pages = query_all(
        get_data_source_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 pages:
        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'."""
    pages = query_all(
        get_data_source_id(READING_LIST_DB),
        filter={
            "or": [
                {
                    "property": "Status",
                    "status": {
                        "equals": "To Read"
                    }
                },
                {
                    "property": "Status",
                    "status": {
                        "equals": "In Progress"
                    }
                }
            ]
        }
    )

    articles = []
    for page in pages:
        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_active_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())