<commit_before>#!/usr/bin/env python
"""
Command line interface to interact with a VNC Server

(c) 2010 <NAME>

MIT License
"""
import getpass
import optparse
import sys
import os
import shlex
import socket
import logging
import logging.handlers

from twisted.python.log import PythonLoggingObserver
from twisted.internet import reactor, protocol
from twisted.internet.error import ConnectionDone
from twisted.python.failure import Failure

from .client import VNCDoToolFactory, VNCDoToolClient, factory_connect
from .loggingproxy import VNCLoggingServerFactory


log = logging.getLogger()

SUPPORTED_FORMATS = ('png', 'jpg', 'jpeg', 'gif', 'bmp')


class TimeoutError(RuntimeError):
    pass


def log_exceptions(type_, value, tb):
    log.critical('Unhandled exception:', exc_info=(type_, value, tb))


def log_connected(pcol):
    log.info('connected to %s' % pcol.name)
    return pcol


class VNCDoCLIClient(VNCDoToolClient):
    def vncRequestPassword(self):
        if self.factory.password is None:
            self.factory.password = getpass.getpass('VNC password:')

        self.sendPassword(self.factory.password)


class VNCDoCLIFactory(VNCDoToolFactory):
    protocol = VNCDoCLIClient

    def clientConnectionLost(self, connector, reason):
        if reason.type == ConnectionDone:
            self.done(0)
        else:
            self.error(reason)

    def clientConnectionFailed(self, connector, reason):
        self.error(reason)

    def error(self, reason):
        log.critical(reason.getErrorMessage())
        self.done(10)

    def done(self, exit_code):
        reactor.exit_status = exit_code
        reactor.callLater(0.1, reactor.stop)


class ExitingProcess(protocol.ProcessProtocol):

    def processExited(self, reason):
        reactor.callLater(0.1, reactor.stop)

    def errReceived(self, data):
        print(data)


class VNCDoToolOptionParser(optparse.OptionParser):
    def format_help(self, **kwargs):
        result = optparse.OptionParser.format_help(self, **kwargs)
        result += '\n'.join(
           ['',
            'Common Commands (CMD):',
            '  key KEY\t\tsend KEY to server, alphanumeric or keysym: ctrl-c, del',
            '  type TEXT\t\tsend alphanumeric string of TEXT',
            '  typefile FILENAME\t\ttype out the contents of FILENAME',
            '  move X Y\t\tmove the mouse cursor to position X,Y',
            '  click BUTTON\t\tsend a mouse BUTTON click',
            '  capture FILE\t\tsave current screen as FILE',
            '  expect FILE FUZZ\twait until screen matches FILE',
            '  pause SECONDS\t\twait SECONDS before sending next command',
            '',
            'Other Commands (CMD):',
            '  keyup KEY\t\tsend KEY released',
            '  keydown KEY\t\tsend KEY pressed',
            '  mousedown BUTTON\tsend BUTTON down',
            '  mousemove X Y\t\talias for move',
            '  mouseup BUTTON\tsend BUTTON up',
            '  drag X Y\t\tmove the mouse to X,Y in small steps',
            '  rcapture FILE X Y W H\tcapture a region of the screen',
            '  rexpect FILE X Y\texpect that matches a region of the screen',
            '',
            'If a filename is given commands will be read from it, or stdin `-`',
            '',
           ])
        return result


