a lc@sdZddlZddlZddlmZmZmZmZmZm Z ddl Z ddl m Z m Z ddlmZddlmZddlmZmZddlmZd d gZGd d d e jZGd d d e jZdS) zI Utility for running a prompt_toolkit application in an asyncssh server. N)Any AwaitableCallableOptionalTextIOcast) AppSessioncreate_app_session)Size)get_event_loop) PipeInputcreate_pipe_input) Vt100_OutputPromptToolkitSSHSessionPromptToolkitSSHServerc@seZdZedgedfeddddZedddZe ddd d Z edd d Z ddd dZ ddddZ eeeeddddZeeddddZdS)rNinteract enable_cprreturncsN|_|_d_d_d_d_d_Gfddd}tt|_ dS)NcsReZdZeddfdd ZedddZdddd Zeedfd d Z dS) z0PromptToolkitSSHSession.__init__..StdoutN)datarcs:z"jdur j|ddWnty4Yn0dS)N z )_chanwritereplaceBrokenPipeError)srself]/var/www/html/django/DPS/env/lib/python3.9/site-packages/prompt_toolkit/contrib/ssh/server.pyr)s   z6PromptToolkitSSHSession.__init__..Stdout.writercSsdSNTrrrrrisatty0sz7PromptToolkitSSHSession.__init__..Stdout.isattycSsdSNrr"rrrflush3sz6PromptToolkitSSHSession.__init__..Stdout.flushcs"jdusJtjjdS)Nr)rstrZ _orig_chanZ get_encodingr"rrrencoding6sz9PromptToolkitSSHSession.__init__..Stdout.encoding) __name__ __module__ __qualname__r&rboolr#r%propertyr'rrrrStdout(s r-) rr interact_taskr app_session_input_outputrrstdout)rrrr-rrr__init__sz PromptToolkitSSHSession.__init__r cCs8|jdurtdddS|j\}}}}t||dSdS)zU Callable that returns the current `Size`, required by Vt100_Output. NO)rowscolumns)rr get_terminal_sizerwidthheightpixwidth pixheightrrr _get_size=s  z!PromptToolkitSSHSession._get_size)chanrcCs ||_dSr$)r)rr?rrrconnection_madeGsz'PromptToolkitSSHSession.connection_madecCsdSr!rrrrrshell_requestedJsz'PromptToolkitSSHSession.shell_requestedcCst||_dSr$)r create_task _interactr.rrrrsession_startedMsz'PromptToolkitSSHSession.session_startedc s|jdurtdt|jdr6|jjdur6|jd|j}t|j|j||j d|_ t |_ t |j |j dv}||_zHz||IdHWntytYn0W|j|j n|j|j 0Wdn1s0YWdn1s0YdS)Nz,`_interact` called before `connection_made`. set_line_modeF)termr)inputoutput)r ExceptionhasattrZ_editorrEZget_terminal_typerr2r>rr1r r0r r/r BaseException traceback print_excclose)rrFsessionrrrrCPs&       z!PromptToolkitSSHSession._interact)r:r;r<r=rcCs|jr|jjr|jjdSr$)r/ZappZ _on_resizer9rrrterminal_size_changedlsz-PromptToolkitSSHSession.terminal_size_changed)rdatatypercCs|jdurdS|j|dSr$)r0Z send_text)rrrQrrr data_receivedss z%PromptToolkitSSHSession.data_received)r(r)r*rrr+r3r r>rr@rArDrCintobjectrPr&rRrrrrrs )  c@sTeZdZdZddeegedfeddddZe edd d Z ed d d Z dS)raX Run a prompt_toolkit application over an asyncssh server. This takes one argument, an `interact` function, which is called for each connection. This should be an asynchronous function that runs the prompt_toolkit applications. This function runs in an `AppSession`, which means that we can have multiple UI interactions concurrently. Example usage: .. code:: python async def interact(ssh_session: PromptToolkitSSHSession) -> None: await yes_no_dialog("my title", "my text").run_async() prompt_session = PromptSession() text = await prompt_session.prompt_async("Type something: ") print_formatted_text('You said: ', text) server = PromptToolkitSSHServer(interact=interact) loop = get_event_loop() loop.run_until_complete( asyncssh.create_server( lambda: MySSHServer(interact), "", port, server_host_keys=["/etc/ssh/..."], ) ) loop.run_forever() :param enable_cpr: When `True`, the default, try to detect whether the SSH client runs in a terminal that responds to "cursor position requests". That way, we can properly determine how much space there is available for the UI (especially for drop down menus) to render. TrNrcCs||_||_dSr$)rr)rrrrrrr3szPromptToolkitSSHServer.__init__)usernamercCsdS)NFr)rrVrrr begin_authsz!PromptToolkitSSHServer.begin_authr cCst|j|jdS)NrU)rrrrrrrsession_requestedsz(PromptToolkitSSHServer.session_requested) r(r)r*__doc__rrrr+r3r&rWrXrrrrr{s) )rYasynciorLtypingrrrrrrZasyncsshZ"prompt_toolkit.application.currentrr Zprompt_toolkit.data_structuresr Zprompt_toolkit.eventloopr Zprompt_toolkit.inputr r Zprompt_toolkit.output.vt100r__all__ZSSHServerSessionrZ SSHServerrrrrrs    h