#!/usr/bin/env python3 from __future__ import annotations import json import os import hashlib import subprocess import urllib.error import urllib.parse import urllib.request from functools import partial from http import HTTPStatus from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo ROOT = Path(__file__).resolve().parents[1] PUBLIC = ROOT / 'public' DIGEST_ROOT = Path(os.environ.get('DORIS_DIGEST_DIR', '/home/fizzlepoof/.openclaw/workspace/doris-digest')) DIGEST_FEEDBACK = DIGEST_ROOT / 'data' / 'feedback.json' PAPERLESS_STATE = ROOT / 'data' / 'paperless_review_state.json' LEAD_WARRANT_STATE = ROOT / 'data' / 'lead_warrant_state.json' DONETICK_SECRET = Path('/home/fizzlepoof/.openclaw/workspace/.secrets/donetick_secrets.json.enc') DONETICK_KEY = Path('/home/fizzlepoof/.openclaw/workspace/.secrets/encryption.key') GENERATOR = ROOT / 'bin' / 'generate_dashboard.py' HOST = os.environ.get('HOST', '0.0.0.0') PORT = int(os.environ.get('PORT', '8787')) LOCAL_TZ = ZoneInfo('America/Chicago') def load_json(path: Path, default): try: return json.loads(path.read_text()) if path.exists() else default except Exception: return default def save_json(path: Path, value) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2) + '\n') def decrypt_donetick() -> dict | None: if not DONETICK_SECRET.exists() or not DONETICK_KEY.exists(): return None try: res = subprocess.run([ 'openssl','enc','-aes-256-cbc','-d','-pbkdf2', '-in',str(DONETICK_SECRET), '-k',DONETICK_KEY.read_text().strip(), ], capture_output=True, text=True, timeout=8, check=True) return json.loads(res.stdout).get('donetick') except Exception: return None def donetick_headers() -> tuple[dict[str, str] | None, dict | None]: sec = decrypt_donetick() if not sec: return None, None 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'] return ({ 'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'Content-Type': 'application/json', }, sec) except Exception: return None, sec def refresh_dashboard() -> bool: try: subprocess.run(['python3', str(GENERATOR)], cwd=str(ROOT), timeout=60, check=True) return True except Exception: return False def load_lead_warrant_state() -> dict: state = load_json(LEAD_WARRANT_STATE, {}) if not isinstance(state, dict): state = {} acknowledged = state.get('acknowledged', []) ignored_sources = state.get('ignored_sources', []) ignored_topics = state.get('ignored_topics', []) if not isinstance(acknowledged, list): acknowledged = [] if not isinstance(ignored_sources, list): ignored_sources = [] if not isinstance(ignored_topics, list): ignored_topics = [] return { 'acknowledged': [str(x).strip() for x in acknowledged if str(x).strip()][:250], 'ignored_sources': [str(x).strip().lower() for x in ignored_sources if str(x).strip()][:100], 'ignored_topics': [str(x).strip().lower() for x in ignored_topics if str(x).strip()][:100], } def parse_dt(value: str | None): if not value: return None try: return datetime.fromisoformat(value.replace('Z', '+00:00')) except Exception: return None def due_state(due_raw: str | None) -> str: dt = parse_dt(due_raw) if not dt: return 'nodue' local = dt.astimezone(LOCAL_TZ) now = datetime.now(LOCAL_TZ) days = (local.date() - now.date()).days if days < 0: return 'overdue' if days == 0: return 'today' if days == 1: return 'tomorrow' return 'upcoming' def fetch_donetick_focus_signature() -> tuple[str, int]: headers, sec = donetick_headers() if not headers or not sec: return 'unavailable', 0 try: req = urllib.request.Request(f"{sec['instance'].rstrip('/')}/api/v1/chores", headers=headers) with urllib.request.urlopen(req, timeout=12) as r: chores = json.loads(r.read() or b'{}').get('res', []) urgent = [] for chore in chores: if chore.get('status') not in (0, None): continue state = due_state(chore.get('nextDueDate')) if state not in {'overdue', 'today'}: continue urgent.append({ 'id': str(chore.get('id') or ''), 'name': str(chore.get('name') or ''), 'due': str(chore.get('nextDueDate') or ''), 'state': state, }) urgent.sort(key=lambda item: (item['due'] or '9999', item['name'].lower(), item['id'])) payload = json.dumps(urgent, separators=(',', ':'), sort_keys=True) return hashlib.sha1(payload.encode('utf-8')).hexdigest()[:16], len(urgent) except Exception: return 'error', 0 class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, directory=None, **kwargs): super().__init__(*args, directory=str(PUBLIC), **kwargs) def do_GET(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == '/api/donetick-focus-state': return self._donetick_focus_state(parsed) self.path = parsed.path return super().do_GET() def do_POST(self): length = int(self.headers.get('Content-Length', '0') or '0') raw = self.rfile.read(length) if length else b'{}' try: payload = json.loads(raw.decode('utf-8') or '{}') except Exception: return self._json(HTTPStatus.BAD_REQUEST, {'ok': False, 'error': 'invalid_json'}) if self.path == '/api/topic-feedback': return self._topic_feedback(payload) if self.path == '/api/topic-feedback/reset': return self._topic_reset() if self.path == '/api/paperless-action': return self._paperless_action(payload) if self.path == '/api/donetick-complete': return self._donetick_complete(payload) if self.path == '/api/lead-warrant-ack': return self._lead_warrant_ack(payload) if self.path == '/api/lead-warrant-ignore': return self._lead_warrant_ignore(payload) return self._json(HTTPStatus.NOT_FOUND, {'ok': False, 'error': 'not_found'}) def _topic_feedback(self, payload: dict): term = str(payload.get('term', '')).strip() vote = str(payload.get('vote', '')).strip().lower() if not term or vote not in {'up', 'down'}: return self._json(HTTPStatus.BAD_REQUEST, {'ok': False, 'error': 'bad_payload'}) items = load_json(DIGEST_FEEDBACK, []) items = [x for x in items if not (str(x.get('origin', '')) == 'dashboard' and str(x.get('term', '')).strip().lower() == term.lower())] items.append({ 'term': term, 'vote': vote, 'note': f"Dashboard {'hide' if vote == 'down' else 'prefer'} topic: {term}", 'origin': 'dashboard', 'source': str(payload.get('source', '')).strip(), 'domain': str(payload.get('domain', '')).strip(), }) save_json(DIGEST_FEEDBACK, items) return self._json(HTTPStatus.OK, {'ok': True, 'count': len(items)}) def _topic_reset(self): items = load_json(DIGEST_FEEDBACK, []) items = [x for x in items if str(x.get('origin', '')) != 'dashboard'] save_json(DIGEST_FEEDBACK, items) return self._json(HTTPStatus.OK, {'ok': True, 'count': len(items)}) def _paperless_action(self, payload: dict): item_id = str(payload.get('id', '')).strip() action = str(payload.get('action', '')).strip().lower() if not item_id or action not in {'needs_attention', 'good_to_go'}: return self._json(HTTPStatus.BAD_REQUEST, {'ok': False, 'error': 'bad_payload'}) state = load_json(PAPERLESS_STATE, {}) current = state.get(item_id, {}) if isinstance(state, dict) else {} current_status = str(current.get('status') or '').lower() if action == 'needs_attention': new_status = 'active' if current_status == 'needs_attention' else 'needs_attention' else: new_status = 'good_to_go' state[item_id] = {'status': new_status} save_json(PAPERLESS_STATE, state) return self._json(HTTPStatus.OK, {'ok': True, 'id': item_id, 'status': new_status}) def _donetick_complete(self, payload: dict): item_id = str(payload.get('id') or payload.get('choreId') or '').strip() if not item_id: return self._json(HTTPStatus.BAD_REQUEST, {'ok': False, 'error': 'bad_payload', 'detail': 'expected id or choreId'}) headers, sec = donetick_headers() if not headers or not sec: return self._json(HTTPStatus.BAD_GATEWAY, {'ok': False, 'error': 'donetick_auth_failed'}) try: req = urllib.request.Request( f"{sec['instance'].rstrip('/')}/api/v1/chores/{item_id}/do", data=b'{}', headers=headers, method='POST', ) with urllib.request.urlopen(req, timeout=15) as r: response = json.loads(r.read() or b'{}') if (r.headers.get('Content-Type') or '').startswith('application/json') else {'status': r.status} refreshed = refresh_dashboard() return self._json(HTTPStatus.OK, {'ok': True, 'id': item_id, 'dashboard_refreshed': refreshed, 'response': response}) except urllib.error.HTTPError as e: detail = e.read().decode('utf-8', 'ignore')[:500] return self._json(HTTPStatus.BAD_GATEWAY, {'ok': False, 'error': 'donetick_complete_failed', 'status': e.code, 'detail': detail}) except Exception as e: return self._json(HTTPStatus.BAD_GATEWAY, {'ok': False, 'error': 'donetick_complete_failed', 'detail': str(e)}) def _lead_warrant_ack(self, payload: dict): key = str(payload.get('key') or '').strip() if not key: return self._json(HTTPStatus.BAD_REQUEST, {'ok': False, 'error': 'bad_payload', 'detail': 'expected key'}) state = load_lead_warrant_state() acknowledged = [x for x in state.get('acknowledged', []) if x != key] acknowledged.append(key) save_json(LEAD_WARRANT_STATE, { 'acknowledged': acknowledged[-250:], 'ignored_sources': state.get('ignored_sources', []), 'ignored_topics': state.get('ignored_topics', []), }) refreshed = refresh_dashboard() return self._json(HTTPStatus.OK, {'ok': True, 'key': key, 'dashboard_refreshed': refreshed}) def _lead_warrant_ignore(self, payload: dict): key = str(payload.get('key') or '').strip() mode = str(payload.get('mode') or '').strip().lower() topic = str(payload.get('topic') or '').strip().lower() source = str(payload.get('source') or '').strip().lower() if not key or mode not in {'topic', 'source'}: return self._json(HTTPStatus.BAD_REQUEST, {'ok': False, 'error': 'bad_payload', 'detail': 'expected key and mode'}) state = load_lead_warrant_state() acknowledged = [x for x in state.get('acknowledged', []) if x != key] acknowledged.append(key) ignored_sources = [x for x in state.get('ignored_sources', []) if x] ignored_topics = [x for x in state.get('ignored_topics', []) if x] if mode == 'topic': if not topic: return self._json(HTTPStatus.BAD_REQUEST, {'ok': False, 'error': 'bad_payload', 'detail': 'expected topic'}) ignored_topics = [x for x in ignored_topics if x != topic] ignored_topics.append(topic) else: if not source: return self._json(HTTPStatus.BAD_REQUEST, {'ok': False, 'error': 'bad_payload', 'detail': 'expected source'}) ignored_sources = [x for x in ignored_sources if x != source] ignored_sources.append(source) save_json(LEAD_WARRANT_STATE, { 'acknowledged': acknowledged[-250:], 'ignored_sources': ignored_sources[-100:], 'ignored_topics': ignored_topics[-100:], }) refreshed = refresh_dashboard() return self._json(HTTPStatus.OK, { 'ok': True, 'key': key, 'mode': mode, 'ignored_source': source if mode == 'source' else '', 'ignored_topic': topic if mode == 'topic' else '', 'dashboard_refreshed': refreshed, }) def _donetick_focus_state(self, parsed): current = urllib.parse.parse_qs(parsed.query).get('current', [''])[0] signature, urgent_count = fetch_donetick_focus_signature() changed = bool(current and signature not in {'error', 'unavailable'} and current != signature) refreshed = refresh_dashboard() if changed else False return self._json(HTTPStatus.OK, { 'ok': True, 'signature': signature, 'urgent_count': urgent_count, 'changed': changed, 'dashboard_refreshed': refreshed, }) def _json(self, status: HTTPStatus, payload: dict): body = json.dumps(payload).encode('utf-8') self.send_response(status) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, fmt, *args): print('%s - - [%s] %s' % (self.address_string(), self.log_date_time_string(), fmt % args)) if __name__ == '__main__': handler = partial(Handler, directory=str(PUBLIC)) ThreadingHTTPServer((HOST, PORT), handler).serve_forever()