cytube_bot package¶

Submodules¶

cytube_bot.bot module¶

class cytube_bot.bot.Bot(domain, channel, user=None, restart_delay=5, loop=None, response_timeout=0.1, get=<function get>, socket_io=<bound method type.connect of <class 'cytube_bot.socket_io.SocketIO'>>)[source]¶

Bases: object

CyTube bot.

get¶: function (url, loop) – HTTP GET coroutine.

socket_io¶: function (url, loop) – socket.io connect coroutine.

response_timeout¶: float – socket.io event response timeout in seconds.

restart_delay¶: None or float – Delay in seconds before reconnection. None or < 0 - do not reconnect.

domain¶: str – Domain.

channel¶: cytube_bot.channel.Channel – Channel.

user¶: cytube_bot.user.User – Bot user.

loop¶: asyncio.events.AbstractEventLoop – Event loop.

server¶: None or str – socket.io server URL.

socket¶: None or cytube_bot.socket_io.SocketIO – socket.io connection.

handlers¶: collections.defaultdict of (str, list of function) – Event handlers.

EVENT_LOG_LEVEL = {'channelCSSJS': 10, 'emoteList': 10, 'mediaUpdate': 10}¶

EVENT_LOG_LEVEL_DEFAULT = 20¶

GUEST_LOGIN_LIMIT = re.compile('guest logins .* ([0-9]+) seconds\\.', re.IGNORECASE)¶

MUTED = re.compile('.*\\bmuted', re.IGNORECASE)¶

SOCKET_CONFIG_URL = '%(domain)s/socketconfig/%(channel)s.json'¶

SOCKET_IO_URL = '%(domain)s/socket.io/'¶

__init__(domain, channel, user=None, restart_delay=5, loop=None, response_timeout=0.1, get=<function get>, socket_io=<bound method type.connect of <class 'cytube_bot.socket_io.SocketIO'>>)[source]¶

Parameters:

domain (str) – Domain.
channel (str or (str, str)) – ‘name’ or (‘name’, ‘password’)
user (None or str or (str, str), optional) – None (anonymous) or ‘name’ (guest) or (‘name’, ‘password’) (registered)
restart_delay (None or float, optional) – Delay in seconds before reconnection. None or < 0 - do not reconnect.
loop (asyncio.events.AbstractEventLoop, optional) – Event loop.
response_timeout (float, optional) – socket.io event response timeout in seconds.
get (function (url, loop), optional) – HTTP GET coroutine.
socket_io (function (url, loop), optional) – socket.io connect coroutine.

coroutine add_media(link, append=True, temp=True)[source]¶

Add media link to playlist.

Parameters:	link (`str` or `cytube_bot.media_link.MediaLink`) – Media link. append (`bool`, optional) – `True` - append, `False` - insert after current item. temp (`bool`, optional) – `True` to add temporary item.
Returns:	Playlist item data.
Return type:	`dict`
Raises:	`cytube_bot.error.ChannelPermissionError` `cytube_bot.error.ChannelError` `ValueError`

coroutine chat(msg, meta=None)[source]¶

Send a chat message.

Parameters:	msg (`str`) – meta (`None` or `dict`, optional) –
Returns:	Message data.
Return type:	`dict`
Raises:	`cytube_bot.error.ChannelError` `cytube_bot.error.ChannelPermissionError`

coroutine clear_chat()[source]¶

Clear chat.

Raises:	`cytube_bot.error.ChannelPermissionError`

coroutine connect()[source]¶

Get server URL and connect.

Raises:	`cytube_bot.error.SocketIOError`

coroutine disconnect()[source]¶: Disconnect.

coroutine get_socket_config()[source]¶

Get server URL.

Raises:	`cytube_bot.error.SocketConfigError`

coroutine kick(user, reason='')[source]¶

Kick a user.

Parameters:	user (`str` or `cytube_bot.user.User`) – reason (`str`, optional) –
Raises:	`cytube_bot.error.ChannelError` `cytube_bot.error.ChannelPermissionError` `ValueError`

logger = <logging.Logger object>¶

coroutine login()[source]¶

Connect, join channel, log in.

Raises:	`cytube_bot.error.LoginError` `cytube_bot.error.SocketIOError`

