#!/usr/bin/env python3 from __future__ import annotations import html, json, math, os, re, shutil, sqlite3, subprocess, time, urllib.error, urllib.request, yaml from collections import Counter from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlparse from zoneinfo import ZoneInfo from typing import Any ROOT=Path(__file__).resolve().parents[1] DIGEST_ROOT=Path(os.environ.get('DORIS_DIGEST_DIR','/home/fizzlepoof/.openclaw/workspace/doris-digest')) PUBLIC=ROOT/'public'; DATA=ROOT/'data'; LOGS=ROOT/'logs' PAPERLESS_STATE_PATH=DATA/'paperless_review_state.json' LAT=36.53; LON=-87.36 LOCAL_TZ=ZoneInfo('America/Chicago') HERMES_HOME=Path.home()/'.hermes' HERMES_STATE_DB=HERMES_HOME/'state.db' HERMES_CONFIG=HERMES_HOME/'config.yaml' LABELS={ 'astronomy':('๐Ÿ”ญ','Astronomy'),'homelab':('๐Ÿ ','Homelab'),'smart_home':('๐Ÿก','Smart Home'), 'local':('๐Ÿ“','Local'),'radio':('๐Ÿ“ป','Ham Radio'),'security':('โš ๏ธ','Security'), 'tech':('๐Ÿงฐ','Tech'),'us':('๐Ÿ“ฐ','News'),'world':('๐ŸŒ','World'),'cooking':('๐Ÿณ','Cooking'),'gaming':('๐ŸŽฎ','Gaming')} WEATHER_CODES={0:'Clear',1:'Mainly clear',2:'Partly cloudy',3:'Overcast',45:'Fog',48:'Rime fog',51:'Light drizzle',53:'Drizzle',55:'Heavy drizzle',61:'Light rain',63:'Rain',65:'Heavy rain',71:'Light snow',73:'Snow',75:'Heavy snow',80:'Rain showers',81:'Rain showers',82:'Heavy showers',95:'Thunderstorm'} LOW_VALUE_LABELS={'shared','duesoon','submitted','waitingsubmission','notes'} PRIORITY_COLORS={1:'#ff6b6b',2:'#ff9f1c',3:'#ffd166',4:'#8ecae6'} def esc(v:Any)->str: return html.escape(str(v or ''),quote=True) def load_json(p:Path,d:Any)->Any: try: return json.loads(p.read_text()) if p.exists() else d except Exception: return d def strip_markup(text:str)->str: text=re.sub(r'<[^>]+>',' ',text or '') return re.sub(r'\s+',' ',html.unescape(text)).strip() def parse_time(v:str)->datetime: try: return datetime.fromisoformat((v or '').replace('Z','+00:00')).astimezone(timezone.utc) except Exception: return datetime.now(timezone.utc) def parse_dt(v:str|None)->datetime|None: if not v: return None try: return datetime.fromisoformat(v.replace('Z','+00:00')) except Exception: return None def c_to_f(c:float|int|None)->str: if c is None: return 'โ€”' return f"{round(c*9/5+32)}ยฐF" def kmh_to_mph(k:float|int|None)->str: if k is None: return 'โ€”' return f"{round(k*0.621371)} mph" def compact_number(value:int|float|None)->str: if value in (None,''): return 'โ€”' try: n=float(value) except Exception: return str(value) abs_n=abs(n) if abs_n >= 1_000_000_000: return f"{n/1_000_000_000:.1f}B" if abs_n >= 1_000_000: return f"{n/1_000_000:.1f}M" if abs_n >= 1_000: return f"{n/1_000:.1f}K" if n.is_integer(): return f"{int(n):,}" return f"{n:,.1f}" def total_tokens_value(row:dict[str,Any] | sqlite3.Row)->int: keys=('input_tokens','output_tokens','cache_read_tokens','cache_write_tokens','reasoning_tokens') total=0 for key in keys: try: total += int((row[key] if isinstance(row, sqlite3.Row) else row.get(key,0)) or 0) except Exception: pass return total def read_hermes_model_config()->dict[str,str]: default='unknown'; provider='unknown' try: data=yaml.safe_load(HERMES_CONFIG.read_text(encoding='utf-8')) or {} except Exception: return {'default_model':default,'provider':provider} model_cfg=data.get('model') if isinstance(data,dict) else {} if isinstance(model_cfg,dict): default=str(model_cfg.get('default') or default) provider=str(model_cfg.get('provider') or provider) return {'default_model':default or 'unknown','provider':provider or 'unknown'} def moon_phase(dt:datetime)->dict[str,str]: # Known new moon: 2000-01-06 18:14 UTC; synodic month 29.53058867 days. known=datetime(2000,1,6,18,14,tzinfo=timezone.utc) days=(dt.astimezone(timezone.utc)-known).total_seconds()/86400 age=days%29.53058867 illum=(1-math.cos(2*math.pi*age/29.53058867))/2 names=[(1.85,'New Moon','๐ŸŒ‘'),(5.54,'Waxing Crescent','๐ŸŒ’'),(9.23,'First Quarter','๐ŸŒ“'),(12.92,'Waxing Gibbous','๐ŸŒ”'),(16.61,'Full Moon','๐ŸŒ•'),(20.30,'Waning Gibbous','๐ŸŒ–'),(23.99,'Last Quarter','๐ŸŒ—'),(27.68,'Waning Crescent','๐ŸŒ˜'),(29.54,'New Moon','๐ŸŒ‘')] name,emoji='New Moon','๐ŸŒ‘' for limit,n,e in names: if agedict[str,Any]: url=(f'https://api.open-meteo.com/v1/forecast?latitude={LAT}&longitude={LON}' '¤t=temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code,apparent_temperature' '&hourly=temperature_2m,precipitation_probability' '&daily=precipitation_sum,sunrise,sunset&timezone=America/Chicago&forecast_days=3') try: with urllib.request.urlopen(url,timeout=12) as r: raw=json.load(r) cur=raw.get('current',{}); daily=raw.get('daily',{}); hourly=raw.get('hourly',{}) probs=hourly.get('precipitation_probability') or [] return {'ok':True,'current':cur,'daily':daily,'max_precip':max(probs[:24] or [0]),'condition':WEATHER_CODES.get(cur.get('weather_code'),'Weather code '+str(cur.get('weather_code'))),'source':'Open-Meteo'} except Exception as e: return {'ok':False,'error':str(e),'source':'Open-Meteo'} def decrypt_donetick()->dict[str,Any]|None: secret=Path('/home/fizzlepoof/.openclaw/workspace/.secrets/donetick_secrets.json.enc') keyfile=Path('/home/fizzlepoof/.openclaw/workspace/.secrets/encryption.key') if not secret.exists() or not keyfile.exists(): return None try: res=subprocess.run(['openssl','enc','-aes-256-cbc','-d','-pbkdf2','-in',str(secret),'-k',keyfile.read_text().strip()],capture_output=True,text=True,timeout=8,check=True) return json.loads(res.stdout).get('donetick') except Exception: return None def fetch_donetick_todos()->list[dict[str,Any]]: sec=decrypt_donetick() if not sec: return [] try: login=urllib.request.Request(sec['instance']+sec['jwt_endpoint'],data=json.dumps({'username':sec['service_account']['username'],'password':sec['service_account']['password']}).encode(),headers={'Content-Type':'application/json','Accept':'application/json'},method='POST') with urllib.request.urlopen(login,timeout=12) as r: token=json.loads(r.read())['access_token'] headers={'Authorization':'Bearer '+token,'Accept':'application/json'} req=urllib.request.Request(sec['instance']+'/api/v1/chores',headers=headers) with urllib.request.urlopen(req,timeout=12) as r: chores=json.loads(r.read()).get('res',[]) users={} try: ureq=urllib.request.Request(sec['instance']+'/api/v1/users',headers=headers) with urllib.request.urlopen(ureq,timeout=12) as r: users={u.get('id'):u.get('displayName') or u.get('username') for u in json.loads(r.read()).get('res',[])} except Exception: pass out=[] for c in chores: if c.get('status') not in (0,None): continue due=c.get('nextDueDate') or '' assignee=users.get(c.get('assignedTo')) or ('Unassigned' if not c.get('assignedTo') else f"User {c.get('assignedTo')}") due_title,state,due_local=due_title_and_state(due) out.append({ 'title':c.get('name','Untitled task'), 'time':due[:16].replace('T',' '), 'body':strip_markup(c.get('description','')), 'assignee':assignee, 'priority':priority_num(c.get('priority')) or 0, 'priority_text':priority_text(c.get('priority')) or '', 'due':due, 'due_title':due_title, 'state':state, 'subtitle':subtitle_text(c), 'chips':visible_label_chips(c), 'source':'Donetick', 'id':c.get('id'), '_sort_overdue':0 if due_local and due_local < datetime.now().astimezone() else (2 if not due_local else 1), '_sort_missing':0 if due_local else 1, '_sort_due':due_local.timestamp() if due_local else 99999999999, '_sort_name':str(c.get('name','')).lower(), }) out.sort(key=lambda x:(x.get('_sort_overdue',9),x.get('_sort_missing',9),x.get('_sort_due',99999999999),x.get('_sort_name',''))) for item in out: for k in ('_sort_overdue','_sort_missing','_sort_due','_sort_name'): item.pop(k,None) return out[:10] except Exception: return [] def priority_text(priority:Any)->str|None: try: p=int(priority) return f'P{p}' if p in (1,2,3,4) else None except Exception: return None def priority_num(priority:Any)->int|None: try: p=int(priority) return p if p in (1,2,3,4) else None except Exception: return None def recurrence_text(task:dict[str,Any])->str|None: ftype=str(task.get('frequencyType','') or '').strip().lower() freq=task.get('frequency') try: freq_num=int(freq) if freq is not None else None except Exception: freq_num=None if not ftype: return None if ftype=='daily': return 'Daily' if freq_num in (None,1) else f'Every {freq_num} days' if ftype=='weekly': return 'Weekly' if freq_num in (None,1) else f'Every {freq_num} weeks' if ftype=='monthly': return 'Monthly' if freq_num in (None,1) else f'Every {freq_num} months' if ftype=='yearly': return 'Yearly' if freq_num in (None,1) else f'Every {freq_num} years' if ftype=='once': return 'One-time' return ftype.capitalize() def due_title_and_state(due_raw:str|None)->tuple[str,str,datetime|None]: dt=parse_dt(due_raw) if not dt: return ('No due date','nodue',None) local=dt.astimezone(LOCAL_TZ) now=datetime.now(LOCAL_TZ) days=(local.date()-now.date()).days time_text=local.strftime('%-I:%M %p') if days < 0: if abs(days)==1: return (f'Overdue โ€ข Yesterday {time_text}','overdue',local) return (f"Overdue โ€ข {local.strftime('%a, %b %-d')} {time_text}",'overdue',local) if days == 0: return (f'Due today โ€ข {time_text}','today',local) if days == 1: return (f'Due tomorrow โ€ข {time_text}','tomorrow',local) if days <= 6: return (f"Due {local.strftime('%a')} โ€ข {time_text}",'upcoming',local) return (f"Due {local.strftime('%b %-d')} โ€ข {time_text}",'upcoming',local) def dedupe_parts(parts:list[str])->list[str]: seen=set(); out=[] for p in parts: if not p: continue key=p.strip().lower() if key in seen: continue seen.add(key); out.append(p) return out def subtitle_text(task:dict[str,Any])->str: parts=[] labels=task.get('labelsV2') or [] label_names=[str(lbl.get('name','')).strip() for lbl in labels if lbl.get('name')] prominent=[x for x in label_names if x.strip().lower() not in LOW_VALUE_LABELS] low=[x for x in label_names if x.strip().lower() in LOW_VALUE_LABELS] parts.extend(prominent[:3]) ptxt=priority_text(task.get('priority')) if ptxt: parts.append(ptxt) rtxt=recurrence_text(task) if rtxt: label_keys={x.strip().lower() for x in label_names} if rtxt.strip().lower() not in label_keys: parts.append(rtxt) if not prominent and low: parts.extend(low[:1]) return ' โ€ข '.join(dedupe_parts(parts)) def visible_label_chips(task:dict[str,Any])->list[dict[str,str]]: chips=[] for lbl in task.get('labelsV2') or []: name=str(lbl.get('name','')).strip() if not name or name.lower() in LOW_VALUE_LABELS: continue color=str(lbl.get('color','')).strip() or '#607d8b' chips.append({'name':name,'color':color}) return chips[:4] def run(cmd:list[str])->str: try: return subprocess.check_output(cmd,stderr=subprocess.DEVNULL,text=True,timeout=8).strip() except Exception: return '' def homelab_pulse(candidates:list[dict[str,Any]])->list[dict[str,str]]: usage=shutil.disk_usage('/home/fizzlepoof/.openclaw/workspace') disk=f'{usage.used/usage.total*100:.0f}% used ({usage.free/1024**3:.1f} GiB free)' dash=run(['systemctl','is-active','doris-dashboard.service']) or 'unknown' cron='present' if 'DORIS DASHBOARD' in run(['bash','-lc','crontab -l 2>/dev/null']) else 'missing' newest=max((parse_time(i.get('collected_at') or i.get('published_at') or '') for i in candidates), default=datetime.now(timezone.utc)) age=(datetime.now(timezone.utc)-newest).total_seconds()/3600 return [ {'label':'Dashboard service','value':dash,'state':'ok' if dash=='active' else 'warn'}, {'label':'Dashboard cron','value':cron,'state':'ok' if cron=='present' else 'warn'}, {'label':'Digest freshness','value':f'{age:.1f}h since newest candidate','state':'ok' if age<8 else 'warn'}, {'label':'Workspace disk','value':disk,'state':'ok' if usage.free/usage.total>.15 else 'warn'}, ] def system_snapshot()->list[dict[str,str]]: uptime_text='unknown' try: uptime_seconds=float(Path('/proc/uptime').read_text().split()[0]) hours=int(uptime_seconds//3600) minutes=int((uptime_seconds%3600)//60) uptime_text=(f'{hours}h {minutes}m' if hours else f'{minutes}m') except Exception: pass load_text='unknown' load_state='ok' try: load1,load5,load15=os.getloadavg() load_text=f'{load1:.2f} / {load5:.2f} / {load15:.2f}' cpu_count=os.cpu_count() or 1 if load1 > cpu_count*1.25: load_state='warn' except Exception: load_state='warn' mem_text='unknown' mem_state='ok' try: parts=Path('/proc/meminfo').read_text().splitlines() raw={line.split(':',1)[0]: int(line.split(':',1)[1].strip().split()[0]) for line in parts if ':' in line} total=raw.get('MemTotal',0) avail=raw.get('MemAvailable',0) used=max(total-avail,0) pct=(used/total*100) if total else 0 mem_text=f'{pct:.0f}% used ({used/1024/1024:.1f} / {total/1024/1024:.1f} GiB)' if pct >= 85: mem_state='warn' except Exception: mem_state='warn' root_usage=shutil.disk_usage('/') root_pct=root_usage.used/root_usage.total*100 if root_usage.total else 0 root_text=f'{root_pct:.0f}% used ({root_usage.free/1024**3:.1f} GiB free)' return [ {'label':'Host','value':run(['hostname']) or 'nomad','state':'ok'}, {'label':'Uptime','value':uptime_text,'state':'ok'}, {'label':'Load avg','value':load_text,'state':load_state}, {'label':'Memory','value':mem_text,'state':mem_state}, {'label':'Root disk','value':root_text,'state':'warn' if root_pct >= 85 else 'ok'}, ] def score_item(item:dict[str,Any],prefs:dict[str,Any])->float: score=float(item.get('priority',1)); text=f"{item.get('title','')} {item.get('summary','')}".lower() if item.get('category') in prefs.get('boost_categories',[]): score+=2 if item.get('category') in prefs.get('quiet_categories',[]): score-=3 for kw in prefs.get('boost_keywords',[]): if kw.lower() in text: score+=1.5 for kw in prefs.get('mute_keywords',[]): if kw.lower() in text: score-=3 if item.get('source')=='CISA Alerts': terms=[t.lower() for t in prefs.get('homelab_relevance_keywords',[])] if terms and not any(t in text for t in terms): score-=8 return score def select_items(items:list[dict[str,Any]],prefs:dict[str,Any],max_items:int=14)->list[dict[str,Any]]: now=datetime.now(timezone.utc); recent=[] for i in items: if (now-parse_time(i.get('published_at',''))).total_seconds()<96*3600: recent.append(i) recent.sort(key=lambda i:(score_item(i,prefs),i.get('published_at','')),reverse=True) out=[]; seen=set(); sources=Counter(); categories=Counter() def consider(limit_per_category:int|None)->None: for i in recent: if len(out)>=max_items: break url=i.get('url',''); src=i.get('source',''); cat=(i.get('category') or 'other').strip() or 'other' if url in seen or sources[src]>=2: continue if limit_per_category is not None and categories[cat] >= limit_per_category: continue out.append(i); seen.add(url); sources[src]+=1; categories[cat]+=1 # First fill favors breadth across categories, then allows a second item per # category, and only then permits a small spillover from the strongest buckets. consider(1) consider(2) consider(3) return out def select_exploration_items(items:list[dict[str,Any]],prefs:dict[str,Any],exclude_urls:set[str]|None=None,max_items:int=4)->list[dict[str,Any]]: now=datetime.now(timezone.utc) exclude_urls=exclude_urls or set() boosted={str(x).strip() for x in prefs.get('boost_categories',[])} quiet={str(x).strip() for x in prefs.get('quiet_categories',[])} preferred=['tech','smart_home','gaming','cooking','security'] recent=[i for i in items if (now-parse_time(i.get('published_at',''))).total_seconds()<96*3600] recent.sort(key=lambda i:(score_item(i,prefs),i.get('published_at','')),reverse=True) out=[]; seen=set(exclude_urls); categories=Counter(); sources=Counter() def consider(allowed_boosted:bool, only_categories:set[str]|None=None, per_category_limit:int=1)->None: for i in recent: if len(out)>=max_items: break url=i.get('url',''); src=i.get('source',''); cat=(i.get('category') or 'other').strip() or 'other' if not url or url in seen or sources[src]>=1 or categories[cat]>=per_category_limit: continue if cat in quiet: continue if only_categories is not None and cat not in only_categories: continue if not allowed_boosted and cat in boosted: continue out.append(i); seen.add(url); categories[cat]+=1; sources[src]+=1 # Keep this lane genuinely exploratory: preferred off-profile categories first, # then broader off-profile picks, allowing a little depth if the pool is narrow. consider(False, set(preferred), 1) consider(False, set(preferred), 3) consider(False, None, 1) consider(False, None, 3) return out def label(cat:str)->str: e,n=LABELS.get(cat,('๐Ÿงฉ',(cat or 'Other').title())); return f'{e} {n}' def slugify(text:str)->str: return re.sub(r'[^a-z0-9]+','-',(text or '').strip().lower()).strip('-') or 'section' def why(i:dict[str,Any])->str: cat=i.get('category'); src=i.get('source','') if cat=='astronomy': return 'Space/astronomy signal from your YouTube + feed tuning.' if cat=='homelab': return 'Practical self-hosting / infrastructure signal.' if cat=='smart_home': return 'Smart-home/Home Assistant adjacent.' if cat=='local': return 'Local or regional relevance.' if cat=='radio': return 'Ham-radio interest signal.' if cat=='security': return 'Kept only because it appears practically relevant.' return f'Interesting signal from {src}.' def card(i:dict[str,Any], compact:bool=False)->str: summary=(i.get('summary') or '').strip()[:120 if compact else 260] when=(i.get('published_at','')[:16].replace('T',' ')) cls='compact-card' if compact else 'card' why_html='' if compact else f"
Why: {esc(why(i))}
" topic=(i.get('category') or 'other').strip() or 'other' source=(i.get('source') or '').strip() domain=(urlparse(i.get('url') or '').netloc or '').lower().removeprefix('www.') feedback_html='' if not compact else ( f"
" f"" f"" f"
" ) return f"""
{esc(label(i.get('category','')))}{esc(i.get('source'))}{esc(when)} UTC

