codekingpro/portable-devtools

1# Urwid web (CGI/Asynchronous Javascript) display module2#    Copyright (C) 2004-2007  Ian Ward3#4#    This library is free software; you can redistribute it and/or5#    modify it under the terms of the GNU Lesser General Public6#    License as published by the Free Software Foundation; either7#    version 2.1 of the License, or (at your option) any later version.8#9#    This library is distributed in the hope that it will be useful,10#    but WITHOUT ANY WARRANTY; without even the implied warranty of11#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU12#    Lesser General Public License for more details.13#14#    You should have received a copy of the GNU Lesser General Public15#    License along with this library; if not, write to the Free Software16#    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA17#18# Urwid web site: https://urwid.org/19 20 21"""22Urwid web application display module23"""24 25from __future__ import annotations26 27import dataclasses28import glob29import html30import os31import pathlib32import random33import selectors34import signal35import socket36import string37import sys38import tempfile39import typing40from contextlib import suppress41 42from urwid.str_util import calc_text_pos, calc_width, move_next_char43from urwid.util import StoppingContext, get_encoding44 45from .common import BaseScreen46 47if typing.TYPE_CHECKING:48    from typing_extensions import Literal49 50    from urwid import Canvas51 52TEMP_DIR = tempfile.gettempdir()53CURRENT_DIR = pathlib.Path(__file__).parent54 55_js_code = CURRENT_DIR.joinpath("_web.js").read_text("utf-8")56 57ALARM_DELAY = 6058POLL_CONNECT = 359MAX_COLS = 20060MAX_ROWS = 10061MAX_READ = 409662BUF_SZ = 1638463 64_code_colours = {65    "black": "0",66    "dark red": "1",67    "dark green": "2",68    "brown": "3",69    "dark blue": "4",70    "dark magenta": "5",71    "dark cyan": "6",72    "light gray": "7",73    "dark gray": "8",74    "light red": "9",75    "light green": "A",76    "yellow": "B",77    "light blue": "C",78    "light magenta": "D",79    "light cyan": "E",80    "white": "F",81}82 83# replace control characters with ?'s84_trans_table = "?" * 32 + "".join([chr(x) for x in range(32, 256)])85 86_css_style = CURRENT_DIR.joinpath("_web.css").read_text("utf-8")87 88# HTML Initial Page89_html_page = [90    """<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN"91 "http://www.w3.org/TR/html4/loose.dtd">92<html>93<head>94<title>Urwid Web Display - """,95    """</title>96<style type="text/css">97"""98    + _css_style99    + r"""100</style>101</head>102<body id="body" onload="load_web_display()">103<div style="position:absolute; visibility:hidden;">104<br id="br"\>105<pre>The quick brown fox jumps over the lazy dog.<span id="testchar">X</span>106<span id="testchar2">Y</span></pre>107</div>108Urwid Web Display - <b>""",109    """</b> -110Status: <span id="status">Set up</span>111<script type="text/javascript">112//<![CDATA[113"""114    + _js_code115    + """116//]]>117</script>118<pre id="text"></pre>119</body>120</html>121""",122]123 124 125class Screen(BaseScreen):126    def __init__(self) -> None:127        super().__init__()128        self.palette = {}129        self.has_color = True130        self._started = False131 132    @property133    def started(self) -> bool:134        return self._started135 136    def register_palette(self, palette) -> None:137        """Register a list of palette entries.138 139        palette -- list of (name, foreground, background) or140                   (name, same_as_other_name) palette entries.141 142        calls self.register_palette_entry for each item in l143        """144 145        for item in palette:146            if len(item) in {3, 4}:147                self.register_palette_entry(*item)148                continue149            if len(item) != 2:150                raise ValueError(f"Invalid register_palette usage: {item!r}")151            name, like_name = item152            if like_name not in self.palette:153                raise KeyError(f"palette entry '{like_name}' doesn't exist")154            self.palette[name] = self.palette[like_name]155 156    def register_palette_entry(157        self,158        name: str | None,159        foreground: str,160        background: str,161        mono: str | None = None,162        foreground_high: str | None = None,163        background_high: str | None = None,164    ) -> None:165        """Register a single palette entry.166 167        name -- new entry/attribute name168        foreground -- foreground colour169        background -- background colour170        mono -- monochrome terminal attribute171 172        See curses_display.register_palette_entry for more info.173        """174        if foreground == "default":175            foreground = "black"176        if background == "default":177            background = "light gray"178        self.palette[name] = (foreground, background, mono)179 180    def set_mouse_tracking(self, enable: bool = True) -> None:181        """Not yet implemented"""182 183    def tty_signal_keys(self, *args, **vargs):184        """Do nothing."""185 186    def start(self) -> StoppingContext:187        """188        This function reads the initial screen size, generates a189        unique id and handles cleanup when fn exits.190 191        web_display.set_preferences(..) must be called before calling192        this function for the preferences to take effect193        """194        if self._started:195            return StoppingContext(self)196 197        client_init = sys.stdin.read(50)198        if not client_init.startswith("window resize "):199            raise ValueError(client_init)200        _ignore1, _ignore2, x, y = client_init.split(" ", 3)201        x = int(x)202        y = int(y)203        self._set_screen_size(x, y)204        self.last_screen = {}205        self.last_screen_width = 0206 207        self.update_method = os.environ["HTTP_X_URWID_METHOD"]208        if self.update_method not in {"multipart", "polling"}:209            raise ValueError(self.update_method)210 211        if self.update_method == "polling" and not _prefs.allow_polling:212            sys.stdout.write("Status: 403 Forbidden\r\n\r\n")213            sys.exit(0)214 215        clients = glob.glob(os.path.join(_prefs.pipe_dir, "urwid*.in"))216        if len(clients) >= _prefs.max_clients:217            sys.stdout.write("Status: 503 Sever Busy\r\n\r\n")218            sys.exit(0)219 220        urwid_id = f"{random.randrange(10**9):09d}{random.randrange(10**9):09d}"  # noqa: S311221        self.pipe_name = os.path.join(_prefs.pipe_dir, f"urwid{urwid_id}")222        os.mkfifo(f"{self.pipe_name}.in", 0o600)223        signal.signal(signal.SIGTERM, self._cleanup_pipe)224 225        self.input_fd = os.open(f"{self.pipe_name}.in", os.O_NONBLOCK | os.O_RDONLY)226        self.input_tail = ""227        self.content_head = (228            f"Content-type: multipart/x-mixed-replace;boundary=ZZ\r\nX-Urwid-ID: {urwid_id}\r\n\r\n\r\n--ZZ\r\n"229        )230        if self.update_method == "polling":231            self.content_head = f"Content-type: text/plain\r\nX-Urwid-ID: {urwid_id}\r\n\r\n\r\n"232 233        signal.signal(signal.SIGALRM, self._handle_alarm)234        signal.alarm(ALARM_DELAY)235        self._started = True236 237        return StoppingContext(self)238 239    def stop(self) -> None:240        """241        Restore settings and clean up.242        """243        if not self._started:244            return245 246        # XXX which exceptions does this actually raise? EnvironmentError?247        with suppress(Exception):248            self._close_connection()249        signal.signal(signal.SIGTERM, signal.SIG_DFL)250        self._cleanup_pipe()251        self._started = False252 253    def set_input_timeouts(self, *args: typing.Any) -> None:254        pass255 256    def _close_connection(self) -> None:257        if self.update_method == "polling child":258            self.server_socket.settimeout(0)259            sock, _addr = self.server_socket.accept()260            sock.sendall(b"Z")261            sock.close()262 263        if self.update_method == "multipart":264            sys.stdout.write("\r\nZ\r\n--ZZ--\r\n")265            sys.stdout.flush()266 267    def _cleanup_pipe(self, *args) -> None:268        if not self.pipe_name:269            return270        # XXX which exceptions does this actually raise? EnvironmentError?271        with suppress(Exception):272            os.remove(f"{self.pipe_name}.in")273            os.remove(f"{self.pipe_name}.update")274 275    def _set_screen_size(self, cols: int, rows: int) -> None:276        """Set the screen size (within max size)."""277 278        cols = min(cols, MAX_COLS)279        rows = min(rows, MAX_ROWS)280        self.screen_size = cols, rows281 282    def draw_screen(self, size: tuple[int, int], canvas: Canvas) -> None:283        """Send a screen update to the client."""284 285        (cols, rows) = size286 287        if cols != self.last_screen_width:288            self.last_screen = {}289 290        sendq = [self.content_head]291 292        if self.update_method == "polling":293            send = sendq.append294        elif self.update_method == "polling child":295            signal.alarm(0)296            try:297                s, _addr = self.server_socket.accept()298            except socket.timeout:299                sys.exit(0)300            send = s.sendall301        else:302            signal.alarm(0)303            send = sendq.append304            send("\r\n")305            self.content_head = ""306 307        if canvas.rows() != rows:308            raise ValueError(rows)309 310        if canvas.cursor is not None:311            cx, cy = canvas.cursor312        else:313            cx = cy = None314 315        new_screen = {}316 317        y = -1318        for row in canvas.content():319            y += 1320            l_row = tuple((attr_, line.decode(get_encoding())) for attr_, _, line in row)321 322            line = []323 324            sig = l_row325            if y == cy:326                sig = (*sig, cx)327            new_screen[sig] = [*new_screen.get(sig, []), y]328 329            if (old_line_numbers := self.last_screen.get(sig, None)) is not None:330                if y in old_line_numbers:331                    old_line = y332                else:333                    old_line = old_line_numbers[0]334                send(f"<{old_line:d}\n")335                continue336 337            col = 0338            for a, run in l_row:339                t_run = run.translate(_trans_table)340                if a is None:341                    fg, bg, _mono = "black", "light gray", None342                else:343                    fg, bg, _mono = self.palette[a]344                if y == cy and col <= cx:345                    run_width = calc_width(t_run, 0, len(t_run))346                    if col + run_width > cx:347                        line.append(code_span(t_run, fg, bg, cx - col))348                    else:349                        line.append(code_span(t_run, fg, bg))350                    col += run_width351                else:352                    line.append(code_span(t_run, fg, bg))353 354            send(f"{''.join(line)}\n")355        self.last_screen = new_screen356        self.last_screen_width = cols357 358        if self.update_method == "polling":359            sys.stdout.write("".join(sendq))360            sys.stdout.flush()361            sys.stdout.close()362            self._fork_child()363        elif self.update_method == "polling child":364            s.close()365        else:  # update_method == "multipart"366            send("\r\n--ZZ\r\n")367            sys.stdout.write("".join(sendq))368            sys.stdout.flush()369 370        signal.alarm(ALARM_DELAY)371 372    def clear(self) -> None:373        """374        Force the screen to be completely repainted on the next375        call to draw_screen().376 377        (does nothing for web_display)378        """379 380    def _fork_child(self) -> None:381        """382        Fork a child to run CGI disconnected for polling update method.383        Force parent process to exit.384        """385        daemonize(f"{self.pipe_name}.err")386        self.input_fd = os.open(f"{self.pipe_name}.in", os.O_NONBLOCK | os.O_RDONLY)387        self.update_method = "polling child"388        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)389        s.bind(f"{self.pipe_name}.update")390        s.listen(1)391        s.settimeout(POLL_CONNECT)392        self.server_socket = s393 394    def _handle_alarm(self, sig, frame) -> None:395        if self.update_method not in {"multipart", "polling child"}:396            raise ValueError(self.update_method)397        if self.update_method == "polling child":398            # send empty update399            try:400                s, _addr = self.server_socket.accept()401                s.close()402            except socket.timeout:403                sys.exit(0)404        else:405            # send empty update406            sys.stdout.write("\r\n\r\n--ZZ\r\n")407            sys.stdout.flush()408        signal.alarm(ALARM_DELAY)409 410    def get_cols_rows(self) -> tuple[int, int]:411        """Return the screen size."""412        return self.screen_size413 414    @typing.overload415    def get_input(self, raw_keys: Literal[False]) -> list[str]: ...416 417    @typing.overload418    def get_input(self, raw_keys: Literal[True]) -> tuple[list[str], list[int]]: ...419 420    def get_input(self, raw_keys: bool = False) -> list[str] | tuple[list[str], list[int]]:421        """Return pending input as a list."""422        pending_input = []423        resized = False424        with selectors.DefaultSelector() as selector:425            selector.register(self.input_fd, selectors.EVENT_READ)426 427            iready = [event.fd for event, _ in selector.select(0.5)]428 429        if not iready:430            if raw_keys:431                return [], []432            return []433 434        keydata = os.read(self.input_fd, MAX_READ).decode(get_encoding())435        os.close(self.input_fd)436        self.input_fd = os.open(f"{self.pipe_name}.in", os.O_NONBLOCK | os.O_RDONLY)437        # sys.stderr.write( repr((keydata,self.input_tail))+"\n" )438        keys = keydata.split("\n")439        keys[0] = self.input_tail + keys[0]440        self.input_tail = keys[-1]441 442        for k in keys[:-1]:443            if k.startswith("window resize "):444                _ign1, _ign2, x, y = k.split(" ", 3)445                x = int(x)446                y = int(y)447                self._set_screen_size(x, y)448                resized = True449            else:450                pending_input.append(k)451        if resized:452            pending_input.append("window resize")453 454        if raw_keys:455            return pending_input, []456        return pending_input457 458 459def code_span(s: str, fg: str, bg: str, cursor: int = -1) -> str:460    code_fg = _code_colours[fg]461    code_bg = _code_colours[bg]462 463    if cursor >= 0:464        c_off, _ign = calc_text_pos(s, 0, len(s), cursor)465        c2_off = move_next_char(s, c_off, len(s))466 467        return (468            code_fg469            + code_bg470            + s[:c_off]471            + "\n"472            + code_bg473            + code_fg474            + s[c_off:c2_off]475            + "\n"476            + code_fg477            + code_bg478            + s[c2_off:]479            + "\n"480        )481 482    return f"{code_fg + code_bg + s}\n"483 484 485def is_web_request() -> bool:486    """487    Return True if this is a CGI web request.488    """489    return "REQUEST_METHOD" in os.environ490 491 492def handle_short_request() -> bool:493    """494    Handle short requests such as passing keystrokes to the application495    or sending the initial HTML page.  If returns True, then this496    function recognized and handled a short request, and the calling497    script should immediately exit.498 499    web_display.set_preferences(..) should be called before calling this500    function for the preferences to take effect501    """502    if not is_web_request():503        return False504 505    if os.environ["REQUEST_METHOD"] == "GET":506        # Initial request, send the HTML and javascript.507        sys.stdout.write("Content-type: text/html\r\n\r\n" + html.escape(_prefs.app_name).join(_html_page))508        return True509 510    if os.environ["REQUEST_METHOD"] != "POST":511        # Don't know what to do with head requests etc.512        return False513 514    if "HTTP_X_URWID_ID" not in os.environ:515        # If no urwid id, then the application should be started.516        return False517 518    urwid_id = os.environ["HTTP_X_URWID_ID"]519    if len(urwid_id) > 20:520        # invalid. handle by ignoring521        # assert 0, "urwid id too long!"522        sys.stdout.write("Status: 414 URI Too Long\r\n\r\n")523        return True524    for c in urwid_id:525        if c not in string.digits:526            # invald. handle by ignoring527            # assert 0, "invalid chars in id!"528            sys.stdout.write("Status: 403 Forbidden\r\n\r\n")529            return True530 531    if os.environ.get("HTTP_X_URWID_METHOD", None) == "polling":532        # this is a screen update request533        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)534        try:535            s.connect(os.path.join(_prefs.pipe_dir, f"urwid{urwid_id}.update"))536            data = f"Content-type: text/plain\r\n\r\n{s.recv(BUF_SZ)}"537            while data:538                sys.stdout.write(data)539                data = s.recv(BUF_SZ)540        except OSError:541            sys.stdout.write("Status: 404 Not Found\r\n\r\n")542            return True543        return True544 545    # this is a keyboard input request546    try:547        fd = os.open((os.path.join(_prefs.pipe_dir, f"urwid{urwid_id}.in")), os.O_WRONLY)548    except OSError:549        sys.stdout.write("Status: 404 Not Found\r\n\r\n")550        return True551 552    # FIXME: use the correct encoding based on the request553    keydata = sys.stdin.read(MAX_READ)554    os.write(fd, keydata.encode("ascii"))555    os.close(fd)556    sys.stdout.write("Content-type: text/plain\r\n\r\n")557 558    return True559 560 561@dataclasses.dataclass562class _Preferences:563    app_name: str = "Unnamed Application"564    pipe_dir: str = TEMP_DIR565    allow_polling: bool = True566    max_clients: int = 20567 568 569_prefs = _Preferences()570 571 572def set_preferences(573    app_name: str,574    pipe_dir: str = TEMP_DIR,575    allow_polling: bool = True,576    max_clients: int = 20,577) -> None:578    """579    Set web_display preferences.580 581    app_name -- application name to appear in html interface582    pipe_dir -- directory for input pipes, daemon update sockets583                and daemon error logs584    allow_polling -- allow creation of daemon processes for585                     browsers without multipart support586    max_clients -- maximum concurrent client connections. This587               pool is shared by all urwid applications588               using the same pipe_dir589    """590    _prefs.app_name = app_name591    _prefs.pipe_dir = pipe_dir592    _prefs.allow_polling = allow_polling593    _prefs.max_clients = max_clients594 595 596class ErrorLog:597    def __init__(self, errfile: str | pathlib.PurePath) -> None:598        self.errfile = errfile599 600    def write(self, err: str) -> None:601        with open(self.errfile, "a", encoding="utf-8") as f:602            f.write(err)603 604 605def daemonize(errfile: str) -> None:606    """607    Detach process and become a daemon.608    """609    if os.fork():610        os._exit(0)611 612    os.setsid()613    signal.signal(signal.SIGHUP, signal.SIG_IGN)614    os.umask(0)615 616    if os.fork():617        os._exit(0)618 619    os.chdir("/")620    for fd in range(20):621        with suppress(OSError):622            os.close(fd)623 624    sys.stdin = open("/dev/null", encoding="utf-8")  # noqa: SIM115  # pylint: disable=consider-using-with625    sys.stdout = open("/dev/null", "w", encoding="utf-8")  # noqa: SIM115  # pylint: disable=consider-using-with626    sys.stderr = ErrorLog(errfile)627