coroutine move_media(item, after)[source]¶

Move a playlist item.

Parameters:	item (`int` or `cytube_bot.playlist.PlaylistItem`) – after (`int` or `cytube_bot.playlist.PlaylistItem`) –
Raises:	`cytube_bot.error.ChannelError` `cytube_bot.error.ChannelPermissionError` `ValueError`

off(event, *handlers)[source]¶

Remove event handlers.

Parameters:	event (`str`) – Event name. handlers (`list` of `function`) – Event handlers.

on(event, *handlers)[source]¶

Add event handlers.

Parameters:	event (`str`) – Event name. handlers (`list` of `function`) – Event handlers.

coroutine pause()[source]¶

Pause current media.

Raises:	`cytube_bot.error.ChannelPermissionError`

coroutine pm(to, msg, meta=None)[source]¶

Send a private chat message.

Parameters:	to (`str`) – msg (`str`) – meta (`None` or `dict`, optional) –
Returns:	Message data.
Return type:	`dict`
Raises:	`cytube_bot.error.ChannelPermissionError` `cytube_bot.error.ChannelError`

coroutine remove_leader()[source]¶: Remove leader.

coroutine remove_media(item)[source]¶

Remove playlist item.

Parameters:	item (`int` or `cytube_bot.playlist.PlaylistItem`) – Item to remove.
Raises:	`cytube_bot.error.ChannelPermissionError` `ValueError`

coroutine run()[source]¶: Main loop.

coroutine set_afk(value=True)[source]¶

Set bot AFK.

Parameters:	value (`bool`, optional) –
Raises:	`cytube_bot.error.ChannelPermissionError`

coroutine set_current_media(item)[source]¶

Set current playlist item.

Parameters:	item (`int` or `cytube_bot.playlist.PlaylistItem`) –
Raises:	`cytube_bot.error.ChannelError` `cytube_bot.error.ChannelPermissionError` `ValueError`

coroutine set_leader(user)[source]¶

Set leader.

Parameters:	user (`None` or `str` or `cytube_bot.user.User`) –
Raises:	`cytube_bot.error.ChannelPermissionError` `cytube_bot.error.ChannelError` `ValueError`

coroutine trigger(event, data)[source]¶

Trigger an event.

Parameters:	event (`str`) – Event name. data (`object`) – Event data.
Raises:	`cytube_bot.error.LoginError` `cytube_bot.error.Kicked`

cytube_bot.channel module¶

class cytube_bot.channel.Channel(name='', password=None)[source]¶

Bases: object

CyTube channel.

name¶: str

password¶: str or None

drink_count¶: int

voteskip_count¶: int

voteskip_need¶: int

motd¶: str

css¶: str

js¶: str

emotes¶: list of dict

permissions¶: dict of (str, float)

options¶: dict

userlist¶: cytube_bot.user.UserList

playlist¶: cytube_bot.playlist.Playlist

RANK_PRECISION = 0.0001¶

__init__(name='', password=None)[source]¶: Initialize self. See help(type(self)) for accurate signature.

check_permission(action, user, throw=True)[source]¶

Check if user has permission.

Parameters:	action (`str`) – Permission to check. user (`cytube_bot.user.User`) – User. throw (`bool`, optional) – `True` to raise exception if user does not have permission.
Returns:	`True` if user has permission.
Return type:	`bool`
Raises:	`ChannelPermissionError` – If user does not have permission. `ValueError` – If permission does not exist.

has_permission(action, user)[source]¶: check_permission(action, user, False)

logger = <logging.Logger object>¶

cytube_bot.error module¶

exception cytube_bot.error.ChannelError[source]¶: Bases: cytube_bot.error.CytubeError

exception cytube_bot.error.ChannelPermissionError[source]¶: Bases: cytube_bot.error.ChannelError

exception cytube_bot.error.ConnectionClosed[source]¶: Bases: cytube_bot.error.SocketIOError

exception cytube_bot.error.ConnectionFailed[source]¶: Bases: cytube_bot.error.SocketIOError

exception cytube_bot.error.CytubeError[source]¶: Bases: Exception

exception cytube_bot.error.Kicked[source]¶: Bases: cytube_bot.error.CytubeError