{esc(i.get('title'))}

{esc(summary)}

{feedback_html}{why_html}
""" def section(title:str,sub:str,items:list[dict[str,Any]],empty:str,compact:bool=True)->str: body=''.join(card(i,compact=compact) for i in items) if items else f"

{esc(empty)}

" style='news-list' if compact else 'grid' return collapsible_section(title,sub,f"
{body}
",slugify(title),open_by_default=True) def collapsible_section(title:str,sub:str,inner_html:str,key:str,open_by_default:bool=True,extra_class:str='')->str: open_attr=' open' if open_by_default else '' sub_html=f"

{esc(sub)}

" if sub else '' return f"

{esc(title)}

{sub_html}
{inner_html}
" def inject_dashboard_todos(items:list[dict[str,Any]])->list[dict[str,Any]]: out=list(items or []) title='Paperless notes/tags audit' if not any((x.get('title') or '').strip().lower()==title.lower() for x in out): out.insert(0,{"title":title,"time":"","body":"Review and correct all Paperless notes and tags. Keep this visible until the wider cleanup is actually done.","assignee":"fizzlepoof","priority":2,"priority_text":"P2","due":"","due_title":"Pinned follow-up","state":"today","subtitle":"Dashboard โ€ข P2","chips":[],"source":"Dashboard"}) return out def load_paperless_review()->list[dict[str,Any]]: p=DATA/'paperless_review.json' remote=os.environ.get('PAPERLESS_TRIAGE_REMOTE_QUEUE','truenas_admin@10.5.1.6:/mnt/docker-ssd/docker/appdata/n8n/paperless-triage/paperless_review.json') ssh_key=os.environ.get('PAPERLESS_TRIAGE_SSH_KEY','/home/fizzlepoof/.ssh/id_ed25519_andys') if remote: tmp=DATA/'paperless_review.remote.tmp' try: cmd=['scp','-q','-o','ConnectTimeout=8'] if ssh_key and Path(ssh_key).exists(): cmd.extend(['-i',ssh_key]) cmd.extend([remote,str(tmp)]) res=subprocess.run(cmd,capture_output=True,text=True,timeout=12) if res.returncode==0 and tmp.exists(): tmp.replace(p) elif tmp.exists(): tmp.unlink() except Exception: try: tmp.unlink() except Exception: pass try: return json.loads(p.read_text()) except Exception: return [] def load_paperless_state()->dict[str,Any]: return load_json(PAPERLESS_STATE_PATH,{}) or {} def paperless_focus_summary(items:list[dict[str,Any]]|None=None,state:dict[str,Any]|None=None)->dict[str,Any]: items=items if items is not None else load_paperless_review() state=state if state is not None else load_paperless_state() active=[]; flagged=[] for item in items: status=((state.get(str(item.get('id'))) or {}).get('status') or '').lower() if status in {'dismissed','good_to_go'}: continue if status in {'acknowledged','needs_attention'}: flagged.append(item) else: active.append(item) headline=(flagged[0] if flagged else (active[0] if active else {})).get('title') or 'No documents waiting' return {'active_count':len(active),'flagged_count':len(flagged),'headline':headline} def paperless_review_section()->str: items=load_paperless_review() state=load_paperless_state() if not items: return collapsible_section('๐Ÿ“„ Paperless Review','No documents waiting.','

No documents waiting.

','paperless-review',open_by_default=False,extra_class='panel') urgency_class={'high':'warn','medium':'info','low':'ok','none':'ok'} active=[]; needs_attention=[] for item in items: status=((state.get(str(item.get('id'))) or {}).get('status') or '').lower() if status in {'dismissed','good_to_go'}: continue if status=='acknowledged': status='needs_attention' (needs_attention if status=='needs_attention' else active).append(item) def render_card(item:dict[str,Any], needs_attention_card:bool=False)->str: triage=item.get('triage',{}) or {} current=item.get('current',{}) or {} urgency=triage.get('urgency','none') summary=strip_markup(triage.get('summary','')) reason=strip_markup(triage.get('reason','')) meta_parts=[x for x in [triage.get('document_type',''), current.get('correspondent','')] if x] meta=' โ€ข '.join(str(x) for x in meta_parts) status_class=' needs-attention' if needs_attention_card else '' primary_label='Clear flag' if needs_attention_card else 'Needs attention' primary_action='needs_attention' return f"""
{esc(str(urgency).upper())}{esc(item.get('original_filename') or 'Document #'+str(item.get('id','')))}
{f"View in Paperless" if item.get('archive_url') else ''}

