Best Python code snippet using playwright-python
mcprotocol.py
Source:mcprotocol.py  
...23# make the handlers a little more obvious by using decorators.24# functions wrapped will automatically be added the the handlers25# dictionary of the protocol.26# look at bravo.packets for the name dict used27def wrap_handler(*types):28    def wrap_it(func):29        for t in types:30            packet_handlers[packets_by_name[t]] = func31        def wrapped(self, *args, **kwargs):32            func(*args, **kwargs)33        return wrapped34    return wrap_it35class ExtrasMixin(object):36    """37    These are all the little things that you would want a bot/client to *do*,38    but the mechanics are tied too closely to the network protocol to be defined39    in the bot/player class.40    These are expected to be called by the bot.  As the protocol changes, these41    may also change, but the bot and related scripts wont have too.42    actions requiring a action number (transaction token) should follow43    twisted protocol.44    functions need extra information for the ai45    ai can then handle more complex cases, such as moving items from one chest to another46    or crafting items, etc.47    """48    open_window = None49    def get_item(self, item, count=1):50        """51        kinda wraps around stacks and what-not52        returns a list of tuples of slots that can be used53        for example, if you have 200 dirt in 3 stacks, and you54        want to put 150 of it in a chest, then this will return55        stacks that you can use to get the 150 needed.56        if the quantity cannot be satisfied, then return None57        a count of -1 will return all the that item58        """59        if count == -1:60            return [ i for i in self.bot.inventory.flat if i[0] == item ] 61            62        ret = []63        current = 064        for i in [ i for i in self.bot.inventory.flat if i[0] == item]:65            if i[2] + current >= count:66                ret.append((item[0], item[1], item[2] - i[2] + current))67                return ret68            else:69                ret.append(item) 70                current += item[2] 71    """72    req: item in inventory73    eff: item not in inventory74    """75    def throw_item(self, item, count=1):76        """77        throw item in players inventory78        count:79            if = -1, throw ALL80            there could be multiple stacks of an item81            in that case, throw the count from the first stack, if not enough82                find another stack to remove from83            if unstackable item, but has count, then find enough to satisfy84            the count.85        """86        for slot, item in self.get_item(item, count):87            self.throw_slot(slot, count, item)88    def throw_slot(self, slot_no, count=1, item=None):89        """90        throw an item in a slot91        not sure if this is the correct way to do this.92        simulates a player opening the inventory, clicking an item93        then closing the window with the item in the cursor94        seems to bug out once in a while95        if slot is empty, there will be no error96        """97        if item == None:98            print "must incl item no"99            return100        count = item[2]101        action_no, d = self._get_next_action_no(sys.exit, None)102        103        clickit = make_packet("window-action", wid=0, slot=slot_no, button=0, \104            token=action_no, primary=item[0], secondary=item[1] , count=count)105        # send the data over the wire106        self.transport.write(clickit)107        # closing the window actualy preforms the throw108        closeit = make_packet("window-close", wid=0)109        self.transport.write(closeit)110        # close the fake window no matter what111        #d.addBoth(self.transport.write, closeit)112    # useful if chaining clicks together.113    def window_click(self, **kwargs):114        kwargs["action_no"], d = get_action_number()115        self.transport.write(make_packet("window-action", kwargs)) 116        return d117    def swap_slots(self, name, from_slot, to_slot, item1=None, item2=None):118        """119        Swap the items in two slots120        if the to_slot contains an item, then place it in the from_slot.121        Simulates user interaction on the client (mouse clicks)122        the minecraft protocol requires the client to know the slot and123        item attributes when sending this packet.  the item1 & item2124        parameters are there *in case* the calling function already knows125        this information (not implimented)126        """127        if name == 0:128            item1 = self.bot.inventory.get_slot(item1)129            item2 = self.bot.inventory.get_slot(item2)130        # simulate the first click131        d1 = self.window_click(name=name, slot=from_slot, button=0, \132            primary=item1[0], secondary=item1[1] , count=item1[2])133        # simulate click on the slot to move to134        d2 = self.window_click(name=name, slot=to_slot, button=0, \135            primary=item2[0], secondary=item2[1] , count=item2[2])136        d1.chainDeferred(d2)137        if have_another:138            # simulate click on the orginal slot139            d3 = self.window_click(name=name, slot=from_slot, button=0, \140                primary=item1[0], secondary=item1[1] , count=item1[2])141            d1.chainDeferred(d3)142    """143    req: item in inventory144    eff: wield item145    """146    def hold_item(self, item):147        """148        Put an item in the bots hand.149        If the item is not in the lower panel (holdables), then it will move150        the item to an available slot in holdables.  If there are no open151        slots in holdables, some item will be swapped out and the new item152        placed in the hand, then equipted.153        """154        try:155            l, item_slot = self.bot.inventory.get_slot(item)156        except:157            print "item not found in enventory"158            return159        # item is not in holdables.  move it.160        if l != self.bot.inventory.holdables:161            try:162                holding_slot = self.bot.inventory.holdables.index(None)163            except ValueError:164                # there is no open slot in holdables165                # just use slot 40166                holding_slot = 40167            self.swap_slots(0, item_slot, holding_slot)168        # finally, change the item in the hand169        make_packet("holding_change", slot=slot_no)170    """171    req: wield item can dig block172    eff: block destroyed173    """174    def dig(self, bx, by, bz, times, rate):175        # use the tool in hand (or fist if nothing)176        make_packet("digging", status=dig_status, x=bx, y=by, z=bz, face=block_face)177    def set_head_tracking(self, entity):178        # set an object that the head should watch179        pass180    def open_inventory(self, entity):181        # open chest, furnace, workbench182        pass183    def place_entity(self, bx, by, bz, face, entity, amt=0, dmg=0):184        # place a block or entity185        make_packet("build", bx, by, bz, face, items)186    """187    req: can open from_what188    req: inventory has free slot * quantity189    eff: inventory has what190    """191    def take_from(self, from_what, what, quantity):192        # take something from something else193        pass194    """195    req: can open what196    req: into_what has free slot * count197    eff: into_what has what198    """199    def put_into(self, into_what, what, count):200        # put something from personal inventory into something else201        # open inventory of entity202        # send the move packet203        # wait for confirmation from server204        pass205    """206    req: inventory has materials in recipe207    req: can open crafting area208    eff: inventory has crafted209    """210    def craft(self, recipe):211        # craft something212        pass213    def jump(self):214        # jump, or swim if under water215        pass216    def move_to(self, entity):217        # move to an entity218        pass219    def set_home(self, location):220        # set the place where the bot idles221        # bot could go here to hang out if needed222        pass223    def play_animation(self, animation):224        # play an animation225        make_packet("animate", eid=0, animation=animation_name)226    def crouch(self):227        # crouch/sneak228        pass229    def uncrouch(self):230        # stand up231        pass232    def chat(self, text):233        # send a chat message (limited to 100 chars)234        self.transport.write(make_packet("chat", message=text[:100]))235class WebAuthMixin(object):236    """237    Impliment the "minecraft.net/jsp" auth currently used with minecraft beta.238    """239    login_url = "http://www.minecraft.net/game/getversion.jsp"240    vfy_url   = "http://www.minecraft.net/game/checkserver.jsp"241    join_url  = "http://www.minecraft.net/game/joinserver.jsp"242    def authenticate(self, username, password, online=False):243        """244        all authenticators should impliment this245        """246        if online:247            # login to minecraft.net (1)248            self.username = self._minecraft_login(username, password)249        else:250            self.username = username251        # might get none if there the minecraft servers are down252        if self.username == None:253            print "problem with authentication"254            reactor.stop()255        # send handshake (2)256        self.transport.write(make_packet("handshake", username=self.username))257    # this is the login for minecraft.net258    def _minecraft_login(self, user, passwd, server_ver=None):259        if server_ver == None:260            server_ver = self.server_version261        o = "?user=%s&password=%s&version=%s" % (user, passwd, server_ver)262        c = urllib2.urlopen(self.login_url + o).read()263        try:264            ver, ticket, user, sid = c.split(":")[:4]265        except:266            return None267        self.sid = sid268        return user269    # this is the login for the server270    def server_login(self):271        # it might look funny abt "unused".  b/c MAD's packet use this name272        # but a client uses this space for a password273        # send "login packet" (3)274        p = make_packet("login", protocol=8, username=self.username, \275            unused=self.bot.password, seed=0, dimension=0)276        self.transport.write(p)277    # ask the server what kind of authentication to use278    def _check_auth(self, hash):279        o = "?user=%s&sessionId=%s&serverID=%s" % (self.username, self.sid, hash)280        c = urllib2.urlopen(self.join_url + o).read()281        return c.lower() == "ok"282    # serverID aka "server hash"283    def _verify_name(self, serverID):284        o = "?user=%s&serverID=%s" % (self.username, serverID)285        c = urllib2.urlopen(self.vfy_url + o).read()286        return c.lower() == "yes"287class MinecraftClientProtocol(Protocol, WebAuthMixin, ExtrasMixin):288    """289    Impliment v.8 of the Minecraft Protocol for clients290    This is the bare, boring network stuff.291    all of the OnXXXX methods are called when a packet comes in.292    outgoing packets are not automatically managed293    """294    server_version = 12295    keep_alive_interval = 60    296    bot_tick_interval = 50.000 / 1000.000297    flying_interval = 200.000 / 1000.000298    # regex to strip names from chat messages299    chat_regex = re.compile("<(.*?)>(.*)")300    def dummy_handler(self, header, packet):301        try:302            print packets[header]303        except KeyError:304            print "unhandled", header305    def _get_next_action_no(self, callback, args):306        d = Deferred()307        self.action_no += 1308        self.pending_actions[self.action_no] = d309        return self.action_no, d310    def __init__(self, bot, world, online=True):311        self.buffer = ""312        # online mode for the client313        self.online_mode = False314        # make sure our bot is properly connect to the protocol315        self.bot = bot316        bot.conn = self317        # one connection per world        318        self.world = world319        # used for handling window actions320        self.action_no = 0321        self.pending_actions = {}322        # after client is ready, sends this back to the server323        self.confirmed_spawn = False324        # are we authenticated?  (will be set later)325        self.authenticated = False326    def OnAuthenticated(self):327        """328        Called when our authenticator is finished.329        """330        self.authenticated = True331    def set_username(self, username):332        self.username = username333    def connectionMade(self):334        # great!  lets get authenticated and move on to the good stuff335        self.authenticate(self.bot.username, self.bot.password, self.online_mode)336    def connectionLost(self, reason):337        print "Lost connection:", reason338        try:339            self.keep_alive.cancel()340            self.bot_tick.cancel()341        except:342            pass343    def send_flying(self):344        self.transport.write(make_packet("flying", flying=self.bot.location.midair))345    def send_keep_alive(self):346        self.transport.write(make_packet("ping"))347    # called by twisted whenever data comes in over the wire348    # parse out as many packets as we can349    def dataReceived(self, data):350        self.buffer += data351        packets, self.buffer = parse_packets(self.buffer)352        for header, payload in packets:353            if header in packet_handlers:354                packet_handlers[header](self, payload)355            else:356                self.dummy_handler(header, payload)357    @wrap_handler("login")358    def OnLoginResponse(self, packet):359        # BUG: we are not really checking if we are allowed to join or not360        self.authenticated = True361        # hack, b/c not implimenting the client protocol362        self.bot.eid = packet.protocol363        self.keep_alive = task.LoopingCall(self.send_keep_alive)364        self.keep_alive.start(self.keep_alive_interval)365        self.flying_tick = task.LoopingCall(self.send_flying)366        self.flying_tick.start(self.flying_interval)367        368        self.bot_tick = task.LoopingCall(self.bot.tick)369        self.bot_tick.start(self.bot_tick_interval)370    # kinda tied into the authenticator...371    @wrap_handler("handshake")372    def OnHandshake(self, packet):373        username = packet.username374        if username == "ok":375            self.auth = True376        elif username == "-":377            self.auth = True378        elif username == "+":379            print "joining password protected servers is not implemented [yet]"380            sys.exit()381        else:382            self.auth = self.check_auth(username)383        # this really shouldn't be here384        self.server_login()385    @wrap_handler("ping")386    def OnPing(self, packet):387        pass388    @wrap_handler("time")389    def OnTime(self, packet):390        pass391#   C H U N K   R E L A T E D   ============================392    @wrap_handler("prechunk")393    def OnPreChunk(self, packet):394        pass395    @wrap_handler("chunk")396    def OnMapChunk(self, packet):397        def add_chunk(chunk, packet):398            chunk.load_from_packet(packet)399            self.bot.world.add_chunk(chunk)400        def chunk_error(failure):401            print "couldn't parse chunk packet"402        # the packet (x, y) is in block coords, so /16403        cx, bx = divmod(packet.x, 16)404        cz, bz = divmod(packet.z, 16)405        # for performance reasons, we will only load the chunk that the bot is on406        # it will fail with AttrubuteError if the player hasn't been properly init'd407        try:408            assert (cx == self.bot.chunk.x) and (cz == self.bot.chunk.z)409        except (AssertionError, AttributeError):410            return411        # we assume the world will give us a clean chunk if it doesn't already exist412        d = self.world.request_chunk(cx, cz)413        d.addCallback(add_chunk, packet)414        d.addErrback(chunk_error)415    @wrap_handler("block")416    def OnBlockChange(self, packet):417        self.world.change_block(packet.x, packet.y, packet.z, \418            packet.type, packet.meta)419    @wrap_handler("batch")420    def OnMultiBlockChange(self, packet):421        for i in xrange(packet.length):422            bx = packet.coords[i] >> 12423            bz = packet.coords[i] >> 8 & 15424            y = packet.coords[i] & 255425            self.world.change_block(bx, y, bz, packet.types[i], \426                packet.metadata[i])427    @wrap_handler("digging")428    def OnPlayerDigging(self, packet):429        print packet430    @wrap_handler("animate")431    def OnAnimation(self, packet):432        pass433    @wrap_handler("health")434    def OnUpdateHealth(self, packet):435        self.bot.hp = packet.hp436    @wrap_handler("spawn")437    def OnPlayer(self, packet):438        pass439    @wrap_handler("position", "orientation", "location")440    def OnPlayerLocationUpdate(self, packet):441        self.bot.update_location_from_packet(packet)442        print packet443        # everytime the player spawns, it must send back the location that it was given444        # this is a check for the server.  not entirely part of authentication, but445        # the bot won't run without it446        if self.confirmed_spawn == False:447            location = Location()448            location.load_from_packet(packet)449            p = location.save_to_packet()450            self.transport.write(p)451            self.confirmed_spawn = True452            self.bot.set_location(location)453            self.bot.OnReady()454    @wrap_handler("destroy")455    def OnDestroy(self, packet):456        self.world.remove_entity_by_id(packet.eid)457    @wrap_handler("create")458    def OnCreate(self, packet):459        self.world.add_entity_by_id(packet.eid)460    @wrap_handler("entity-position", "entity-orientation", "entity-location")461    def OnEntityLocationUpdate(self, packet):462        #print packet463        pass464 465    @wrap_handler("teleport")466    def OnEntityTeleport(self, packet):467        #print packet468        pass469    @wrap_handler("chat")470    def OnChat(self, packet):471        """472        Parse chat messages.473        The Notch server sends messages back that the client sends,474        so we check and skip any messages that we sent.475        Return who sent the message with the text.476        I don't know anything about the color codes, so they are477        not processed.478        """479        match = self.chat_regex.match(packet.message)480        # not really sure why this would fail.  just in case...481        if match == None:482            return483        who, text = match.groups()484        if who != self.username:485            self.bot.OnChatIn(who, text)486    @wrap_handler("window-open")487    def OnWindowOpen(self, packet):488        print packet489    @wrap_handler("window-close")490    def OnWindowClose(self, packet):491        print packet492    @wrap_handler("window-action")493    def OnWindowAction(self, packet):494        print packet495    # also used when bot picks up items496    @wrap_handler("window-slot")497    def OnWindowSlot(self, packet):498        self.bot.inventory.update_from_packet(packet)499        if packet.primary != -1:500            self.throw_slot(packet.slot, packet.count, (packet.primary, packet.secondary, packet.count))501    @wrap_handler("window-progress")502    def OnWindowProgress(self, packet):503        print packet504    @wrap_handler("window-token")505    def OnWindowToken(self, packet):506        try:507            d = self.pending_actions[packet.token]508            d.callback(packet)509        except KeyError:510            print "huh....got action back that didnt send?"511            return512    # sent by server to init the player's inventory513    @wrap_handler("inventory")514    def OnInventory(self, packet):515        if packet.name == 0:516            self.bot.inventory.load_from_packet(packet)517    @wrap_handler("error")518    def OnKick(self, packet):...chat_users.py
Source:chat_users.py  
...22            groups.append(info)23    return event, users, groups24@dp.longpoll_event_register('лÑди')25@dp.my_signal_event_register('лÑди')26@dp.wrap_handler(users_getter)27def list_users(event: MySignalEvent, users: List[dict], _):28    try:29        page = int(get_index(event.args, 0, 1)) - 130        if page < 0:31            raise ValueError32    except ValueError:33        page = 034    count = len(users)35    pages = ceil(count/20)36    msg = ''37    for i, user in enumerate(users[page*20:page*20+20], 1 + page*20):38        msg += f"\n{i}. [id{user['id']}|{user['first_name']} {user['last_name']}]"  # noqa39    if msg == '':40        msg = f'\nСÑÑаниÑа {page + 1} пÑÑÑа'41    else:42        msg = f'\nУÑаÑÑники беÑÐµÐ´Ñ (ÑÑÑаниÑа {page + 1} из {pages}):' + msg43    event.msg_op(1, msg, disable_mentions=1, reply_to=event.msg['id'])44    return "ok"45@dp.longpoll_event_register('беÑеда', 'ÑаÑ')46@dp.my_signal_event_register('беÑеда', 'ÑаÑ')47@dp.wrap_handler(users_getter)48def chat_info(event: MySignalEvent, users: List[dict], groups: List[dict]):49    admins = []50    owner = None51    for member in users + groups:52        if member.get('is_owner') is True:53            owner = member54        elif member.get('is_admin') is True:55            admins.append('\n-- ' + format_push(member))56    msg = f"""57    ÐеÑеда ð·ðº: {event.chat.name}58    СоздаÑÐµÐ»Ñ ð·ðº: {format_push(owner)}59    Iris IDð·ðº : {event.chat.iris_id}60    Я дежÑÑнÑй в ÑаÑе ð·ðº: {'â
' if event.chat.installed else 'â'}61    ÐаÑеление ÑаÑа ð·ðº: {len(users) + len(groups)}62    УÑаÑÑников ð·ðº: {len(users)}63    ÐоÑов ð·ðº: {len(groups)}64    ÐдминиÑÑÑаÑоÑÑ:{''.join(admins) if admins else 'ÐÐ´Ð¼Ð¸Ð½Ñ Ð½ÐµÐ²Ð¸Ð´Ð¸Ð¼Ñе ðð'}65    """.replace('    ', '')66    event.msg_op(1, msg, disable_mentions=1, reply_to=event.msg['id'])67    return "ok"68@dp.longpoll_event_register('боÑÑ')69@dp.my_signal_event_register('боÑÑ')70@dp.wrap_handler(users_getter)71def list_groups(event: MySignalEvent, _, groups: List[dict]):72    try:73        page = int(get_index(event.args, 0, 1)) - 174        if page < 0:75            raise ValueError76    except ValueError:77        page = 078    count = len(groups)79    pages = ceil(count/20)80    msg = ''81    for i, group in enumerate(groups[page*20:page*20+20], 1 + page*20):82        msg += f"\n{i}. [public{group['id']}|{group['name']}] ð·ðº"83    if msg == '':84        msg = f'СÑÑаниÑа {page + 1} пÑÑÑа'85    else:86        msg = f'ÐÑÑÐ¿Ð¿Ñ Ð±ÐµÑÐµÐ´Ñ (ÑÑÑаниÑа {page + 1} из {pages}):' + msg87    event.msg_op(2, msg)88    return "ok"89@dp.longpoll_event_register('ÑÑаÑ', 'ÑÑаÑÑники')90@dp.my_signal_event_register('ÑÑаÑ', 'ÑÑаÑÑники')91@dp.wrap_handler(users_getter)92def chat_info(event: MySignalEvent, users: List[dict], groups: List[dict]):93    admins = []94    owner = None95    for member in users + groups:96        if member.get('is_owner') is True:97            owner = member98        elif member.get('is_admin') is True:99            admins.append('\n-- ' + format_push(member))100    msg = f"""101    ЧиÑло лÑдей ð·ðº: {len(users)} ð¤102    ÐаÑеление ÑаÑа: {len(users) + len(groups)} ðª103    ÐоÑов: {len(groups)} (â¡_â¡)104    """.replace('    ', '')105    event.msg_op(1, msg, disable_mentions=1, reply_to=event.msg['id'])106    return "ok"107@dp.longpoll_event_register('ад','ÐдминÑ')108@dp.my_signal_event_register('ад','админÑ')109@dp.wrap_handler(users_getter)110def chat_info(event: MySignalEvent, users: List[dict], groups: List[dict]):111    admins = []112    owner = None113    for member in users + groups:114        if member.get('is_owner') is True:115            owner = member116        elif member.get('is_admin') is True:117            admins.append('\n-- ' + format_push(member))118    msg = f"""119    СоздаÑÐµÐ»Ñ ð·ðº: {format_push(owner)} ð120    ÐдминиÑÑÑаÑоÑÑ:\n{' '.join(admins) if admins else 'ÐÐ´Ð¼Ð¸Ð½Ñ Ð½ÐµÐ²Ð¸Ð´Ð¸Ð¼Ñе ðð'}121    """.replace('    ', '')122    event.msg_op(1, msg, disable_mentions=1, reply_to=event.msg['id'])123    return "ok"124@dp.longpoll_event_register('имÑ', 'название')125@dp.my_signal_event_register('имÑ', 'название')126@dp.wrap_handler(users_getter)127def chat_info(event: MySignalEvent, users: List[dict], groups: List[dict]):128    admins = []129    owner = None130    for member in users + groups:131        if member.get('is_owner') is True:132            owner = member133        elif member.get('is_admin') is True:134            admins.append('\n-- ' + format_push(member))135    msg = f"""136    ÐеÑеда ð·ðº: {event.chat.name}137    СоздаÑелÑ: {format_push(owner)} ð138    """.replace('    ', '')139    event.msg_op(1, msg, disable_mentions=1, reply_to=event.msg['id'])140    return "ok"...day_achi.py
Source:day_achi.py  
...11    KEY_DAY, KEY_STATE, KEY_DAY_TIME, KEY_COUNT, KEY_NEED,12    STATE_0, STATE_1, STATE_213    )1415def wrap_handler(func):16    """handler修饰"""17    def _func(self, *args, **kw):18        _data = kw.pop('_data')19        #æ¯æ¥é¶ç¹æ´æ°æ°æ®20        player = _data[0]21        if not is_today(player.achievement.data[KEY_DAY_TIME]):22            new = copy.deepcopy(player._game.achi_mgr.get_new())23            player.achievement.data[KEY_DAY] = new[KEY_DAY]24            player.achievement.data[KEY_DAY_TIME] = int(time.time())25        kw['_data'] = self.get_data(_data)26        if kw['_data'] is None:27            return28        return func(self, *args, **kw)29    return _func
...flaskapp.py
Source:flaskapp.py  
...51    logger=app.logger,52)53db_connector.check_docs_table()54db_connector.check_users_table()55def wrap_handler(handler):56    @wraps(handler)57    def wrapped():58        app.logger.info('Start handling %s', handler.__name__)59        try:60            response = handler()61        except Exception as exc:62            app.logger.exception('Handling failed: %s', exc)63            response = json.dumps({'message': 'Internal error'}), 50064        app.logger.info('Stop handling %s', handler.__name__)65        return response66    return wrapped67@app.route('/login', methods=['POST'])68@wrap_handler69def login():...LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!