exception cytube_bot.error.LoginError[source]¶: Bases: cytube_bot.error.CytubeError

exception cytube_bot.error.PingTimeout[source]¶: Bases: cytube_bot.error.ConnectionClosed

exception cytube_bot.error.ProxyConfigError[source]¶: Bases: cytube_bot.error.CytubeError

exception cytube_bot.error.SocketConfigError[source]¶: Bases: cytube_bot.error.CytubeError

exception cytube_bot.error.SocketIOError[source]¶: Bases: Exception

cytube_bot.media_link module¶

class cytube_bot.media_link.MediaLink(type_, id_)[source]¶

Bases: object

Media link.

type¶: str – Link type.

id¶: str – Link ID.

FILE_TYPES¶: list of str – Supported raw file extensions.

URL_TO_LINK¶: list of (str, str, str) – (url regexp, type format string, id format string)

LINK_TO_URL¶: dict of (str, str) – (type, url format string)

FILE_TYPES = ['.mp4', '.flv', '.webm', '.ogg', '.ogv', '.mp3', '.mov', '.m4a']

LINK_TO_URL = {'cm': '{0}', 'dm': 'https://dailymotion.com/video/{0}', 'fi': '{0}', 'gd': 'https://drive.google.com/file/d/{0}', 'hb': 'https://smashcast.tv/{0}', 'hl': '{0}', 'im': 'https://imgur.com/a/{0}', 'li': 'https://livestream.com/{0}', 'rt': '{0}', 'sb': 'https://streamable.com/{0}', 'sc': 'https://soundcloud.com/{0}', 'tc': 'https://clips.twitch.tv/{0}', 'tw': 'https://twitch.tv/{0}', 'us': 'https://www.ustream.tv/{0}', 'vi': 'https://vimeo.com/{0}', 'vm': 'https://vid.me/{0}', 'yp': 'https://youtube.com/playlist?list={0}', 'yt': 'https://youtube.com/watch?v={0}'}

URL_TO_LINK = [('youtube\\.com/watch\\?([^#]+)', 'yt', '{v}'), ('youtu\\.be/([^\\?&#]+)', 'yt', '{0}'), ('youtube\\.com/playlist\\?([^#]+)', 'yp', '{list}'), ('clips\\.twitch\\.tv/([A-Za-z]+)', 'tc', '{0}'), ('twitch\\.tv/(?:.*?)/([cv])/(\\d+)', 'tv', '{0}{1}'), ('twitch\\.tv/videos/(\\d+)', 'tv', 'v{0}'), ('twitch\\.tv/([\\w-]+)', 'tw', '{0}'), ('livestream\\.com/([^\\?&#]+)', 'li', '{0}'), ('ustream\\.tv/([^\\?&#]+)', 'us', '{0}'), ('(?:hitbox|smashcast)\\.tv/([^\\?&#]+)', 'hb', '{0}'), ('vimeo\\.com/([^\\?&#]+)', 'vi', '{0}'), ('dailymotion\\.com/video/([^\\?&#_]+)', 'dm', '{0}'), ('imgur\\.com/a/([^\\?&#]+)', 'im', '{0}'), ('soundcloud\\.com/([^\\?&#]+)', 'sc', '{url}'), ('(?:docs|drive)\\.google\\.com/file/d/([a-zA-Z0-9_-]+)', 'gd', '{0}'), ('drive\\.google\\.com/open\\?id=([a-zA-Z0-9_-]+)', 'gd', '{0}'), ('vid\\.me/embedded/([\\w-]+)', 'vm', '{0}'), ('vid\\.me/([\\w-]+)', 'vm', '{0}'), ('(.*\\.m3u8)', 'hl', '{url}'), ('streamable\\.com/([\\w-]+)', 'sb', '{0}'), ('^dm:([^\\?&#_]+)', 'dm', '{0}'), ('^fi:(.*)', 'fi', '{0}'), ('^cm:(.*)', 'cm', '{0}'), ('^([a-z]{2}):([^\\?&#]+)', '{0}', '{1}')]

__init__(type_, id_)[source]¶: Initialize self. See help(type(self)) for accurate signature.

