from browser import document, ajax, prompt, alert, window, timer, confirm, aio
from browser.template import Template
import browser.html
from javascript import JSON
import hashlib
import random
import time
import html
import re
import urllib.parse
import base64
import jupiter
def if_else(c, x, y):
if c:
return x
else:
return y
def get_random_pcode(s='1234567890qwertyuiopasdfghjklzxcvbnmQWERTYUIOPASDFGHJKLZXCVBNM', n=6):
return ''.join([random.choice(s) for i in range(n)])
def if_no_key_error(func, *args, **kwargs):
try:
return func(*args, **kwargs) or True
except KeyError:
return False
def get_uuid(data=None, email='joe@example.com', nrandbytes=16):
h = hashlib.sha256(bytes(f'{email},{time.time()}', 'utf-8'))
if data:
h.update(bytes(data, 'utf-8'))
if nrandbytes > 0:
h.update(random.randbytes(nrandbytes))
return h.hexdigest()
def standardize_uuid(s):
s = ''.join(s.split('-')) # remove any pre-existing dash
return '-'.join([s[:8], s[8:12], s[12:16], s[16:20], s[20:32]])
defaultAttrRules = [
('.insertbefore-vert', {'b-on': 'drop:prepend_block_h;dragenter:drag_past_h;dragover:drag_past_h;dragleave:drag_past_h'}),
('.std-editor', {'contenteditable': 'true',
'b-on': 'input:changed_text_h;keydown:key_command_h;click:click_command_h;pointerup:on_selection_h;drop:add_child_h;dragenter:drag_past_h;dragover:drag_past_h;dragleave:drag_past_h'}),
('.editBlockMenu', {'b-on': 'dragstart:start_drag_h;pointerdown:pointerdown_h;pointerup:pointerup_h', 'draggable': 'true'}),
('.addBlockMenu', {'b-on': 'click:plus_menu_h'}),
]
def apply_attr_rules(m, attrRules):
'apply selector-based attribute rules analogous to CSS, but for non-style attributes'
for selector, rules in attrRules:
for el in m.select(selector):
for attr, v in rules.items():
if attr == 'b-on' and el.attrs.get(attr, False):
el.attrs[attr] = el.attrs[attr] + ';' + v
else:
el.attrs[attr] = v
def get_var_or_empty_string(path, d, default='', htmlEscape=False):
'convenience function for safe variable access in templates'
for k in path.split('.'):
try:
d = d[k]
except KeyError:
return default
if htmlEscape:
d = html.escape(d)
return d
def copy_to_container(templateID, uuid, container, variables=None, insertBeforeID=None,
insertAfterID=None, attrs={}, defaultTemplate=None,
handlers=[], attrRules=defaultAttrRules, addClass=None):
'clone a template, inject content, and append to container'
try:
template = document[templateID]
except KeyError:
if defaultTemplate:
template = defaultTemplate
else:
raise
message = template.cloneNode(True) # clone its full subtree including all descendants
if uuid:
message.id = uuid
else:
message.id = get_uuid(templateID + repr(attrs))
for k,v in attrs.items():
message.attrs[k] = v
insert_to_dom(container, message, insertBeforeID, insertAfterID)
if attrRules:
apply_attr_rules(message, attrRules)
#print(f'after attr rules: {message.html}')
if variables:
#print(f'render: {template}, {variables}')
Template(message, handlers.copy()).render(_safe=get_var_or_empty_string, **variables)
#print(f'after templating: {document[message.id].html}')
message = document[message.id] # Template.render() replaced message in the DOM, so re-get!!
if addClass:
add_class(message, addClass)
bind_handlers(message, handlers.copy(), 'b-on-var') # bind handlers injected by variables
return message
def insert_to_dom(container, message, insertBeforeID=None, insertAfterID=None, checkDuplicates=True):
'place message at desired location in DOM'
if checkDuplicates and getattr(message, 'id', 'NoNeXiStEnT') in document:
raise KeyError(f'cannot insert duplicate DOM element ID {message.id}')
if message.parent and message.parent.parent and message.parent.parent.tagName.lower() == 'details':
oldContainer = message.parent
else:
oldContainer = None
if insertBeforeID: # insert at this specific location
container.insertBefore(message, document[insertBeforeID])
elif insertAfterID:
document[insertAfterID].after(message)
else:
container <= message # append to container's children
if container.parent and container.parent.tagName.lower() == 'details':
container.parent.style.display = 'block'
if oldContainer and not oldContainer.children:
oldContainer.parent.style.display = 'none'
def move_element_by_id(el_id, parent_id, selector='[data-name=ChildrenContainer]', **kwargs):
'move specified DOM element to specified subdiv of new parent'
container = select_closest(document[parent_id], selector)
el = document[el_id]
insert_to_dom(container, el, checkDuplicates=False, **kwargs)
def move_within_container(container, ori, dest=None):
if dest:
dest = container.select_one(dest)
if dest:
dest = dest.id
insert_to_dom(container, container.select_one(ori), dest, checkDuplicates=False)
def copy_items(labels, templateID, containerID, values=(), event=None, func=None, attr='data-name',
withTitle=False):
container = document[containerID]
attrs = {}
for i,labelHTML in enumerate(labels):
if values:
attrs[attr] = values[i]
m = copy_to_container(templateID, None, container, attrs=attrs)
m.html = labelHTML
if withTitle and values:
m.attrs['title'] = values[i]
if event:
m.bind(event, func)
def bind_handlers(el, handlers, attr, selector=None):
'find subnodes with specified attr, and bind handlers just as Brython b-on template does'
hdict = {h.__name__:h for h in handlers}
if not selector:
selector = f'[{attr}]'
for node in el.select(selector):
for s in node.attrs[attr].split(';'):
try:
ev, hname = s.split(':')
node.bind(ev, hdict[hname])
except (ValueError, KeyError):
pass
class UITool(object):
'UI toolbar displayed only until next mouse click'
def __init__(self, containerID, top_of_doc='wholebody'):
self.id = containerID
self.justOpened = False
el = document[top_of_doc]
el.bind('click', self.hide_menu)
def show_menu(self, ev, action, data=None):
'show menu right by current mouse click position'
self.action = action
self.data = data
self.justOpened = True
el = document[self.id]
el.left = int(ev.currentTarget.abs_left)
el.top = int(ev.currentTarget.abs_top) + 20
el.style.display = 'block'
def hide_menu(self, ev):
el = document[self.id]
if not self.justOpened:
el.style.display = 'none'
self.justOpened = False
class Menu(UITool):
'very very basic popup menu'
def __init__(self, containerID, labels, values, templateID='MenuItemTemplate',
top_of_doc='wholebody', withTitle=False):
UITool.__init__(self, containerID, top_of_doc)
self.action = None
copy_items(labels, templateID, containerID, values, 'click', self.do_action, withTitle=withTitle)
def do_action(self, ev):
if isinstance(self.action, dict): # use dict to dispatch specific menu action
self.action[ev.currentTarget.attrs['data-name']](ev, self.data)
else:
self.action(ev, ev.currentTarget.attrs['data-name'])
self.justOpened = False
self.hide_menu(ev)
menuIndex = {
'addBlockMenu': Menu('addBlockMenu', ('Text', 'List', 'To Do', 'Quote',
'Callout', 'Toggle', 'URL', 'Image',
'Equation', 'Code', 'Heading', 'Subheading', 'Subheading 3',
'PDF document', 'File', 'Audio File'),
('TextBlock', 'ListBlock', 'ToDoBlock', 'QuoteBlock',
'CalloutBlock', 'ToggleBlock', 'LinkBlock', 'ImageBlock',
'MathBlock', 'CodeBlock', 'H1Block', 'H2Block', 'H3Block',
'PdfBlock', 'FileBlock', 'AudioBlock')),
'editBlockMenu': Menu('editBlockMenu', ('Lots', 'of', 'Cool', 'Things'),
('opt1', 'opt2', 'opt3', 'opt4')),
'xrefMenu': Menu('xrefMenu', ('Open', 'Unlink'), ('open', 'unlink')),
'relationMenu': Menu('relationMenu', ('Link', 'New'), ('link', 'new')),
'addXRefMenu': Menu('addXRefMenu', ('Link Existing Block', 'Link New Block',),
('link', 'create',)),
'formatToolbar': Menu('formatToolbar', ('B', 'It', '<>',
'∑', 'Annotate'),
('bold', 'italic', 'code', 'equation', 'annotate'), 'ButtonTemplate',
withTitle=True),
}
def get_text(el):
try:
return el.value
except AttributeError:
return el.text
def is_contenteditable(el):
'True iff this node or one of its parents is contenteditable'
while el:
if el.attrs.get('contenteditable', False) == 'true':
return True
el = el.parent
def rm_element_keeping_children(el):
parent = el.parent
for c in list(el.child_nodes):
parent.insertBefore(c, el)
el.remove()
parent.normalize() # merge adjacent text nodes if any
return parent
def closest_parent_safe(el, selector):
'equivalent to el.closest(selector) but safe on non-Node objects such as Text'
while el:
try:
return el.closest(selector)
except:
el = el.parent
def parent_details_open(el):
'make el visible by setting details.open true from here up'
try:
el = el.closest('details')
while el:
el.open = True
el = getattr(el, 'parent', False) and el.parent.closest('details')
except KeyError:
pass
def find_parent_element(el, parents):
'if el has ancestor in parents, return it, else None'
while el:
for p in parents:
if p == el:
return p
el = el.parent
def path_to_parent(el, parent):
l = []
while el and el != parent:
l.append(el)
el = el.parent
if not el:
raise KeyError('el is not a child of parent')
return l
def select_closest(el, selector='[data-name=ChildrenContainer]'):
'finds match that is closest descendant of el'
hits = el.select(selector)
if not hits:
return None
l = [(len(path_to_parent(hit, el)), hit) for hit in hits]
l.sort(key=lambda x:x[0])
return l[0][1]
def add_class(el, cname):
el.class_name = ' '.join([c for c in ((el.class_name or '').split() + [cname])])
def remove_class(el, cname):
el.class_name = ' '.join([c for c in (el.class_name or '').split() if c != cname])
# In theory this could work for input and textarea
# using additional selectors '[data-block-attr]', '[data-mutation-obs]' and attributes=True
# But in practice MutationObserver just does not work for input and textarea.
def create_mutation_observer(node, func, selectors=('[contenteditable]', ),
options = dict(subtree=True, childList=True, characterData=True)):
'begin observing mutations in selected subtrees of node'
observer = window.MutationObserver.new(func)
editables = []
for selector in selectors:
for el in node.select(selector):
if not find_parent_element(el, editables):
el.attrs['data-mutation-obs'] = 'editable'
observer.observe(el, options)
editables.append(el)
return observer
#######################################################################
# mathjax handling
def tex2html(texString, **kwargs):
el = window.MathJax.tex2chtml(texString, kwargs)
return el.outerHTML
def refresh_mathjax():
window.MathJax.startup.document.clear()
window.MathJax.startup.document.updateDocument()
##########################################################################
# new-style ivals text markup system
# convert to DOM (HTML)
def make_ivals_tree(ivals, end):
'turn sorted list of ivals into nested-containment list, by adding children key to each ival dict'
l = []
while ivals and ivals[0]['i'][1] <= end:
ival = ivals[0]
children, ivals = make_ivals_tree(ivals[1:], ival['i'][1])
ival['children'] = children
l.append(ival)
return l, ivals
def convert_ival_to_html(ival, s, ipos):
'return html representing region s[ipos:ival["i"][1]]'
prefix = html.escape(s[ipos:ival['i'][0]])
ipos = ival['i'][0]
innerHTML = ''
for c in ival.get('children', ()):
t, ipos = convert_ival_to_html(c, s, ipos)
innerHTML += t
innerHTML += html.escape(s[ipos:ival['i'][1]])
out = prefix + format_ival_element(ival, innerHTML)
return out, ival['i'][1]
def convert_ivals_to_html(s, ivals):
's is the entire text string; applies formatting ivals and returns HTML'
ivals = ivals.copy() # do not alter the input list
ivals.sort(key=lambda x:x['i'][1], reverse=True)
ivals.sort(key=lambda x:x['i'][0]) # stable sort, so now ordered by (start, end DESC)
root = dict(i=[0, len(s)], children=make_ivals_tree(ivals, len(s))[0])
return convert_ival_to_html(root, s, 0)[0]
def format_ival_element(ival, innerHTML):
'note that innerHTML is already sterilized'
defaultFunc = lambda tag,s,ival: f'<{tag}>{s}'
try:
tag = formatCodes[ival['t']]
return formatFuncs.get(ival['t'], defaultFunc)(tag, innerHTML, ival)
except KeyError:
return innerHTML # if formatting func fails, just show the content
# convert back from DOM
def convert_dom_to_ivals(el, s=''):
'return text and format ivals list for specified DOM node'
if el.nodeName == '#text': # just append text, no children ivals
return (s + el.text, [])
elif el.tagName.lower() in ('textarea', 'input'):
return (el.value, []) # just unformatted text
start = len(s)
data = get_element_dict(el)
child_ivals = []
if data and 'nodetext' in data:
s += data['nodetext']
del data['nodetext']
if data and 'skipChildren' in data:
del data['skipChildren']
else: # recurse to DOM children
for c in el.child_nodes:
s, subvals = convert_dom_to_ivals(c, s)
child_ivals += subvals
end = len(s)
if data:
data['i'] = [start, end] # interval coords
child_ivals = [data] + child_ivals
return s, child_ivals
def get_element_dict(el):
'return a dict of the form {"t": "a", "href": "https://example.com"}'
tag = el.tagName.lower()
if tag == 'a':
return dict(t=tag, href=el.href)
elif tag == 'span' and el.attrs.get('class', '').startswith('highlight-'):
return dict(t='h', color=el.attrs['class'].split('-')[-1])
elif tag == 'span' and 'data-dest-id' in el.attrs:
return dict(t='IVAL', ID=el.attrs['data-dest-id'], pred=el.attrs.get('data-ival-pred', 'use'),
is_deleted=el.attrs.get('data-dest-deleted', False))
elif tag == 'span' and 'data-inline-latex' in el.attrs:
return dict(t='e', nodetext=el.attrs['data-inline-latex'], # nodetext will be re-inserted in text
skipChildren=True)
elif tag in formatRev:
return dict(t=formatRev[tag])
formatCodes = dict(b='b', c='code', i='i', _='u', a='a', h='h', e='e', s='s', IVAL='span')
formatRev = {t[1]:t[0] for t in formatCodes.items() if t[1] != 'span'}
formatFuncs = dict(a=lambda tag,s,ival: f'''{s}''',
h=lambda tag,s,ival: f'''{s}''',
e=lambda tag,s,ival: f'''{tex2html(s, display=False)}''',
IVAL=lambda tag,s,ival: f'''<{tag} class="ival-annotation" data-dest-id="{ival['ID']}" data-ival-pred="{ival.get('pred', 'use')}" title="{ival.get('text', 'minilanguage term')}: double-click to open definition">{s}''')
def set_inline_latex(span, latex):
span.attrs['data-inline-latex'] = latex
span.html = tex2html(latex, display=False)
# Regular vs. edge ivals are stored separately on the backend.
# So they have to be separated before sending to backend API:
def separate_edge_ivals(ivals, attr='text'):
'return separate dicts for regular ivals vs. edges_ivals'
nodeIvals = []
edgeIvals = {}
for ival in ivals:
dest = ival.get('ID', None)
if dest: # reference to another block must be stored as edge
l = edgeIvals.setdefault(dest, {}).setdefault(attr, [])
if not ival['is_deleted']: # if deleted, leave list empty
del ival['is_deleted'] # don't save these temp data
del ival['ID']
l.append(ival) # ival not deleted, so add to list
else:
nodeIvals.append(ival)
return {attr: nodeIvals}, edgeIvals
# and re-merged after retrieval from backend API:
def get_edges_ivals_dict(nodedata):
'returns {attr: ivals} from api.edges data for this node'
xrefsDict = nodedata.get('properties', {}).get('ivals', {})
attrsDict = {}
for dest, xrefData in xrefsDict.items(): # get ivals xrefs for each dest
for attr, ivals in xrefData.get('edge_data', {}).get('attrs', {}).items():
#print(f'get_edges_ivals_dict: {nodedata["id"], dest, attr, ivals}')
l = []
for ival in ivals: # inject dest ID into each ival
d = ival.copy()
d['ID'] = dest
for a, val in xrefData.items(): # copy other attrs like text, page_icon
if a != 'edge_data':
d[a] = val
l.append(d)
#print(f'attrsDict: {attrsDict, attr, l}')
l2 = attrsDict.setdefault(attr, [])
l2 += l
return attrsDict
##########################################################################
# Notion-style format interval list handling
class DOM2FormattedText(object):
'convert DOM elements back into format-interval list a la Notion'
_funcs = {'a': 'generate_a', 'span': 'generate_span', 'mjx-container': 'generate_math'}
def __init__(self, formats):
self.tags = {t[1]:t[0] for t in formats.items()}
self.funcs = {t[0]:getattr(self, t[1]) for t in self._funcs.items()}
def get_formatted_text(self, el):
'get Notion-style format-interval list for contents of el'
if el.tagName.lower() in ('textarea', 'input'):
return [[el.value]] # just unformatted text
else:
return list(self.generate_children(el, None, False))
def generate_children(self, el, tagInfo, addTag):
for c in el.child_nodes:
for l in self.generate_formatted_text(c, True):
if addTag and tagInfo: # add format info from el
l[1:] = [[tagInfo] + (l[1:] and l[1])]
yield l
def generate_formatted_text(self, el, addTag=False):
'walk the subtree generating intervals of the form [text, [optional format(s)]]'
if el.nodeName == '#text':
yield [el.text]
return
tag = el.tagName.lower()
try:
f = self.funcs[tag]
except KeyError:
c = self.tags.get(tag, None)
it = self.generate_children(el, c and [c], addTag)
else:
it = f(el, addTag)
for l in it:
yield l
def generate_a(self, el, addTag):
return self.generate_children(el, ['a', el.href], addTag)
def generate_span(self, el, addTag):
if el.attrs.get('class', '').startswith('highlight-'):
return self.generate_children(el, ['h', el.attrs['class'].split('-')[-1]], addTag)
def generate_math(self, el, addTag):
'just a stub until I implement inline equation editing'
if addTag == 'math is just ignored now...':
yield []
class FormattedText(object):
'handle text stored as format-interval list a la Notion'
_formats = dict(b='b', c='code', i='i', _='u', a='a', h='h', e='e', s='s')
_funcs = dict(a=lambda s,href: f'{s}',
h=lambda s,color: f'{s}',
e=lambda s,texString: tex2html(texString, display=False))
def __init__(self, data, block=None, attr='text'):
self.block = block
if isinstance(data, str):
data = [[data]]
self.data = data
self.attr = attr
self.inverse = DOM2FormattedText(self._formats)
def get_html(self, stripTags=False):
return (self.data and ''.join([self.get_html_entry(t, stripTags) for t in self.data])) or ''
def get_html_entry(self, t, stripTags=False):
s = html.escape((t and t[0]) or '') # prevent HTML injection attacks
if len(t) == 1 or stripTags:
return s
flist = t[1:] and t[1] # properly handle missing arg
for i in range(len(flist) - 1, -1, -1): # iterate in reverse
try:
tag = self._formats[flist[i][0]]
try:
arg2 = (flist[i][1:] and flist[i][1]) or '' # if no 2nd arg, get '' instead of IndexError
s = self._funcs[tag](s, arg2)
except KeyError:
s = f'<{tag}>{s}'
except KeyError:
pass
return s
def update(self, el):
self.data = list(self.inverse.generate_formatted_text(el))
def get_data(self):
return self.data
# convenience for converting DOM content back to FormattedText
dom2ft = DOM2FormattedText(FormattedText._formats)
def updater(d, **kwargs):
'make a copy of dict and update it with kwargs'
d = d.copy()
d.update(kwargs)
return d
def convert_nodedata_to_html(nodedata, variables={}):
'convert attrs, ivals, format etc. to dict of html-formatted variables for templating'
variables = variables.copy()
ivalsDict = nodedata.get('ivals', {}).get('attrs', {})
ivalsEdges = get_edges_ivals_dict(nodedata)
for k,v in nodedata.get('attrs', {}).items():
ivals = ivalsDict.get(k, []) + ivalsEdges.get(k, []) # merge regular ivals and edge ivals
if ivals: # ivals format representation
variables[k] = convert_ivals_to_html(v, ivals)
elif k.endswith('_f_'): # for properties, which still use FormattedText
variables[k[:-3]] = FormattedText(v).get_html()
elif isinstance(v, str):
variables[k] = html.escape(v) # must protect against HTML injection attacks
else:
variables[k] = v
for k in ('format',):
variables[k] = nodedata.get(k, {})
variables['tex2html'] = tex2html
try:
variables['apiURL'] = dataLoader.url
except (NameError, AttributeError):
pass
return variables
class BlockNode(object):
'very basic way of displaying a document as JSON dict with type, ftext, children'
_containers = dict(children=dict(selector='[data-name=ChildrenContainer]'),
header=dict(selector='[data-name=HeaderContainer]'),
properties=dict(replace_id=True, selector='[data-name=PropertiesContainer]'))
_propertyTypes = dict(text='Text', number='Number', relation='Relation', select='Choice (1 of N)',
multi_select='Choices (any of N)', date='Date', url='URL')
addBlockMenu = 'addBlockMenu'
editBlockMenu = 'editBlockMenu'
def __init__(self, container, nodedata, variables={}, parent=None, uuid=None, addChildrenKwargs={}, **kwargs):
self.parent = parent
try:
addClass = nodedata['addClass']
del nodedata['addClass']
except KeyError:
addClass = None
self.data = nodedata
if parent and not container: # default to use parent as container
container = select_closest(document[parent.id])
nodetype = nodedata.get('type', 'TextBlock')
template = f'{nodetype}Template'
defaultTemplate = document['TextBlockTemplate']
if uuid == 'RANDOM':
uuid = standardize_uuid(get_uuid(repr(nodedata)))
elif not uuid:
uuid = nodedata.get('id', None)
variables = convert_nodedata_to_html(self.data, variables)
variables['propertyTypes'] = self._propertyTypes
handlers = [getattr(self, a) for a in dir(self) if a.endswith('_h')]
try:
node = copy_to_container(template, uuid, container, variables, addClass=addClass,
defaultTemplate=defaultTemplate, handlers=handlers, **kwargs)
except Exception:
print(f'render error: {template}, {uuid}, {container}, {variables}, {kwargs}')
raise
self.id = node.id
self.observer = create_mutation_observer(node, self.content_changed)
schemas = nodedata.get('schemas', ())
self.make_child_dicts() # apply makerules for subcontent of this block
self.children = add_children(node, nodedata.get('children', ()), self, **addChildrenKwargs)
for attr, kwargs in self._containers.items(): # add children to subcontainers
makedata = nodedata.get('_make', {}).get(attr, ())
if makedata:
setattr(self, attr, getattr(self, attr, []) +
add_children(node, makedata, self, **kwargs))
def make_child_dicts(self):
'apply makerules for auto-generated children of this block'
rule = getattr(self, 'makerule', None)
if rule:
for c in rule():
try:
container = c['container']
del c['container']
except KeyError:
container = 'children'
self.data.setdefault('_make', {}).setdefault(container, []).append(c)
def get_id(self):
'safe method for getting ID of this block, even if shown more than once on this page'
return self.data['id']
def changed_text_h(self, ev, el):
attr = ev.currentTarget.attrs.get('data-block-attr', 'text')
dataLoader.queue_update(f'{self.id}:{attr}', self.save_text, ev.currentTarget)
def content_changed(self, mutations, observer):
'process MutationObserver mutations'
for m in mutations:
el = closest_parent_safe(m.target, '[data-mutation-obs]')
if el:
attr = el.attrs.get('data-block-attr', 'text')
dataLoader.queue_update(f'{self.id}:{attr}', self.save_text, el)
def hide_modal_h(self, ev, templateObj=None):
document[self.id].style.display = 'none'
def plus_menu_h(self, ev, el):
'launch the menu for adding a new block'
menu = menuIndex[self.addBlockMenu]
menu.show_menu(ev, self.create_block)
def pointerdown_h(self, ev, el):
self.last_pointerdown = dict(time=time.time(), x=ev.pageX, y=ev.pageY)
def pointerup_h(self, ev, el):
if time.time() - self.last_pointerdown['time'] < 0.3:
menu = menuIndex[self.editBlockMenu]
menu.show_menu(ev, self.noop)
def create_block(self, ev, blockType):
self.copy(blockType, empty=True)
def noop(self, ev, data):
print(f'noop: {data}')
def key_command_h(self, ev, el):
if ev.key == 'Enter':
ev.preventDefault() # block default key action
self.copy(empty=True) # clone an empty version of this block
elif ev.key == 'Delete' and confirm('Do you want to delete this block and its subcontents from the page? At present there is no UNDO function.'):
ev.preventDefault() # block default key action
self.delete()
def on_selection_h(self, event, templateObj):
selection = get_nonzero_selection()
if selection:
menu = menuIndex['formatToolbar']
menu.show_menu(event, dict(bold=self.bold_selection, italic=self.italic_selection,
code=self.code_selection, annotate=self.show_annotation_tool, equation=self.latex_selection),
dict(selection=selection))
def bold_selection(self, event, data):
document.execCommand('bold', False, None)
def italic_selection(self, event, data):
document.execCommand('italic', False, None)
def code_selection(self, event, data):
node = browser.html.CODE()
embed_selection_in_node(node)
def latex_selection(self, event, data):
node = browser.html.SPAN()
embed_selection_in_node(node, set_inline_latex)
timer.set_timeout(refresh_mathjax, 500) # force mathjax to actually show the change
def rm_latex_h(self, event, templateObj=None):
el = event.currentTarget
el.replaceWith(el.attrs['data-inline-latex']) # remove span, replace with text string
def checkbox_h(self, ev, el):
self.save_attr('checked', ev.currentTarget.checked)
def show_annotation_tool(self, event, data):
target = document['annotation-tool-test']
target.scrollIntoView()
def stop_event_propagation_h(self, event, templateObj):
event.stopPropagation()
def click_command_h(self, event, templateObj):
if event.detail == 2 and (el := get_selected_annotation()) and (
dest := el.attrs.get('data-dest-id', False)):
dataLoader.load_page(dest)
def upload_file_h(self, event, templateObj):
aio.run(self.upload_file_async(event, templateObj))
async def upload_file_async(self, event, templateObj):
reader = LocalFileUIBase(asBase64=True)
reader.open(event) # pass on the input event
base64data = await reader.get_text()
data = dict(ech_id=reader.ech_id, base64data=base64data,
b_data=dict(filename=reader.name))
fut = aio.Future()
dataLoader.insert('rpc/upload_ech_file', data, lambda req:fut.set_result(req))
req = await fut
if self.report_response(req):
data = dict(b_id=self.id, attrs=dict(ech_id=reader.ech_id), ivals={}, edges_ivals={})
self.data.setdefault('attrs', {})['ech_id'] = reader.ech_id # save local attr
if not self.data.get('attrs', {}).get('text', False):
self.data.setdefault('attrs', {})['text'] = reader.name # ensure some text description
data['attrs']['text'] = reader.name
dataLoader.insert('rpc/update_attrs', data, self.report_response) # save to backend
#templateObj.data.ech_id = reader.ech_id # trigger brython auto re-render FAILS???
self.rerender_h(event, templateObj) # so do it manually
def rerender_h(self, event, templateObj, **kwargs):
variables = templateObj.data.to_dict() # needed to preserve low-level vars like _safe
variables.update(kwargs)
variables = convert_nodedata_to_html(self.data, variables)
templateObj.render(**variables) # manually re-render!!
def block_details_toggle_h(self, event, templateObj):
undo_stupid_toggle(event)
def save_text_h(self, ev, el):
self.save_text(ev.currentTarget)
def save_input_h(self, ev, el):
'save value as simple string rather than FormattedText'
self.save_text(ev.currentTarget, False, '')
def save_text(self, srcElement, isFormatted=True, attrSuffix=''):
attr = srcElement.attrs.get('data-block-attr', 'text') + attrSuffix
if srcElement.tagName.lower() in ('textarea', 'input'): # just treat as string
data = srcElement.value
ivalsDict = edgesIvals = None # no intervals to save
else: # save with ivals formatting
data, ivals = convert_dom_to_ivals(srcElement)
ivalsDict, edgesIvals = separate_edge_ivals(ivals, attr)
self.save_attr(attr, data, ivalsDict, edgesIvals)
def save_attr(self, attr, data, ivalsDict=None, edgesIvals=None):
self.data.setdefault('attrs', {})[attr] = data # save to local data
dataLoader.insert('rpc/update_attrs', dict(b_id=self.id, attrs={attr: data}, ivals=ivalsDict,
edges_ivals=edgesIvals), self.report_response)
def report_response(self, req):
if req.status in (200, 201, 203, 204, 205, 206):
return True
else:
alert(f'Error: {req.status} {req.text}')
def get_insertion_position(self, position=0, step=1024*1024, attr='children'):
'find position value to insert before first node > specified position'
l = getattr(self, attr)
for i, block in enumerate(l):
if position < block.data['position']: # insert before this block
if i == 0:
return block.data['position'] - step, block.id, None
else: # midpoint between predecessor
return int((l[i - 1].data['position'] + block.data['position']) / 2), block.id, None
elif position == block.data['position']: # insert after this block
if i == len(l) - 1:
return block.data['position'] + step, None, block.id
else: # midpoint between successor
return int((l[i + 1].data['position'] + block.data['position']) / 2), None, block.id
return position, None, None # for empty list, just use starting position
def copy(self, blockType=None, empty=False, nodedata=None, position=None, parent=None,
insertBeforeID=None, insertAfterID=None, **attrs):
'clone this block and save new copy to backend database'
attrs = attrs.copy()
if not parent:
parent = self.parent or self # for top-level container, the buck stops here
if position is None:
position, insertBeforeID, insertAfterID = parent.get_insertion_position(
self.data['position'])
if not blockType: # default: use same block type
blockType = self.data['type']
if blockType in ('ToDoBlock', 'ToggleBlock'): # work-around for initial zero-width CSS nuttiness
attrs.setdefault('text', 'start')
elif blockType == 'LinkBlock': # must have an initial URL; user will change
attrs.setdefault('url', 'https://courselets.org')
if not nodedata:
nodedata = dict(type=blockType, position=position, parent_id=parent.get_id(),
page_id=self.data['page_id'])
if empty:
attrs.setdefault('text', '') # ensure text attribute is set
nodedata['attrs'] = attrs
else:
nodedata['attrs'] = self.data.get('attrs', {}).copy()
nodedata['attrs'].update(attrs)
nodedata['id'] = standardize_uuid(get_uuid(repr(nodedata)))
klass = get_domnode_class(blockType)
block = klass(None, nodedata, parent=parent,
insertBeforeID=insertBeforeID, insertAfterID=insertAfterID)
parent.add_child_to_list(block, position)
block.save_new_block()
def save_new_block(self):
'insert this block as new block in database'
data = dict(id=self.id, parent_id=self.parent.id, page_id=self.data['page_id'],
position=self.data['position'], data=self.data)
dataLoader.insert('blockinsert', data, self.report_response)
def add_child_to_list(self, block, position):
'add block to self.children at specified position'
for i, b2 in enumerate(self.children):
if position < b2.data['position']:
self.children.insert(i, block)
return i
self.children.append(block)
return len(self.children) - 1
def start_drag_h(self, ev, el):
ev.stopPropagation()
ev.dataTransfer.setData('uuid', self.id)
blockIndex[self.id] = self
ev.dataTransfer.effectAllowed = 'move'
def drag_past_h(self, ev, el):
if ev.type == 'dragleave':
remove_class(ev.currentTarget, 'drag-over')
else:
ev.preventDefault() # allow drops on this target
add_class(ev.currentTarget, 'drag-over')
ev.dataTransfer.dropEffect = 'move'
def prepend_block_h(self, ev, el):
'insert right before this block'
self.parent.move_block(ev, self.data['position'] - 1)
def add_child_h(self, ev, el):
'insert as last child of this block'
self.move_block(ev, (self.children and self.children[-1].data['position']) or 0)
def move_block(self, ev, position=0):
'insert as child of this block at specified position'
block = blockIndex.get(ev.dataTransfer.getData('uuid'), None)
if block:
ev.preventDefault() # block default paste action
if 'hidden_type' in block.data: # copy a search result as new XRefBlock
position, insertBeforeID, insertAfterID = self.get_insertion_position(position)
self.copy(block.data['type'], True, position=position, parent=self,
insertBeforeID=insertBeforeID, insertAfterID=insertAfterID,
**(block.data['attrs']))
else: # a normal block-reposition event
block.reposition(self, position)
remove_class(ev.currentTarget, 'drag-over')
def reposition(self, parent, position=0):
'move this block to new parent and position'
position, insertBeforeID, insertAfterID = parent.get_insertion_position(position)
move_element_by_id(self.id, parent.id, insertBeforeID=insertBeforeID,
insertAfterID=insertAfterID)
self.parent.remove_child_from_list(self.id)
parent.add_child_to_list(self, position)
self.parent = parent
self.data['parent_id'] = parent.id
self.data['position'] = position
dataLoader.insert('rpc/upsert_location', dict(b_parent_id=parent.id, b_position=position,
b_id=self.id, b_page_id=self.data['page_id']), self.report_response)
def add_child_dict(self, nodedata, position, save=True, replace_id=False):
'create new BlockNode from nodedata, and add as child at specified position'
block = add_children(document[self.id], [nodedata], self, replace_id=replace_id)[0]
self.add_child_to_list(block, position)
if save:
dataLoader.insert('rpc/upsert_location', dict(b_parent_id=self.id, b_position=position,
b_id=block.get_id(), b_page_id=self.data['page_id']), self.report_response)
return block
def remove_child_from_list(self, child_id):
'remove specified ID from children of this block'
for i, c in enumerate(self.children):
if c.id == child_id:
del self.children[i]
return True
def empty_children(self, attr='children', selector='[data-name=ChildrenContainer]'):
'reset this container to empty'
target = document[self.id]
if selector:
target = select_closest(target, selector)
target.html = '' # empty the DOM container
setattr(self, attr, []) # empty the object container
def reload_children(self, data, **kwargs):
self.empty_children()
self.children = add_children(None, data, parent=self, **kwargs)
def delete(self, b_id=None, dest_id=None, relation='in_page', node_id=None):
if not b_id: # default: remove block from current page
for c in tuple(self.children): # first remove children as well
c.delete()
b_id = self.get_id()
node_id = self.id
dest_id = self.data['page_id']
self.parent.remove_child_from_list(self.id)
if not node_id:
node_id = b_id
document[node_id].remove() # remove from DOM
dataLoader.insert('rpc/delete_edge', dict(b_id=b_id, b_dest_id=dest_id,
b_relation=relation), self.report_response)
def create_row(self, collection_id, text='Untitled', properties={}, func=None, **kwargs):
'create new block and add as row of specified collection'
b_data = dict(attrs=dict(text=text, **kwargs), properties=properties, type='PageBlock')
page_id = standardize_uuid(get_uuid(repr(b_data)))
b_data['page_id'] = page_id # required for proper xref formating
rowdata = dict(b_id=page_id, b_dest_id=collection_id, b_data=b_data)
if not func:
func = self.report_response
dataLoader.insert('rpc/new_dbrow', rowdata, func)
return page_id, b_data
def create_relation(self, dest_id, pcode, pcode2, rtype='xref', data=None):
link = dict(b_id=self.get_id(), b_dest_id=dest_id, b_relation=pcode,
b_relation2=pcode2, b_rtype=rtype, b_data=data)
dataLoader.insert('rpc/insert_edge', link, self.report_response)
return link
def set_visibility(self, visible=True, selector=None):
target = document[self.id]
if selector:
target = select_closest(target, selector)
if visible:
target.style.display = 'block'
else:
target.style.display = 'none'
# lookup block object by ID, e.g. for drag-and-drop
blockIndex = {}
def add_children(node, children, parent=None, selector='[data-name=ChildrenContainer]',
replace_id=False, resetBlank=False, **kwargs):
'add domnodes as children of DOM node'
if parent and not node: # sensible default: node is just DOM element representing parent
node = document[parent.id]
if node.attrs.get('data-name', None) == 'ChildrenContainer':
subcont = node # insert children directly in this node
else:
subcont = select_closest(node, selector)
if resetBlank: # reset container to empty
subcont.html = ''
if replace_id: # avoid potential block ID collisions by reassigning random ID
uuid = 'RANDOM'
else:
uuid = None
subnodes = []
if subcont and children:
for c in children:
klass = get_domnode_class(c.get('type', ''))
subnodes.append(klass(subcont, c, parent=parent, uuid=uuid, **kwargs))
return subnodes
def get_domnode_class(btype):
'get BlockNode subclass for specified block type'
subclasses = dict(TableView=TableViewNode, MathBlock=MathBlockNode, ViewBlock=ViewBlockNode,
TableHeader=TableHeaderNode, RowBlock=RowBlockNode, RelationCell=RelationCellNode,
RowCell=RowCellNode, XRefBlock=XRefBlockNode, XRefCell=XRefCellNode,
XRefProperty=XRefPropertyNode, TextProperty=TextPropertyNode,
Properties=PropertiesNode, RelationProperty=RelationPropertyNode,
PageBlock=PageBlockNode, XRefPropLink=XRefCellNode, TextProperty2=RowCellNode,
CodeBlock=CodeBlockNode, ListChooser=ListChooserNode, ItemChooser=ItemChooserNode,
ModalChooser=ModalChooserNode, SchemaChooser=SchemaChooserNode
)
return subclasses.get(btype, BlockNode)
class PageBlockNode(BlockNode):
def makerule(self):
properties = self.data['properties']
tableFormat = {}
for cid,props in self.data['schemas'].items():
l = []
for pid,pschema in props.items():
if pid != 'title':
l.append(dict(property=pid, visible=(pid in properties),
name=pschema.get('name', '(no name)')))
l.sort(key=lambda x:x['name'].lower()) # sort props in alphabetical order
tableFormat[cid] = l
if tableFormat:
self.tableFormat = tableFormat
l = self.get_annotation_children(properties)
self.schema_children = add_children(None, l, self, selector='[data-name=PropertiesContainer]')
l = []
for cid in tableFormat:
l.append(dict(id=self.get_id(), type='Properties', collection_id=cid, properties=properties,
attrs=self.data.get('collections', {}).get(cid, {}), container='properties'))
return l
else:
return ()
def get_table_format(self, collection_id):
return self.tableFormat[collection_id]
def get_property_schema(self, pcode):
for cid,props in self.data['schemas'].items():
try:
return props[pcode]
except KeyError:
pass
raise KeyError(f'pcode {pcode} not found in schemas')
def set_property_schema(self, collection_id, pcode, pdata):
self.data['schemas'][collection_id][pcode] = pdata
def get_annotation_children(self, properties): # TODO: convert this to RelationProperty style
xrefcss = 'xref-bold'
pschema = dict(name='(Annotations)')
l = []
for tid,tdata in properties.get('ivals2', {}).items():
if not sum([sum([len(ivals) for ivals in d.values()])
for d in tdata.get('edge_data', {}).values()]):
continue # skip deleted (zero ivals) annotation
tdata = tdata.copy()
if tid != tdata.get('page_id', None):
tdata['scrollTarget'] = tid
ivals = [dict(t='h', color='yellow_background', i=iv['i'])
for iv in tdata.get('edge_data', {}).get('attrs', {}).get('text', [])]
tdata.update({'xrefcss': xrefcss, 'property': pschema, 'no_edit': True})
l.append({'type': 'XRefProperty', 'attrs': tdata, 'ivals': dict(attrs=dict(text=ivals))})
xrefcss = 'xref-invis'
return l
class TextPropertyNode(BlockNode):
'editable database property'
def save_text(self, srcElement):
data = dom2ft.get_formatted_text(srcElement)
pcode = self.data['attrs']['property']['pcode']
payload = dict(b_id=self.parent.get_id(), props_data={pcode: data})
dataLoader.insert('rpc/update_block_property', payload, self.report_response)
class XRefPropertyNode(BlockNode):
addBlockMenu = 'addXRefMenu'
def load_page_h(self, event, templateObj):
'handle click on an XRefProperty'
dataLoader.load_page(templateObj.data.page_id,
scrollTarget=self.data['attrs'].get('scrollTarget', None))
class XRefBlockNode(BlockNode):
'clickable link to another Conrad page'
def load_page_h(self, event, templateObj):
dataLoader.load_page(self.data['attrs']['page_id'],
scrollTarget=self.data['attrs'].get('scrollTarget', None))
class MathBlockNode(BlockNode):
def refresh_math_block_h(self, event, templateObj):
templateObj.data.text = event.currentTarget.value # this automatically triggers re-running the template!
timer.set_timeout(refresh_mathjax, 500) # force mathjax to actually show the change
# use a single interpreter for different codeblocks so they share globals namespace
brythonInterpreter = jupiter.BrythonInterpreter()
class CodeBlockNode(BlockNode):
def run_code_h(self, event, templateObj):
el = select_closest(event.currentTarget.parent, '[data-name=CodeLanguage]')
codeLanguage = el.value
el = select_closest(event.currentTarget.parent, '[data-name=SourceCode]')
sourceCode = el.value
el = select_closest(event.currentTarget.parent, '.interpreter-stdout')
plotlyDiv = select_closest(event.currentTarget.parent, '[data-name=PlotlyDisplay]')
if codeLanguage == 'mermaid':
svg_id = standardize_uuid(get_uuid(sourceCode))
window.render_mermaid(self.id, '[data-name=MermaidDisplay]', sourceCode, svg_id)
elif codeLanguage == 'python':
brythonInterpreter.run_code(sourceCode, el, plotlyDiv=plotlyDiv)
el.parent.style.display = 'block'
else:
alert(f'No support for running language {codeLanguage} yet...')
class ViewBlockNode(BlockNode):
def __init__(self, container, nodedata, variables={}, parent=None, uuid=None, **kwargs):
variables = variables.copy()
variables['viewchoices'] = [vdict['title'] for vdict in nodedata.get('children', [])]
BlockNode.__init__(self, container, nodedata, variables, parent, uuid, **kwargs)
def show_dbview_h(self, event, templateObj):
try:
iview = int(event.currentTarget.value)
except ValueError:
return # ignore "NO-SELECTION" option
tv = self.children[iview]
tv.load_rows()
tv.set_visibility(True) # show new view
try:
self.currentView.set_visibility(False) # hide last view
except AttributeError:
pass
self.currentView = tv
def create_default_format(viewblock_data):
'create a generic table display format'
props = [dict(width=200, visible=True, property=pcode)
for pcode in viewblock_data['table_schema']['schema']]
viewblock_data.setdefault('format', {})['table_properties'] = props
class TableViewNode(BlockNode):
def __init__(self, container, nodedata, variables={}, parent=None, uuid=None, **kwargs):
variables = variables.copy()
l = [(f'$.properties."{t[0]}"', t[1])
for t in parent.data.get('table_schema', {}).get('schema', {}).items() if t[0] != 'title']
l.append(('$.last_edited_time', dict(name='Modified', type='last_edited_time')))
l.append(('$.attrs.text', dict(name='Title', type='text')))
l.sort(key=lambda t:t[1]['name'].lower())
variables['schema'] = l
BlockNode.__init__(self, container, nodedata, variables, parent, uuid, **kwargs)
def load_rows(self, offset=0, limit=20):
#print(f'load_rows {self.parent.data}')
attrs = self.data.get('attrs', {})
view_filter = attrs.get('filter_by', False) or 'DEFAULT'
view_order = attrs.get('order_by', False) or 'NULL' # get_dbview_fo5() deserializes this as SQL NULL
orderByDir = attrs.get('order_by_dir', 'asc')
dataLoader.load_data('rpc/get_dbview_fo5', self.read_rows, offset=offset, limit=limit, view_id=self.get_id(),
view_filter=view_filter, view_order=view_order, order=f'sortkey1.{orderByDir}')
def makerule(self):
formatDict = self.data.get('format', dict(property='title', visible=True, width=200))
return [updater(formatDict, type='TableHeader', container='header')]
def read_rows(self, req):
rows, nextOffset = read_row_data(req)
if rows:
self.children += add_children(document[self.id], rows, self, replace_id=True)
self.nextOffset = nextOffset
def get_table_format(self, collection_id=None):
'stub API -- TableView only has a single collection_id'
return self.data.get('format', {}).get('table_properties', [])
def get_property_schema(self, pcode):
return self.parent.data['table_schema']['schema'][pcode]
def get_property_schema_safe(self, pcode):
try:
return self.get_property_schema(pcode)
except KeyError:
return {}
def load_more_h(self, event, templateObj):
self.load_rows(getattr(self, 'nextOffset', 0))
def run_filter_h(self, event, templateObj):
're-run the filter_by order_by rules'
self.empty_children() # revert to empty
self.load_rows()
def add_child_to_list(self, block, position):
self.children.append(block)
def show_er_diagram_h(self, event, templateObj):
if c_id := self.parent.data.get('attrs', {}).get('collection_id', False):
dataLoader.load_data('rpc/get_schemas', lambda req:read_schemas(req, self.id), c_id=c_id)
class TableHeaderNode(BlockNode):
def makerule(self):
'generate header cells UI dicts'
l = []
for c in self.data.get('table_properties', []):
if c.get('visible', False):
try:
text = self.parent.get_property_schema(c['property'])['name']
except KeyError:
text = 'Untitled'
l.append(dict(attrs=updater(c, text=text), type='HeaderCell'))
return l
class RowBlockNode(BlockNode):
_rowCellTypes = dict(relation='RelationCell', default='RowCell')
def makerule(self):
'generate row cells UI dicts'
l = []
for c in self.parent.get_table_format(self.data['collection_id']):
try:
celltype = self._rowCellTypes[self.parent.get_property_schema(c['property'])['type']]
kwargs = {}
except KeyError:
celltype = self._rowCellTypes['default']
kwargs = dict(text_f_=self.data.get('properties', {})
.get(c.get('property', 'notfound'), [[]]))
d = dict(attrs=updater(c, **kwargs), type=celltype)
if not c.get('visible', False):
d['addClass'] = 'hidden-property'
if c.get('visible', False) or getattr(self, '_showHiddenProperties', False):
l.append(d)
return l
def copy(self, **kwargs):
nodedata = dict(type=self.data['type'], collection_id=self.data['collection_id'],
properties=dict(title=[[]]))
BlockNode.copy(self, nodedata=nodedata, position=True, insertAfterID=self.id, **kwargs)
def save_new_block(self):
'insert this block as new row in a TableSchema'
b_data = self.data.copy()
try:
del b_data['_make']
except KeyError:
pass
data = dict(b_id=self.get_id(), b_dest_id=self.data['collection_id'], b_data=b_data)
dataLoader.insert('rpc/new_dbrow', data, self.report_response)
def delete(self):
'remove RowBlock from its TableSchema'
BlockNode.delete(self, b_id=self.get_id(), dest_id=self.data['collection_id'], relation='in_schema')
def more_props_h(self, event, templateObj):
event.currentTarget.style.display = 'none'
for el in document[self.id].select('.hidden-property'):
remove_class(el, 'hidden-property')
def add_property_h(self, event, templateObj):
el = document[self.id].select_one('[data-name=NewPropertyName]')
el2 = document[self.id].select_one('[data-name=NewPropertyType]')
try:
pname = el.value.strip()
ptype = el2.value
el.value = '' # reset to blank string
el2.value = 'NO-SELECTION' # reset to no option chosen
if not pname or ptype not in self._propertyTypes:
raise AttributeError
except AttributeError:
alert('Please give a name and choose a type for the new property to be created.')
return
if ptype == 'relation':
self.chooser.set_action(lambda block:self.create_db_relation(block, pname, ptype))
self.chooser.launch_modal(event, self.data['collection_id'])
else:
self.create_db_property(self.data['collection_id'], pname, ptype)
def create_db_relation(self, block, pname, ptype):
collection_id = block.get_id()
pdata = dict(collection_id=collection_id, property=get_random_pcode())
self.create_db_property(self.data['collection_id'], pname, ptype, pdata=pdata)
def create_db_property(self, collection_id, pname, ptype, pcode=None, pdata={}):
pdata = pdata.copy()
pdata.update(dict(name=pname, type=ptype))
while not pcode or if_no_key_error(self.parent.get_property_schema, pcode):
pcode = get_random_pcode()
data = dict(collection_id=self.data['collection_id'], pcode=pcode, pdata=pdata)
dataLoader.insert('rpc/create_db_property', data, self.report_response)
self.parent.set_property_schema(self.data['collection_id'], pcode, data['pdata'])
self.add_cell(pcode, data['pdata'])
def add_cell(self, pcode, pdata):
b_data = dict(type=self._rowCellTypes.get(pdata['type'], self._rowCellTypes['default']),
attrs=dict(name=pdata['name'], text_f_=[[]], property=pcode))
self.children += add_children(document[self.id], [b_data], self) # add to DOM
def show_er_diagram_h(self, event, templateObj):
if c_id := self.data.get('collection_id', False):
dataLoader.load_data('rpc/get_schemas', lambda req:read_schemas(req, self.id), c_id=c_id)
class RowCellNode(BlockNode):
def save_text(self, srcElement):
pcode = self.data['attrs']['property']
if pcode == 'title':
text, ivals = convert_dom_to_ivals(srcElement)
ivalsDict, edgesIvals = separate_edge_ivals(ivals, 'text')
dataLoader.insert('rpc/update_attrs', dict(b_id=self.parent.get_id(), attrs=dict(text=text),
ivals=ivalsDict, edges_ivals=edgesIvals), self.report_response)
else:
data = dom2ft.get_formatted_text(srcElement)
payload = dict(b_id=self.parent.get_id(), props_data={pcode: data})
dataLoader.insert('rpc/update_block_property', payload, self.report_response)
def key_command_h(self, ev, el):
if ev.key == 'Enter' and ev.altKey: # open this row as a separate page
ev.preventDefault() # block default key action
dataLoader.load_page(self.parent.get_id())
else:
BlockNode.key_command_h(self, ev, el)
def copy(self, **kwargs):
self.parent.copy(**kwargs)
def delete(self):
self.parent.delete()
class RelationCellNode(BlockNode):
_containers = dict(children=dict(replace_id=True, selector='[data-name=ChildrenContainer]'))
_rowCellTypes = dict(default='XRefCell')
def __init__(self, container, nodedata, variables={}, parent=None, uuid=None, **kwargs):
BlockNode.__init__(self, container, nodedata, variables, parent, uuid, **kwargs)
document[self.id].bind('click', self.show_menu)
def makerule(self):
'add clickable links to targets'
l = []
try:
relations = self.parent.data['properties'][self.data['attrs']['property']]
except KeyError:
return []
for page_id, data in relations.items():
data['page_id'] = page_id # required for click action on XRef
l.append(dict(id=page_id, attrs=data, type=self._rowCellTypes['default']))
return l
def show_menu(self, ev):
'launch the menu for adding a new link'
ev.stopPropagation()
menu = menuIndex['relationMenu']
menu.show_menu(ev, dict(link=self.link_modal, new=self.create_page))
def link_modal(self, ev, data):
pschema, pcode = self.get_pschema()
collection_id = pschema['collection_id']
self.chooser.set_action(self.link_block)
self.chooser.launch_modal(ev, collection_id)
def link_block(self, block):
if isinstance(block, str): # just a title for creating a new block
aio.run(self.create_page_async(block))
else:
page_id = block.get_id()
self.save_relation(page_id, dict(attrs=block.data['attrs'].copy()))
def create_page(self, ev, data):
'create a new page in the target TableSchema, and link to it from this property'
aio.run(self.create_page_async())
def get_pschema(self):
pcode = self.data['attrs']['property']
return (self.parent.parent.get_property_schema(pcode), pcode)
async def create_page_async(self, title='Untitled'):
pschema, pcode = self.get_pschema()
blockSaved = aio.Future()
page_id, b_data = self.create_row(pschema['collection_id'], title,
func=lambda x:blockSaved.set_result(self.report_response(x)))
if await blockSaved:
self.save_relation(page_id, b_data)
def save_relation(self, page_id, b_data):
pschema, pcode = self.get_pschema()
self.parent.create_relation(page_id, pcode, pschema['property'])
b_data['id'] = page_id
b_data['type'] = 'XRefCell' # display on this page as this type
b_data['attrs']['page_id'] = page_id # required for click action on XRef
self.children += add_children(document[self.id], [b_data], self, replace_id=True) # add to DOM
class XRefCellNode(XRefBlockNode):
'cell-embedded link to another Conrad page'
def xref_menu_h(self, ev, templateObj):
'launch the menu for adding a new block'
ev.stopPropagation()
menu = menuIndex['xrefMenu']
menu.show_menu(ev, dict(open=self.load_page_h, unlink=self.rm_link_h), data=templateObj)
def rm_link_h(self, ev, templateObj):
'delete this link'
relation = self.parent.data['attrs']['property']
self.delete(b_id=self.parent.parent.get_id(), dest_id=self.data['attrs']['page_id'],
relation=relation, node_id=self.id)
class PropertiesNode(RowBlockNode):
_rowCellTypes = dict(relation='RelationProperty', url='URLProperty2', default='TextProperty2')
_showHiddenProperties = True
class RelationPropertyNode(RelationCellNode):
_rowCellTypes = dict(default='XRefPropLink')
#######################################################################################
# read data from API responses
def read_row_data(req, rowtype='RowBlock'):
'read database row data from API and return as (list of dicts, nextOffset)'
if req.status != 200 and req.status != 206:
alert(f'Error: {req.status} {req.text}')
return
first, last, tablesize = get_pagination_headers(req)
if last is None: # empty result set
return (None, 0)
data = JSON.parse(req.text)
rows = []
for t in data:
r = t['data']
r['id'] = t['id']
r['collection_id'] = t['collection_id']
r['type'] = rowtype
rows.append(r)
return (rows, last + 1)
def read_page(req, historySaver=None):
'read JSON block list from XHR req, and inject into DOM'
if req.status != 200 and req.status != 206:
alert(f'Error: {req.status} {req.text}')
return
blocks = JSON.parse(req.text)
page, d = get_blocktree(blocks)
pageroot = add_blocktree_to_dom(page)
dataLoader.set_page(pageroot) # give dataLoader a handle for manipulating current page
refresh_mathjax()
document['searchresults'].parent.open = False # when new page loaded, hide old search
init_annotation_tool() # (re)load the annotation prototype
if historySaver:
historySaver.pushState()
try:
document.title = page['attrs']['text']
except (KeyError, IndexError):
pass
def get_blocktree(blocks):
'return tree root, and dict of all IDs in the blocks data'
d = {}
schemas = {}
page = None
for b in blocks:
data = b['data']
data['id'] = b['id'] # always provide id and title as simple strings
try:
data['title'] = data['attrs']['text'] or 'Untitled'
except (KeyError, TypeError, IndexError):
data['title'] = 'Untitled'
if data['type'] == 'PageBlock':
if page is None: # first record describes page being loaded
page = data
else: # force subpage to be displayed as XRefBlock
data['type'] = 'XRefBlock'
data['attrs']['page_id'] = b['id']
if data['type'] == 'TableSchema':
for view_id in data['view_blocks']:
schemas[view_id] = data
#print(f'TableSchema {b["id"]}')
else:
d[b['id']] = data
for b in blocks:
try: # connect ViewBlock to its schema
schema = schemas[b['id']]
#print(f'read_page {collection_id, schema}')
d[b['id']]['table_schema'] = schema
d[b['id']].setdefault('attrs', {})['collection_id'] = schema['id']
except KeyError:
pass
try: # add block as child to its parent
d[b['data']['parent_id']].setdefault('children', []).append(d[b['id']])
except KeyError:
pass
return page, d
def add_blocktree_to_dom(page, target='wholebody'):
'turn blocktree data into BlockNodes and inject into DOM'
wholebody = document[target]
wholebody.html = '' # empty the document
root = PageBlockNode(wholebody, page)
add_rescaling(wholebody)
return root
def add_rescaling(container, selector='.rescalable'):
for el in container.select(selector):
el.attrs['title'] = 'Click on this to zoom in at higher magnification, or Shift-Click to zoom out'
el.bind('click', increase_scale)
def get_pagination_headers(req):
'report first, last, tablesize counters for current pagination tranche'
rangeHeader = req.getResponseHeader('Content-Range')
m = re.match(r'(\d+)-(\d+)/(\d+)', rangeHeader)
if not m: # empty result set
return None, None, None
return [int(m.group(i)) for i in (1,2,3)]
class PreventDetailsToggle(object):
'since Chrome vs. Firefox etc behave differently, just override the result after 1 ms'
def __init__(self, el, ms=1):
self.el = el
self.open = el.open
self.t = timer.set_timeout(self.restore, ms)
def restore(self):
self.el.open = self.open
def undo_stupid_toggle(event):
'any click or spacebar inside will toggle -- very bad for contenteditable. Block that.'
if event.type == 'click' or (event.type == 'keyup' and getattr(event, 'keyCode', 0) == 32):
return PreventDetailsToggle(event.currentTarget.closest('details'))
class HistoryAPI(object):
'provides interface between javascript history API and DataLoader'
def __init__(self, loader):
self.loader = loader
window.bind('popstate', self.pop_state)
def pushState(self):
print(f'pushState: {self.loader.get_state()}')
window.history.pushState(self.loader.get_state(), '', '')
def pop_state(self, event):
if event.state:
state = event.state.to_dict()
print(f'pop_state: {state}')
if 'page_id' in state:
self.loader.load_page(state['page_id'], scrollTarget=state.get('scrollTarget', None),
addToHistory=False)
class LocalFileUIBase(object):
'base class for reading contents of local file chosen by '
def __init__(self, asBase64=False):
self.asBase64 = asBase64
self.fut = aio.Future()
def open(self, event):
fileHandle = event.currentTarget.files[0]
self.name = fileHandle.name
reader = window.FileReader.new()
if self.asBase64:
reader.readAsDataURL(fileHandle)
else:
reader.readAsText(fileHandle)
reader.bind('load', self.read)
def read(self, event):
if self.asBase64:
i = event.target.result.index(';base64,') + 8
self.data = event.target.result[i:]
#as_uint8 = window.Uint8Array.new(event.target.result)
#as_bytes = bytes(as_uint8)
#self.data = base64.standard_b64encode(as_bytes).decode('ascii') # bython buggy
self.ech_id = hashlib.sha256(base64.standard_b64decode(self.data)).hexdigest()
else:
self.data = event.target.result
self.fut.set_result(self.data)
async def get_text(self):
return await self.fut
class LocalFileUI(LocalFileUIBase):
'read contents of local file chosen by '
def __init__(self, inputID, asBase64=False):
LocalFileUIBase.__init__(self, asBase64)
self.inputID = inputID
document[inputID].bind('input', self.open)
class DataLoader(object):
'data access via postgrest API'
def __init__(self, jwtoken=None, url='http://localhost:3000/', searchinput='searchinput',
searchmore='searchmore', searchadv='searchjp', addblock='addblock'):
self.url = url
self.updates = {}
self.scrollTarget = None
if searchinput:
document[searchinput].bind('change', self.search_text)
document[searchinput].bind('keyup', lambda ev:undo_stupid_toggle(ev))
if searchmore:
document[searchmore].bind('click', self.search_more)
if searchadv:
document[searchadv].bind('change', lambda ev:self.search_text(ev, 'rpc/search_jsonpath'))
if addblock:
document[addblock].bind('click', self.add_block)
self._searchmore = searchmore
if not jwtoken:
jwtoken = prompt('Please enter your JWT to login:')
self.jwtoken = jwtoken
self.historySaver = HistoryAPI(self)
self.load_data('rpc/set_httponly_auth', lambda req:None, v='this var is ignored');
def get_state(self):
d = dict(page_id=self.page.get_id())
if self.scrollTarget:
d['scrollTarget'] = self.scrollTarget
return d
def set_page(self, blockNode):
self.page = blockNode
def load_page(self, page_id, view='rpc/get_page', scrollTarget=None, addToHistory=True):
'load page with JWT auth'
if not page_id:
print('load_page: ignored empty page_id')
return
self.scrollTarget = scrollTarget
self.flush_queue()
return self.load_data(view, lambda x:self.flush_and_read(x, addToHistory), page_id=page_id, order='data->position')
def flush_and_read(self, req, addToHistory=True):
'ensure that any pending updates are sent before loading new doc'
self.flush_queue()
read_page(req, addToHistory and self.historySaver)
if self.scrollTarget:
try:
target = document[self.scrollTarget]
parent_details_open(target) # make all ancestor details.open true
target.scrollIntoView()
except KeyError:
pass
def prepare_headers(self, headers, offset=0, limit=None):
headers = headers.copy()
headers['Authorization'] = 'Bearer ' + self.jwtoken
if limit:
headers['Range-Unit'] = 'items'
headers['Range'] = f'{offset}-{offset + limit - 1}'
headers['Prefer'] = 'count=estimated'
return headers
def load_data(self, view, reader, headers={}, offset=0, limit=None, **kwargs):
headers = self.prepare_headers(headers, offset, limit)
send_ajax(self.url + view, reader, headers=headers, **kwargs)
def update(self, view, updates, reader, headers={}, **kwargs):
headers = self.prepare_headers(headers)
send_ajax(self.url + view, reader, updates, method='PATCH', headers=headers, **kwargs)
def insert(self, view, data, reader, headers={}):
headers = self.prepare_headers(headers)
send_ajax(self.url + view, reader, data, method='POST', headers=headers)
def queue_update(self, targetID, func, data, ms=10000):
try: # clear any existing timeout for this target, then reset
timer.clear_timeout(self.updates[targetID][0])
except KeyError:
pass
self.updates[targetID] = (timer.set_timeout(self.dequeue, ms, targetID), func, data)
def dequeue(self, targetID):
'runs the scheduled update function and removes it from the updates dict'
try:
func, data = self.updates[targetID][1:]
del self.updates[targetID]
func(data)
except KeyError:
pass
def flush_queue(self):
'send all updates now in the queue, and empty the queue'
for t,func,data in self.updates.values():
timer.clear_timeout(t)
func(data)
self.updates.clear()
def search_text(self, ev, endpoint='rpc/search_text'):
self.query = ev.currentTarget.value
self.search_endpoint = endpoint
self.load_data(self.search_endpoint, self.read_search_results, searchstring=self.query, limit=20)
def search_more(self, ev):
offset = getattr(self, 'nextOffset', 0)
self.load_data(self.search_endpoint, self.read_search_results, searchstring=self.query, offset=offset, limit=20)
def add_block(self, ev):
'add an empty block to current page'
self.page.copy(blockType='ListBlock', empty=True, position=1024*1024)
def read_search_results(self, req, targetID='searchresults', parent=None, blockType=None):
if req.status != 200 and req.status != 206:
alert(f'Error: {req.status} {req.text}')
return False
first, last, tablesize = get_pagination_headers(req)
if last is None: # empty result set
return None
self.nextOffset = last + 1
l = []
for d in JSON.parse(req.text):
data = d['data']
data['id'] = d['id'] # must copy block ID into block data
data['attrs']['scrollTarget'] = d['id'] # for scrolling after page load
if blockType:
data['type'] = blockType
l.append(data)
resetBlank = (first == 0) # if start of list, then empty the results div
if parent:
holder = document[parent.id]
else:
holder = document[targetID]
holder.parent.open = True # make the results visible
if self._searchmore: # show the pagination More button
document[self._searchmore].style.display = 'block'
# avoid subtle bugs by not linking search results to their actual block IDs
return add_children(holder, l, parent, replace_id=True, resetBlank=resetBlank)
################################################################
# AJAX utilities
def send_ajax(url, receiver, data=None, method='GET', headers={}, asyncMode=True,
withCredentials=True, **kwargs):
'only browser.ajax.Ajax supports withCredentials, so we have to use that'
req = ajax.Ajax()
req.bind('complete', receiver)
if kwargs:
url = url + '?' + '&'.join([f'{t[0]}={urllib.parse.quote(t[1])}'
for t in kwargs.items()])
req.open(method, url, asyncMode)
req.withCredentials = withCredentials # no cross-origin cookies/auth work without this!!!!
for h, v in headers.items():
req.set_header(h, v)
if data:
req.set_header('Content-Type', 'application/json')
req.send(JSON.stringify(data))
else:
req.send()
#################################################################
# app initialization
async def start_conrad(jwtUIID='get-jwt'):
global dataLoader
jwtreader = LocalFileUI('jwt-file')
s = await jwtreader.get_text()
data = JSON.parse(s)
jwtkey = data['jwtkey']
page = data.get('page', 'da0a2f81-a1e6-4707-998d-6f06827091da')
# localhost dev server
#dataLoader = DataLoader(jwtkey)
# production server
dataLoader = DataLoader(jwtkey, url='https://api.blindspots.institute/')
dataLoader.load_page(page)
document[jwtUIID].style.display = 'none'
add_rescaling(document['erdiagram-prototype'])
chooser = create_modal_chooser('ModalChooser')
RelationCellNode.chooser = chooser
chooser = create_modal_chooser('SchemaChooser')
RowBlockNode.chooser = chooser
# load an example page
#dataLoader.load_page('487e464f-b30b-433d-a928-a87254bfc6a5')
#dataLoader.load_page('4a2eb6af-646d-4876-94cb-0a6da3dce1ae')
#dataLoader.load_page('80a6d54d-ece3-4a42-80fc-72f7f5cf5c00')
#dataLoader.load_page('2fd47d0c-945d-497e-80f7-464d642d09d6')
#dataLoader.load_page('3ba00a8e-ee8a-4cdf-8353-0e2452c881fd')
#dataLoader.load_page('40cbe866-13ad-4ce1-be62-894d7b3bb6de')
#dataLoader.load_page('7bdb7ce5-cf90-4ded-bb19-8661e2dd52c2')
#dataLoader.load_page('7349fa6c-dc5e-43cd-8c8c-58275e7965a7')
#dataLoader.load_page('b83e275a-4456-4746-b273-ec113a0c976e')
#dataLoader.load_page('56180cc3-6550-40ec-983b-fb55749c00e2')
#dataLoader.load_page('000525a6-6e18-4340-bae9-893910556b51')
#dataLoader.load_page('f895bca8-5b32-4b2d-9e9a-6a1e1b7ad935')
def init_annotation_tool(view_id='1263be77-ede7-432e-b9b7-fd24afc7bbe7', limit=1000,
target='annotation-tool-test', rmTriggerID='annotation-rm-trigger'):
document[target].html = '' # empty the current annotation tool
dataLoader.load_data('rpc/get_dbview_fo5', build_annotation_tool, limit=limit, view_id=view_id,
view_filter='NULL', view_order='NULL', order=f'sortkey1')
document[rmTriggerID].bind('click', rm_annotation_span)
document['er-diagram-button'].bind('click', show_er_diagram)
def build_annotation_tool(req, target='annotation-tool-test', pcode='._fB', triggerID='annotation-tool-trigger'):
rows, nextOffset = read_row_data(req, rowtype='ItemChooser')
l = [dict(id='mini-language-list', type='ListChooser', attrs=dict(text='Mini-Languages'), children=rows),
dict(id='mini-language-words', type='ListChooser', attrs=dict(text='Terms in this mini-language'))]
mlChooser, wordChooser = add_children(document[target], l, addChildrenKwargs=dict(replace_id=True))
mlChooser.set_action(lambda block: set_word_list(block, pcode, wordChooser))
annotationButton = AnnotationButton(triggerID)
wordChooser.set_action(annotationButton.set_dest)
def set_word_list(itemBlock, pcode, targetList):
wordsData = itemBlock.data['properties'].get(pcode, {})
l = []
for block_id, data in wordsData.items():
l.append(dict(id=block_id, type='ItemChooser', attrs=data))
targetList.reload_items(l)
class ListChooserNode(BlockNode):
def reload_items(self, items, sortFunc=lambda x:x['attrs']['text'].lower()):
if sortFunc:
items.sort(key=sortFunc)
self.reload_children(items, replace_id=True)
def set_action(self, func):
self._action_f = func
def run_action(self, block):
return self._action_f(block)
class ItemChooserNode(BlockNode):
def choose_item_h(self, ev, templateObj):
for el in document[self.parent.id].select('.ival-annotation'): # remove any existing item highlighting
remove_class(el, 'ival-annotation')
add_class(ev.currentTarget, 'ival-annotation') # highlight this item as chosen
self.parent._action_f(self)
def show_page_h(self, ev, templateObj):
ev.stopPropagation()
dataLoader.load_page(self.get_id())
class AnnotationButton(object):
def __init__(self, triggerID):
self.triggerID = triggerID
document[triggerID].bind('click', self.create_annotation)
def set_dest(self, block):
self.dest = block
el = document[self.triggerID]
el.parent.select_one('.minilanguage-term').text = self.dest.data['attrs']['text']
el.parent.style.display = 'block'
def create_annotation(self, ev):
title=f"{self.dest.data['attrs']['text']}: double-click to open definition"
insert_annotation_span(self.dest.get_id(), title)
def insert_annotation_span(dest, title='linked to annotation',
err='Please select a single phrase within editable text, before clicking this button.'):
span = browser.html.SPAN(Class='ival-annotation', data_dest_id=dest, data_ival_pred='use', title=title)
embed_selection_in_node(span, err=err)
def embed_selection_in_node(span, func=None,
err='currently only can be applied to selection not already formatted.'):
el, iv = get_simple_selection()
if el and is_contenteditable(el):
s = el.text
prefix, text, suffix = s[:iv[0]], s[iv[0]:iv[1]], s[iv[1]:]
if func:
func(span, text)
else:
span.text = text
el.parent.replaceChild(span, el)
if prefix:
span.before(prefix)
if suffix:
span.after(suffix)
else:
alert(err)
def rm_annotation_span(ev):
'remove the selected annotation'
el = get_selected_annotation() or get_selected_annotation('focusNode')
parent = None
if el:
dest = el.attrs['data-dest-id']
parent = rm_element_keeping_children(el)
if parent:
span = browser.html.SPAN(data_dest_id=dest, data_dest_deleted=True)
parent.appendChild(span) # this ensures update_attr will delete this ival from backend edges_ivals
else:
alert('Please select some text within an annotation you want to remove, before clicking this button.')
def get_selected_annotation(attr='anchorNode', selector='.ival-annotation[data-dest-id]'):
'get the ival span enclosing the current selection endpoint, or None'
target = document.getSelection()
return closest_parent_safe(getattr(target, attr), selector)
def get_nonzero_selection():
'return selection if not empty, else None'
target = document.getSelection()
if target.anchorNode != target.focusNode or target.anchorOffset != target.focusOffset:
return target
def get_simple_selection():
'return selection interval only if contained in a single text element'
target = document.getSelection()
interval = (min(target.anchorOffset, target.focusOffset), max(target.anchorOffset, target.focusOffset))
if target.anchorNode == target.focusNode and target.anchorNode.nodeName == '#text' and interval[0] < interval[1]:
return (target.anchorNode, interval)
else:
return None, None
# er diagram prototype
def show_er_diagram(ev):
dataLoader.load_data('rpc/get_schemas', read_schemas, c_id='a4ddbea8-62ec-4ae4-9e09-0dba9d893500')
def read_schemas(req, targetID='erdiagram-prototype'):
if req.status != 200 and req.status != 206:
alert(f'Error: {req.status} {req.text}')
return
tables = JSON.parse(req.text)
sourceCode = get_mermaid_erdiagram_code(tables)
svg_id = standardize_uuid(get_uuid(sourceCode))
window.render_mermaid(targetID, '[data-name=MermaidDisplay]', sourceCode, svg_id)
def get_mermaid_erdiagram_code(tables):
'produces mermaid source code for ER Diagram for the set of tables'
s = 'erDiagram\n'
for t in tables: # display nodes with attributes
data = t['data']
attrs = [(d['type'], d['name'])
for d in data['schema'].values()
if d['type'] not in ('relation', 'rollup')]
attrs.sort(key=lambda x:x[1].lower()) # order by attr name
s += f'''{t['id']}["{data['attrs']['text']}"] {{
{"\n".join([(a[0] + ' attr "' + a[1] + '"') for a in attrs])}
}}
'''
edges = {}
for t in tables: # collect edges
for pcode,pdata in t['data']['schema'].items():
if pdata['type'] == 'relation':
t_id = min(t['id'], pdata['collection_id'])
e_id = max(t['id'], pdata['collection_id'])
edges.setdefault(t_id, {}).setdefault(e_id, []).append(pdata['name'])
for t_id,edata in edges.items(): # display edges
for e_id, enames in edata.items():
s += f'''{t_id} }}o--o{{ {e_id} : "{' / '.join(enames)}"
'''
return s
def increase_scale(event):
event.preventDefault() # block default action
el = event.currentTarget.children[0]
try:
scale = float(el.attrs['data-scale-factor'])
except KeyError:
scale = 1.0
el.attrs['data-size-initial'] = f'{el.width},{el.height}'
setattr(event.currentTarget.style, 'max-width', f'{event.currentTarget.width}px') # prevent grid-column auto-expansion
if event.shiftKey:
scale /= 1.5
else:
scale *= 1.5
size0 = [float(x) for x in el.attrs['data-size-initial'].split(',')]
el.style.width = f'{scale * size0[0]}' # rescale this element
el.style.height = f'{scale * size0[1]}'
el.attrs['data-scale-factor'] = f'{scale}'
##################################################################
# modal dialog prototype
class ModalChooserNode(BlockNode):
def makerule(self):
return [dict(type='ListChooser', attrs=dict(text='Choose an item'))]
def launch_modal(self, ev, data=None, left=100, attr='collection_id'):
if data:
setattr(self, attr, data)
el = document[self.id]
el.select_one('input').value = '' # start with blank search string
self.query = self.waiting = False
self.emptyQuery = 'iMpOsSiBlE'
self.refilter_search_h()
el.left = left
el.top = int(ev.currentTarget.abs_top)
el.style.display = 'block'
self.show_chooser_xor_message()
def show_chooser_xor_message(self, showResults=True):
'iff results then display chooser else message'
self.set_visibility(showResults, '[data-name=ChildrenContainer]')
self.set_visibility(not showResults, '[data-name=NoResultsMessage]')
def refilter_search_h(self, ev=None, templateObj=None, ms=5000):
el = document[self.id]
query = el.select_one('input').value
if query != self.query and not self.waiting and not query.startswith(self.emptyQuery):
view_filter = (query and f'$ ? (@.attrs.text like_regex "{query}" flag "i")') or 'NULL'
self.load_modal_list(view_filter)
self.query = query
self.waiting = timer.set_timeout(self.refilter_if_needed, ms)
def refilter_if_needed(self):
self.waiting = False
self.refilter_search_h()
def load_modal_list(self, view_filter='NULL', limit=10):
'load the targetList with specified view results'
dataLoader.load_data('rpc/get_dbview5', self.read_modal_list,
limit=limit, collection_id=self.collection_id, view_filter=view_filter, view_order='NULL',
order=f'sortkey1')
def read_modal_list(self, req):
rows, nextOffset = read_row_data(req, rowtype='ItemChooser')
if rows:
self.children[0].reload_items(rows)
else:
self.emptyQuery = self.query
self.show_chooser_xor_message(rows)
self.refilter_if_needed()
def create_row_h(self, ev, templateObj):
el = document[self.id]
title = el.select_one('input').value
self.children[0].run_action(title)
el.style.display = 'none'
def set_action(self, func):
self.children[0].set_action(func)
def create_modal_chooser(blockType, target='modal-container', top_of_doc='wholebody'):
l = [dict(type=blockType, attrs=dict(text='Choose an item'))]
chooser = add_children(document[target], l)[0]
document[top_of_doc].bind('click', chooser.hide_modal_h)
return chooser
class SchemaChooserNode(ModalChooserNode):
def load_modal_list(self, view_filter='$', limit=10):
if view_filter == 'NULL':
view_filter = '$'
dataLoader.load_data('rpc/search_jsonpath', self.read_modal_schemas,
searchstring=f'$ ? (@.type == "TableSchema"){view_filter[1:]}', limit=limit)
def read_modal_schemas(self, req):
rows = dataLoader.read_search_results(req, parent=self.children[0], blockType='ItemChooser')
document[self.id].style.display = 'block'
self.show_chooser_xor_message(rows)
if not rows:
self.emptyQuery = self.query
self.refilter_if_needed()
def create_db_h(self, ev, templateObj):
el = document[self.id]
data = dict(ref_id=self.collection_id, cname=el.select_one('input').value)
data['collection_id'] = self.targetID = standardize_uuid(get_uuid(repr(data))) # ID of new collection
dataLoader.insert('rpc/create_db', data, self.create_relation)
el.style.display = 'none'
def create_relation(self, req):
if self.report_response(req):
self.children[0].run_action(DummyBlock(self.targetID))
class DummyBlock(object):
def __init__(self, block_id):
self.id = block_id
def get_id(self):
return self.id
aio.run(start_conrad())