Upload to Roam via the backend API instead of a headless browser

conftest.py

@@ -0,0 +1,17 @@
+"""Shared pytest configuration.
+
+secure_settings.py holds per-device secrets and is not checked in. Modules
+that import gonotego.settings.settings need it to exist, so when it is absent
+the tests get an empty stand-in module instead.
+"""
+import importlib
+import sys
+import types
+
+try:
+  importlib.import_module('gonotego.settings.secure_settings')
+except ImportError:
+  stub = types.ModuleType('gonotego.settings.secure_settings')
+  sys.modules['gonotego.settings.secure_settings'] = stub
+  import gonotego.settings
+  gonotego.settings.secure_settings = stub

gonotego/settings-server/src/App.tsx

@@ -32,6 +32,7 @@ const SettingsUI = () => {
     NOTE_TAKING_SYSTEM: '',
     BLOB_STORAGE_SYSTEM: '',
     ROAM_GRAPH: '',
+    ROAM_API_TOKEN: '',
     ROAM_USER: '',
     ROAM_PASSWORD: '',
     REMNOTE_USER_ID: '',
@@ -701,8 +702,9 @@ const SettingsUI = () => {
       {/* Conditional Settings based on Note Taking System */}
       {renderSettingGroup('Roam Research', 'Roam Research integration settings', [
         { key: 'ROAM_GRAPH', label: 'Graph Name' },
-        { key: 'ROAM_USER', label: 'Username' },
-        { key: 'ROAM_PASSWORD', label: 'Password', type: 'password' },
+        { key: 'ROAM_API_TOKEN', label: 'API Token (recommended; Roam Settings > Graph > API tokens)', type: 'password' },
+        { key: 'ROAM_USER', label: 'Username (browser fallback)' },
+        { key: 'ROAM_PASSWORD', label: 'Password (browser fallback)', type: 'password' },
       ], shouldShowSection('roam'))}
 
       {renderSettingGroup('RemNote', 'RemNote integration settings', [

gonotego/settings/secure_settings_template.py

@@ -3,6 +3,7 @@
 BLOB_STORAGE_SYSTEM = '<BLOB_STORAGE_SYSTEM>'
 
 ROAM_GRAPH = '<ROAM_GRAPH>'
+ROAM_API_TOKEN = '<ROAM_API_TOKEN>'
 ROAM_USER = '<ROAM_USER>'
 ROAM_PASSWORD = '<ROAM_PASSWORD>'
 

gonotego/settings/server.py

@@ -22,6 +22,7 @@
 
 # Sensitive keys that should be masked
 SENSITIVE_KEYS = [
+    'ROAM_API_TOKEN',
     'ROAM_PASSWORD',
     'REMNOTE_API_KEY',
     'IDEAFLOW_PASSWORD',

gonotego/settings/settings.py

@@ -17,12 +17,21 @@ def get_redis_key(key):
   return f'{SETTINGS_KEY}:{key}'
 
 
-def get(key):
+_MISSING = object()
+
+
+def get(key, default=_MISSING):
+  """Returns a setting, preferring the value set on the device over secure_settings.
+
+  Raises AttributeError for an unknown key unless a default is given.
+  """
   r = interprocess.get_redis_client()
   value_bytes = r.get(get_redis_key(key))
   if value_bytes is None:
     # If the setting isn't set in redis, fall back to the value from secure_settings.
-    return getattr(secure_settings, key)
+    if default is _MISSING:
+      return getattr(secure_settings, key)
+    return getattr(secure_settings, key, default)
   value_repr = value_bytes.decode('utf-8')
   value = ast.literal_eval(value_repr)
   return value

gonotego/uploader/roam/roam_api_uploader.py

@@ -0,0 +1,118 @@
+"""Uploads notes to Roam Research through the Roam backend API.
+
+Produces the same structure as the browser-based uploader (roam_uploader.py):
+
+  Daily Notes page for the note's date
+    - [[Go Note Go Notes]]:
+      - 06:30 AM                (one block per writing session)
+        - a note
+        - another note
+          - an indented note
+
+Unlike the browser uploader this needs no Firefox and no Roam password, and it
+is unaffected by the bot-verification challenge in front of Roam's web app.
+It is used whenever ROAM_API_TOKEN is set.
+"""
+from datetime import datetime
+import os
+
+from gonotego.common import events
+from gonotego.settings import settings
+from gonotego.uploader.blob import blob_uploader
+from gonotego.uploader.roam import roam_backend_api
+
+SECTION_TITLE = '[[Go Note Go Notes]]:'
+UNVERIFIED_TAG = '#[[unverified transcription]]'
+
+
+def note_datetime(note_event):
+  """The note's effective time (clock + alleged-time offset), or now if unknown."""
+  timestamp = note_event.effective_timestamp if note_event is not None else None
+  return datetime.fromtimestamp(timestamp) if timestamp else datetime.now()
+
+
+class Uploader:
+
+  def __init__(self, client=None):
+    self._client = client
+    self.session_uid = None
+    self.last_note_uid = None
+    self.stack = []
+
+  def get_client(self):
+    if self._client is None:
+      self._client = roam_backend_api.RoamBackendClient(
+          token=settings.get('ROAM_API_TOKEN'),
+          graph=settings.get('ROAM_GRAPH'))
+    return self._client
+
+  def new_session(self, note_event):
+    """Creates the block that this writing session's notes are nested under."""
+    client = self.get_client()
+    dt = note_datetime(note_event)
+    title = roam_backend_api.daily_note_title(dt)
+    page_uid = client.get_or_create_page(title, uid=roam_backend_api.daily_note_uid(dt))
+    section_uid = client.get_or_create_block_on_page(page_uid, SECTION_TITLE)
+    self.session_uid = client.create_block(section_uid, dt.strftime('%H:%M %p'))
+    print(f'Started session (({self.session_uid})) on "{title}"')
+
+  def upload(self, note_events):
+    """Uploads the note events. Returns True on success, False if a request failed."""
+    try:
+      self._upload(note_events)
+    except roam_backend_api.RoamAPIError as e:
+      print(f'Roam API upload failed: {e}')
+      return False
+    return True
+
+  def _upload(self, note_events):
+    client = self.get_client()
+    blob_client = None
+    for note_event in note_events:
+      if note_event.action == events.INDENT:
+        # When you press tab, that adds your most-recent note to a stack.
+        if self.last_note_uid and self.last_note_uid not in self.stack:
+          self.stack.append(self.last_note_uid)
+      elif note_event.action == events.UNINDENT:
+        # When you shift-tab, that pops from the stack.
+        if self.stack:
+          self.stack.pop()
+      elif note_event.action == events.CLEAR_EMPTY:
+        # When you shift-delete from an empty note, that clears the stack.
+        self.stack = []
+      elif note_event.action == events.ENTER_EMPTY:
+        # When you submit from an empty note, that pops from the stack.
+        if self.stack:
+          self.stack.pop()
+      elif note_event.action == events.END_SESSION:
+        self.end_session()
+      elif note_event.action == events.SUBMIT:
+        if self.session_uid is None:
+          self.new_session(note_event)
+        text = note_event.text.strip()
+        has_audio = bool(note_event.audio_filepath) and os.path.exists(note_event.audio_filepath)
+        if has_audio:
+          text = f'{text} {UNVERIFIED_TAG}'
+        parent_uid = self.stack[-1] if self.stack else self.session_uid
+        block_uid = client.create_block(parent_uid, text)
+        self.last_note_uid = block_uid
+        print(f'Inserted: "{text}" at block (({block_uid}))')
+        if has_audio:
+          if blob_client is None:
+            blob_client = blob_uploader.make_client()
+          embed_url = blob_uploader.upload_blob(note_event.audio_filepath, blob_client)
+          if embed_url:
+            embed_text = '{{audio: ' + embed_url + '}}'
+            print(f'Audio embed: {embed_text}')
+            client.create_block(block_uid, embed_text)
+
+  def handle_inactivity(self):
+    self.end_session()
+
+  def handle_disconnect(self):
+    self.end_session()
+
+  def end_session(self):
+    self.session_uid = None
+    self.last_note_uid = None
+    self.stack = []

gonotego/uploader/roam/roam_backend_api.py

@@ -0,0 +1,196 @@
+"""Client for the Roam Research backend API.
+
+Talks to Roam over HTTPS instead of driving Roam in a browser. Roam's web app
+sits behind a bot-verification challenge that headless browsers on the Pi fail,
+so the browser-based uploader can no longer reach the graph; the backend API
+has no such check.
+
+Create a token in Roam under Settings > Graph > API tokens (edit access), then
+run ":set ROAM_API_TOKEN <token>" on Go Note Go or add it to secure_settings.
+
+Reference: https://github.com/Roam-Research/backend-sdks
+Docs: https://roamresearch.com/#/app/developer-documentation/page/bmYYKQ4vf
+"""
+import random
+import re
+import string
+import time
+
+import requests
+
+BASE_URL = 'https://api.roamresearch.com'
+UID_ALPHABET = string.ascii_letters + string.digits + '-_'
+UID_LENGTH = 9
+PEER_RE = re.compile(r'https://(peer-\d+)[^:/]*:(\d+)')
+MAX_REDIRECTS = 5
+RETRY_DELAY_SECONDS = 3
+
+PAGE_UID_QUERY = """
+[:find ?uid
+ :in $ ?title
+ :where
+ [?page :node/title ?title]
+ [?page :block/uid ?uid]]
+"""
+
+BLOCK_ON_PAGE_QUERY = """
+[:find ?uid
+ :in $ ?page-uid ?string
+ :where
+ [?page :block/uid ?page-uid]
+ [?block :block/page ?page]
+ [?block :block/string ?string]
+ [?block :block/uid ?uid]]
+"""
+
+
+class RoamAPIError(Exception):
+  """The Roam backend API rejected a request or could not be reached."""
+
+
+class RoamNotReadyError(RoamAPIError):
+  """The graph is still starting up on Roam's side; the request can be retried."""
+
+
+def generate_uid():
+  """Returns a new random block uid in Roam's 9-character format."""
+  return ''.join(random.choice(UID_ALPHABET) for _ in range(UID_LENGTH))
+
+
+def ordinal_suffix(day):
+  """Returns 'st', 'nd', 'rd', or 'th' for a day of the month."""
+  if 3 < day < 21:
+    return 'th'
+  return {1: 'st', 2: 'nd', 3: 'rd'}.get(day % 10, 'th')
+
+
+def daily_note_title(dt):
+  """Returns the title Roam uses for the Daily Notes page of a date, e.g. 'September 5th, 2026'."""
+  return f'{dt.strftime("%B")} {dt.day}{ordinal_suffix(dt.day)}, {dt.year}'
+
+
+def daily_note_uid(dt):
+  """Returns the uid Roam uses for the Daily Notes page of a date, e.g. '09-05-2026'."""
+  return dt.strftime('%m-%d-%Y')
+
+
+def normalize_graph_name(graph):
+  """Strips the 'app/' prefix the browser uploader accepts in ROAM_GRAPH."""
+  if graph.startswith('app/'):
+    return graph[len('app/'):]
+  return graph
+
+
+class RoamBackendClient:
+  """A minimal client for Roam's q and write endpoints."""
+
+  def __init__(self, token, graph, session=None, timeout=60, retries=3):
+    if not token:
+      raise ValueError('A Roam API token is required.')
+    self._token = token
+    self.graph = normalize_graph_name(graph)
+    self._session = session or requests.Session()
+    self._timeout = timeout
+    self._retries = retries
+    # Roam redirects the first request to a graph-specific peer; we remember it.
+    self._base_url = BASE_URL
+
+  def _headers(self):
+    return {
+        'Content-Type': 'application/json; charset=utf-8',
+        'Accept': 'application/json',
+        'Authorization': f'Bearer {self._token}',
+        'x-authorization': f'Bearer {self._token}',
+    }
+
+  def _call_once(self, endpoint, body):
+    for _ in range(MAX_REDIRECTS):
+      url = f'{self._base_url}/api/graph/{self.graph}/{endpoint}'
+      response = self._session.post(
+          url, headers=self._headers(), json=body,
+          allow_redirects=False, timeout=self._timeout)
+      if response.is_redirect or response.is_permanent_redirect:
+        location = response.headers.get('Location', '')
+        match = PEER_RE.search(location)
+        if not match:
+          raise RoamAPIError(f'Unexpected redirect from Roam API: {location!r}')
+        peer, port = match.groups()
+        self._base_url = f'https://{peer}.api.roamresearch.com:{port}'
+        continue
+      break
+    else:
+      raise RoamAPIError('Too many redirects from the Roam API.')
+
+    if response.status_code == 401:
+      raise RoamAPIError('Roam API token is invalid or lacks permission for this graph.')
+    if response.status_code == 503:
+      raise RoamNotReadyError('Roam graph is not ready yet; retry in a few seconds.')
+    if not response.ok:
+      raise RoamAPIError(f'Roam API error (HTTP {response.status_code}): {response.text[:500]}')
+    return response
+
+  def call(self, endpoint, body):
+    """POSTs body to the graph's endpoint ('q', 'pull', or 'write') and returns the response.
+
+    Retries transient failures (connection problems, graph not ready) a few times.
+    Raises RoamAPIError if the request ultimately fails.
+    """
+    attempt = 0
+    while True:
+      try:
+        return self._call_once(endpoint, body)
+      except (requests.RequestException, RoamNotReadyError) as e:
+        attempt += 1
+        if attempt > self._retries:
+          if isinstance(e, RoamAPIError):
+            raise
+          raise RoamAPIError(f'Could not reach the Roam API: {e!r}') from e
+        print(f'Roam API request failed ({e!r}); retrying in {RETRY_DELAY_SECONDS}s.')
+        time.sleep(RETRY_DELAY_SECONDS)
+
+  def q(self, query, args=None):
+    """Runs a datalog query and returns its result rows."""
+    body = {'query': query}
+    if args is not None:
+      body['args'] = list(args)
+    return self.call('q', body).json()['result']
+
+  def write(self, body):
+    return self.call('write', body)
+
+  def get_page_uid(self, title):
+    """Returns the uid of the page with this title, or None if it doesn't exist."""
+    results = self.q(PAGE_UID_QUERY, [title])
+    return results[0][0] if results else None
+
+  def create_page(self, title, uid=None):
+    """Creates a page and returns its uid."""
+    page = {'title': title}
+    if uid:
+      page['uid'] = uid
+    self.write({'action': 'create-page', 'page': page})
+    return uid or self.get_page_uid(title)
+
+  def get_or_create_page(self, title, uid=None):
+    return self.get_page_uid(title) or self.create_page(title, uid=uid)
+
+  def get_block_on_page_uid(self, page_uid, text):
+    """Returns the uid of a block anywhere on the page with exactly this text, or None."""
+    results = self.q(BLOCK_ON_PAGE_QUERY, [page_uid, text])
+    return results[0][0] if results else None
+
+  def create_block(self, parent_uid, text, order='last', uid=None):
+    """Creates a child block under parent_uid and returns the new block's uid.
+
+    order is an integer position or 'last'.
+    """
+    uid = uid or generate_uid()
+    self.write({
+        'action': 'create-block',
+        'location': {'parent-uid': parent_uid, 'order': order},
+        'block': {'string': text, 'uid': uid},
+    })
+    return uid
+
+  def get_or_create_block_on_page(self, page_uid, text, order='last'):
+    return self.get_block_on_page_uid(page_uid, text) or self.create_block(page_uid, text, order=order)