classmethod from_url(url)[source]¶

Create a media link from URL.

Parameters:	url (`str`) – Media URL.
Returns:
Return type:	MediaLink
Raises:	`ValueError` – If media URL is not supported.

logger = <logging.Logger object>¶

url¶: Media URL.

cytube_bot.playlist module¶

class cytube_bot.playlist.Playlist[source]¶

Bases: object

CyTube playlist.

time¶: int – Playlist duration in seconds.

current¶: None or cytube_bot.playlist.PlaylistItem – Current playlist item.

current_time¶: int – Current playlist item time in seconds.

locked¶: bool – True if playlist is locked.

paused¶: bool – True if playlist is paused.

queue¶: list of cytube_bot.playlist.PlaylistItem

__init__()[source]¶: Initialize self. See help(type(self)) for accurate signature.

add(after, item)[source]¶

Add playlist item.

Parameters:	after (`int` or `None`) – `int` - insert after item with ID, `None` - append. item (`dict` or `cytube_bot.playlist.PlaylistItem`) – Playlist item or data.

clear()[source]¶: Clear playlist.

current

get(uid)[source]¶

Get playlist item by ID.

Parameters:	item (`int`) – Playlist item ID.
Returns:
Return type:	`cytube_bot.playlist.PlaylistItem`
Raises:	`ValueError` – If item does not exist.

index(item)[source]¶

Get playlist item index by ID.

Parameters:	item (`int`) – Playlist item ID.
Returns:
Return type:	`int`
Raises:	`ValueError` – If item does not exist.

move(item, after)[source]¶

Move playlist item.

Parameters:	after (`int`) – item (`int`) –

remove(item)[source]¶

Remove playlist item.

Parameters:	item (`int` or `cytube_bot.playlist.PlaylistItem`) – Playlist item or ID.
Raises:	`ValueError` – If item does not exist.

class cytube_bot.playlist.PlaylistItem(data)[source]¶

Bases: object

CyTube playlist item.

link¶: cytube_bot.media_link.MediaLink – Media link.

uid¶: int – Playlist item ID.

temp¶: bool – True if item is temporary.

duration¶: int – Duration in seconds.

title¶: str

username¶: str

__init__(data)[source]¶: Initialize self. See help(type(self)) for accurate signature.

cytube_bot.proxy module¶

class cytube_bot.proxy.Socket(family=<AddressFamily.AF_INET: 2>, type=<SocketType.SOCK_STREAM: 1>, proto=0, fileno=None)[source]¶

Bases: socks.socksocket

SOCKS enabled socket (no proxy for localhost).

__init__(family=<AddressFamily.AF_INET: 2>, type=<SocketType.SOCK_STREAM: 1>, proto=0, fileno=None)[source]¶: Initialize self. See help(type(self)) for accurate signature.

bind(addr, *args, **kwargs)[source]¶

Implements proxy connection for UDP sockets.

Happens during the bind() phase.

connect(addr, *args, **kwargs)[source]¶

Connects to the specified destination through a proxy. Uses the same API as socket’s connect(). To select the proxy server, use set_proxy().

dest_pair - 2-tuple of (IP/hostname, port).

sendto(data, [flags, ]address) → count[source]¶: Like send(data, flags) but allows specifying the destination address. For IP sockets, the address is a pair (hostaddr, port).

set_proxy_for_address(addr)[source]¶

Unset proxy for localhost.

Parameters:	host (`str`) –

cytube_bot.proxy.getaddrinfo(host, port, *args, **kwargs)[source]¶

cytube_bot.proxy.set_proxy(addr, port, proxy_type=2, modules=None)[source]¶

Set SOCKS proxy for all connections.

Parameters:	addr (`str`) – Proxy IP. port (`int`) – Proxy port. proxy_type (`int`, optional) – `socks.SOCKS4` or `socks.SOCKS5`. modules (`None` or `list` of `types.ModuleType`, optional) – Modules to wrap (default: (sys.modules[__name__],)).
Raises:	`cytube_bot.error.ProxyConfigError` – If pysocks is not installed.

cytube_bot.proxy.wrap_module(module)[source]¶

cytube_bot.socket_io module¶

