diff options
Diffstat (limited to 'mod.py')
| -rwxr-xr-x | mod.py | 279 |
1 files changed, 279 insertions, 0 deletions
@@ -0,0 +1,279 @@ +#!/usr/bin/env python3 + +from itertools import zip_longest +from sys import stdout +import time + +from textual.app import App, ComposeResult +from textual.widgets import DataTable, Header, Footer, TextLog, Static, Input +from textual.containers import Horizontal, Vertical +from textual.screen import Screen +from textual import log +from rich.text import Text +from jinja2.filters import do_filesizeformat + +from fhost import db, File, su, app as fhost_app, in_upload_bl +from modui import * + +fhost_app.app_context().push() + +class NullptrMod(Screen): + BINDINGS = [ + ("q", "quit_app", "Quit"), + ("f1", "filter(1, 'Lookup name:')", "Lookup name"), + ("f2", "filter(2, 'Filter IP address:')", "Filter IP"), + ("f3", "filter(3, 'Filter MIME Type:')", "Filter MIME"), + ("f4", "filter(4, 'Filter extension:')", "Filter Ext."), + ("f5", "refresh", "Refresh"), + ("f6", "filter_clear", "Clear filter"), + ("r", "remove_file(False)", "Remove file"), + ("ctrl+r", "remove_file(True)", "Ban file"), + ("p", "ban_ip(False)", "Ban IP"), + ("ctrl+p", "ban_ip(True)", "Nuke IP"), + ] + + async def action_quit_app(self): + self.mpvw.shutdown() + await self.app.action_quit() + + def action_refresh(self): + ftable = self.query_one("#ftable") + ftable.watch_query(None, None) + + def action_filter_clear(self): + self.query_one("#filter_container").display = False + ftable = self.query_one("#ftable") + ftable.focus() + ftable.query = ftable.base_query + + def action_filter(self, fcol: int, label: str): + self.query_one("#filter_label").update(label) + finput = self.query_one("#filter_input") + self.filter_col = fcol + self.query_one("#filter_container").display = True + finput.focus() + self._refresh_layout() + + if self.current_file: + match fcol: + case 1: finput.value = "" + case 2: finput.value = self.current_file.addr + case 3: finput.value = self.current_file.mime + case 4: finput.value = self.current_file.ext + + def on_input_submitted(self, message: Input.Submitted) -> None: + self.query_one("#filter_container").display = False + ftable = self.query_one("#ftable") + ftable.focus() + + if len(message.value): + match self.filter_col: + case 1: + try: ftable.query = ftable.base_query.filter(File.id == su.debase(message.value)) + except ValueError: pass + case 2: ftable.query = ftable.base_query.filter(File.addr == message.value) + case 3: ftable.query = ftable.base_query.filter(File.mime.like(message.value)) + case 4: ftable.query = ftable.base_query.filter(File.ext.like(message.value)) + else: + ftable.query = ftable.base_query + + def action_remove_file(self, permanent: bool) -> None: + if self.current_file: + self.current_file.delete(permanent) + db.session.commit() + self.mount(Notification(f"{'Banned' if permanent else 'Removed'} file {self.current_file.getname()}")) + self.action_refresh() + + def action_ban_ip(self, nuke: bool) -> None: + if self.current_file: + if not fhost_app.config["FHOST_UPLOAD_BLACKLIST"]: + self.mount(Notification("Failed: FHOST_UPLOAD_BLACKLIST not set!")) + return + else: + if in_upload_bl(self.current_file.addr): + txt = f"{self.current_file.addr} is already banned" + else: + with fhost_app.open_instance_resource(fhost_app.config["FHOST_UPLOAD_BLACKLIST"], "a") as bl: + print(self.current_file.addr.lstrip("::ffff:"), file=bl) + txt = f"Banned {self.current_file.addr}" + + if nuke: + tsize = 0 + trm = 0 + for f in File.query.filter(File.addr == self.current_file.addr): + if f.getpath().is_file(): + tsize += f.size or f.getpath().stat().st_size + trm += 1 + f.delete(True) + db.session.commit() + txt += f", removed {trm} {'files' if trm != 1 else 'file'} totaling {jinja2.filters.do_filesizeformat(tsize, True)}" + self.mount(Notification(txt)) + self._refresh_layout() + ftable = self.query_one("#ftable") + ftable.watch_query(None, None) + + def on_update(self) -> None: + stdout.write("\033[?25l") + stdout.flush() + + def compose(self) -> ComposeResult: + yield Header() + yield Horizontal( + FileTable(id="ftable", zebra_stripes=True), + Vertical( + DataTable(id="finfo", show_header=False), + MpvWidget(id="mpv"), + TextLog(id="ftextlog"), + id="infopane")) + yield Horizontal(Static("Filter:", id="filter_label"), Input(id="filter_input"), id="filter_container") + yield Footer() + + def on_mount(self) -> None: + self.current_file = None + + self.ftable = self.query_one("#ftable") + self.ftable.focus() + + self.finfo = self.query_one("#finfo") + self.finfo.add_columns("key", "value") + + self.mpvw = self.query_one("#mpv") + self.ftlog = self.query_one("#ftextlog") + + self.mimehandler = mime.MIMEHandler() + self.mimehandler.register(mime.MIMECategory.Archive, self.handle_libarchive) + self.mimehandler.register(mime.MIMECategory.Text, self.handle_text) + self.mimehandler.register(mime.MIMECategory.AV, self.handle_mpv) + self.mimehandler.register(mime.MIMECategory.Document, self.handle_mupdf) + self.mimehandler.register(mime.MIMECategory.Fallback, self.handle_libarchive) + self.mimehandler.register(mime.MIMECategory.Fallback, self.handle_mpv) + self.mimehandler.register(mime.MIMECategory.Fallback, self.handle_raw) + + def handle_libarchive(self, cat): + import libarchive + with libarchive.file_reader(str(self.current_file.getpath())) as a: + self.ftlog.write("\n".join(e.path for e in a)) + return True + + def handle_text(self, cat): + with open(self.current_file.getpath(), "r") as sf: + data = sf.read(1000000).replace("\033","") + self.ftlog.write(data) + return True + + def handle_mupdf(self, cat): + import fitz + with fitz.open(self.current_file.getpath(), + filetype=self.current_file.ext.lstrip(".")) as doc: + p = doc.load_page(0) + pix = p.get_pixmap(dpi=72) + imgdata = pix.tobytes("ppm").hex() + + self.mpvw.styles.height = "40%" + self.mpvw.start_mpv("hex://" + imgdata, 0) + + self.ftlog.write(Text.from_markup(f"[bold]Pages:[/bold] {doc.page_count}")) + self.ftlog.write(Text.from_markup("[bold]Metadata:[/bold]")) + for k, v in doc.metadata.items(): + self.ftlog.write(Text.from_markup(f" [bold]{k}:[/bold] {v}")) + toc = doc.get_toc() + if len(toc): + self.ftlog.write(Text.from_markup("[bold]TOC:[/bold]")) + for lvl, title, page in toc: + self.ftlog.write(f"{' ' * lvl} {page}: {title}") + return True + + def handle_mpv(self, cat): + if cat == mime.MIMECategory.AV or self.current_file.nsfw_score >= 0: + self.mpvw.styles.height = "20%" + self.mpvw.start_mpv(str(self.current_file.getpath()), 0) + + import av + with av.open(str(self.current_file.getpath())) as c: + self.ftlog.write(Text("Format:", style="bold")) + self.ftlog.write(f" {c.format.long_name}") + if len(c.metadata): + self.ftlog.write(Text("Metadata:", style="bold")) + for k, v in c.metadata.items(): + self.ftlog.write(f" {k}: {v}") + for s in c.streams: + self.ftlog.write(Text(f"Stream {s.index}:", style="bold")) + self.ftlog.write(f" Type: {s.type}") + if s.base_rate: + self.ftlog.write(f" Frame rate: {s.base_rate}") + if len(s.metadata): + self.ftlog.write(Text(" Metadata:", style="bold")) + for k, v in s.metadata.items(): + self.ftlog.write(f" {k}: {v}") + return True + return False + + def handle_raw(self, cat): + def hexdump(binf, length): + def fmt(s): + if isinstance(s, str): + c = chr(int(s, 16)) + else: + c = chr(s) + s = c + if c.isalpha(): return f"\0[chartreuse1]{s}\0[/chartreuse1]" + if c.isdigit(): return f"\0[gold1]{s}\0[/gold1]" + if not c.isprintable(): + g = "grey50" if c == "\0" else "cadet_blue" + return f"\0[{g}]{s if len(s) == 2 else '.'}\0[/{g}]" + return s + return Text.from_markup("\n".join(f"{' '.join(map(fmt, map(''.join, zip(*[iter(c.hex())] * 2))))}" + f"{' ' * (16 - len(c))}" + f" {''.join(map(fmt, c))}" + for c in map(lambda x: bytes([n for n in x if n != None]), + zip_longest(*[iter(binf.read(min(length, 16 * 10)))] * 16)))) + + with open(self.current_file.getpath(), "rb") as binf: + self.ftlog.write(hexdump(binf, self.current_file.size)) + if self.current_file.size > 16*10*2: + binf.seek(self.current_file.size-16*10) + self.ftlog.write(" [...] ".center(64, '─')) + self.ftlog.write(hexdump(binf, self.current_file.size - binf.tell())) + + return True + + def on_file_table_selected(self, message: FileTable.Selected) -> None: + f = message.file + self.current_file = f + self.finfo.clear() + self.finfo.add_rows([ + ("ID:", str(f.id)), + ("File name:", f.getname()), + ("URL:", f.geturl() if fhost_app.config["SERVER_NAME"] else "⚠ Set SERVER_NAME in config.py to display"), + ("File size:", do_filesizeformat(f.size, True)), + ("MIME type:", f.mime), + ("SHA256 checksum:", f.sha256), + ("Uploaded by:", Text(f.addr)), + ("Management token:", f.mgmt_token), + ("Secret:", f.secret), + ("Is NSFW:", ("Yes" if f.is_nsfw else "No") + f" (Score: {f.nsfw_score:0.4f})"), + ("Is banned:", "Yes" if f.removed else "No"), + ("Expires:", time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(File.get_expiration(f.expiration, f.size)/1000))) + ]) + + self.mpvw.stop_mpv(True) + self.ftlog.remove() + self.query_one("#infopane").mount(TextLog(id="ftextlog")) + self.ftlog = self.query_one("#ftextlog") + + if f.getpath().is_file(): + self.mimehandler.handle(f.mime, f.ext) + self.ftlog.scroll_home(animate=False) + +class NullptrModApp(App): + CSS_PATH = "mod.css" + + def on_mount(self) -> None: + self.title = "0x0 File Moderation Interface" + self.main_screen = NullptrMod() + self.install_screen(self.main_screen, name="main") + self.push_screen("main") + +if __name__ == "__main__": + app = NullptrModApp() + app.run() |