commit 6110e268cf603a82ce7b37c0337a9fe976c46b2c
parent 9cf2ace7cb33ff1770cdc05259b89ff833c8cc62
Author: luke8086 <55237178+luke8086@users.noreply.github.com>
Date: Tue, 2 Aug 2022 13:57:51 +0000
Rearrange functions for easier navigation
Diffstat:
| M | retronews.py | | | 485 | +++++++++++++++++++++++++++++++++++++++---------------------------------------- |
1 file changed, 242 insertions(+), 243 deletions(-)
diff --git a/retronews.py b/retronews.py
@@ -548,50 +548,158 @@ def db_load_starred_thread_ids(db: DB, page: int = 1) -> list[str]:
return [row["thread_id"] for row in db.execute(sql, (page_size, offset))]
-def app_safe_run(app: AppState, fn: Callable[[], T], flash: Optional[str]) -> Optional[T]:
- if flash is not None:
- app_show_flash(app, flash)
+def msg_flatten_thread(msg: Message, prefix: str = "", is_last_child: bool = False) -> Generator[Message, None, None]:
+ msg.index_tree = "" if msg.is_thread else f"{prefix}{'└─' if is_last_child else '├─'}> "
+ yield msg
- ret = None
+ child_count = len(msg.children)
+ child_prefix = "" if msg.is_thread else f"{prefix}{' ' if is_last_child else '│ '}"
- try:
- ret = fn()
- except Exception as e:
- app_show_flash(app, f"Error: {e}")
- else:
- if flash is not None:
- app_show_flash(app, None)
+ for i, child_node in enumerate(msg.children):
+ child_is_last = i == child_count - 1
+ for child in msg_flatten_thread(child_node, prefix=child_prefix, is_last_child=child_is_last):
+ yield child
- return ret
+def msg_build_lines(msg: Message) -> list[str]:
+ lines = [
+ f"Content-Location: {msg.content_location}",
+ f"Date: {msg.date.strftime('%Y-%m-%d %H:%M')}",
+ f"From: {msg.author}",
+ f"Subject: {msg.title}",
+ "",
+ ]
-def app_show_help_screen(app: AppState) -> None:
- app.screen.erase()
- app.screen.addstr(0, 0, HELP_SCREEN)
- app.screen.refresh()
- app.screen.getch()
+ text = parse_html(msg.body or "")
+ for p in text.split("\n"):
+ lines += wrap_paragraph(p)
-def app_show_flash(app: AppState, flash: Optional[str]) -> None:
- app.flash = flash
- app_render(app)
+ return lines
-def app_prompt(app: AppState, prompt: str) -> str:
- lt = app.layout
+def msg_unload(msg: Message) -> Message:
+ msg.children = []
+ msg.body = None
+ return msg
- app.screen.insstr(lt.flash_menu_row, 0, prompt.ljust(lt.cols))
- app.screen.refresh()
- curses.curs_set(1)
- win = curses.newwin(1, lt.cols - len(prompt), lt.flash_menu_row, len(prompt))
+def hn_parse_search_hit(hit: HNSearchHit) -> Message:
+ return Message(
+ msg_id=f"{hit['objectID']}@hn",
+ thread_id=f"{hit['objectID']}@hn",
+ content_location=f"https://news.ycombinator.com/item?id={hit['objectID']}",
+ date=datetime.fromtimestamp(hit["created_at_i"]),
+ author=hit["author"],
+ title=html.unescape(hit["title"]),
+ total_comments=(hit["num_comments"] or 0) + 1,
+ )
- textbox = curses.textpad.Textbox(win)
- textbox.stripspaces = True
- ret = textbox.edit().strip()
- del win
- curses.curs_set(0)
+def hn_parse_entry(entry: HNEntry, thread_id: str = "", parent_title: str = "") -> Message:
+ thread_id = thread_id or str(entry["id"])
+
+ my_title = html.unescape(entry["title"]) if entry["title"] else None
+
+ body = f"<p>{entry['url']}</p>" if entry["url"] else ""
+ body = f"{body}{entry['text']}" if entry["text"] else body
+
+ return Message(
+ msg_id=f"{entry['id']}@hn",
+ thread_id=f"{thread_id}@hn",
+ content_location=f"https://news.ycombinator.com/item?id={entry['id']}",
+ date=datetime.fromtimestamp(entry["created_at_i"]),
+ author=entry["author"] or "unknown",
+ title=my_title or f"Re: {parent_title}",
+ body=body,
+ children=[hn_parse_entry(child, thread_id, my_title or parent_title) for child in entry["children"]],
+ )
+
+
+def hn_fetch_threads_by_id(thread_ids: list[str]) -> list[Message]:
+ story_tags = ",".join(f"story_{x}" for x in thread_ids)
+ url = f"https://hn.algolia.com/api/v1/search_by_date?hitsPerPage={len(thread_ids)}&tags=story,({story_tags})"
+ hits = json.loads(fetch(url))["hits"]
+
+ return [hn_parse_search_hit(hit) for hit in hits]
+
+
+def hn_fetch_threads(group: str = "news", page: int = 1) -> list[Message]:
+ rex = re.compile(r'href="item\?id=(\d+)"')
+
+ html = fetch(f"https://news.ycombinator.com/{group}?p={page}")
+ thread_ids = list(set(match.group(1) for match in rex.finditer(html)))
+
+ return hn_fetch_threads_by_id(thread_ids)
+
+
+def hn_fetch_new_threads(page: int = 1) -> list[Message]:
+ url = f"https://hn.algolia.com/api/v1/search_by_date?tags=story&hitsPerPage=30&page={page}"
+ hits = json.loads(fetch(url))["hits"]
+
+ return [hn_parse_search_hit(hit) for hit in hits]
+
+
+def hn_fetch_thread(entry_id: Union[str, int]) -> Message:
+ resp = fetch(f"http://hn.algolia.com/api/v1/items/{entry_id}")
+ entry: HNEntry = json.loads(resp)
+ return hn_parse_entry(entry)
+
+
+def group_set_page(group: Group, page: int) -> Group:
+ return dataclasses.replace(group, page=page)
+
+
+def group_advance_page(group: Group, offset: int = 1) -> Group:
+ return group_set_page(group, page=max(1, group.page + offset))
+
+
+def group_fetch_starred_threads(db: DB, page: int = 1) -> list[Message]:
+ thread_ids = db_load_starred_thread_ids(db, page)
+ threads_by_provider: dict[str, list[str]] = {}
+ threads = []
+
+ for (source_id, provider) in (t.split("@") for t in thread_ids):
+ threads_by_provider.setdefault(provider, list()).append(source_id)
+
+ for provider, thread_ids in threads_by_provider.items():
+ if provider == "hn":
+ threads += hn_fetch_threads_by_id(thread_ids)
+
+ threads.sort(key=lambda x: x.date, reverse=True)
+
+ return threads
+
+
+def group_fetch_threads(group: Group, db: DB) -> list[Message]:
+ if group.provider == "hn":
+ return hn_fetch_threads(group.name, group.page)
+ elif group.provider == "hn-new":
+ return hn_fetch_new_threads(group.page)
+ elif group.provider == "starred":
+ return group_fetch_starred_threads(db, group.page)
+ else:
+ return []
+
+
+def group_fetch_thread(thread_id: str) -> Message:
+ (source_id, provider) = thread_id.split("@")
+ return {"hn": hn_fetch_thread}[provider](source_id)
+
+
+def app_safe_run(app: AppState, fn: Callable[[], T], flash: Optional[str]) -> Optional[T]:
+ if flash is not None:
+ app_show_flash(app, flash)
+
+ ret = None
+
+ try:
+ ret = fn()
+ except Exception as e:
+ app_show_flash(app, f"Error: {e}")
+ else:
+ if flash is not None:
+ app_show_flash(app, None)
return ret
@@ -677,53 +785,54 @@ def app_open_thread(app: AppState, thread_message: Message) -> None:
app_load_messages(app, messages, selected_message_id=thread_message.msg_id, show_pager=True)
-def app_get_pager_line_attr(app: AppState, line: str) -> int:
- if line.startswith("Content-Location: "):
- return app.colors.tree
- elif line.startswith("Date: "):
- return app.colors.date
- elif line.startswith("From: "):
- return app.colors.author
- elif line.startswith("Subject: "):
- return app.colors.subject
- elif line.startswith(">>") or line.startswith("> >"):
- return app.colors.nested_quote
- elif line.startswith(">"):
- return app.colors.quote
- elif line.startswith(" "):
- return app.colors.code
- elif line == "~":
- return app.colors.empty_pager_line
- else:
- return 0
+def app_update_layout(app: AppState) -> None:
+ lt = app.layout
+ (lt.lines, lt.cols) = app.screen.getmaxyx()
-def app_render_pager_line(app: AppState, row: int, line: str) -> None:
- line_attr = app_get_pager_line_attr(app, line)
+ if lt.lines < 25 or lt.cols < 80:
+ raise Exception("At least 80x25 terminal is required")
- app.screen.move(row, 0)
- app.screen.clrtoeol()
- app.screen.move(row, 0)
+ max_index_height = lt.lines - 3
+ lt.index_height = (max_index_height // 3) if app.pager_visible else max_index_height
- for word in line.split(" "):
- is_url = word.startswith("http://") or word.startswith("https://")
- word_attr = app.colors.url if is_url and line_attr == 0 else line_attr
- app.screen.addstr(word, word_attr)
- app.screen.addstr(" ")
+ lt.middle_menu_row = lt.index_start + lt.index_height if app.pager_visible else None
+ lt.pager_start = lt.index_start + lt.index_height + 1 if app.pager_visible else None
+ lt.pager_height = lt.lines - lt.pager_start - 2 if lt.pager_start is not None else None
+ lt.bottom_menu_row = lt.lines - 2
+ lt.flash_menu_row = lt.lines - 1
-def app_render_pager(app: AppState) -> None:
- message = app.selected_message
- start = app.layout.pager_start
- height = app.layout.pager_height
- if message is None or start is None or height is None:
- return
+def app_show_help_screen(app: AppState) -> None:
+ app.screen.erase()
+ app.screen.addstr(0, 0, HELP_SCREEN)
+ app.screen.refresh()
+ app.screen.getch()
- for i in range(height):
- line = list_get(message.lines, i + app.pager_offset)
- line = "~" if line is None else line
- app_render_pager_line(app, i + start, line)
+
+def app_show_flash(app: AppState, flash: Optional[str]) -> None:
+ app.flash = flash
+ app_render(app)
+
+
+def app_prompt(app: AppState, prompt: str) -> str:
+ lt = app.layout
+
+ app.screen.insstr(lt.flash_menu_row, 0, prompt.ljust(lt.cols))
+ app.screen.refresh()
+
+ curses.curs_set(1)
+ win = curses.newwin(1, lt.cols - len(prompt), lt.flash_menu_row, len(prompt))
+
+ textbox = curses.textpad.Textbox(win)
+ textbox.stripspaces = True
+ ret = textbox.edit().strip()
+
+ del win
+ curses.curs_set(0)
+
+ return ret
def app_render_index_row(app: AppState, row: int, message: Message) -> None:
@@ -769,6 +878,55 @@ def app_render_index(app: AppState) -> None:
app_render_index_row(app, app.layout.index_start + i, app.messages[i + offset])
+def app_get_pager_line_attr(app: AppState, line: str) -> int:
+ if line.startswith("Content-Location: "):
+ return app.colors.tree
+ elif line.startswith("Date: "):
+ return app.colors.date
+ elif line.startswith("From: "):
+ return app.colors.author
+ elif line.startswith("Subject: "):
+ return app.colors.subject
+ elif line.startswith(">>") or line.startswith("> >"):
+ return app.colors.nested_quote
+ elif line.startswith(">"):
+ return app.colors.quote
+ elif line.startswith(" "):
+ return app.colors.code
+ elif line == "~":
+ return app.colors.empty_pager_line
+ else:
+ return 0
+
+
+def app_render_pager_line(app: AppState, row: int, line: str) -> None:
+ line_attr = app_get_pager_line_attr(app, line)
+
+ app.screen.move(row, 0)
+ app.screen.clrtoeol()
+ app.screen.move(row, 0)
+
+ for word in line.split(" "):
+ is_url = word.startswith("http://") or word.startswith("https://")
+ word_attr = app.colors.url if is_url and line_attr == 0 else line_attr
+ app.screen.addstr(word, word_attr)
+ app.screen.addstr(" ")
+
+
+def app_render_pager(app: AppState) -> None:
+ message = app.selected_message
+ start = app.layout.pager_start
+ height = app.layout.pager_height
+
+ if message is None or start is None or height is None:
+ return
+
+ for i in range(height):
+ line = list_get(message.lines, i + app.pager_offset)
+ line = "~" if line is None else line
+ app_render_pager_line(app, i + start, line)
+
+
def app_render_top_menu(app: AppState) -> None:
lt = app.layout
cols = lt.cols
@@ -813,25 +971,6 @@ def app_render_bottom_menu(app: AppState) -> None:
app.screen.insstr(lt.bottom_menu_row, lt.cols - len(page_text), page_text, app.colors.menu | curses.A_BOLD)
-def app_update_layout(app: AppState) -> None:
- lt = app.layout
-
- (lt.lines, lt.cols) = app.screen.getmaxyx()
-
- if lt.lines < 25 or lt.cols < 80:
- raise Exception("At least 80x25 terminal is required")
-
- max_index_height = lt.lines - 3
- lt.index_height = (max_index_height // 3) if app.pager_visible else max_index_height
-
- lt.middle_menu_row = lt.index_start + lt.index_height if app.pager_visible else None
- lt.pager_start = lt.index_start + lt.index_height + 1 if app.pager_visible else None
- lt.pager_height = lt.lines - lt.pager_start - 2 if lt.pager_start is not None else None
-
- lt.bottom_menu_row = lt.lines - 2
- lt.flash_menu_row = lt.lines - 1
-
-
def app_render(app: AppState) -> None:
app_update_layout(app)
app.screen.erase()
@@ -856,150 +995,17 @@ def app_init(screen: Window, db: DB) -> AppState:
return app
-def msg_flatten_thread(msg: Message, prefix: str = "", is_last_child: bool = False) -> Generator[Message, None, None]:
- msg.index_tree = "" if msg.is_thread else f"{prefix}{'└─' if is_last_child else '├─'}> "
- yield msg
-
- child_count = len(msg.children)
- child_prefix = "" if msg.is_thread else f"{prefix}{' ' if is_last_child else '│ '}"
-
- for i, child_node in enumerate(msg.children):
- child_is_last = i == child_count - 1
- for child in msg_flatten_thread(child_node, prefix=child_prefix, is_last_child=child_is_last):
- yield child
-
-
-def msg_build_lines(msg: Message) -> list[str]:
- lines = [
- f"Content-Location: {msg.content_location}",
- f"Date: {msg.date.strftime('%Y-%m-%d %H:%M')}",
- f"From: {msg.author}",
- f"Subject: {msg.title}",
- "",
- ]
-
- text = parse_html(msg.body or "")
-
- for p in text.split("\n"):
- lines += wrap_paragraph(p)
-
- return lines
-
-
-def msg_unload(msg: Message) -> Message:
- msg.children = []
- msg.body = None
- return msg
-
-
-def hn_parse_search_hit(hit: HNSearchHit) -> Message:
- return Message(
- msg_id=f"{hit['objectID']}@hn",
- thread_id=f"{hit['objectID']}@hn",
- content_location=f"https://news.ycombinator.com/item?id={hit['objectID']}",
- date=datetime.fromtimestamp(hit["created_at_i"]),
- author=hit["author"],
- title=html.unescape(hit["title"]),
- total_comments=(hit["num_comments"] or 0) + 1,
- )
-
-
-def hn_parse_entry(entry: HNEntry, thread_id: str = "", parent_title: str = "") -> Message:
- thread_id = thread_id or str(entry["id"])
-
- my_title = html.unescape(entry["title"]) if entry["title"] else None
-
- body = f"<p>{entry['url']}</p>" if entry["url"] else ""
- body = f"{body}{entry['text']}" if entry["text"] else body
-
- return Message(
- msg_id=f"{entry['id']}@hn",
- thread_id=f"{thread_id}@hn",
- content_location=f"https://news.ycombinator.com/item?id={entry['id']}",
- date=datetime.fromtimestamp(entry["created_at_i"]),
- author=entry["author"] or "unknown",
- title=my_title or f"Re: {parent_title}",
- body=body,
- children=[hn_parse_entry(child, thread_id, my_title or parent_title) for child in entry["children"]],
- )
-
-
-def hn_fetch_threads_by_id(thread_ids: list[str]) -> list[Message]:
- story_tags = ",".join(f"story_{x}" for x in thread_ids)
- url = f"https://hn.algolia.com/api/v1/search_by_date?hitsPerPage={len(thread_ids)}&tags=story,({story_tags})"
- hits = json.loads(fetch(url))["hits"]
-
- return [hn_parse_search_hit(hit) for hit in hits]
-
-
-def hn_fetch_threads(group: str = "news", page: int = 1) -> list[Message]:
- rex = re.compile(r'href="item\?id=(\d+)"')
-
- html = fetch(f"https://news.ycombinator.com/{group}?p={page}")
- thread_ids = list(set(match.group(1) for match in rex.finditer(html)))
-
- return hn_fetch_threads_by_id(thread_ids)
-
-
-def hn_fetch_new_threads(page: int = 1) -> list[Message]:
- url = f"https://hn.algolia.com/api/v1/search_by_date?tags=story&hitsPerPage=30&page={page}"
- hits = json.loads(fetch(url))["hits"]
-
- return [hn_parse_search_hit(hit) for hit in hits]
-
-
-def hn_fetch_thread(entry_id: Union[str, int]) -> Message:
- resp = fetch(f"http://hn.algolia.com/api/v1/items/{entry_id}")
- entry: HNEntry = json.loads(resp)
- return hn_parse_entry(entry)
-
-
-def group_set_page(group: Group, page: int) -> Group:
- return dataclasses.replace(group, page=page)
-
-
-def group_advance_page(group: Group, offset: int = 1) -> Group:
- return group_set_page(group, page=max(1, group.page + offset))
-
-
-def group_fetch_starred_threads(db: DB, page: int = 1) -> list[Message]:
- thread_ids = db_load_starred_thread_ids(db, page)
- threads_by_provider: dict[str, list[str]] = {}
- threads = []
-
- for (source_id, provider) in (t.split("@") for t in thread_ids):
- threads_by_provider.setdefault(provider, list()).append(source_id)
-
- for provider, thread_ids in threads_by_provider.items():
- if provider == "hn":
- threads += hn_fetch_threads_by_id(thread_ids)
-
- threads.sort(key=lambda x: x.date, reverse=True)
-
- return threads
-
-
-def group_fetch_threads(group: Group, db: DB) -> list[Message]:
- if group.provider == "hn":
- return hn_fetch_threads(group.name, group.page)
- elif group.provider == "hn-new":
- return hn_fetch_new_threads(group.page)
- elif group.provider == "starred":
- return group_fetch_starred_threads(db, group.page)
- else:
- return []
-
-
-def group_fetch_thread(thread_id: str) -> Message:
- (source_id, provider) = thread_id.split("@")
- return {"hn": hn_fetch_thread}[provider](source_id)
-
+def app_main(screen: Window, db: DB) -> None:
+ app = app_init(screen, db)
-def argparse_formatter_class(prog):
- return argparse.ArgumentDefaultsHelpFormatter(prog, max_help_position=32)
+ while True:
+ app_render(app)
+ app.flash = ""
+ c = app.screen.getch()
+ KEY_BINDINGS.get(c, cmd_unknown)(app)
-def logging_init(path: Optional[str]) -> None:
+def setup_logging(path: Optional[str]) -> None:
if path is None:
return logging.disable()
@@ -1009,23 +1015,16 @@ def logging_init(path: Optional[str]) -> None:
logging.debug("Session started")
-def main(screen: Window, db: DB) -> None:
- app = app_init(screen, db)
-
- while True:
- app_render(app)
- app.flash = ""
- c = app.screen.getch()
- KEY_BINDINGS.get(c, cmd_unknown)(app)
-
-
if __name__ == "__main__":
- ap = argparse.ArgumentParser(formatter_class=argparse_formatter_class)
+ ap = argparse.ArgumentParser(
+ formatter_class=lambda prog: argparse.ArgumentDefaultsHelpFormatter(prog, max_help_position=32)
+ )
ap.add_argument("-d", "--db", metavar="PATH", default="~/.retronews.db", help="database path")
ap.add_argument("-l", "--logfile", metavar="PATH", default=None, help="debug logfile path")
args = ap.parse_args()
- logging_init(args.logfile)
+ setup_logging(args.logfile)
+
db = db_init(args.db)
- curses.wrapper(main, db)
+ curses.wrapper(app_main, db)