class cytube_bot.socket_io.SocketIO(websocket, config, qsize, loop)[source]¶

Bases: object

Asynchronous socket.io connection.

websocket¶: websockets.client.WebSocketClientProtocol – Websocket connection.

ping_interval¶: float – Ping interval in seconds.

ping_timeout¶: float – Ping timeout in seconds.

error¶: None or Exception

events¶: asyncio.Queue of ((str, object) or None) – Event queue.

response¶: list of cytube_bot.socket_io.SocketIOResponse

response_lock¶: asyncio.Lock

ping_task¶: asyncio.tasks.Task

recv_task¶: asyncio.tasks.Task

close_task¶: asyncio.tasks.Task

closing¶: asyncio.Event

closed¶: asyncio.Event

ping_response¶: asyncio.Event

loop¶: asyncio.events.AbstractEventLoop – Event loop.

__init__(websocket, config, qsize, loop)[source]¶

Parameters:	websocket (`websockets.client.WebSocketClientProtocol`) – Websocket connection. config (`dict`) – Websocket configuration. qsize (`int`) – Event queue size. loop (`asyncio.events.AbstractEventLoop`) – Event loop.

coroutine close()[source]¶: Close the connection.

coroutine classmethod connect(url, retry=-1, retry_delay=1, qsize=0, loop=None, get=<function get>, connect=<function connect>)[source]¶

Create a connection.

Parameters:	url (`str`) – socket.io URL. retry (`int`) – Maximum number of tries. retry_delay (`float`) – Delay between tries in seconds. qsize (`int`) – Event queue size. loop (`None` or `asyncio.events.AbstractEventLoop`) – Event loop. get (`function`) – HTTP GET request coroutine. connect (`function`) – Websocket connect coroutine.
Returns:
Return type:	`SocketIO`
Raises:	`ConnectionFailed` `asyncio.CancelledError`

coroutine emit(event, data, match_response=False, response_timeout=None)[source]¶

Send an event.

Parameters:	event (`str`) – Event name. data (`object`) – Event data. match_response (`function` or `None`, optional) – Response match function. response_timeout (`float` or `None`, optional) – Response timeout in seconds.
Returns:	Response data if `get_response` is `True`.
Return type:	`object`
Raises:	`asyncio.CancelledError` `SocketIOError`

error

logger = <logging.Logger object>¶

coroutine recv()[source]¶

Receive an event.

Returns:	Event name and data.
Return type:	(`str`, `object`)
Raises:	`ConnectionClosed`

class cytube_bot.socket_io.SocketIOResponse(match)[source]¶

Bases: object

socket.io event response.

id¶: int

match¶: function`(`str, object)

future¶: asyncio.Future

MAX_ID = 4294967296¶

__init__(match)[source]¶: Initialize self. See help(type(self)) for accurate signature.

cancel(ex=None)[source]¶

last_id = 0¶

static match_event(ev=None, data=None)[source]¶

set(value)[source]¶

cytube_bot.user module¶

class cytube_bot.user.User(name='', password=None, rank=-1, profile=None, meta=None)[source]¶

Bases: object

CyTube user.

name¶: str – Name.

password¶: None or str – Password.

uncloaked_ip¶: None or list of str – Uncloaked IP.

rank¶: float – Rank.

image¶: str – Profile image.

text¶: str – Profile text.

afk¶: bool – True if user is AFK.

muted¶: bool – True if user is muted.

smuted¶: bool – True if user is shadow muted.

__init__(name='', password=None, rank=-1, profile=None, meta=None)[source]¶: Initialize self. See help(type(self)) for accurate signature.

ip¶: Cloaked IP.

meta¶

profile¶

update(name=None, rank=None, profile=None, meta=None)[source]¶

Update user data.

Parameters:	name (`str` or `None`) – rank (`float` or `None`) – profile (`dict` or `None`) – meta (`dict` or `None`) –

class cytube_bot.user.UserList[source]¶

Bases: dict

CyTube user list.

count¶: int

leader¶: cytube_bot.user.User or None

__init__()[source]¶: Initialize self. See help(type(self)) for accurate signature.

add(user)[source]¶

Add a user.

Parameters:	user (`cytube_bot.user.User`) –
Raises:	`ValueError` – If user exists.