def build_command_list(factory, args, delay=None, warp=1.0):
    client = VNCDoCLIClient

    if delay:
        delay = float(delay) / 1000.0

    while args:
        cmd = args.pop(0)
        if cmd == 'key':
            key = args.pop(0)
            factory.deferred.addCallback(client.keyPress, key)
        elif cmd in ('kdown', 'keydown'):
            key = args.pop(0)
            factory.deferred.addCallback(client.keyDown, key)
        elif cmd in ('kup', 'keyup'):
            key = args.pop(0)
            factory.deferred.addCallback(client.keyUp, key)
        elif cmd in ('move', 'mousemove'):
            x, y = int(args.pop(0)), int(args.pop(0))
            factory.deferred.addCallback(client.mouseMove, x, y)
        elif cmd == 'click':
            button = int(args.pop(0))
            factory.deferred.addCallback(client.mousePress, button)
        elif cmd in ('mdown', 'mousedown'):
            button = int(args.pop(0))
            factory.deferred.addCallback(client.mouseDown, button)
        elif cmd in ('mup', 'mouseup'):
            button = int(args.pop(0))
            factory.deferred.addCallback(client.mouseUp, button)
        elif cmd == 'type':
            for key in args.pop(0):
                factory.deferred.addCallback(client.keyPress, key)
                if delay:
                    factory.deferred.addCallback(client.pause, delay)
        elif cmd == 'typefile':
            with open(args.pop(0)) as f:
                content = f.read()
                for key in content:
                    if key == '\r':
                        continue
                    if key == '\n':
                        key = 'enter'
                    if key == '\t':
                        key = 'tab'
                    factory.deferred.addCallback(client.keyPress, key)
                    if delay:
                        factory.deferred.addCallback(client.pause, delay)
        elif cmd == 'pastefile':
            with open(args.pop(0)) as f:
                content = f.read().replace('\r\n', '\n')
                factory.deferred.addCallback(client.paste, content)
        elif cmd == 'capture':
            filename = args.pop(0)
            imgformat = os.path.splitext(filename)[1][1:]
            if imgformat not in SUPPORTED_FORMATS:
                print('unsupported image format "%s", choose one of %s' % (
                        imgformat, SUPPORTED_FORMATS))
            else:
                factory.deferred.addCallback(client.captureScreen, filename)
        elif cmd == 'expect':
            filename = args.pop(0)
            rms = int(args.pop(0))
            factory.deferred.addCallback(client.expectScreen, filename, rms)
        elif cmd == 'rcapture':
            filename = args.pop(0)
            x = int(args.pop(0))
            y = int(args.pop(0))
            w = int(args.pop(0))
            h = int(args.pop(0))
            imgformat = os.path.splitext(filename)[1][1:]
            if imgformat not in SUPPORTED_FORMATS:
                print('unsupported image format "%s", choose one of %s' % (
                        imgformat, SUPPORTED_FORMATS))
            else:
                factory.deferred.addCallback(client.captureRegion, filename, x, y, w, h)
        elif cmd == 'rexpect':
            filename = args.pop(0)
            x = int(args.pop(0))
            y = int(args.pop(0))
            rms = int(args.pop(0))
            factory.deferred.addCallback(client.expectRegion, filename, x, y, rms)
        elif cmd in ('pause', 'sleep'):
            duration = float(args.pop(0)) / warp
            factory.deferred.addCallback(client.pause, duration)
        elif cmd in 'drag':
            x, y = int(args.pop(0)), int(args.pop(0))
            factory.deferred.addCallback(client.mouseDrag, x, y)
        elif os.path.isfile(cmd):
            lex = shlex.shlex(open(cmd), posix=True)
            lex.whitespace_split = True
            args = list(lex) + args
        else:
            print('unknown cmd "%s"' % cmd)
<commit_msg>write to stdout and and exit with error code on bad arguments<commit_after>#!/usr/bin/env python
"""
Command line interface to interact with a VNC Server

(c) 2010 <NAME>

MIT License
"""
from __future__ import print_function

import getpass
import optparse
import sys
import os
import shlex
import socket
import logging
import logging.handlers

from twisted.python.log import PythonLoggingObserver
from twisted.internet import reactor, protocol
from twisted.internet.error import ConnectionDone
from twisted.python.failure import Failure

from .client import VNCDoToolFactory, VNCDoToolClient, factory_connect
from .loggingproxy import VNCLoggingServerFactory


log = logging.getLogger()

SUPPORTED_FORMATS = ('png', 'jpg', 'jpeg', 'gif', 'bmp')


class TimeoutError(RuntimeError):
    pass


def log_exceptions(type_, value, tb):
    log.critical('Unhandled exception:', exc_info=(type_, value, tb))


def log_connected(pcol):
    log.info('connected to %s' % pcol.name)
    return pcol


class VNCDoCLIClient(VNCDoToolClient):
    def vncRequestPassword(self):
        if self.factory.password is None:
            self.factory.password = getpass.getpass('VNC password:')

        self.sendPassword(self.factory.password)


class VNCDoCLIFactory(VNCDoToolFactory):
    protocol = VNCDoCLIClient

    def clientConnectionLost(self, connector, reason):
        if reason.type == ConnectionDone:
            self.done(0)
        else:
            self.error(reason)

    def clientConnectionFailed(self, connector, reason):
        self.error(reason)

    def error(self, reason):
        log.critical(reason.getErrorMessage())
        self.done(10)

    def done(self, exit_code):
        reactor.exit_status = exit_code
        reactor.callLater(0.1, reactor.stop)


class ExitingProcess(protocol.ProcessProtocol):

    def processExited(self, reason):
        reactor.callLater(0.1, reactor.stop)

    def errReceived(self, data):
        print(data)


class VNCDoToolOptionParser(optparse.OptionParser):
    def format_help(self, **kwargs):
        result = optparse.OptionParser.format_help(self, **kwargs)
        result += '\n'.join(
           ['',
            'Common Commands (CMD):',
            '  key KEY\t\tsend KEY to server, alphanumeric or keysym: ctrl-c, del',
            '  type TEXT\t\tsend alphanumeric string of TEXT',
            '  typefile FILENAME\t\ttype out the contents of FILENAME',
            '  move X Y\t\tmove the mouse cursor to position X,Y',
            '  click BUTTON\t\tsend a mouse BUTTON click',
            '  capture FILE\t\tsave current screen as FILE',
            '  expect FILE FUZZ\twait until screen matches FILE',
            '  pause SECONDS\t\twait SECONDS before sending next command',
            '',
            'Other Commands (CMD):',
            '  keyup KEY\t\tsend KEY released',
            '  keydown KEY\t\tsend KEY pressed',
            '  mousedown BUTTON\tsend BUTTON down',
            '  mousemove X Y\t\talias for move',
            '  mouseup BUTTON\tsend BUTTON up',
            '  drag X Y\t\tmove the mouse to X,Y in small steps',
            '  rcapture FILE X Y W H\tcapture a region of the screen',
            '  rexpect FILE X Y\texpect that matches a region of the screen',
            '',
            'If a filename is given commands will be read from it, or stdin `-`',
            '',
           ])
        return result