{esc(item.get('title') or triage.get('suggested_title','Untitled'))}

{f"

{esc(summary)}

" if summary else ''} {f"

{esc(meta)}

" if meta else ''} {f"

{esc(reason)}

" if reason else ''}
""" visible_active=active[:3] overflow_active=active[3:] active_cards=''.join(render_card(item,False) for item in visible_active) flagged_cards=''.join(render_card(item,True) for item in needs_attention[:8]) count=len(active) flag_count=len(needs_attention) empty_active="

No active documents waiting.

" body=f"
{active_cards or empty_active}
" if overflow_active: body += f"
Show {esc(str(len(overflow_active)))} more active items
{''.join(render_card(item,False) for item in overflow_active[:8])}
" if flag_count: body += f"
Needs attention {esc(str(flag_count))}
{flagged_cards}

Flagged items stay here so I can revisit them later and, if needed, ask you about the issue.

" if count+flag_count>3: body += f"

Showing {esc(str(min(count,3)))} active up front and keeping the rest tucked away to cut page bloat.

" return collapsible_section(f'๐Ÿ“„ Paperless Review ({count} active)',f'{flag_count} flagged for follow-up ยท review queue written by n8n paperless-intake-triage workflow.',body,'paperless-review',open_by_default=(count+flag_count)>0,extra_class='panel') def list_panel(title:str,items:list[dict[str,Any]],empty:str)->str: if not items: body=f"
  • {esc(empty)}
    • " else: rows=[] for x in items[:8]: kicker=x.get('time') or x.get('title') or x.get('label') headline=x.get('title') if x.get('time') else '' assignee=x.get('assignee') or '' detail=x.get('body') or x.get('summary') or x.get('value') or '' assignee_html=f"{esc(assignee)}" if assignee else '' headline_html=f"
    {esc(headline)}
    " if headline else '' detail_html=f"
    {esc(detail)}
    " if detail else '' rows.append(f"
  • {esc(kicker)}{assignee_html}
    {headline_html}{detail_html}
    • ") body=''.join(rows) return f"

    {esc(title)}

    " def donetick_panel(items:list[dict[str,Any]],empty:str)->str: if not items: return f"

    Donetick

    {esc(empty)}
    " rows=[] for x in items[:6]: chips=''.join(f"{esc(c.get('name',''))}" for c in (x.get('chips') or [])) subtitle_text=(x.get('subtitle') or '').strip() chip_names={str(c.get('name','')).strip().lower() for c in (x.get('chips') or []) if c.get('name')} subtitle_parts=[] for part in [p.strip() for p in subtitle_text.split('โ€ข') if p.strip()]: if part.lower() in chip_names: continue subtitle_parts.append(part) subtitle=' โ€ข '.join(subtitle_parts) state=esc(x.get('state') or 'upcoming') due_style='' if state=='overdue': due_style=' style="color:#ff6b6b;"' elif x.get('priority') in PRIORITY_COLORS: due_style=f' style="color:{esc(PRIORITY_COLORS[x.get("priority")])};"' assignee_html=f"{esc(x.get('assignee') or '')}" if x.get('assignee') and x.get('assignee')!='Unassigned' else '' chips_html=f"
    {chips}
    " if chips else '' subtitle_html=f"
    {esc(subtitle)}
    " if subtitle else '' rows.append(f"
    {esc(x.get('due_title') or x.get('time') or 'No due date')}
    {assignee_html}
    {esc(x.get('title') or 'Untitled task')}
    {chips_html}{subtitle_html}
    ") return f"

    Donetick

    {''.join(rows)}
    " def feedback_bootstrap()->str: return """""" def metric_card(title:str,value:str,sub:str,cls:str='')->str: return f"
    {esc(title)}{esc(value)}{esc(sub)}
    " def forecast_panel(weather:dict[str,Any])->str: if not weather.get('ok'): return "" daily=weather.get('daily',{}) or {} days=daily.get('time') or [] precip=daily.get('precipitation_sum') or [] sunrise=daily.get('sunrise') or [] sunset=daily.get('sunset') or [] cards=[] for idx, day in enumerate(days[:3]): try: label=datetime.fromisoformat(day).strftime('%a') except Exception: label=day rain=precip[idx] if idx < len(precip) else 'โ€”' rise=(sunrise[idx][-5:] if idx < len(sunrise) and sunrise[idx] else 'โ€”') set_=(sunset[idx][-5:] if idx < len(sunset) and sunset[idx] else 'โ€”') cards.append(f"""
    {esc(label)} {esc(str(rain))} mm rain {esc(rise)} sunrise ยท {esc(set_)} sunset
    """) if not cards: return "" return collapsible_section('3-day forecast','Rain, sunrise, and sunset over the next three days.',f"
    {''.join(cards)}
    ",'forecast',open_by_default=False,extra_class='panel forecast-panel') def fetch_hermes_usage(days:int=30)->dict[str,Any]: config=read_hermes_model_config() stats={ 'ok':False, 'period_days':days, 'default_model':config.get('default_model','unknown'), 'provider':config.get('provider','unknown'), 'sessions':0, 'messages':0, 'tool_calls':0, 'user_messages':0, 'assistant_messages':0, 'total_tokens':0, 'input_tokens':0, 'output_tokens':0, 'cache_read_tokens':0, 'cache_write_tokens':0, 'reasoning_tokens':0, 'models':[], 'platforms':[], 'top_tools':[], 'recent_sessions':[], 'generated_at':datetime.now(LOCAL_TZ).isoformat(), } if not HERMES_STATE_DB.exists(): stats['error']='Hermes state DB not found' return stats cutoff=time.time()-(days*86400) try: conn=sqlite3.connect(f"file:{HERMES_STATE_DB}?mode=ro", uri=True) conn.row_factory=sqlite3.Row cur=conn.cursor() overview=cur.execute(""" SELECT count(*) AS sessions, sum(coalesce(input_tokens,0)) AS input_tokens, sum(coalesce(output_tokens,0)) AS output_tokens, sum(coalesce(cache_read_tokens,0)) AS cache_read_tokens, sum(coalesce(cache_write_tokens,0)) AS cache_write_tokens, sum(coalesce(reasoning_tokens,0)) AS reasoning_tokens, sum(coalesce(api_call_count,0)) AS api_calls FROM sessions WHERE started_at >= ? """,(cutoff,)).fetchone() stats.update({ 'sessions':int((overview['sessions'] or 0) if overview else 0), 'input_tokens':int((overview['input_tokens'] or 0) if overview else 0), 'output_tokens':int((overview['output_tokens'] or 0) if overview else 0), 'cache_read_tokens':int((overview['cache_read_tokens'] or 0) if overview else 0), 'cache_write_tokens':int((overview['cache_write_tokens'] or 0) if overview else 0), 'reasoning_tokens':int((overview['reasoning_tokens'] or 0) if overview else 0), }) stats['total_tokens']=stats['input_tokens']+stats['output_tokens']+stats['cache_read_tokens']+stats['cache_write_tokens']+stats['reasoning_tokens'] row=cur.execute("SELECT count(*) AS c FROM messages WHERE role != 'session_meta' AND timestamp >= ?",(cutoff,)).fetchone() stats['messages']=int((row['c'] or 0) if row else 0) row=cur.execute("SELECT count(*) AS c FROM messages WHERE role='tool' AND timestamp >= ?",(cutoff,)).fetchone() stats['tool_calls']=int((row['c'] or 0) if row else 0) row=cur.execute("SELECT count(*) AS c FROM messages WHERE role='user' AND timestamp >= ?",(cutoff,)).fetchone() stats['user_messages']=int((row['c'] or 0) if row else 0) row=cur.execute("SELECT count(*) AS c FROM messages WHERE role='assistant' AND timestamp >= ?",(cutoff,)).fetchone() stats['assistant_messages']=int((row['c'] or 0) if row else 0) stats['models']=[dict(r) for r in cur.execute(""" SELECT model, count(*) AS sessions, sum(coalesce(input_tokens,0)+coalesce(output_tokens,0)+coalesce(cache_read_tokens,0)+coalesce(cache_write_tokens,0)+coalesce(reasoning_tokens,0)) AS total_tokens FROM sessions WHERE started_at >= ? GROUP BY model ORDER BY total_tokens DESC, sessions DESC LIMIT 5 """,(cutoff,))] stats['platforms']=[dict(r) for r in cur.execute(""" SELECT source, count(*) AS sessions, sum(coalesce(input_tokens,0)+coalesce(output_tokens,0)+coalesce(cache_read_tokens,0)+coalesce(cache_write_tokens,0)+coalesce(reasoning_tokens,0)) AS total_tokens FROM sessions WHERE started_at >= ? GROUP BY source ORDER BY total_tokens DESC, sessions DESC LIMIT 5 """,(cutoff,))] stats['top_tools']=[dict(r) for r in cur.execute(""" SELECT tool_name, count(*) AS calls FROM messages WHERE role='tool' AND tool_name IS NOT NULL AND tool_name != '' AND timestamp >= ? GROUP BY tool_name ORDER BY calls DESC, tool_name ASC LIMIT 8 """,(cutoff,))] stats['recent_sessions']=[dict(r) for r in cur.execute(""" SELECT source, model, started_at, (coalesce(input_tokens,0)+coalesce(output_tokens,0)+coalesce(cache_read_tokens,0)+coalesce(cache_write_tokens,0)+coalesce(reasoning_tokens,0)) AS total_tokens, coalesce(api_call_count,0) AS api_calls FROM sessions WHERE started_at >= ? ORDER BY started_at DESC LIMIT 5 """,(cutoff,))] for item in stats['recent_sessions']: started=item.get('started_at') try: item['started_local']=datetime.fromtimestamp(float(started), LOCAL_TZ).strftime('%a %-I:%M %p') except Exception: item['started_local']='recent' stats['ok']=True conn.close() return stats except Exception: stats['error']='Hermes telemetry is temporarily unavailable.' return stats def focus_card(title:str,kicker:str,headline:str,body:str,tone:str='')->str: return f"
    {esc(kicker)}

    {esc(title)}

    {esc(headline)}

    {esc(body)}

    " def build_focus_cards(local_items:list[dict[str,Any]], todos:list[dict[str,Any]], pulse:list[dict[str,str]], paperless:dict[str,Any])->str: urgent_todos=[item for item in todos if (item.get('state') or '') in {'overdue','today'} and (item.get('source') or '') != 'Dashboard'] warn_items=[item for item in pulse if item.get('state')=='warn'] local_story=local_items[0] if local_items else None if urgent_todos: chores_headline=f"{len(urgent_todos)} urgent / due today" first_urgent=urgent_todos[0] chores_body=(first_urgent.get('title') or 'You have work waiting') + (f" โ€ข {first_urgent.get('due_title')}" if first_urgent.get('due_title') else '') else: chores_headline='No urgent chores' chores_body='Nothing due soon enough to nag you here.' cards=[ focus_card( 'Local watch', 'Useful nearby', (local_story.get('title') if local_story else 'Local lane quiet'), (f"{local_story.get('source','Local feed')} โ€ข {label(local_story.get('category','local'))}" if local_story else 'Nothing local survived filtering hard enough to earn the homepage.'), 'local' if local_story else '' ), focus_card( 'Chores with heat', 'Donetick', chores_headline, chores_body, 'warn' if urgent_todos else 'ok' ), focus_card( 'Paperless follow-up', 'Queue check', (f"{paperless.get('active_count',0)} active ยท {paperless.get('flagged_count',0)} flagged" if paperless.get('active_count',0) or paperless.get('flagged_count',0) else 'No documents waiting'), str(paperless.get('headline') or 'No documents waiting'), 'warn' if paperless.get('flagged_count',0) else ('ok' if not paperless.get('active_count',0) else '') ), focus_card( 'Homelab pulse', 'NOMAD checks', ('Stable' if not warn_items else f"{len(warn_items)} warning{'s' if len(warn_items)!=1 else ''}"), ('; '.join(f"{item.get('label')}: {item.get('value')}" for item in warn_items[:2]) if warn_items else 'Dashboard service, cron, digest freshness, and disk all look fine.'), 'warn' if warn_items else 'ok' ), ] return "
    " + ''.join(cards) + "
    " def build_hero_summary(local_items:list[dict[str,Any]], todos:list[dict[str,Any]], pulse:list[dict[str,str]], paperless:dict[str,Any], hermes:dict[str,Any])->str: urgent_todos=len([item for item in todos if (item.get('state') or '') in {'overdue','today'} and (item.get('source') or '') != 'Dashboard']) warn_count=len([item for item in pulse if item.get('state')=='warn']) local_status='local lane has signal' if local_items else 'local lane is quiet' paperless_text=f"{paperless.get('active_count',0)} Paperless item{'s' if paperless.get('active_count',0)!=1 else ''}" hermes_model=(hermes.get('models') or [{}])[0].get('model') or hermes.get('default_model') or 'your main model' return f"{urgent_todos} urgent chore{'s' if urgent_todos!=1 else ''}, {paperless_text}, {warn_count} homelab warning{'s' if warn_count!=1 else ''}, and Hermes has mostly been running on {hermes_model} while the {local_status}." def list_rows(rows:list[str])->str: return ''.join(f"
  • {row}
    • " for row in rows) if rows else "
  • Nothing to show.
    • " def build_system_section(snapshot:list[dict[str,str]])->str: metrics=''.join(metric_card(item.get('label','Metric'), item.get('value','โ€”'), 'live host state', item.get('state','')) for item in snapshot) subtitle='Live NOMAD host state: uptime, load, memory pressure, and root disk usage.' return collapsible_section('System snapshot',subtitle,f"
    {metrics}
    ",'system-snapshot',open_by_default=True) def build_hermes_section(hermes:dict[str,Any])->str: if not hermes.get('ok'): body="

    Hermes telemetry unavailable

    Local Hermes usage stats could not be read right now.

    " return collapsible_section('Hermes usage',f"Could not read the last {hermes.get('period_days',30)} days of local Hermes state.",body,'hermes-usage',open_by_default=True) metrics=''.join([ metric_card('Default model',str(hermes.get('default_model','unknown')),str(hermes.get('provider','unknown')),'space'), metric_card(f"{hermes.get('period_days',30)}-day tokens",compact_number(hermes.get('total_tokens',0)),'all session rows','space'), metric_card('Sessions / messages',f"{compact_number(hermes.get('sessions',0))} / {compact_number(hermes.get('messages',0))}",'state.db window','space'), metric_card('Tool calls',compact_number(hermes.get('tool_calls',0)),f"{compact_number(hermes.get('user_messages',0))} user ยท {compact_number(hermes.get('assistant_messages',0))} assistant",'space'), ]) model_rows=[] for item in hermes.get('models',[])[:5]: model_rows.append(f"{esc(item.get('model') or 'unknown')} ยท {compact_number(item.get('sessions',0))} sessions ยท {compact_number(item.get('total_tokens',0))} tokens") platform_rows=[] for item in hermes.get('platforms',[])[:5]: platform_rows.append(f"{esc((item.get('source') or 'unknown').title())} ยท {compact_number(item.get('sessions',0))} sessions ยท {compact_number(item.get('total_tokens',0))} tokens") tool_rows=[] for item in hermes.get('top_tools',[])[:8]: tool_rows.append(f"{esc(item.get('tool_name') or 'tool')} ยท {compact_number(item.get('calls',0))} calls") recent_rows=[] for item in hermes.get('recent_sessions',[])[:5]: source=(item.get('source') or 'session').title() model=item.get('model') or 'unknown' recent_rows.append(f"{esc(source)} ยท {esc(item.get('started_local') or 'recent')} ยท {esc(model)} ยท {compact_number(item.get('total_tokens',0))} tokens") detail_html=f"""

    Model mix

      {list_rows(model_rows)}

    Platform split

      {list_rows(platform_rows)}

    Top tools

      {list_rows(tool_rows)}

    Recent sessions

      {list_rows(recent_rows)}
    """ body=f"
    {metrics}
    {detail_html}" subtitle=f"Live read from ~/.hermes/state.db over the last {hermes.get('period_days',30)} days." return collapsible_section('Hermes usage',subtitle,body,'hermes-usage',open_by_default=True) def render()->str: candidates=load_json(DIGEST_ROOT/'data/candidates.json',[]); prefs=load_json(DIGEST_ROOT/'data/preferences.json',{}) donetick_todos=fetch_donetick_todos(); todos=inject_dashboard_todos(donetick_todos or load_json(DATA/'todos.json',[])); calendar=load_json(DATA/'calendar.json',[]); notes=load_json(DATA/'notes.json',[]) selected=select_items(candidates,prefs); top_briefing=selected[:4]; top_urls={i.get('url','') for i in top_briefing if i.get('url')}; groups={'space':[],'homelab':[],'local':[],'signals':[]} exploration=select_exploration_items(candidates,prefs,{i.get('url','') for i in selected if i.get('url')},4) for i in selected: if i.get('url','') in top_urls: continue cat=i.get('category') if cat=='astronomy': groups['space'].append(i) elif cat in {'homelab','smart_home','radio'}: groups['homelab'].append(i) elif cat=='local': groups['local'].append(i) else: groups['signals'].append(i) weather=fetch_weather(); moon=moon_phase(datetime.now(timezone.utc)); pulse=homelab_pulse(candidates); hermes=fetch_hermes_usage(30); system=system_snapshot() paperless=paperless_focus_summary() now=datetime.now(LOCAL_TZ); counts=Counter(i.get('category','other') for i in candidates) if weather['ok']: cur=weather['current']; daily=weather.get('daily',{}) weather_metrics=''.join([ metric_card('Clarksville',c_to_f(cur.get('temperature_2m')),weather.get('condition',''), 'weather'), metric_card('Feels / humidity',f"{c_to_f(cur.get('apparent_temperature'))} / {cur.get('relative_humidity_2m','โ€”')}%",'right now','weather'), metric_card('Wind',kmh_to_mph(cur.get('wind_speed_10m')),'Open-Meteo','weather'), metric_card('Rain chance',f"{weather.get('max_precip',0)}%",'next 24h','weather'), metric_card('Sunrise / sunset',f"{(daily.get('sunrise') or ['โ€”'])[0][-5:]} / {(daily.get('sunset') or ['โ€”'])[0][-5:]}",'local time','weather')]) else: weather_metrics=metric_card('Weather','Unavailable',weather.get('error','API issue'),'warn') moon_metrics=''.join([ metric_card('Moon',f"{moon['emoji']} {moon['name']}",f"{moon['illumination']} lit, age {moon['age']}",'space'), metric_card('Sky note','Space feeds live','Moon phase is computed locally','space')]) pulse_metrics=''.join(metric_card(x['label'],x['value'],x['state'],x['state']) for x in pulse) forecast_html=forecast_panel(weather) hero_summary=build_hero_summary(groups['local'],todos,pulse,paperless,hermes) focus_html=build_focus_cards(groups['local'],todos,pulse,paperless) hermes_html=build_hermes_section(hermes) system_html=build_system_section(system) hero_tokens=compact_number(hermes.get('total_tokens',0)) if hermes.get('ok') else 'โ€”' primary_html=''.join([ section('Top briefing','Compact scan of what is worth your time.',top_briefing,'Nothing worth bothering you with.',compact=True), section('Homelab / Smart Home / Radio','Infrastructure, self-hosting, smart home, and ham radio.',groups['homelab'][:4],'No homelab signal right now.',compact=True), section('Sky / Space','SpaceX, NASA, planetary science, and astronomy.',groups['space'][:4],'No sky signal right now.',compact=True), section('Local','Clarksville / Middle Tennessee items that seem useful.',groups['local'][:3],'No local signal right now.',compact=True), section('New to You','Intentional oddballs outside your usual top lanes, so you can see if any new categories deserve a slot.',exploration[:4],'No worthwhile experimental picks right now.',compact=True), section('Other signals','Everything else that survived filtering.',groups['signals'][:3],'No extra signals.',compact=True), ]) secondary_html=''.join([ collapsible_section('Focus now','What actually deserves attention before you go wandering through feeds.',focus_html,'focus-now',open_by_default=True), system_html, hermes_html, collapsible_section('Agenda','Calendar, Donetick, and pinned notes.',f"
    {list_panel('Calendar',calendar,'Calendar integration not wired yet.')}{donetick_panel(todos,'No open Donetick tasks found; JSON placeholder active.')}{list_panel('Notes',notes,'No notes.')}
    ",'today-overview',open_by_default=False), collapsible_section('Environment','Weather, sky, and outside-the-window context.',f"
    {weather_metrics}{moon_metrics}
    ",'at-a-glance',open_by_default=False), forecast_html, paperless_review_section(), collapsible_section('Homelab pulse','Safe local checks on NOMAD only.',f"
    {pulse_metrics}
    ",'homelab-pulse',open_by_default=False), ]) return f"""Doris Dashboard

    PC Doris Thatcher

    Doris Dashboard

    {esc(hero_summary)}

    Hidden topics: 0
    {esc(now.strftime('%a %b %-d, %-I:%M %p'))}Central time ยท refreshes every 30m
    {len(candidates)}Candidates
    {len(set(i.get('source','') for i in candidates))}Sources
    {esc(hero_tokens)}30-day Hermes tokens
    {primary_html}
    {feedback_bootstrap()}""" def main()->int: PUBLIC.mkdir(parents=True,exist_ok=True); LOGS.mkdir(parents=True,exist_ok=True); DATA.mkdir(parents=True,exist_ok=True) candidates=load_json(DIGEST_ROOT/'data/candidates.json',[]); prefs=load_json(DIGEST_ROOT/'data/preferences.json',{}) weather=fetch_weather(); moon=moon_phase(datetime.now(timezone.utc)); pulse=homelab_pulse(candidates); articles=select_items(candidates,prefs,20); todos=inject_dashboard_todos(fetch_donetick_todos()); hermes=fetch_hermes_usage(30) generated={'generated_at':datetime.now(timezone.utc).isoformat(),'generated_at_local':datetime.now(LOCAL_TZ).isoformat(),'timezone':'America/Chicago','candidate_count':len(candidates),'source_count':len(set(i.get('source','') for i in candidates))} (DATA/'dashboard.json').write_text(json.dumps(generated,indent=2)+'\n') (DATA/'weather.json').write_text(json.dumps(weather,indent=2)+'\n') (DATA/'sky.json').write_text(json.dumps({'moon':moon},indent=2)+'\n') (DATA/'homelab.json').write_text(json.dumps(pulse,indent=2)+'\n') (DATA/'articles.json').write_text(json.dumps(articles,indent=2)+'\n') (DATA/'hermes_usage.json').write_text(json.dumps(hermes,indent=2)+'\n') if todos: (DATA/'todos.json').write_text(json.dumps(todos,indent=2)+'\n') (PUBLIC/'index.html').write_text(render(),encoding='utf-8') return 0 if __name__=='__main__': raise SystemExit(main())