get(name)[source]¶

Get user by name.

Parameters:	name (`str`) – User name.
Returns:
Return type:	`cytube_bot.user.User`
Raises:	`ValueError` – If user does not exist.

leader

cytube_bot.util module¶

class cytube_bot.util.MessageParser(markup=[('code', None, '`', '`'), ('strong', None, '*', '*'), ('em', None, '_', '_'), ('s', None, '~~', '~~'), (None, {'class': 'spoiler'}, '[sp]', '[/sp]')])[source]¶

Bases: html.parser.HTMLParser

Chat message parser.

markup¶: None or list of (str, None or dict of (str, str), None or str, None or str)

message¶: str

tags¶: list of (str, str)

DEFAULT_MARKUP = [('code', None, '`', '`'), ('strong', None, '*', '*'), ('em', None, '_', '_'), ('s', None, '~~', '~~'), (None, {'class': 'spoiler'}, '[sp]', '[/sp]')]¶

__init__(markup=[('code', None, '`', '`'), ('strong', None, '*', '*'), ('em', None, '_', '_'), ('s', None, '~~', '~~'), (None, {'class': 'spoiler'}, '[sp]', '[/sp]')])[source]¶

Initialize and reset this instance.

If convert_charrefs is True (default: False), all character references are automatically converted to the corresponding Unicode characters. If strict is set to False (the default) the parser will parse invalid markup, otherwise it will raise an error. Note that the strict mode and argument are deprecated.

get_tag_markup(tag, attr)[source]¶

handle_data(data)[source]¶

handle_endtag(tag)[source]¶

handle_starttag(tag, attr)[source]¶

parse(msg)[source]¶

Parse a message.

Parameters:	msg (`str`) – Message to parse.
Returns:	Parsed message.
Return type:	`str`

class cytube_bot.util.Queue(maxsize=0, *, loop=None)[source]¶

Bases: asyncio.queues.Queue

coroutine join()[source]¶

task_done()[source]¶

cytube_bot.util.cloak_ip(ip, start=0)[source]¶

Cloak IP.

Parameters:	ip (`str`) – IP to cloak. start (`int`, optional) – Index of first cloaked part (0-3).
Returns:
Return type:	`str`

Examples

>>> cloak_ip('127.0.0.1')
'yFA.j8g.iXh.gvS'
>>> cloak_ip('127.0.0.1', 2)
'127.0.ou9.RBl'

cytube_bot.util.get(url, loop)[source]¶

Asynchronous HTTP GET request.

Parameters:	url (`str`) – loop (`asyncio.events.AbstractEventLoop`) –
Returns:
Return type:	`asyncio.futures.Future`

cytube_bot.util.ip_hash(string, length)[source]¶

cytube_bot.util.to_sequence(obj)[source]¶

Convert an object to sequence.

Parameters:	obj (`object`) –
Returns:
Return type:	`collections.Sequence`

Examples

>>> to_sequence(None)
()
>>> to_sequence(1)
(1,)
>>> to_sequence('str')
('str',)
>>> x = [0, 1, 2]
>>> to_sequence(x)
[0, 1, 2]
>>> to_sequence(x) is x
True

cytube_bot.util.uncloak_ip(ip, start=0)[source]¶

Uncloak IP.

Parameters:	ip (`str`) – Cloaked IP. start (`int` or `None`, optional) – Index of first cloaked part (0-3) (`None` - detect).
Returns:
Return type:	`list` of `str`

Examples

>>> uncloak_ip('yFA.j8g.iXh.gvS')
['127.0.0.1']
>>> uncloak_ip('127.0.ou9.RBl')
[]
>>> uncloak_ip('127.0.ou9.RBl', 2)
['127.0.0.1']
>>> uncloak_ip('127.0.ou9.RBl', None)
['127.0.0.1']

cytube_bot package¶

Submodules¶

cytube_bot.bot module¶

cytube_bot.channel module¶

cytube_bot.error module¶

cytube_bot.media_link module¶

cytube_bot.playlist module¶

cytube_bot.proxy module¶

cytube_bot.socket_io module¶

cytube_bot.user module¶

cytube_bot.util module¶

Module contents¶