530 lines
14 KiB
Python
530 lines
14 KiB
Python
import objgraph, mimetypes
|
|
|
|
from jinja2.exceptions import TemplateNotFound
|
|
from izzylib import Color, class_name, fuzzy_string_match
|
|
from threading import Thread
|
|
|
|
from . import ProtocolHandler, ProtocolRequest, handle_remote_file, list_directory
|
|
from .functions import error, finish_request, finish_request_error, redirect
|
|
|
|
from .. import var
|
|
from ..functions import TimeoutCallback, get_app, run_in_gui_thread
|
|
|
|
|
|
class LocalRequest(ProtocolRequest):
|
|
def ok_or_redirect(self, *message, level='INFO'):
|
|
if redir := self.query.get('redir'):
|
|
return self.redirect(redir, ' '.join(message), level)
|
|
|
|
self.window.notification(' '.join(message), level=level)
|
|
return self.response('ok')
|
|
|
|
|
|
def handle_refresh_account(self, row):
|
|
row.refresh()
|
|
self.ok_or_redirect(f'Refreshed account info: {row.fullhandle}')
|
|
|
|
|
|
def handle_return_account(self, row):
|
|
html = self._render_template('account_row.haml', {'row': row})
|
|
return self.response(html)
|
|
|
|
|
|
Local = ProtocolHandler('pyweb', LocalRequest)
|
|
|
|
|
|
@Local.route('/')
|
|
def home(handler, request):
|
|
context = DotDict(
|
|
links = []
|
|
)
|
|
|
|
with handler.db.session as s:
|
|
for row in s.fetch('links'):
|
|
if bookmark := s.fetch('bookmarks', id=row.bookmarkid):
|
|
context.links.append(bookmark)
|
|
|
|
context.search = s.get_search()
|
|
|
|
return request.page('home', context)
|
|
|
|
|
|
@Local.route('/search-web')
|
|
def search_web(handler, request):
|
|
keyword = request.query.get('keyword')
|
|
|
|
try:
|
|
text = request.query['text']
|
|
|
|
except KeyError:
|
|
request.error('No search text provided', 400)
|
|
|
|
with handler.db.session as s:
|
|
row = s.get_search(keyword)
|
|
|
|
return request.redirect(row.compile(text))
|
|
|
|
|
|
### Bookmarks ###
|
|
@Local.route('/bookmarks')
|
|
def bookmarks_home(handler, request):
|
|
bookmarks = DotDict()
|
|
|
|
with handler.db.session as s:
|
|
for row in s.fetch('bookmarks'):
|
|
category = row.category
|
|
|
|
if category not in bookmarks.keys():
|
|
bookmarks[category] = []
|
|
|
|
row.name = row.name.replace('<', '<').replace('>', '>')
|
|
bookmarks[category].append(row)
|
|
|
|
for category in bookmarks:
|
|
bookmarks[category] = sorted(bookmarks[category], key=lambda x:x['name'])
|
|
|
|
return request.page('bookmarks', dict(bookmarks = {key: bookmarks[key] for key in sorted(bookmarks)}))
|
|
|
|
|
|
@Local.route('/bookmarks/edit/{markid:int}')
|
|
def bookmarks_edit(handler, request, markid):
|
|
with handler.db.session as s:
|
|
if markid:
|
|
if not (row := s.fetch('bookmarks', id=markid).one()):
|
|
return request.error(f'Invalid bookmark ID: {markid}', 404)
|
|
|
|
else:
|
|
row = None
|
|
|
|
context = {
|
|
'bookmark': row,
|
|
'categories': s.get_categories()
|
|
}
|
|
|
|
return request.page('bookmark-edit', context)
|
|
|
|
|
|
@Local.route('/bookmarks/update/{markid:int}')
|
|
def bookmarks_update(handler, request, markid):
|
|
with handler.db.session as s:
|
|
if markid > 0:
|
|
if not (row := s.fetch('bookmarks', id=markid).one()):
|
|
return request.error(f'Invalid bookmark ID: {markid}', 404)
|
|
|
|
s.put_bookmark_row(row, **request.query)
|
|
msg = 'Updated bookmark'
|
|
|
|
else:
|
|
s.put_bookmark(**request.query)
|
|
msg = 'Created new bookmark'
|
|
|
|
return request.redirect('/bookmarks', msg)
|
|
|
|
|
|
@Local.route('/bookmarks/delete/{markid:int}')
|
|
def bookmarks_delete(handler, request, markid):
|
|
with handler.db.session as s:
|
|
if not (row := s.fetch('bookmarks', id=markid).one()):
|
|
return request.error(f'Invalid bookmark ID: {markid}', 404)
|
|
|
|
s.del_bookmark_row(row)
|
|
|
|
return request.redirect('/bookmarks', 'Deleted bookmark')
|
|
|
|
|
|
### Downloads ###
|
|
@Local.route('/downloads')
|
|
def downloads_home(handler, request):
|
|
return request.error('Not implemented yet', 500)
|
|
|
|
|
|
### Extensions ###
|
|
@Local.route('/extensions')
|
|
def extensions_home(handler, request):
|
|
return request.error('Not implemented yet', 500)
|
|
|
|
|
|
### Fediverse ###
|
|
@Local.route('/fediverse')
|
|
def fediverse_home(handler, request):
|
|
accounts = ','.join([str(acct.id) for acct in handler.app.accounts])
|
|
context = DotDict(
|
|
accounts = f'[{accounts}]'
|
|
)
|
|
|
|
return request.page('fediverse', context)
|
|
|
|
|
|
@Local.route('/fediverse/avatar/{acctid:int}')
|
|
def fediverse_avatar(handler, request, acctid):
|
|
try:
|
|
row = handler.app.get_account_by_id(acctid)
|
|
|
|
except IndexError:
|
|
return request.error(f'Account with ID not found: {acctid}', 404)
|
|
|
|
try:
|
|
return request.file(row.avatar)
|
|
|
|
except FileNotFoundError:
|
|
return request.error(f'Avatar for account does not exist: @{row.handle}@{row.domain}', 404)
|
|
|
|
|
|
@Local.route('/fediverse/refresh/{acctid:int}')
|
|
def fediverse_refresh(handler, request, acctid):
|
|
try:
|
|
row = handler.app.get_account_by_id(acctid)
|
|
|
|
except IndexError:
|
|
return request.error(f'Account with ID not found: {acctid}', 404)
|
|
|
|
return request.finish_thread(request.handle_refresh_account, row)
|
|
|
|
|
|
@Local.route('/fediverse/set_active/{acctid:int}')
|
|
def fediverse_set_active(handler, request, acctid):
|
|
try:
|
|
row = handler.app.get_account_by_id(acctid)
|
|
|
|
except IndexError:
|
|
return request.error(f'Account with ID not found: {acctid}', 404)
|
|
|
|
row.set_active()
|
|
return request.ok_or_redirect(f'Account set as active: {row.fullhandle}')
|
|
|
|
|
|
@Local.route('/fediverse/logout/{acctid:int}')
|
|
def fediverse_logout(handler, request, acctid):
|
|
try:
|
|
row = handler.app.get_account_by_id(acctid)
|
|
|
|
except IndexError:
|
|
return request.error(f'Account with ID not found: {acctid}', 404)
|
|
|
|
with handler.db.session as s:
|
|
s.remove_row(row)
|
|
handler.app.accounts.remove(row)
|
|
|
|
return request.ok_or_redirect(f'Logged out of {row.fullhandle}')
|
|
|
|
|
|
@Local.route('/fediverse/acct_info/{acctid:int}')
|
|
def fediverse_acct_info(handler, request, acctid):
|
|
try:
|
|
row = handler.app.get_account_by_id(acctid)
|
|
|
|
except IndexError:
|
|
return request.error(f'Account with ID not found: {acctid}', 404)
|
|
|
|
return request.finish_thread(request.handle_return_account, row)
|
|
|
|
|
|
### Help ###
|
|
@Local.route('/help')
|
|
def help_home(handler, request):
|
|
keybinds = {}
|
|
|
|
for description, keydata in handler.app.keybinds.items():
|
|
keybinds[description] = [bind.replace('<','').replace('>','+') for bind in keydata[1]]
|
|
|
|
return request.page('help', dict(keybinds = keybinds))
|
|
|
|
|
|
### History ###
|
|
@Local.route('/history')
|
|
def history_home(handler, request):
|
|
context = DotDict(
|
|
history = {}
|
|
)
|
|
|
|
with handler.db.session as s:
|
|
for row in s.fetch('history', orderby='domain'):
|
|
if (search_text := request.query.get('text')) and True not in fuzzy_string_match(search_text, row.domain, row.url, row.title):
|
|
continue
|
|
|
|
if row.domain not in context.history:
|
|
context.history[row.domain] = []
|
|
|
|
row.title = row.title.replace('<', '<').replace('>', '>')
|
|
context.history[row.domain].append(row)
|
|
|
|
return request.page('history', context)
|
|
|
|
|
|
@Local.route('/history/delete/{rowid:int}')
|
|
def history_delete(handler, request, rowid):
|
|
with handler.db.session as s:
|
|
s.remove('history', id=rowid)
|
|
|
|
return request.ok_or_redirect('Removed history item')
|
|
|
|
|
|
@Local.route('/history/delete_host/{host}')
|
|
def history_delete_host(handler, request, host):
|
|
with handler.db.session as s:
|
|
s.remove('history', domain=host)
|
|
|
|
return request.ok_or_redirect('Removed history items for', host)
|
|
|
|
|
|
@Local.route('/history/delete_days/{days:int}')
|
|
def history_delete_days(handler, request, days):
|
|
assert days > 0
|
|
|
|
with handler.db.session as s:
|
|
rows = s.del_history_days(days)
|
|
|
|
return request.ok_or_redirect(f'Cleared history {len(rows)} item(s) older than', str(days), 'days' if days > 1 else 'day')
|
|
|
|
|
|
@Local.route('/history/delete_all')
|
|
def history_clear(handler, request):
|
|
with handler.db.session as s:
|
|
s.execute('DELETE FROM history')
|
|
s.cache['history'].clear()
|
|
|
|
return request.ok_or_redirect('History cleared')
|
|
|
|
|
|
### Passwords ###
|
|
@Local.route('/passwords')
|
|
def passwords_home(handler, request):
|
|
domain = request.query.get('domain')
|
|
|
|
with handler.db.session as s:
|
|
if (search_text := request.query.get('text')):
|
|
items = []
|
|
|
|
for row in s.fetch('passwords', domain=domain):
|
|
if True in fuzzy_string_match(search_text, row.domain, row.url, row.username, row.note, row.label):
|
|
items.append(row)
|
|
|
|
elif domain:
|
|
items = s.fetch('passwords', domain=domain).all()
|
|
|
|
else:
|
|
items = s.fetch('passwords').all()
|
|
|
|
return request.page('passwords', {'items': items})
|
|
|
|
|
|
@Local.route('/passwords/edit', '/passwords/edit/{rowid}')
|
|
def passwords_edit(handler, request, rowid=None):
|
|
if not rowid:
|
|
return request.page('password-edit', {'password': None})
|
|
|
|
with handler.db.session as s:
|
|
if not (item := s.fetch('passwords', id=rowid).one()):
|
|
return request.error(f'Cannot find password: {rowid}', 404)
|
|
|
|
return request.page('password-edit', {'password': item})
|
|
|
|
|
|
@Local.route('/passwords/update', '/passwords/update/{rowid}')
|
|
def passwords_update(handler, request, rowid=None):
|
|
query = request.query.copy()
|
|
query.pop('redir', None)
|
|
|
|
print(query.get('label'))
|
|
|
|
if not query.get('label'):
|
|
query['label'] = f'{query.username} @ {query.domain}'
|
|
|
|
print(query.get('label'))
|
|
|
|
for key, value in list(query.items()):
|
|
if isinstance(value, str) and not value:
|
|
del query[key]
|
|
|
|
with handler.db.session as s:
|
|
if not rowid:
|
|
row = s.put_password(**query)
|
|
return request.ok_or_redirect(f'Created new password: {row.label}')
|
|
|
|
if not (row := s.fetch('passwords', id=rowid).one()):
|
|
return request.error(f'Cannot find password: {rowid}', 404)
|
|
|
|
row = s.put_password_row(row, **query)
|
|
|
|
return request.ok_or_redirect(f'updated password: {row.label}')
|
|
|
|
|
|
@Local.route('/passwords/copy/{rowid}')
|
|
def passwords_copy(handler, request, rowid):
|
|
with handler.db.session as s:
|
|
if not (row := s.fetch('passwords', id=rowid).one()):
|
|
return request.error(f'Cannot find password: {rowid}', 404)
|
|
|
|
row.copy_password()
|
|
return request.ok_or_redirect(f'Copied password for 60 seconds')
|
|
|
|
|
|
@Local.route('/passwords/delete/{rowid}')
|
|
def passwords_update(handler, request, rowid):
|
|
with handler.db.session as s:
|
|
try:
|
|
s.remove('passwords', id=rowid)
|
|
return request.ok_or_redirect(f'Deleted password: {label}')
|
|
|
|
except KeyError:
|
|
return request.error(f'Cannot find password: {rowid}', 404)
|
|
|
|
|
|
### Preferences ###
|
|
@Local.route('/preferences')
|
|
def preferences_home(handler, request):
|
|
context = DotDict(
|
|
settings = {
|
|
'font': 'sans undertale',
|
|
'font-size': '14'
|
|
}
|
|
)
|
|
|
|
return request.page('preferences', context)
|
|
|
|
|
|
@Local.route('/preferences/update')
|
|
def preferences_update(handler, request):
|
|
if not len([key for key in request.query.keys() if key != 'redir']):
|
|
return request.error('No key/value pairs provided', 400)
|
|
|
|
with handler.db.session as s:
|
|
for key, value in request.query.items():
|
|
if key == 'redir':
|
|
continue
|
|
|
|
row = s.put_config(key, value)
|
|
logging.verbose(f'Updated config: {row.key} = {row.value}')
|
|
|
|
return request.ok_or_redirect('Updated preferences')
|
|
|
|
|
|
### Search ###
|
|
@Local.route('/search')
|
|
def search_home(handler, request):
|
|
with handler.db.session as s:
|
|
context = {'searches': sorted(s.fetch('search').all(), key=lambda x:x['name'])}
|
|
return request.page('search', context)
|
|
|
|
|
|
@Local.route('/search/edit/{searchid:int}')
|
|
def search_edit(handler, request, searchid):
|
|
context = DotDict(
|
|
search = None
|
|
)
|
|
|
|
if searchid:
|
|
with handler.db.session as s:
|
|
if not (row := s.fetch('search', id=searchid).one()):
|
|
return request.error(f'Invalid search engine ID: {searchid}', 404)
|
|
|
|
context.search = row
|
|
|
|
return request.page('search-edit', context)
|
|
|
|
|
|
@Local.route('/search/update/{searchid:int}')
|
|
def search_update(handler, request, searchid):
|
|
if searchid > 0:
|
|
msg = 'Updated search engine'
|
|
|
|
else:
|
|
msg = 'Created new search engine'
|
|
|
|
with handler.db.session as s:
|
|
if not s.fetch('search', id=searchid):
|
|
return request.error(f'Invalid search engine ID: {searchid}', 404)
|
|
|
|
s.put_search(**request.query)
|
|
|
|
return request.redirect('/search', msg)
|
|
|
|
|
|
@Local.route('/search/delete/{searchid:int}')
|
|
def search_delete(handler, request, searchid):
|
|
with handler.db.session as s:
|
|
if not (row := s.fetch('search', id=searchid).one()):
|
|
return request.error(f'Invalid search engine ID: {searchid}', 404)
|
|
|
|
s.remove_row(row)
|
|
|
|
return request.ok_or_redirect('Removed search:', searchid)
|
|
|
|
|
|
@Local.route('/search/default/{searchid:int}')
|
|
def search_default(handler, request, searchid):
|
|
with handler.db.session as s:
|
|
if not (row := s.fetch('search', id=searchid).one()):
|
|
return request.error(f'Invalid search engine ID: {searchid}', 404)
|
|
|
|
s.put_config('default_search', row.keyword)
|
|
|
|
return request.ok_or_redirect('Set default search')
|
|
|
|
|
|
### Resources ###
|
|
@Local.route('/css/{filename}')
|
|
def css(handler, request, filename):
|
|
context = DotDict(
|
|
primary = Color('#e7a'),
|
|
secondary = Color('#e7a'),
|
|
background = Color('#191919'),
|
|
positive = Color('#ada'),
|
|
negative = Color('#daa'),
|
|
speed = 350
|
|
)
|
|
|
|
return request.template(f'css/{filename}', context)
|
|
|
|
|
|
@Local.route('/js/{filename}')
|
|
def javascript(handler, request, filename):
|
|
with handler.app.path.localweb.join('js', filename).open() as fd:
|
|
return request.response(fd.read(), 'application/javascript')
|
|
|
|
|
|
@Local.route('/font/{name}/{filename}')
|
|
def font(handler, request, name, filename):
|
|
with handler.app.path.localweb.join('fonts', name, filename).open('rb') as fd:
|
|
return request.response(fd.read(), request.mimetype)
|
|
|
|
|
|
@Local.route('/icon/{name}')
|
|
def icon(handler, request, name):
|
|
with handler.app.path.resources.join('icons', name).open('rb') as fd:
|
|
return request.response(fd.read(), 'image/svg+xml')
|
|
|
|
|
|
@Local.route('/debug')
|
|
def debug_home(handler, request):
|
|
return request.page('debug')
|
|
|
|
|
|
@Local.route('/debug/objects')
|
|
def debug_objects(handler, request):
|
|
objects = objgraph.most_common_types(
|
|
limit = int(request.query.get('minimum', 100)),
|
|
shortnames = False
|
|
)
|
|
|
|
context = dict(objects = objects)
|
|
return request.page('debug', context)
|
|
|
|
|
|
@Local.route('/debug/leaking')
|
|
def debug_leaking(handler, request):
|
|
objects = []
|
|
|
|
for obj in objgraph.get_leaking_objects():
|
|
try:
|
|
objects.append((class_name(obj), str(obj)))
|
|
|
|
except:
|
|
pass
|
|
|
|
context = dict(
|
|
objects = objects
|
|
)
|
|
|
|
return request.page('debug', context)
|