def mentions_feed(self):
return MentionsFeed(self)
+ def ego_feed(self):
+ return EgoFeed(self)
+
class Feed:
"""Base class that encapsulates some kind of collection of _things_ we
can get from the server, with both the ability to go backwards in
super().__init__(client, "notifications", {"types[]":['mention']},
get=lambda item: item['status'])
+class EgoFeed(IncrementalServerFeed):
+ def __init__(self, client):
+ super().__init__(client, "notifications", {
+ "types[]":['reblog','follow','favourite']})
+
+def parse_creation_time(created_at):
+ date, suffix = created_at.split(".", 1)
+ if suffix.lstrip(string.digits) != "Z":
+ raise ValueError(f"{self.post_id}: bad creation date {date!r}")
+ tm = time.strptime(date, "%Y-%m-%dT%H:%M:%S")
+ return calendar.timegm(tm)
+
class Status:
def __init__(self, data, client):
rb = data.get('reblog')
self.post_id = data['id']
- date, suffix = data['created_at'].split(".", 1)
- if suffix.lstrip(string.digits) != "Z":
- raise ValueError(f"{self.post_id}: bad creation date {date!r}")
- tm = time.strptime(date, "%Y-%m-%dT%H:%M:%S")
- self.datestamp = calendar.timegm(tm)
+ self.datestamp = parse_creation_time(data['created_at'])
self.account = data['account']
yield text.BlankLine()
for media in self.media:
yield text.Media(media['url'], media.get('description'))
+
+class Notification:
+ def __init__(self, data, client):
+ self.ntype = data.get('type')
+ self.account = data['account']
+ self.post_id = data['id']
+ self.datestamp = parse_creation_time(data['created_at'])
+ st = data.get('status')
+ if st is not None:
+ hp = text.HTMLParser()
+ hp.feed(st['content'])
+ hp.done()
+ self.content = hp.paras
+ else:
+ self.content = []
+ self.client = client
+
+ def text(self):
+ yield text.NotificationLog(
+ self.datestamp, self.client.fq(self.account['acct']),
+ self.account['display_name'], self.ntype, self.content)
def run(self):
home_feed = self.home_timeline_feed()
mentions_feed = self.mentions_feed()
+ ego_feed = self.ego_feed()
def extend_both():
home_feed.extend_future()
self, mentions_feed,
text.ColouredString("Mentions [ESC][R]",
"HHHHHHHHHHHHKKKHHKH"))
+ self.ego_timeline = NotificationsFile(
+ self, ego_feed,
+ text.ColouredString("Ego Log [ESC][L][L][E]",
+ "HHHHHHHHHHHKKKHHKHHKHHKH"))
self.add_sigwinch_selfpipe()
self.escape_menu = EscMenu(self)
+ self.log_menu = LogMenu(self)
+ self.log_menu_2 = LogMenu2(self)
self.activity_stack = [self.home_timeline]
sl_rendered = util.exactly_one(sl.render(self.cc.scr_w))
self.cc.print_at(self.cc.scr_h - 1, 0, sl_rendered)
+ def chain_to(self, activity):
+ assert self.cc.activity_stack[-1] is self
+ self.cc.activity_stack[-1] = activity
+
def handle_key(self, ch):
if ch in {ord('q'), ord('Q'), 13}:
return 'quit'
"Utilities [ESC]",
"HHHHHHHHHHHKKKH")
- def chain_to(self, activity):
- assert self.cc.activity_stack[-1] is self
- self.cc.activity_stack[-1] = activity
-
def handle_key(self, ch):
if ch in {ord('r'), ord('R')}:
self.chain_to(self.cc.mentions_timeline)
+ elif ch in {ord('l'), ord('L')}:
+ self.chain_to(self.cc.log_menu)
+ else:
+ super().handle_key(ch)
+
+class LogMenu(Menu):
+ def __init__(self, cc):
+ super().__init__(cc)
+ self.title = text.ColouredString(
+ "Client Logs [ESC][L]",
+ "HHHHHHHHHHHHHKKKHHKH")
+
+ def handle_key(self, ch):
+ if ch in {ord('l'), ord('L')}:
+ self.chain_to(self.cc.log_menu_2)
+ else:
+ super().handle_key(ch)
+
+class LogMenu2(Menu):
+ def __init__(self, cc):
+ super().__init__(cc)
+ self.title = text.ColouredString(
+ "Server Logs [ESC][L][L]",
+ "HHHHHHHHHHHHHKKKHHKHHKH")
+
+ def handle_key(self, ch):
+ if ch in {ord('e'), ord('E')}:
+ self.chain_to(self.cc.ego_timeline)
else:
super().handle_key(ch)
elif ch in {ord('q'), ord('Q')}:
return 'quit'
-class StatusFile(File):
- def __init__(self, cc, feed, title):
+class ObjectFile(File):
+ def __init__(self, cc, constructor, feed, title):
super().__init__(cc)
self.feed = feed
self.feed.start()
self.header = text.FileHeader(title)
+ self.constructor = constructor
self.history_closed = False
self.minpos = self.feed.min_index()
self.maxpos = self.feed.max_index()
- self.statuses = {i: client.Status(self.feed[i], cc)
+ self.statuses = {i: self.constructor(self.feed[i], cc)
for i in range(self.minpos, self.maxpos)}
self.itempos = self.maxpos - 1
new_minpos = self.feed.min_index()
while self.minpos > new_minpos:
self.minpos -= 1
- self.statuses[self.minpos] = client.Status(
+ self.statuses[self.minpos] = self.constructor(
self.feed[self.minpos], self.cc)
got_any = True
new_maxpos = self.feed.max_index()
while self.maxpos < new_maxpos:
- self.statuses[self.maxpos] = client.Status(
+ self.statuses[self.maxpos] = self.constructor(
self.feed[self.maxpos], self.cc)
self.maxpos += 1
got_any = True
def up_line(self): self.move_by(-1)
def goto_top(self): self.move_to(0)
def goto_bottom(self): self.move_to(len(self.lines))
+
+class StatusFile(ObjectFile):
+ def __init__(self, cc, feed, title):
+ super().__init__(cc, client.Status, feed, title)
+
+class NotificationsFile(ObjectFile):
+ def __init__(self, cc, feed, title):
+ super().__init__(cc, client.Notification, feed, title)
import html.parser
import io
import itertools
+import sys
import time
import unittest
import wcwidth
self.space_colours = []
self.unfinished_word = ColouredString('')
- def render(self, width):
+ def render(self, width, laterwidth=None):
+ if laterwidth is None:
+ laterwidth = width
+
# For the moment, greedy algorithm. We can worry about cleverness later
line, space = ColouredString(''), ColouredString('')
for word, space_colour in zip(self.words, self.space_colours):
if line != "" and (line + space + word).width >= width:
yield line
line, space = ColouredString(''), ColouredString('')
+ width = laterwidth
line += space + word
space = ColouredString(' ', space_colour)
# FIXME: wrap explicitly?
yield line
line, space = ColouredString(''), ColouredString('')
+ width = laterwidth
if len(line) != 0 or len(self.words) == 0:
yield line
else:
self.unfinished_word += c
+ def add_para(self, para):
+ self.end_word()
+ self.words.extend(para.words)
+ self.space_colours.extend(para.space_colours)
+
def __repr__(self):
return f"Paragraph({self.words!r}, unfinished={self.unfinished_word!r})"
+class NotificationLog:
+ def __init__(self, timestamp, account, nameline, ntype, cparas):
+ self.timestamp = timestamp
+ self.account = account
+ self.nameline = nameline
+
+ date = time.strftime("%Y-%m-%d %H:%M:%S",
+ time.localtime(self.timestamp))
+ sentence = f"{self.nameline} ({self.account})"
+ want_content = False
+ if ntype == 'reblog':
+ sentence += ' boosted:'
+ want_content = True
+ elif ntype == 'favourite':
+ sentence += ' favourited:'
+ want_content = True
+ elif ntype == 'follow':
+ sentence += ' followed you'
+
+ self.para = Paragraph()
+ self.para.add(ColouredString(date + " " + sentence))
+ if want_content:
+ for cpara in cparas:
+ self.para.add_para(cpara)
+
+ def render(self, width):
+ it = self.para.render(width, max(0, width-2))
+ yield next(it)
+ try:
+ line = next(it)
+ except StopIteration:
+ return
+ line = ColouredString(" ") + line
+ try:
+ next(it)
+ except StopIteration:
+ yield line
+ return
+ line = next(line.split(width-3)) + ColouredString("...")
+ yield line
+
class HTMLParser(html.parser.HTMLParser):
def __init__(self):
super().__init__()
~ian / mastodonochrome.git / commitdiff