Jerry/mau_mau_bot_bot
1
0
Fork 0
Code Issues Pull requests Releases 1 Wiki Activity

Compare commits

base: Jerry:old
Branches Tags
Jerry:master
Jerry:old
Jerry:artifacts
..
compare: Jerry:master
Branches Tags
Jerry:master
Jerry:old
Jerry:artifacts

No commits in common. "old" and "master" have entirely different histories.
old ... master

8 changed files with 316 additions and 955 deletions
Show stats Expand all files Collapse all files

7
README.md
View file

@ -1,12 +1,11 @@
# mau_mau_bot_bot # mau_mau_bot_bot
[![License: GPL v3](https://img.shields.io/badge/License-GPL%20v3-blue.svg)](./LICENSE) [![License: GPL v3](https://img.shields.io/badge/License-GPL%20v3-blue.svg)](./LICENSE)
This user bot can play uno with other humans. (Python3 only) This user bot can play uno with other humans.
## Installation ## Installation
The bot uses [Telethon](https://github.com/LonamiWebs/Telethon), a Pure Python 3 MTProto API Telegram client library. The bot uses [Telethon](https://github.com/LonamiWebs/Telethon), a Pure Python 3 MTProto API Telegram client library.
Please follow [this](https://telethon.readthedocs.io/en/stable/extra/basic/installation.html#installation) guide to install Telethon, and [create your api id and hash](https://telethon.readthedocs.io/en/stable/extra/basic/creating-a-client.html). Place your settings in `config.py`.
Then place your settings in `config.py`.
Run the bot with `python3 bot.py` or `./bot.py`. Run the bot with `python3 bot.py` or `./bot.py`.
## Usage ## Usage
@ -14,4 +13,4 @@ Add yourself and unobot to a group. Create a new game with `/new`.
The bot will join the game and play its cards automatically. The bot will join the game and play its cards automatically.
## Contributing ## Contributing
* Copyright © 2018 JerryXiao * Copyright © 2023 JerryXiao

266
bot.py
View file

@ -6,18 +6,20 @@ import re
import logging import logging
import time import time
import sys import sys
from telethon.tl.functions.messages import GetInlineBotResultsRequest, SendInlineBotResultRequest, SendMessageRequest, SetTypingRequest from telethon.tl.functions.messages import GetInlineBotResultsRequest, SendInlineBotResultRequest, \
SendMessageRequest, SetTypingRequest
from telethon.tl.types import SendMessageTypingAction, SendMessageCancelAction from telethon.tl.types import SendMessageTypingAction, SendMessageCancelAction
from telethon.tl.types import PeerUser, PeerChat, PeerChannel, User as _User, Chat as _Chat, Channel as _Channel from telethon.tl.types import PeerUser, PeerChat, PeerChannel, User as _User, Chat as _Chat, Channel as _Channel, \
InputPeerChannel, InputPeerChat, InputPeerUser
from telethon.errors.rpcbaseerrors import RPCError from telethon.errors.rpcbaseerrors import RPCError
import asyncio import asyncio
from game import Game from game import Game
from card import GREY_SET_ID
game = Game() game = Game()
from config import api_id, api_hash, PHONE, session_name, unobot_username, default_delay, print_cards, disable_all_commands, game_autostart, unogroup_chatname, game_consts from config import api_id, api_hash, PHONE, session_name, unobot_username, unobot_usernames, \
default_delay, print_cards, disable_all_commands, game_consts
game.delay = default_delay game.delay = default_delay
@ -37,49 +39,16 @@ my_firstname = client.get_me().first_name
# parse game_consts # parse game_consts
for key in game_consts: for key in game_consts:
game_consts[key] = game_consts[key].replace('%username%', my_username) game_consts[key] = game_consts[key].replace('{username}', my_username)
game_consts[key] = game_consts[key].replace('%firstname%', my_firstname) game_consts[key] = game_consts[key].replace('{firstname}', my_firstname)
def _print(*args, **kwargs):
print(*args, **kwargs)
sys.stdout.flush()
logger.info("Getting dialogs") logger.info("Getting dialogs")
for dialog in client.iter_dialogs(): for dialog in client.iter_dialogs():
pass pass
#_print(dialog)
logger.info("Done getting dialogs") logger.info("Done getting dialogs")
unobot = client.get_input_entity(unobot_username) unobot = client.get_input_entity(unobot_username)
unochat = None unochat = None
if unogroup_chatname:
unochat = client.get_input_entity(unogroup_chatname)
async def startgame_task():
logger.info("startgame_task run")
if not game.is_playing:
await client(SendMessageRequest(unochat, "/new@{}".format(unobot_username)))
await asyncio.sleep(60)
game.delay = 8
#try starting the game
await client(SendMessageRequest(unochat, "/start@{}".format(unobot_username)))
logger.info("startgame_task game started")
async def task_run():
import schedule_async as schedule
""" To start game automatically
"""
at_times = ("9:00", "12:00", "16:00", "18:00", "19:00", "20:00", "21:30")
logger.info("Coroutine task_run started")
for at_time in at_times:
schedule.every().day.at(at_time).do(startgame_task)
# for debug only
#schedule.every(20).seconds.do(testjob)
while True:
await schedule.run_pending()
await asyncio.sleep(60)
async def inline_query(fail=None): async def inline_query(fail=None):
""" This part handles interaction with unobot. """ This part handles interaction with unobot.
@ -145,13 +114,12 @@ async def inline_query(fail=None):
if len(result_id) == 36: if len(result_id) == 36:
# uuid result for grey cards # uuid result for grey cards
sset = result.document.attributes[1].stickerset sset = result.document.attributes[1].stickerset
if str(sset.id) == GREY_SET_ID: game.add_grey_card(sset.id, result.document.id)
game.add_grey_card(result.document.id) continue
continue
except (AttributeError, IndexError): except (AttributeError, IndexError):
pass pass
except Exception as err: except Exception as err:
logger.critical('Exception while getting grey cards, {}'.format(str(err))) logger.exception("while getting grey cards")
# get ordinary cards # get ordinary cards
try: try:
(result_id, anti_cheat) = result.id.split(':') (result_id, anti_cheat) = result.id.split(':')
@ -172,11 +140,12 @@ async def inline_query(fail=None):
if game.delay: if game.delay:
await typing_sleep(game.delay) await typing_sleep(game.delay)
if print_cards: if print_cards:
_print(game.print_cards()) print(game.print_cards())
callback_id = game.play_card() callback_id = game.play_card()
if not callback_id: if not callback_id:
callback_id = 'draw' callback_id = 'draw'
await client(SendMessageRequest(unochat, 'Error: No card can be played. Leaving game')) #await client(SendMessageRequest(unochat, 'Error: No card can be played. Leaving game'))
print(unochat, 'Error: No card can be played. Leaving game')
await client(SendMessageRequest(unochat, "/leave@{}".format(unobot_username))) await client(SendMessageRequest(unochat, "/leave@{}".format(unobot_username)))
return return
for tries in range(6): for tries in range(6):
@ -196,13 +165,13 @@ async def inline_query(fail=None):
return True return True
else: else:
logger.critical('Bad inline result from bot') logger.critical('Bad inline result from bot')
_print(bot_results) print(bot_results)
return None return None
def safety_check(chat_id, force=False): def safety_check(chat_id, force=False):
if SAFE_MODE or force: if SAFE_MODE or force:
safe_ids = [-1001000100100, ] safe_ids = [-100000000000]
if chat_id in safe_ids: if chat_id in safe_ids:
return True return True
else: else:
@ -253,23 +222,23 @@ def get_peer_id(peer, reverse=False):
return peer return peer
else: else:
peerid = None peerid = None
if type(peer) is PeerChannel: if type(peer) in (PeerChannel, InputPeerChannel):
peerid = getattr(peer, 'channel_id') peerid = getattr(peer, 'channel_id')
peerid = int('-100{}'.format(peerid)) peerid = int('-100{}'.format(peerid))
elif type(peer) in [ _Channel, _Chat]: elif type(peer) in [ _Channel, _Chat]:
peerid = getattr(peer, 'id') peerid = getattr(peer, 'id')
peerid = int('-100{}'.format(peerid)) peerid = int('-100{}'.format(peerid))
elif type(peer) is PeerChat: elif type(peer) in (PeerChat, InputPeerChat):
peerid = getattr(peer, 'chat_id') peerid = getattr(peer, 'chat_id')
peerid = int('-100{}'.format(peerid)) peerid = int('-100{}'.format(peerid))
elif type(peer) is PeerUser: elif type(peer) in (PeerUser, InputPeerUser):
peerid = getattr(peer, 'user_id') peerid = getattr(peer, 'user_id')
elif type(peer) is _User: elif type(peer) is _User:
peerid = getattr(peer, 'id') peerid = getattr(peer, 'id')
else: else:
_print("Error: ", peer) print("Error: ", peer)
if not peerid: if not peerid:
_print('W: Peer_id is none') print('W: Peer_id is none')
return peerid return peerid
@ -297,142 +266,36 @@ def display_username(user, atuser=False, shorten=False):
name += " ({})".format(user.username) name += " ({})".format(user.username)
return name return name
max_items = 10000
cached_ids = list()
cached_entity = list()
# It's a mess, but works
# Welcome to pr
async def mwt_get_entity(entity_type, client, peer, retry=0, from_group=None):
global max_items, cached_ids, cached_entity
def get(unique_id):
global cached_ids, cached_entity
try:
my_index = cached_ids.index(unique_id)
entity = cached_entity[my_index]
return entity
except ValueError:
return None
def store(unique_id, entity):
global cached_ids, cached_entity
cached_ids.append(unique_id)
cached_entity.append(entity)
while len(cached_ids) > max_items:
cached_ids.pop(0)
cached_entity.pop(0)
try:
if entity_type == 'group':
unique_id = get_peer_id(peer)
entity = get(unique_id)
#_print("cache")
if not entity:
#_print("new")
entity = await client.get_entity(peer)
elif entity_type == 'user':
unique_id = peer
entity = get(unique_id)
#_print("cache")
if not entity:
#_print("new")
entity = await client.get_entity(PeerUser(user_id=peer))
else:
return None
store(unique_id, entity)
return entity
except (ValueError, KeyError) as err:
if retry < 1:
retry += 1
if entity_type == 'group':
sys.stdout.write("[Fetching user from group] Error while getting chat: {}".format(err))
sys.stdout.flush()
elif entity_type == 'user':
sys.stdout.write("[Fetching user from group] Error while getting user: {}".format(err))
sys.stdout.flush()
if from_group:
for user in await client.get_participants(from_group):
unique_id = user.id
entity = user
if unique_id not in cached_ids:
store(unique_id, entity)
entity = await mwt_get_entity(entity_type, client, peer, retry=retry)
store(unique_id, entity)
return entity
else:
if entity_type == 'group':
sys.stdout.write("[Give up] Error while getting chat: {}".format(err))
sys.stdout.flush()
entity = EmptyChat(str(id))
elif entity_type == 'user':
sys.stdout.write("[Give up] Error while getting user: {}".format(err))
sys.stdout.flush()
entity = EmptyUser(first_name="PeerUser(user_id={})".format(peer))
store(unique_id, entity)
return entity
async def get_full_info(event):
'''
# full_user = client(GetFullUserRequest(id=PeerUser(user_id=msg.from_id)))
# full_chat = client(GetFullChatRequest(chat_id=chat_id))
# full_channel = client(GetFullChannelRequest(channel=PeerChannel(channel_id=None)))
# first_name = full_user.user.first_name
# last_name = full_user.user.last_name
# username = full_user.user.username
# title = full_chat.chats[0].title
# title = full_channel.chats[0].title
'''
orig_user_id = event.message.from_id
user_id = get_peer_id(PeerUser(user_id=orig_user_id))
if event.is_channel:
#channel = client.get_entity(event.message.to_id)
#user = client.get_entity(PeerUser(user_id=event.message.from_id))
channel = await mwt_get_entity('group', client, event.message.to_id)
user = await mwt_get_entity('user', client, event.message.from_id, from_group=event.message.to_id)
channel_id = get_peer_id(channel)
full_info = ['Channel', channel, user, channel_id, user_id]
elif event.is_group:
group = await mwt_get_entity('group', client, event.message.to_id)
group_id = get_peer_id(group)
user = await mwt_get_entity('user', client, event.message.from_id, from_group=event.message.to_id)
full_info = ['Group', group, user, group_id, user_id]
elif event.is_private:
user = await mwt_get_entity('user', client, event.message.from_id)
full_info = ['User', EmptyChat(), user, user_id, user_id]
else:
return None
return full_info
@client.on(events.NewMessage) @client.on(events.NewMessage)
async def new_msg_handler(event): async def new_msg_handler(event):
global unochat global unochat
#print(event) global unobot_username, unobot
#sys.stdout.flush()
full_info = await get_full_info(event)
msg = event.message msg = event.message
if msg.message and (not msg.media): group = await event.get_chat()
user = await event.get_sender()
if event.is_channel and msg.message and (not msg.media):
# Text handler # Text handler
logger.info("{} - {} - {}: {}".format(full_info[0], full_info[1].title, display_username(full_info[2]), msg.message))
if not safety_check(get_peer_id(msg.to_id)): if not safety_check(get_peer_id(msg.to_id)):
return return
logger.info("Group - {} - {}: {}".format(group.title, display_username(user), msg.message))
# react to commands # react to commands
if not disable_all_commands: if not disable_all_commands:
c = commandify(event.raw_text, wild_card=False) c = commandify(event.raw_text, wild_card=False)
if c[0]: if c[0]:
if c[0] == 'hello': if c[0] == 'hello':
await event.reply('hi!') await event.reply('hi!')
if c[0] in ['startgame', 'start', 'join'] and full_info[0] == "Channel": if c[0] in ['startgame', 'start', 'join']:
if game.is_playing: if game.is_playing:
await event.reply("I'm playing right now.") await event.reply("I'm playing right now.")
else: else:
unochat = msg.to_id unochat = msg.to_id
# get unobot name from args
if len(c) == 3 and len(c[2]) >= 1 and c[2][0].endswith('bot'):
unobot_username = c[2][0]
unobot = await client.get_input_entity(unobot_username)
game.join_game(get_peer_id(unochat)) game.join_game(get_peer_id(unochat))
await client(SendMessageRequest(unochat, "/join@{}".format(unobot_username))) await client(SendMessageRequest(unochat, "/join@{}".format(unobot_username)))
elif c[0] in ['stopgame', 'stop', 'leave'] and full_info[0] == "Channel": elif c[0] in ['stopgame', 'stop', 'leave']:
if game.is_playing: if game.is_playing:
game.leave_game(get_peer_id(unochat)) game.leave_game(get_peer_id(unochat))
game.stop_game() game.stop_game()
@ -442,13 +305,13 @@ async def new_msg_handler(event):
await client(SendMessageRequest(unochat, "/leave@{}".format(unobot_username))) await client(SendMessageRequest(unochat, "/leave@{}".format(unobot_username)))
else: else:
await event.reply("I'm not playing right now.") await event.reply("I'm not playing right now.")
elif c[0] in ['wait', 'delay'] and full_info[0] == "Channel": elif c[0] in ['wait', 'delay']:
if game.delay or (not game.is_playing): if game.delay or (not game.is_playing):
await event.reply("Nothing to do.") await event.reply("Nothing to do.")
else: else:
game.delay = 8 game.delay = 8
await event.reply("OK. {} seconds of delay has been set.".format(game.delay)) await event.reply("OK. {} seconds of delay has been set.".format(game.delay))
elif c[0] in ['nowait', 'nodelay'] and full_info[0] == "Channel": elif c[0] in ['nowait', 'nodelay']:
if game.delay and game.is_playing: if game.delay and game.is_playing:
myreply = "OK. {} seconds of delay has been removed.".format(game.delay) myreply = "OK. {} seconds of delay has been removed.".format(game.delay)
game.delay = None game.delay = None
@ -457,24 +320,38 @@ async def new_msg_handler(event):
await event.reply("Nothing to do.") await event.reply("Nothing to do.")
return return
# react to unobot # react to unobot
if full_info[2].username and full_info[2].username == unobot_username: if user.username and user.username in unobot_usernames:
if re.search(game_consts['myturn'], msg.message): if re.search(game_consts['myturn'], msg.message):
if re.search(game_consts['start'], msg.message): if re.search(game_consts['start'], msg.message):
# I'm the first player # I'm the first player
if not game.is_playing: if not game.is_playing:
unochat = msg.to_id unochat = msg.to_id
unobot_username = user.username
unobot = await client.get_input_entity(unobot_username)
if game.joined and get_peer_id(unochat) in game.joined: if game.joined and get_peer_id(unochat) in game.joined:
logger.info('Bot: Game started. I\'m the first player.') logger.info('Bot: Game started. I\'m the first player.')
game.start_game() game.start_game()
logger.info('Bot: It\'s my turn.') logger.info('Bot: It\'s my turn.')
if game.joined and get_peer_id(unochat) in game.joined: if game.joined and get_peer_id(msg.to_id) in game.joined:
if not game.is_playing: if not game.is_playing:
logger.info('Bot: Game started a long time ago. I joined midway.') logger.info('Bot: Game started a long time ago. I joined midway.')
unochat = msg.to_id
unobot_username = user.username
unobot = await client.get_input_entity(unobot_username)
game.start_game() game.start_game()
await inline_query() await inline_query()
else: else:
logger.info('I\'m not playing in {} - {}'.format(full_info[1].title, get_peer_id(unochat))) if not game.is_playing:
await client(SendMessageRequest(unochat, "/leave@{}".format(unobot_username))) logger.info('I\'m not playing in {} - {}, play anyway.'.format(group.title, get_peer_id(msg.to_id)))
unochat = msg.to_id
unobot_username = user.username
unobot = await client.get_input_entity(unobot_username)
game.join_game(get_peer_id(unochat))
game.start_game()
await inline_query()
else:
logger.info('I\'m not playing in {} - {}'.format(group.title, get_peer_id(msg.to_id)))
await client(SendMessageRequest(msg.to_id, "/leave@{}".format(user.username)))
elif re.search(game_consts['end'], msg.message): elif re.search(game_consts['end'], msg.message):
if game.is_playing: if game.is_playing:
logger.info('Bot: Game ended.') logger.info('Bot: Game ended.')
@ -488,11 +365,14 @@ async def new_msg_handler(event):
if default_delay: if default_delay:
await asyncio.sleep(default_delay) await asyncio.sleep(default_delay)
unochat = msg.to_id unochat = msg.to_id
unobot_username = user.username
game.join_game(get_peer_id(unochat)) game.join_game(get_peer_id(unochat))
await client(SendMessageRequest(unochat, "/join@{}".format(unobot_username))) await client(SendMessageRequest(unochat, "/join@{}".format(unobot_username)))
elif re.search(game_consts['start'], msg.message): elif re.search(game_consts['start'], msg.message):
if not game.is_playing: if not game.is_playing:
unochat = msg.to_id unochat = msg.to_id
unobot_username = user.username
unobot = await client.get_input_entity(unobot_username)
if game.joined and get_peer_id(unochat) in game.joined: if game.joined and get_peer_id(unochat) in game.joined:
logger.info('Bot: Game started.') logger.info('Bot: Game started.')
game.start_game() game.start_game()
@ -505,47 +385,11 @@ async def new_msg_handler(event):
game.leave_game(get_peer_id(unochat)) game.leave_game(get_peer_id(unochat))
game.stop_game() game.stop_game()
elif msg.media:
# Has media, interesting.
media = msg.media
type = None
if hasattr(media, 'photo'):
type = "photo"
logger.info("{} - {} - {}: [Photo]".format(full_info[0], full_info[1].title, display_username(full_info[2])))
elif hasattr(media, 'document'):
try:
if hasattr(media.document.attributes[1], 'stickerset'):
type = "Sticker"
logger.info("{} - {} - {}: [Sticker]:{}".format(full_info[0], full_info[1].title, display_username(full_info[2]), media.document.attributes[1].alt))
else:
type = "Document (file)"
logger.info("{} - {} - {}: [Document (file)]".format(full_info[0], full_info[1].title, display_username(full_info[2])))
except (AttributeError, IndexError):
type = "Document"
logger.info("{} - {} - {}: [Document]".format(full_info[0], full_info[1].title, display_username(full_info[2])))
else:
type = "Unknown media"
logger.info("{} - {} - {}: [Unknown media]".format(full_info[0], full_info[1].title, display_username(full_info[2])))
logger.debug("Media Type: {}".format(type))
# Handler complete
# for debug only
async def testjob():
print('aaa')
async def main(): async def main():
# requires python 3.7 +
#if game_autostart:
#asyncio.create_task(task_run())
await client.run_until_disconnected() await client.run_until_disconnected()
#await client.disconnected
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
# this can run on python 3.5
if game_autostart:
loop.create_task(task_run())
try: try:
loop.run_until_complete(main()) loop.run_until_complete(main())
except KeyboardInterrupt: except KeyboardInterrupt:

420
card.py
View file

@ -11,197 +11,239 @@ CHOOSE = 'colorchooser'
DRAW_FOUR = 'draw_four' DRAW_FOUR = 'draw_four'
SPECIALS = (CHOOSE, DRAW_FOUR) SPECIALS = (CHOOSE, DRAW_FOUR)
_ALL = {
"STICKERS": {
# those stickers below are only used for grey cards. "b_0": 43161736970240473,
"b_1": 43161736970240475,
# NOGREY is not used now "b_2": 43161736970240477,
# can't be used for userbot "b_3": 43161736970240479,
NOGREY = { "b_4": 43161736970240481,
'BQADBAAD-AIAAl9XmQABxEjEcFM-VHIC': 'option_draw', "b_5": 43161736970240483,
'BQADBAAD-gIAAl9XmQABcEkAAbaZ4SicAg': 'option_pass', "b_6": 43161736970240485,
'BQADBAADygIAAl9XmQABJoLfB9ntI2UC': 'option_bluff', "b_7": 43161736970240487,
'BQADBAADxAIAAl9XmQABC5v3Z77VLfEC': 'option_info' "b_8": 43161736970240489,
"b_9": 43161736970240491,
"b_draw": 43161736970240493,
"b_skip": 43161736970240497,
"b_reverse": 43161736970240495,
"g_0": 43161736970240503,
"g_1": 43161736970240505,
"g_2": 43161736970240507,
"g_3": 43161736970240509,
"g_4": 43161736970240511,
"g_5": 43161736970240513,
"g_6": 43161736970240515,
"g_7": 43161736970240517,
"g_8": 43161736970240519,
"g_9": 43161736970240521,
"g_draw": 43161736970240523,
"g_skip": 43161736970240527,
"g_reverse": 43161736970240525,
"r_0": 43161736970240529,
"r_1": 43161736970240531,
"r_2": 43161736970240533,
"r_3": 43161736970240535,
"r_4": 43161736970240537,
"r_5": 43161736970240539,
"r_6": 43161736970240541,
"r_7": 43161736970240543,
"r_8": 43161736970240545,
"r_9": 43161736970240547,
"r_draw": 43161736970240549,
"r_skip": 43161736970240553,
"r_reverse": 43161736970240551,
"y_0": 43161736970240555,
"y_1": 43161736970240557,
"y_2": 43161736970240559,
"y_3": 43161736970240561,
"y_4": 43161736970240563,
"y_5": 43161736970240565,
"y_6": 43161736970240567,
"y_7": 43161736970240569,
"y_8": 43161736970240571,
"y_9": 43161736970240573,
"y_draw": 43161736970240575,
"y_skip": 43161736970240579,
"y_reverse": 43161736970240577,
"draw_four": 43161736970240501,
"colorchooser": 43161736970240499
},
"STICKERS_GREY": {
"b_0": 43161736970240581,
"b_1": 43161736970240583,
"b_2": 43161736970240585,
"b_3": 43161736970240587,
"b_4": 43161736970240589,
"b_5": 43161736970240591,
"b_6": 43161736970240593,
"b_7": 43161736970240595,
"b_8": 43161736970240597,
"b_9": 43161736970240599,
"b_draw": 43161736970240601,
"b_skip": 43161736970240605,
"b_reverse": 43161736970240603,
"g_0": 43161736970240611,
"g_1": 43161736970240613,
"g_2": 43161736970240615,
"g_3": 43161736970240617,
"g_4": 43161736970240619,
"g_5": 43161736970240621,
"g_6": 43161736970240623,
"g_7": 43161736970240625,
"g_8": 43161736970240627,
"g_9": 43161736970240629,
"g_draw": 43161736970240631,
"g_skip": 43161736970240635,
"g_reverse": 43161736970240633,
"r_0": 43161736970240637,
"r_1": 43161736970240639,
"r_2": 43161736970240641,
"r_3": 43161736970240643,
"r_4": 43161736970240645,
"r_5": 43161736970240647,
"r_6": 43161736970240649,
"r_7": 43161736970240651,
"r_8": 43161736970240653,
"r_9": 43161736970240655,
"r_draw": 43161736970240657,
"r_skip": 43161736970240661,
"r_reverse": 43161736970240659,
"y_0": 43161736970240663,
"y_1": 43161736970240665,
"y_2": 43161736970240667,
"y_3": 43161736970240669,
"y_4": 43161736970240671,
"y_5": 43161736970240673,
"y_6": 43161736970240675,
"y_7": 43161736970240677,
"y_8": 43161736970240679,
"y_9": 43161736970240681,
"y_draw": 43161736970240683,
"y_skip": 43161736970240687,
"y_reverse": 43161736970240685,
"draw_four": 43161736970240609,
"colorchooser": 43161736970240607
},
"CB_STICKERS": {
"colorchooser": 5880002320636317358,
"draw_four": 5879982271728980065,
"r_0": 5877721426714169320,
"r_1": 5879880386514783717,
"r_2": 5879770246373445078,
"r_3": 5879933837382783126,
"r_4": 5877760081419835913,
"r_5": 5879900358112710487,
"r_6": 5877642725733436957,
"r_7": 5877749137843162933,
"r_8": 5877483266482639060,
"r_9": 5877311506445504217,
"r_draw": 5877319172962127805,
"r_reverse": 5879797923142701285,
"r_skip": 5879761158222646854,
"g_0": 5879503756537630286,
"g_1": 5879484605278457410,
"g_2": 5879660312390538712,
"g_3": 5879657340273168039,
"g_4": 5877333445138452550,
"g_5": 5879673223062228738,
"g_6": 5879937299126425174,
"g_7": 5879480885836780249,
"g_8": 5879703210523888872,
"g_9": 5879841577190297875,
"g_draw": 5879828352985993092,
"g_reverse": 5879536127706140722,
"g_skip": 5877663328691558252,
"b_0": 5877474899886346488,
"b_1": 5879595475564237681,
"b_2": 5879838149806396627,
"b_3": 5879644472551148854,
"b_4": 5879829976483630741,
"b_5": 5880005537566821550,
"b_6": 5879891184062567694,
"b_7": 5879693400818585891,
"b_8": 5877543314420403675,
"b_9": 5879815124486720957,
"b_draw": 5877337362148626071,
"b_reverse": 5879952477540847522,
"b_skip": 5879687933325217017,
"y_0": 5879826162552671917,
"y_1": 5879882976380063602,
"y_2": 5879670177930415760,
"y_3": 5879704855496363257,
"y_4": 5879532202106031757,
"y_5": 5879476148487851945,
"y_6": 5879621820893629813,
"y_7": 5879449949187345641,
"y_8": 5877679301674930177,
"y_9": 5879515370129198532,
"y_draw": 5879758787400699599,
"y_reverse": 5879737007621541967,
"y_skip": 5877368079754729022
},
"CB_STICKERS_GREY": {
"colorchooser": 5879664091961757349,
"draw_four": 5879690097988735793,
"r_0": 5877371103411704752,
"r_1": 5877302078992289364,
"r_2": 5879920475739524527,
"r_3": 5877448455772705743,
"r_4": 5877436661792509909,
"r_5": 5877206279746752859,
"r_6": 5879858589555757822,
"r_7": 5879564526029902234,
"r_8": 5879894658691108279,
"r_9": 5879865895295127863,
"r_draw": 5877318610321412183,
"r_reverse": 5880000104433193795,
"r_skip": 5877425756870544996,
"g_0": 5879665023969660659,
"g_1": 5879945807456636378,
"g_2": 5879827012956197138,
"g_3": 5879609584531804090,
"g_4": 5877240360312246189,
"g_5": 5877455293360640086,
"g_6": 5879591305150992178,
"g_7": 5877481174833566141,
"g_8": 5877624205834456665,
"g_9": 5879594152714311387,
"g_draw": 5877599861959822892,
"g_reverse": 5879475796300532820,
"g_skip": 5879569061515365986,
"b_0": 5877424945121727831,
"b_1": 5877244191423074517,
"b_2": 5879753839598374502,
"b_3": 5880000465210445948,
"b_4": 5879663598040518716,
"b_5": 5879889964291853901,
"b_6": 5879767643623264510,
"b_7": 5877307168528535701,
"b_8": 5879524471164899173,
"b_9": 5879799864467919022,
"b_draw": 5879549871601487285,
"b_reverse": 5879465080357130570,
"b_skip": 5879806680581018111,
"y_0": 5879699624226197743,
"y_1": 5879640207648623653,
"y_2": 5879913706871066164,
"y_3": 5879537051124109020,
"y_4": 5877545436134248386,
"y_5": 5879513377264374502,
"y_6": 5879961466907398077,
"y_7": 5879549613903449302,
"y_8": 5877716118134591342,
"y_9": 5877734590788931385,
"y_draw": 5879740731358187243,
"y_reverse": 5879596270133186486,
"y_skip": 5879469203525734731
},
"STICKERS_SET": 43161736970240002,
"STICKERS_GREY_SET": 43161736970240002,
"CB_STICKERS_SET": 43161736970240012,
"CB_STICKERS_GREY_SET": 43161736970240013
} }
# STICKERS is not used now
# can't be used for userbot
STICKERS = {
'BQADBAAD2QEAAl9XmQAB--inQsYcLTsC': 'b_0',
'BQADBAAD2wEAAl9XmQABBzh4U-rFicEC': 'b_1',
'BQADBAAD3QEAAl9XmQABo3l6TT0MzKwC': 'b_2',
'BQADBAAD3wEAAl9XmQAB2y-3TSapRtIC': 'b_3',
'BQADBAAD4QEAAl9XmQABT6nhOuolqKYC': 'b_4',
'BQADBAAD4wEAAl9XmQABwRfmekGnpn0C': 'b_5',
'BQADBAAD5QEAAl9XmQABQITgUsEsqxsC': 'b_6',
'BQADBAAD5wEAAl9XmQABVhPF6EcfWjEC': 'b_7',
'BQADBAAD6QEAAl9XmQABP6baig0pIvYC': 'b_8',
'BQADBAAD6wEAAl9XmQAB0CQdsQs_pXIC': 'b_9',
'BQADBAAD7QEAAl9XmQAB00Wii7R3gDUC': 'b_draw',
'BQADBAAD8QEAAl9XmQAB_RJHYKqlc-wC': 'b_skip',
'BQADBAAD7wEAAl9XmQABo7D0B9NUPmYC': 'b_reverse',
'BQADBAAD9wEAAl9XmQABb8CaxxsQ-Y8C': 'g_0',
'BQADBAAD-QEAAl9XmQAB9B6ti_j6UB0C': 'g_1',
'BQADBAAD-wEAAl9XmQABYpLjOzbRz8EC': 'g_2',
'BQADBAAD_QEAAl9XmQABKvc2ZCiY-D8C': 'g_3',
'BQADBAAD_wEAAl9XmQABJB52wzPdHssC': 'g_4',
'BQADBAADAQIAAl9XmQABp_Ep1I4GA2cC': 'g_5',
'BQADBAADAwIAAl9XmQABaaMxxa4MihwC': 'g_6',
'BQADBAADBQIAAl9XmQABv5Q264Crz8gC': 'g_7',
'BQADBAADBwIAAl9XmQABjMH-X9UHh8sC': 'g_8',
'BQADBAADCQIAAl9XmQAB26fZ2fW7vM0C': 'g_9',
'BQADBAADCwIAAl9XmQAB64jIZrgXrQUC': 'g_draw',
'BQADBAADDwIAAl9XmQAB17yhhnh46VQC': 'g_skip',
'BQADBAADDQIAAl9XmQAB_xcaab0DkegC': 'g_reverse',
'BQADBAADEQIAAl9XmQABiUfr1hz-zT8C': 'r_0',
'BQADBAADEwIAAl9XmQAB5bWfwJGs6Q0C': 'r_1',
'BQADBAADFQIAAl9XmQABHR4mg9Ifjw0C': 'r_2',
'BQADBAADFwIAAl9XmQABYBx5O_PG2QIC': 'r_3',
'BQADBAADGQIAAl9XmQABTQpGrlvet3cC': 'r_4',
'BQADBAADGwIAAl9XmQABbdLt4gdntBQC': 'r_5',
'BQADBAADHQIAAl9XmQABqEI274p3lSoC': 'r_6',
'BQADBAADHwIAAl9XmQABCw8u67Q4EK4C': 'r_7',
'BQADBAADIQIAAl9XmQAB8iDJmLxp8ogC': 'r_8',
'BQADBAADIwIAAl9XmQAB_HCAww1kNGYC': 'r_9',
'BQADBAADJQIAAl9XmQABuz0OZ4l3k6MC': 'r_draw',
'BQADBAADKQIAAl9XmQAC2AL5Ok_ULwI': 'r_skip',
'BQADBAADJwIAAl9XmQABu2tIeQTpDvUC': 'r_reverse',
'BQADBAADKwIAAl9XmQAB_nWoNKe8DOQC': 'y_0',
'BQADBAADLQIAAl9XmQABVprAGUDKgOQC': 'y_1',
'BQADBAADLwIAAl9XmQABqyT4_YTm54EC': 'y_2',
'BQADBAADMQIAAl9XmQABGC-Xxg_N6fIC': 'y_3',
'BQADBAADMwIAAl9XmQABbc-ZGL8kApAC': 'y_4',
'BQADBAADNQIAAl9XmQAB67QJZIF6XAcC': 'y_5',
'BQADBAADNwIAAl9XmQABJg_7XXoITsoC': 'y_6',
'BQADBAADOQIAAl9XmQABVrd7OcS2k34C': 'y_7',
'BQADBAADOwIAAl9XmQABRpJSahBWk3EC': 'y_8',
'BQADBAADPQIAAl9XmQAB9MwJWKLJogYC': 'y_9',
'BQADBAADPwIAAl9XmQABaPYK8oYg84cC': 'y_draw',
'BQADBAADQwIAAl9XmQABO_AZKtxY6IMC': 'y_skip',
'BQADBAADQQIAAl9XmQABZdQFahGG6UQC': 'y_reverse',
'BQADBAAD9QEAAl9XmQABVlkSNfhn76cC': 'draw_four',
'BQADBAAD8wEAAl9XmQABl9rUOPqx4E4C': 'colorchooser',
'BQADBAAD-AIAAl9XmQABxEjEcFM-VHIC': 'option_draw',
'BQADBAAD-gIAAl9XmQABcEkAAbaZ4SicAg': 'option_pass',
'BQADBAADygIAAl9XmQABJoLfB9ntI2UC': 'option_bluff',
'BQADBAADxAIAAl9XmQABC5v3Z77VLfEC': 'option_info'
}
# can't be used for userbot
STICKERS_GREY = {
'BQADBAADRQIAAl9XmQAB1IfkQ5xAiK4C': 'b_0',
'BQADBAADRwIAAl9XmQABbWvhTeKBii4C': 'b_1',
'BQADBAADSQIAAl9XmQABS1djHgyQokMC': 'b_2',
'BQADBAADSwIAAl9XmQABwQ6VTbgY-MIC': 'b_3',
'BQADBAADTQIAAl9XmQABAlKUYha8YccC': 'b_4',
'BQADBAADTwIAAl9XmQABMvx8xVDnhUEC': 'b_5',
'BQADBAADUQIAAl9XmQABDEbhP1Zd31kC': 'b_6',
'BQADBAADUwIAAl9XmQABXb5XQBBaAnIC': 'b_7',
'BQADBAADVQIAAl9XmQABgL5HRDLvrjgC': 'b_8',
'BQADBAADVwIAAl9XmQABtO3XDQWZLtYC': 'b_9',
'BQADBAADWQIAAl9XmQAB2kk__6_2IhMC': 'b_draw',
'BQADBAADXQIAAl9XmQABEGJI6CaH3vcC': 'b_skip',
'BQADBAADWwIAAl9XmQAB_kZA6UdHXU8C': 'b_reverse',
'BQADBAADYwIAAl9XmQABGD5a9oG7Yg4C': 'g_0',
'BQADBAADZQIAAl9XmQABqwABZHAXZIg0Ag': 'g_1',
'BQADBAADZwIAAl9XmQABTI3mrEhojRkC': 'g_2',
'BQADBAADaQIAAl9XmQABVi3rUyzWS3YC': 'g_3',
'BQADBAADawIAAl9XmQABZIf5ThaXnpUC': 'g_4',
'BQADBAADbQIAAl9XmQABNndVJSQCenIC': 'g_5',
'BQADBAADbwIAAl9XmQABpoy1c4ZkrvwC': 'g_6',
'BQADBAADcQIAAl9XmQABDeaT5fzxwREC': 'g_7',
'BQADBAADcwIAAl9XmQABLIQ06ZM5NnAC': 'g_8',
'BQADBAADdQIAAl9XmQABel-mC7eXGsMC': 'g_9',
'BQADBAADdwIAAl9XmQABOHEpxSztCf8C': 'g_draw',
'BQADBAADewIAAl9XmQABDaQdMxjjPsoC': 'g_skip',
'BQADBAADeQIAAl9XmQABek1lGz7SJNAC': 'g_reverse',
'BQADBAADfQIAAl9XmQABWrxoiXcsg0EC': 'r_0',
'BQADBAADfwIAAl9XmQABlav-bkgSgRcC': 'r_1',
'BQADBAADgQIAAl9XmQABDjZkqfJ4AdAC': 'r_2',
'BQADBAADgwIAAl9XmQABT7lH7VVcy3MC': 'r_3',
'BQADBAADhQIAAl9XmQAB1arPC5x0LrwC': 'r_4',
'BQADBAADhwIAAl9XmQABWvs7xkCDldkC': 'r_5',
'BQADBAADiQIAAl9XmQABjwABH5ZonWn8Ag': 'r_6',
'BQADBAADiwIAAl9XmQABjekJfm4fBDIC': 'r_7',
'BQADBAADjQIAAl9XmQABqFjchpsJeEkC': 'r_8',
'BQADBAADjwIAAl9XmQAB-sKdcgABdNKDAg': 'r_9',
'BQADBAADkQIAAl9XmQABtw9RPVDHZOQC': 'r_draw',
'BQADBAADlQIAAl9XmQABtG2GixCxtX4C': 'r_skip',
'BQADBAADkwIAAl9XmQABz2qyEbabnVsC': 'r_reverse',
'BQADBAADlwIAAl9XmQABAb3ZwTGS1lMC': 'y_0',
'BQADBAADmQIAAl9XmQAB9v5qJk9R0x8C': 'y_1',
'BQADBAADmwIAAl9XmQABCsgpRHC2g-cC': 'y_2',
'BQADBAADnQIAAl9XmQAB3kLLXCv-qY0C': 'y_3',
'BQADBAADnwIAAl9XmQAB7R_y-NexNLIC': 'y_4',
'BQADBAADoQIAAl9XmQABl-7mwsjD-cMC': 'y_5',
'BQADBAADowIAAl9XmQABwbVsyv2MfPkC': 'y_6',
'BQADBAADpQIAAl9XmQABoBqC0JsemVwC': 'y_7',
'BQADBAADpwIAAl9XmQABpkwAAeh9ldlHAg': 'y_8',
'BQADBAADqQIAAl9XmQABpSBEUfd4IM8C': 'y_9',
'BQADBAADqwIAAl9XmQABMt-2zW0VYb4C': 'y_draw',
'BQADBAADrwIAAl9XmQABIDf-_TuuxtEC': 'y_skip',
'BQADBAADrQIAAl9XmQABm9M0Zh-_UwkC': 'y_reverse',
'BQADBAADYQIAAl9XmQAB_HWlvZIscDEC': 'draw_four',
'BQADBAADXwIAAl9XmQABY_ksDdMex-wC': 'colorchooser'
}
# sticker_id
#sticker_id = [['b_0', '43161736970240581'], ['b_1', '43161736970240583'], ['b_2', '43161736970240585'], ['b_3', '43161736970240587'], ['b_4', '43161736970240589'], ['b_5', '43161736970240591'], ['b_6', '43161736970240593'], ['b_7', '43161736970240595'], ['b_8', '43161736970240597'], ['b_9', '43161736970240599'], ['b_draw', '43161736970240601'], ['b_skip', '43161736970240605'], ['b_reverse', '43161736970240603'], ['g_0', '43161736970240611'], ['g_1', '43161736970240613'], ['g_2', '43161736970240615'], ['g_3', '43161736970240617'], ['g_4', '43161736970240619'], ['g_5', '43161736970240621'], ['g_6', '43161736970240623'], ['g_7', '43161736970240625'], ['g_8', '43161736970240627'], ['g_9', '43161736970240629'], ['g_draw', '43161736970240631'], ['g_skip', '43161736970240635'], ['g_reverse', '43161736970240633'], ['r_0', '43161736970240637'], ['r_1', '43161736970240639'], ['r_2', '43161736970240641'], ['r_3', '43161736970240643'], ['r_4', '43161736970240645'], ['r_5', '43161736970240647'], ['r_6', '43161736970240649'], ['r_7', '43161736970240651'], ['r_8', '43161736970240653'], ['r_9', '43161736970240655'], ['r_draw', '43161736970240657'], ['r_skip', '43161736970240661'], ['r_reverse', '43161736970240659'], ['y_0', '43161736970240663'], ['y_1', '43161736970240665'], ['y_2', '43161736970240667'], ['y_3', '43161736970240669'], ['y_4', '43161736970240671'], ['y_5', '43161736970240673'], ['y_6', '43161736970240675'], ['y_7', '43161736970240677'], ['y_8', '43161736970240679'], ['y_9', '43161736970240681'], ['y_draw', '43161736970240683'], ['y_skip', '43161736970240687'], ['y_reverse', '43161736970240685'], ['draw_four', '43161736970240609'], ['colorchooser', '43161736970240607']]
GREY_SET_ID = '43161736970240002'
# only grey ordinary cards are useful # only grey ordinary cards are useful
GREY_ID = { GREY_ID = {
'43161736970240581': 'b_0', _ALL["STICKERS_GREY_SET"]: {v: k for k, v in _ALL["STICKERS_GREY"].items()},
'43161736970240583': 'b_1', _ALL["CB_STICKERS_GREY_SET"]: {v: k for k, v in _ALL["CB_STICKERS_GREY"].items()},
'43161736970240585': 'b_2',
'43161736970240587': 'b_3',
'43161736970240589': 'b_4',
'43161736970240591': 'b_5',
'43161736970240593': 'b_6',
'43161736970240595': 'b_7',
'43161736970240597': 'b_8',
'43161736970240599': 'b_9',
'43161736970240601': 'b_draw',
'43161736970240605': 'b_skip',
'43161736970240603': 'b_reverse',
'43161736970240611': 'g_0',
'43161736970240613': 'g_1',
'43161736970240615': 'g_2',
'43161736970240617': 'g_3',
'43161736970240619': 'g_4',
'43161736970240621': 'g_5',
'43161736970240623': 'g_6',
'43161736970240625': 'g_7',
'43161736970240627': 'g_8',
'43161736970240629': 'g_9',
'43161736970240631': 'g_draw',
'43161736970240635': 'g_skip',
'43161736970240633': 'g_reverse',
'43161736970240637': 'r_0',
'43161736970240639': 'r_1',
'43161736970240641': 'r_2',
'43161736970240643': 'r_3',
'43161736970240645': 'r_4',
'43161736970240647': 'r_5',
'43161736970240649': 'r_6',
'43161736970240651': 'r_7',
'43161736970240653': 'r_8',
'43161736970240655': 'r_9',
'43161736970240657': 'r_draw',
'43161736970240661': 'r_skip',
'43161736970240659': 'r_reverse',
'43161736970240663': 'y_0',
'43161736970240665': 'y_1',
'43161736970240667': 'y_2',
'43161736970240669': 'y_3',
'43161736970240671': 'y_4',
'43161736970240673': 'y_5',
'43161736970240675': 'y_6',
'43161736970240677': 'y_7',
'43161736970240679': 'y_8',
'43161736970240681': 'y_9',
'43161736970240683': 'y_draw',
'43161736970240687': 'y_skip',
'43161736970240685': 'y_reverse'
} }

26
config.py
View file

@ -1,27 +1,23 @@
api_id = 123456 api_id = 50322
api_hash = 'ffffffffffffffffffffffffffffff' api_hash = '9ff1a639196c0779c86dd661af8522ba'
PHONE = '+10001000000' #PHONE = '+10000000000'
session_name = 'session1' PHONE = lambda: input('Please enter your phone: ')
session_name = 'session'
unobot_username = 'unobot' unobot_usernames = ['unobot', 'ffee1822_bot']
unobot_username = unobot_usernames[-1]
# should be a int or None # should be a int or None
default_delay = None default_delay = None
# print all of the bot's cards # print all of the bot's cards
print_cards = True print_cards = True
# if true, the bot will not react to any command-like things # if true, the bot will not react to any command-like things
disable_all_commands = False disable_all_commands = True
game_autostart = False # {username} {firstname} is available
# if game_autostart is True, the following fields are needed
# can also be a link
unogroup_chatname = None
#unogroup_chatname = 'https://t.me/xxxx'
# %username% %firstname% is available
game_consts = { game_consts = {
'create' : 'Created a new game!|创建新游戏成功！|已經開始咗新一盤!|已開始新的遊戲!|Novo jogo criado!', #en_US, zh_CN, zh_HK, zh_TW, pt_BR 'create' : 'Created a new game!|创建新游戏成功！|已經開始咗新一盤!|已開始新的遊戲!|Novo jogo criado!', #en_US, zh_CN, zh_HK, zh_TW, pt_BR
'end' : 'Game ended', 'end' : 'Game ended',
'start' : 'First player:', 'start' : 'First player:',
'win' : '%firstname% won!', 'win' : '{firstname} won!',
'myturn' : '(@%username%)' 'myturn' : '(@{username})'
} }

12
game.py
View file

@ -21,7 +21,9 @@ def cards_sum(deck):
# r, b, g, y # r, b, g, y
card_count = {RED: 0, BLUE: 0, GREEN: 0, YELLOW: 0} card_count = {RED: 0, BLUE: 0, GREEN: 0, YELLOW: 0}
for card in deck: for card in deck:
card_count[color_from_str(card)] += 1 c = color_from_str(card)
if c in card_count:
card_count[c] += 1
return card_count return card_count
def color_choice(deck, greydeck): def color_choice(deck, greydeck):
@ -114,9 +116,9 @@ class Game():
return str(self.choose_color + self.old_deck + ["[u]" + s for s in self.old_greydeck] + self.special + self.functional) return str(self.choose_color + self.old_deck + ["[u]" + s for s in self.old_greydeck] + self.special + self.functional)
else: else:
return str(self.deck + ["[u]" + s for s in self.greydeck] + self.special + self.functional) return str(self.deck + ["[u]" + s for s in self.greydeck] + self.special + self.functional)
def add_grey_card(self, card_id): def add_grey_card(self, set_id, card_id):
"""get grey_cards from id, and add them""" """get grey_cards from set_id + card_id, and add them"""
grey_card = GREY_ID.get(str(card_id), None) grey_card = GREY_ID.get(int(set_id), {}).get(int(card_id), None)
if grey_card: if grey_card:
self.greydeck.append(grey_card) self.greydeck.append(grey_card)
else: else:
@ -167,7 +169,7 @@ class Game():
return 'pass' return 'pass'
# still no? call his bluff! # still no? call his bluff!
else: else:
if 'call_bluff' in self.functional and randchance(0.4): if 'call_bluff' in self.functional and randchance(0.05):
return 'call_bluff' return 'call_bluff'
# what is left? probably draw( # what is left? probably draw(
else: else:

2
requirements.txt
View file

@ -1 +1 @@
telethon>=1.4.3 Telethon==1.28.1

7
run_bot_windows.py Normal file
View file

@ -0,0 +1,7 @@
import traceback, sys, pathlib, telethon, telethon.sync
sys.argv[0]='bot.py'
try:
exec(pathlib.Path("bot.py").read_text())
except:
traceback.print_exc()
input("Press Enter to exit.")

529
schedule_async.py
View file

@ -1,529 +0,0 @@
"""
Modified to work with asyncio by Jerry Xiao <jerry at mail.jerryxiao.cc>
Don't tell me there's aiocron. I don't know.
Python job scheduling for humans.
github.com/dbader/schedule
An in-process scheduler for periodic jobs that uses the builder pattern
for configuration. Schedule lets you run Python functions (or any other
callable) periodically at pre-determined intervals using a simple,
human-friendly syntax.
Inspired by Addam Wiggins' article "Rethinking Cron" [1] and the
"clockwork" Ruby module [2][3].
Features:
- A simple to use API for scheduling jobs.
- Very lightweight and no external dependencies.
- Excellent test coverage.
- Tested on Python 2.7, 3.5 to 3.7
Usage:
>>> import schedule_async
>>> import time
>>> def job(message='stuff'):
>>> print("I'm working on:", message)
>>> schedule.every(10).minutes.do(job)
>>> schedule.every(5).to(10).days.do(job)
>>> schedule.every().hour.do(job, message='things')
>>> schedule.every().day.at("10:30").do(job)
>>> while True:
>>> await schedule.run_pending()
>>> await asyncio.sleep(1)
[1] https://adam.herokuapp.com/past/2010/4/13/rethinking_cron/
[2] https://github.com/Rykian/clockwork
[3] https://adam.herokuapp.com/past/2010/6/30/replace_cron_with_clockwork/
"""
import collections
import datetime
import functools
import logging
import random
import time
import asyncio
logger = logging.getLogger('schedule')
class CancelJob(object):
"""
Can be returned from a job to unschedule itself.
"""
pass
class Scheduler(object):
"""
Objects instantiated by the :class:`Scheduler <Scheduler>` are
factories to create jobs, keep record of scheduled jobs and
handle their execution.
"""
def __init__(self):
self.jobs = []
async def run_pending(self):
"""
Run all jobs that are scheduled to run.
Please note that it is *intended behavior that run_pending()
does not run missed jobs*. For example, if you've registered a job
that should run every minute and you only call run_pending()
in one hour increments then your job won't be run 60 times in
between but only once.
"""
runnable_jobs = (job for job in self.jobs if job.should_run)
for job in sorted(runnable_jobs):
await self._run_job(job)
async def run_all(self, delay_seconds=0):
"""
Run all jobs regardless if they are scheduled to run or not.
A delay of `delay` seconds is added between each job. This helps
distribute system load generated by the jobs more evenly
over time.
:param delay_seconds: A delay added between every executed job
"""
logger.info('Running *all* %i jobs with %is delay inbetween',
len(self.jobs), delay_seconds)
for job in self.jobs[:]:
await self._run_job(job)
await asyncio.sleep(delay_seconds)
def clear(self, tag=None):
"""
Deletes scheduled jobs marked with the given tag, or all jobs
if tag is omitted.
:param tag: An identifier used to identify a subset of
jobs to delete
"""
if tag is None:
del self.jobs[:]
else:
self.jobs[:] = (job for job in self.jobs if tag not in job.tags)
def cancel_job(self, job):
"""
Delete a scheduled job.
:param job: The job to be unscheduled
"""
try:
self.jobs.remove(job)
except ValueError:
pass
def every(self, interval=1):
"""
Schedule a new periodic job.
:param interval: A quantity of a certain time unit
:return: An unconfigured :class:`Job <Job>`
"""
job = Job(interval, self)
return job
async def _run_job(self, job):
ret = await job.run()
if isinstance(ret, CancelJob) or ret is CancelJob:
self.cancel_job(job)
@property
def next_run(self):
"""
Datetime when the next job should run.
:return: A :class:`~datetime.datetime` object
"""
if not self.jobs:
return None
return min(self.jobs).next_run
@property
def idle_seconds(self):
"""
:return: Number of seconds until
:meth:`next_run <Scheduler.next_run>`.
"""
return (self.next_run - datetime.datetime.now()).total_seconds()
class Job(object):
"""
A periodic job as used by :class:`Scheduler`.
:param interval: A quantity of a certain time unit
:param scheduler: The :class:`Scheduler <Scheduler>` instance that
this job will register itself with once it has
been fully configured in :meth:`Job.do()`.
Every job runs at a given fixed time interval that is defined by:
* a :meth:`time unit <Job.second>`
* a quantity of `time units` defined by `interval`
A job is usually created and returned by :meth:`Scheduler.every`
method, which also defines its `interval`.
"""
def __init__(self, interval, scheduler=None):
self.interval = interval # pause interval * unit between runs
self.latest = None # upper limit to the interval
self.job_func = None # the job job_func to run
self.unit = None # time units, e.g. 'minutes', 'hours', ...
self.at_time = None # optional time at which this job runs
self.last_run = None # datetime of the last run
self.next_run = None # datetime of the next run
self.period = None # timedelta between runs, only valid for
self.start_day = None # Specific day of the week to start on
self.tags = set() # unique set of tags for the job
self.scheduler = scheduler # scheduler to register with
def __lt__(self, other):
"""
PeriodicJobs are sortable based on the scheduled time they
run next.
"""
return self.next_run < other.next_run
def __repr__(self):
def format_time(t):
return t.strftime('%Y-%m-%d %H:%M:%S') if t else '[never]'
timestats = '(last run: %s, next run: %s)' % (
format_time(self.last_run), format_time(self.next_run))
if hasattr(self.job_func, '__name__'):
job_func_name = self.job_func.__name__
else:
job_func_name = repr(self.job_func)
args = [repr(x) for x in self.job_func.args]
kwargs = ['%s=%s' % (k, repr(v))
for k, v in self.job_func.keywords.items()]
call_repr = job_func_name + '(' + ', '.join(args + kwargs) + ')'
if self.at_time is not None:
return 'Every %s %s at %s do %s %s' % (
self.interval,
self.unit[:-1] if self.interval == 1 else self.unit,
self.at_time, call_repr, timestats)
else:
fmt = (
'Every %(interval)s ' +
('to %(latest)s ' if self.latest is not None else '') +
'%(unit)s do %(call_repr)s %(timestats)s'
)
return fmt % dict(
interval=self.interval,
latest=self.latest,
unit=(self.unit[:-1] if self.interval == 1 else self.unit),
call_repr=call_repr,
timestats=timestats
)
@property
def second(self):
assert self.interval == 1, 'Use seconds instead of second'
return self.seconds
@property
def seconds(self):
self.unit = 'seconds'
return self
@property
def minute(self):
assert self.interval == 1, 'Use minutes instead of minute'
return self.minutes
@property
def minutes(self):
self.unit = 'minutes'
return self
@property
def hour(self):
assert self.interval == 1, 'Use hours instead of hour'
return self.hours
@property
def hours(self):
self.unit = 'hours'
return self
@property
def day(self):
assert self.interval == 1, 'Use days instead of day'
return self.days
@property
def days(self):
self.unit = 'days'
return self
@property
def week(self):
assert self.interval == 1, 'Use weeks instead of week'
return self.weeks
@property
def weeks(self):
self.unit = 'weeks'
return self
@property
def monday(self):
assert self.interval == 1, 'Use mondays instead of monday'
self.start_day = 'monday'
return self.weeks
@property
def tuesday(self):
assert self.interval == 1, 'Use tuesdays instead of tuesday'
self.start_day = 'tuesday'
return self.weeks
@property
def wednesday(self):
assert self.interval == 1, 'Use wedesdays instead of wednesday'
self.start_day = 'wednesday'
return self.weeks
@property
def thursday(self):
assert self.interval == 1, 'Use thursdays instead of thursday'
self.start_day = 'thursday'
return self.weeks
@property
def friday(self):
assert self.interval == 1, 'Use fridays instead of friday'
self.start_day = 'friday'
return self.weeks
@property
def saturday(self):
assert self.interval == 1, 'Use saturdays instead of saturday'
self.start_day = 'saturday'
return self.weeks
@property
def sunday(self):
assert self.interval == 1, 'Use sundays instead of sunday'
self.start_day = 'sunday'
return self.weeks
def tag(self, *tags):
"""
Tags the job with one or more unique indentifiers.
Tags must be hashable. Duplicate tags are discarded.
:param tags: A unique list of ``Hashable`` tags.
:return: The invoked job instance
"""
if not all(isinstance(tag, collections.Hashable) for tag in tags):
raise TypeError('Tags must be hashable')
self.tags.update(tags)
return self
def at(self, time_str):
"""
Schedule the job every day at a specific time.
Calling this is only valid for jobs scheduled to run
every N day(s).
:param time_str: A string in `XX:YY` format.
:return: The invoked job instance
"""
assert self.unit in ('days', 'hours') or self.start_day
hour, minute = time_str.split(':')
minute = int(minute)
if self.unit == 'days' or self.start_day:
hour = int(hour)
assert 0 <= hour <= 23
elif self.unit == 'hours':
hour = 0
assert 0 <= minute <= 59
self.at_time = datetime.time(hour, minute)
return self
def to(self, latest):
"""
Schedule the job to run at an irregular (randomized) interval.
The job's interval will randomly vary from the value given
to `every` to `latest`. The range defined is inclusive on
both ends. For example, `every(A).to(B).seconds` executes
the job function every N seconds such that A <= N <= B.
:param latest: Maximum interval between randomized job runs
:return: The invoked job instance
"""
self.latest = latest
return self
def do(self, job_func, *args, **kwargs):
"""
Specifies the job_func that should be called every time the
job runs.
Any additional arguments are passed on to job_func when
the job runs.
:param job_func: The function to be scheduled
:return: The invoked job instance
"""
self.job_func = functools.partial(job_func, *args, **kwargs)
try:
functools.update_wrapper(self.job_func, job_func)
except AttributeError:
# job_funcs already wrapped by functools.partial won't have
# __name__, __module__ or __doc__ and the update_wrapper()
# call will fail.
pass
self._schedule_next_run()
self.scheduler.jobs.append(self)
return self
@property
def should_run(self):
"""
:return: ``True`` if the job should be run now.
"""
return datetime.datetime.now() >= self.next_run
async def run(self):
"""
Run the job and immediately reschedule it.
:return: The return value returned by the `job_func`
"""
logger.info('Running job %s', self)
ret = await self.job_func()
self.last_run = datetime.datetime.now()
self._schedule_next_run()
return ret
def _schedule_next_run(self):
"""
Compute the instant when this job should run next.
"""
assert self.unit in ('seconds', 'minutes', 'hours', 'days', 'weeks')
if self.latest is not None:
assert self.latest >= self.interval
interval = random.randint(self.interval, self.latest)
else:
interval = self.interval
self.period = datetime.timedelta(**{self.unit: interval})
self.next_run = datetime.datetime.now() + self.period
if self.start_day is not None:
assert self.unit == 'weeks'
weekdays = (
'monday',
'tuesday',
'wednesday',
'thursday',
'friday',
'saturday',
'sunday'
)
assert self.start_day in weekdays
weekday = weekdays.index(self.start_day)
days_ahead = weekday - self.next_run.weekday()
if days_ahead <= 0: # Target day already happened this week
days_ahead += 7
self.next_run += datetime.timedelta(days_ahead) - self.period
if self.at_time is not None:
assert self.unit in ('days', 'hours') or self.start_day is not None
kwargs = {
'minute': self.at_time.minute,
'second': self.at_time.second,
'microsecond': 0
}
if self.unit == 'days' or self.start_day is not None:
kwargs['hour'] = self.at_time.hour
self.next_run = self.next_run.replace(**kwargs)
# If we are running for the first time, make sure we run
# at the specified time *today* (or *this hour*) as well
if not self.last_run:
now = datetime.datetime.now()
if (self.unit == 'days' and self.at_time > now.time() and
self.interval == 1):
self.next_run = self.next_run - datetime.timedelta(days=1)
elif self.unit == 'hours' and self.at_time.minute > now.minute:
self.next_run = self.next_run - datetime.timedelta(hours=1)
if self.start_day is not None and self.at_time is not None:
# Let's see if we will still make that time we specified today
if (self.next_run - datetime.datetime.now()).days >= 7:
self.next_run -= self.period
# The following methods are shortcuts for not having to
# create a Scheduler instance:
#: Default :class:`Scheduler <Scheduler>` object
default_scheduler = Scheduler()
#: Default :class:`Jobs <Job>` list
jobs = default_scheduler.jobs # todo: should this be a copy, e.g. jobs()?
def every(interval=1):
"""Calls :meth:`every <Scheduler.every>` on the
:data:`default scheduler instance <default_scheduler>`.
"""
return default_scheduler.every(interval)
async def run_pending():
"""Calls :meth:`run_pending <Scheduler.run_pending>` on the
:data:`default scheduler instance <default_scheduler>`.
"""
await default_scheduler.run_pending()
def run_all(delay_seconds=0):
"""Calls :meth:`run_all <Scheduler.run_all>` on the
:data:`default scheduler instance <default_scheduler>`.
"""
default_scheduler.run_all(delay_seconds=delay_seconds)
def clear(tag=None):
"""Calls :meth:`clear <Scheduler.clear>` on the
:data:`default scheduler instance <default_scheduler>`.
"""
default_scheduler.clear(tag)
def cancel_job(job):
"""Calls :meth:`cancel_job <Scheduler.cancel_job>` on the
:data:`default scheduler instance <default_scheduler>`.
"""
default_scheduler.cancel_job(job)
def next_run():
"""Calls :meth:`next_run <Scheduler.next_run>` on the
:data:`default scheduler instance <default_scheduler>`.
"""
return default_scheduler.next_run
def idle_seconds():
"""Calls :meth:`idle_seconds <Scheduler.idle_seconds>` on the
:data:`default scheduler instance <default_scheduler>`.
"""
return default_scheduler.idle_seconds