#!/usr/bin/env python3 import json import os import datetime import urllib.request from pathlib import Path from zoneinfo import ZoneInfo ROOT = Path('/Users/openclaw/.openclaw/workspace') OUT = ROOT / 'knowledge/projects/dashboard-feed.json' KEY_PATH = Path.home() / '.config/notion/api_key' NOTION_VERSION = '2025-09-03' DS = os.environ.get('NOTION_TASKS_DS_ID', '98ac8c59-4c40-4672-a628-2c35392eecfa') TZ = ZoneInfo('America/Los_Angeles') NOW = datetime.datetime.now(TZ) TODAY = NOW.date() def notion_request(method, path, payload=None): key = KEY_PATH.read_text(encoding='utf-8').strip() data = None if payload is None else json.dumps(payload).encode('utf-8') req = urllib.request.Request( f'https://api.notion.com/v1{path}', data=data, method=method, headers={ 'Authorization': f'Bearer {key}', 'Notion-Version': NOTION_VERSION, 'Content-Type': 'application/json', }, ) with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode('utf-8')) def text(prop): return ''.join(x.get('plain_text', '') for x in (prop or [])).strip() def pick_date(props): d = (props.get('Reminder At') or {}).get('date') or (props.get('Due') or {}).get('date') if not d or not d.get('start'): return None s = d['start'] try: if 'T' in s: dt = datetime.datetime.fromisoformat(s.replace('Z', '+00:00')) if dt.tzinfo is None: dt = dt.replace(tzinfo=TZ) return dt.astimezone(TZ).date() return datetime.date.fromisoformat(s[:10]) except Exception: return None def in_work_hours(now): # Monday=0 ... Sunday=6 return now.weekday() <= 4 and 6 <= now.hour < 16 def collect_property_texts(props): vals = [] for v in (props or {}).values(): if not isinstance(v, dict): continue t = v.get('type') if t == 'title': vals.append(text(v.get('title'))) elif t == 'rich_text': vals.append(text(v.get('rich_text'))) elif t == 'select': vals.append((v.get('select') or {}).get('name', '')) elif t == 'multi_select': vals.extend([x.get('name', '') for x in (v.get('multi_select') or [])]) elif t == 'status': vals.append((v.get('status') or {}).get('name', '')) elif t == 'relation': # relation has ids only; skip pass elif t == 'url': vals.append(v.get('url') or '') elif t == 'email': vals.append(v.get('email') or '') elif t == 'phone_number': vals.append(v.get('phone_number') or '') elif t == 'formula': f = v.get('formula') or {} for key in ('string',): if f.get(key): vals.append(str(f.get(key))) return [x for x in vals if x] def _prop_values(prop): if not isinstance(prop, dict): return [] t = prop.get('type') if t == 'title': return [text(prop.get('title'))] if t == 'rich_text': return [text(prop.get('rich_text'))] if t == 'select': return [((prop.get('select') or {}).get('name', '') or '')] if t == 'multi_select': return [x.get('name', '') for x in (prop.get('multi_select') or [])] if t == 'status': return [((prop.get('status') or {}).get('name', '') or '')] if t == 'formula': f = prop.get('formula') or {} out = [] if f.get('string'): out.append(str(f.get('string'))) return out return [] def property_equals(props, prop_name, target): vals = [v.strip().lower() for v in _prop_values((props or {}).get(prop_name)) if v] return target.strip().lower() in vals def priority_bucket(name, props): # Work-hours prioritization is based on the explicit Notion Project field, # not free-text matching in task names/descriptions. if property_equals(props, 'Project', 'Triune Pools'): return 0 if property_equals(props, 'Project', 'Process'): return 1 return 2 rows = [] cursor = None while True: body = {'page_size': 100} if cursor: body['start_cursor'] = cursor page = notion_request('POST', f'/data_sources/{DS}/query', body) rows.extend(page.get('results', [])) if not page.get('has_more'): break cursor = page.get('next_cursor') active = [] blocked = [] needs_review = [] completed_today = [] next_actions = [] def item_obj(name, url, due=None): out = {'text': name, 'url': url} if due: out['due'] = due.isoformat() return out for r in rows: p = r.get('properties', {}) name = text((p.get('Name') or {}).get('title')) if not name: continue status_name = ((p.get('Completed') or {}).get('status') or {}).get('name', '') is_done = status_name == 'Done' labels = [x.get('name', '') for x in ((p.get('Labels') or {}).get('multi_select') or [])] due = pick_date(p) pri = ((p.get('Priority') or {}).get('select') or {}).get('name', '') if is_done: completed_at = ((p.get('Completed At') or {}).get('date') or {}).get('start') if completed_at and completed_at[:10] == TODAY.isoformat(): completed_today.append(item_obj(name, r.get('url', ''))) continue page_url = r.get('url', '') bucket = priority_bucket(name, p) item = (bucket, due or datetime.date.max, pri, name, page_url) if status_name == 'Needs review': needs_review.append(item) elif any('blocked' in l.lower() or 'waiting' in l.lower() for l in labels): blocked.append(item) else: active.append(item) # next actions = top 3 non-done by due then priority # During work hours (M-F 06:00-16:00 PT), prioritize Triune Pools and Process. if in_work_hours(NOW): sort_key = lambda x: (x[0], x[1], x[2], x[3]) else: sort_key = lambda x: (x[1], x[2], x[3]) all_open = sorted(active + blocked, key=sort_key) next_actions = [item_obj(x[3], x[4], None if x[1] == datetime.date.max else x[1]) for x in all_open[:3]] active_work = [ item_obj( x[3] + (f' (due {x[1].isoformat()})' if x[1] != datetime.date.max else ''), x[4], None if x[1] == datetime.date.max else x[1], ) for x in sorted(active, key=sort_key)[:8] ] blocked_work = [ item_obj( x[3] + (f' (due {x[1].isoformat()})' if x[1] != datetime.date.max else ''), x[4], None if x[1] == datetime.date.max else x[1], ) for x in sorted(blocked, key=sort_key)[:8] ] needs_review_work = [ item_obj( x[3] + (f' (due {x[1].isoformat()})' if x[1] != datetime.date.max else ''), x[4], None if x[1] == datetime.date.max else x[1], ) for x in sorted(needs_review, key=sort_key)[:8] ] payload = { 'activeWork': active_work, 'blocked': blocked_work, 'needsReview': needs_review_work, 'completedToday': completed_today[:12], 'nextActions': next_actions, } OUT.parent.mkdir(parents=True, exist_ok=True) OUT.write_text(json.dumps(payload, indent=2), encoding='utf-8') print(OUT)