def build_command_list(factory, args, delay=None, warp=1.0):
    client = VNCDoCLIClient

    if delay:
        delay = float(delay) / 1000.0

    while args:
        cmd = args.pop(0)
        if cmd == 'key':
            key = args.pop(0)
            factory.deferred.addCallback(client.keyPress, key)
        elif cmd in ('kdown', 'keydown'):
            key = args.pop(0)
            factory.deferred.addCallback(client.keyDown, key)
        elif cmd in ('kup', 'keyup'):
            key = args.pop(0)
            factory.deferred.addCallback(client.keyUp, key)
        elif cmd in ('move', 'mousemove'):
            x, y = int(args.pop(0)), int(args.pop(0))
            factory.deferred.addCallback(client.mouseMove, x, y)
        elif cmd == 'click':
            button = int(args.pop(0))
            factory.deferred.addCallback(client.mousePress, button)
        elif cmd in ('mdown', 'mousedown'):
            button = int(args.pop(0))
            factory.deferred.addCallback(client.mouseDown, button)
        elif cmd in ('mup', 'mouseup'):
            button = int(args.pop(0))
            factory.deferred.addCallback(client.mouseUp, button)
        elif cmd == 'type':
            for key in args.pop(0):
                factory.deferred.addCallback(client.keyPress, key)
                if delay:
                    factory.deferred.addCallback(client.pause, delay)
        elif cmd == 'typefile':
            with open(args.pop(0)) as f:
                content = f.read()
                for key in content:
                    if key == '\r':
                        continue
                    if key == '\n':
                        key = 'enter'
                    if key == '\t':
                        key = 'tab'
                    factory.deferred.addCallback(client.keyPress, key)
                    if delay:
                        factory.deferred.addCallback(client.pause, delay)
        elif cmd == 'pastefile':
            with open(args.pop(0)) as f:
                content = f.read().replace('\r\n', '\n')
                factory.deferred.addCallback(client.paste, content)
        elif cmd == 'capture':
            filename = args.pop(0)
            imgformat = os.path.splitext(filename)[1][1:]
            if imgformat not in SUPPORTED_FORMATS:
                print('unsupported image format "%s", choose one of %s' % (
                        imgformat, SUPPORTED_FORMATS), file=sys.stderr)
                sys.exit(1)
            else:
                factory.deferred.addCallback(client.captureScreen, filename)
        elif cmd == 'expect':
            filename = args.pop(0)
            rms = int(args.pop(0))
            factory.deferred.addCallback(client.expectScreen, filename, rms)
        elif cmd == 'rcapture':
            filename = args.pop(0)
            x = int(args.pop(0))
            y = int(args.pop(0))
            w = int(args.pop(0))
            h = int(args.pop(0))
            imgformat = os.path.splitext(filename)[1][1:]
            if imgformat not in SUPPORTED_FORMATS:
                print('unsupported image format "%s", choose one of %s' % (
                        imgformat, SUPPORTED_FORMATS), file=sys.stderr)
                sys.exit(1)
            else:
                factory.deferred.addCallback(client.captureRegion, filename, x, y, w, h)
        elif cmd == 'rexpect':
            filename = args.pop(0)
            x = int(args.pop(0))
            y = int(args.pop(0))
            rms = int(args.pop(0))
            factory.deferred.addCallback(client.expectRegion, filename, x, y, rms)
        elif cmd in ('pause', 'sleep'):
            duration = float(args.pop(0)) / warp
            factory.deferred.addCallback(client.pause, duration)
        elif cmd in 'drag':
            x, y = int(args.pop(0)), int(args.pop(0))
            factory.deferred.addCallback(client.mouseDrag, x, y)
        elif os.path.isfile(cmd):
            lex = shlex.shlex(open(cmd), posix=True)
            lex.whitespace_split = True
            args = list(lex) + args
        else:
            print('unknown cmd "%s"' % cmd, file=sys.stderr)
            sys.exit(1)

{'fraction_non_alphanumeric': 0.11155913978494623, 'fraction_numerical': 0.00678763440860215, 'mean_word_length': 1.9625721680270753, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7280227', 'n_tokens_mistral': 4433, 'n_tokens_neox': 4306, 'n_words': 1107}