Hangupsbot Duckhunt

duckhunt.py

import asyncio
import re
import operator
import random

from time import time
from collections import defaultdict
from utils import remove_accents

import hangups
import plugins

duck_tail = "・゜゜・。。・゜゜"
duck = ["\_o< ", "\_O< ", "\_0< ", "\_\u00f6< ", "\_\u00f8< ", "\_\u00f3< "]
duck_noise = ["QUACK!", "FLAP FLAP!", "quack!"]

MSG_DELAY = 10
MASK_REQ = 3

def _initialise(bot):
    global pattern
    pattern = re.compile("[\W_]+") #Filter non alphanumeric
    plugins.register_handler(_on_message, "message")
    plugins.register_user_command(["duckfriends", "duckkillers", "ducks", "duckstats"])
    plugins.register_admin_command(["starthunt", "stophunt"])

    for conv_id, conv_data in bot.conversations.get().items():
        if conv_data["type"] == "GROUP" and get_game_status(bot, conv_id) != None:
            asyncio.ensure_future(deploy_duck(bot, conv_id))

def _on_message(bot, event, command):
    game_status = get_game_status(bot, event.conv_id)
    message = pattern.sub('', event.text).lower()
    if message == "bang":
        miss = ["WHOOSH! You missed the duck completely!", "Your gun jammed!", "Better luck next time.", "WTF!? Who are you Dick Cheney?" ]
        noduck = "There is no duck. What are you shooting at?"
        success = "{} you shot a duck in {} seconds! You have killed {} {}."
        action = "kill"
    elif message == "befriend":
        miss = ["The duck didn't want to be friends, maybe next time.", "Well this is awkward, the duck needs to think about it.", "The duck said no, maybe bribe it with some pizza? Ducks love pizza don't they?", "Who knew ducks could be so picky?"]
        noduck = "You tried befriending a non-existent duck, that's fucking creepy."
        success = "{} you befriended a duck in {} seconds! You have made friends with {} {}."
        action = "befriend"
    else:
        return

    if not game_status["game_on"]:
        yield from bot.coro_send_message(event.conv_id,
            _("There is no active hunt right now. Use <i>/bot starthunt</i> to start a game"))
        return
    elif game_status["duck_status"] != 1:
        yield from bot.coro_send_message(event.conv_id,
            _(noduck))
        return
    else:
        game_status["shoot_time"] = time()
        deploy = game_status["duck_time"]
        shoot = game_status["shoot_time"]
        chance = hit_or_miss(deploy, shoot)
        if not random.random() <= chance:
            yield from bot.coro_send_message(event.conv_id,
                _(random.choice(miss)) + _(" Try again."))
            return
        game_status["duck_status"] = 2
        uid = event.user.id_.chat_id
        if uid in game_status["users"]:
            if action in game_status["users"][uid]:
                game_status["users"][uid][action] += 1
            else:
                game_status["users"][uid][action] = 1
        else:
            game_status["users"][uid] = {action: 1}
        set_game_status(bot, event.conv_id, game_status)
        timer = "{:.3f}".format(shoot - deploy)
        duck = "duck" if game_status["users"][uid][action] == 1 else "ducks"
        yield from bot.coro_send_message(event.conv_id,
            _(success).format(event.user.full_name, timer, game_status["users"][uid][action], duck))
        set_ducktime(bot, event.conv_id)

def duckfriends(bot, event, *args):
    friends = {}
    out = "Duck friend scores:\n"
    game_status = get_game_status(bot, event.conv_id)
    for user in game_status["users"]:
        if "befriend" in game_status["users"][user]:
            friends[user] = game_status["users"][user]["befriend"]
    if not friends:
        yield from bot.coro_send_message(event.conv_id,
            _("It appears no one has friended any ducks yet."))
    else:
        topfriends = sorted(friends.items(), key=operator.itemgetter(1), reverse = True)
        out += "".join(["{}: {}\n".format(bot.get_hangups_user(k).full_name, str(v))  for k, v in topfriends])
        out = smart_truncate(out)
        yield from bot.coro_send_message(event.conv_id,
            _(out))

def duckkillers(bot, event, *args):
    killers = {}
    out = _("<b>Duck killer scores:</b>\n")
    game_status = get_game_status(bot, event.conv_id)
    for user in game_status["users"]:
        if "kill" in game_status["users"][user]:
            killers[user] = game_status["users"][user]["kill"]
    if not killers:
        yield from bot.coro_send_message(event.conv_id,
            _("It appears no one has killed any ducks yet."))
    else:
        topkillers = sorted(killers.items(), key=operator.itemgetter(1), reverse = True)
        out += "".join([_("{}: {}\n").format(bot.get_hangups_user(k).full_name, str(v))  for k, v in topkillers])
        out = smart_truncate(out)
        yield from bot.coro_send_message(event.conv_id, out)

def ducks(bot, event, *args):
    if len(args) > 0:
        users = get_users_by_name(bot, event.conv, " ".join(args))
        out = ""
        if len(users) > 1:
            out += _("<b>Multiple matching users:</b>")
        game_status = get_game_status(bot, event.conv_id)
        for user in users:
            uid = user.id_.chat_id
            if uid in game_status["users"]:
                if "kill" in game_status["users"][uid]:
                    kills = game_status["users"][uid]["kill"]
                else:
                    kills = 0
                if "befriend" in game_status["users"][uid]:
                    friends = game_status["users"][uid]["befriend"]
                else:
                    friends = 0
                out += _("{} has killed {} and befriended {} ducks.").format(user.full_name, kills, friends)
            else:
                out += _("{} has not participated in the duck hunt")
    else:
        out = _("Usage: /bot ducks <username>")
    yield from bot.coro_send_message(event.conv_id, out)

def duckstats(bot, event, *args):
    kills = 0
    friends = 0
    game_status = get_game_status(bot, event.conv_id)
    if len(game_status["users"]) > 0:
        for uid in game_status["users"]:
            if "kill" in game_status["users"][uid]:
                 kills += game_status["users"][uid]["kill"]
            if "befriend" in game_status["users"][uid]:
                friends += game_status["users"][uid]["befriend"]
        out = _("<b>Duck Stats:</b>\n{} killed and {} befriended in this Hangout").format(kills, friends)
    else:
        out = _("It looks like there has been no duck activity on this Hangout.") 
    
    yield from bot.coro_send_message(event.conv_id, out)

def starthunt(bot, event, *args):
    game_status = get_game_status(bot, event.conv_id)
    if bot.conversations.catalog[event.conv_id]["type"] == "ONE_TO_ONE":
        yield from bot.coro_send_message(event.conv_id,
            _("No hunting by yourself, that isn't safe."))
        return
    check = game_status['game_on']
    if check:
        yield from bot.coro_send_message(event.conv_id,
            _("There is already a game running."))
        return
    else:
        game_status["game_on"] = 1
    yield from bot.coro_send_message(event.conv_id,
        _("Ducks have been spotted nearby. See how many you can shoot or save. Use <i>bang</i> to shoot or <i>befriend</i> to save them."))
    set_game_status(bot, event.conv_id, game_status)
    set_ducktime(bot, event.conv_id)


def stophunt(bot, event, *args):
    game_status = get_game_status(bot, event.conv_id)
    if game_status["game_on"]:
        game_status["game_on"] = 0
        yield from bot.coro_send_message(event.conv_id,
            _("The game has been stopped."))
    else:
        yield from bot.coro_send_message(event.conv_id,
            _("There is no game running."))

def set_ducktime(bot, conv_id):
    game_status = get_game_status(bot, conv_id)
    game_status["next_duck_time"] = random.randint(int(time()) + 480, int(time()) + 3600)
    #game_status["next_duck_time"] = random.randint(int(time()) + 10, int(time()) + 60)
    game_status["duck_status"] = 0
    set_game_status(bot, conv_id, game_status)
    asyncio.ensure_future(deploy_duck(bot, conv_id))

def deploy_duck(bot, conv_id):
    game_status = get_game_status(bot, conv_id)
    sleeptime = max(game_status["next_duck_time"]-time(), 0)
    yield from asyncio.sleep(sleeptime)

    game_status = get_game_status(bot, conv_id)
    active = game_status["game_on"]
    duck_status = game_status["duck_status"]
    
    if active == 1 and duck_status == 0:
        game_status["duck_status"] = 1
        game_status["duck_time"] = time()
        dtail, dbody, dnoise = generate_duck()
        yield from bot.coro_send_message(conv_id,
            _("{}{}{}").format(dtail, dbody, dnoise))

def hit_or_miss(deploy, shoot):
    """This function calculates if the befriend or bang will be successful."""
    if shoot - deploy < 1:
        return .05
    elif 1 <= shoot - deploy <= 7:
        out = random.uniform(.60, .75)
        return out
    else:
        return 1

def get_game_status(bot, conv_id):
    game_status = bot.conversation_memory_get(conv_id, "duckhunt")
    if game_status == None:
        game_status = {
                "game_on": 0,
                "next_duck_time": 0,
                "duck_status": 0,
                "shoot_time": 0,
                "duck_time": 0,
                "users": {}
                }
    return game_status


def set_game_status(bot, conv_id, game_status):
    bot.conversation_memory_set(conv_id, "duckhunt", game_status)

def generate_duck():
    """Try and randomize the duck message so people can't highlight on it/script against it."""
    rt = random.randint(1, len(duck_tail) - 1)
    dtail = duck_tail[:rt] + u' \u200b ' + duck_tail[rt:]
    dbody = random.choice(duck)
    rb = random.randint(1, len(dbody) - 1)
    dbody = dbody[:rb] + u'\u200b' + dbody[rb:]
    dnoise = random.choice(duck_noise)
    rn = random.randint(1, len(dnoise) - 1)
    dnoise = dnoise[:rn] + u'\u200b' + dnoise[rn:]
    return (dtail, dbody, dnoise)

def smart_truncate(content, length=320, suffix='...'):
    if len(content) <= length:
        return content
    else:
        return content[:length].rsplit(' • ', 1)[0]+suffix

def get_users_by_name(bot, conv, name):
    """Mostly copied from mention plugin"""
    username = ''.join(c for c in name if c.isalnum() or c in ["-"])

    users_in_chat = conv.users
    mention_chat_ids = []

    username_lower = username.lower()
    username_upper = username.upper()

    exact_nickname_matches = []
    exact_fragment_matches = []
    mention_list = []
    for u in users_in_chat:
        nickname = ""
        nickname_lower = ""
        if bot.memory.exists(['user_data', u.id_.chat_id, "nickname"]):
            nickname = bot.memory.get_by_path(['user_data', u.id_.chat_id, "nickname"])
            nickname_lower = nickname.lower()

        _normalised_full_name_upper = remove_accents(u.full_name.upper())
        if (username_lower in u.full_name.replace(" ", "").lower() or
                username_upper in _normalised_full_name_upper.replace(" ", "") or
                username_lower in u.full_name.replace(" ", "_").lower() or
                username_upper in _normalised_full_name_upper.replace(" ", "_") or
                username_lower == nickname_lower or
                username in u.full_name.split(" ")):
            if username_lower == nickname_lower:
                if u not in exact_nickname_matches:
                    exact_nickname_matches.append(u)
            if (username in u.full_name.split(" ") or
                    username_upper in _normalised_full_name_upper.split(" ")):
                if u not in exact_fragment_matches:
                    exact_fragment_matches.append(u)
            if u not in mention_list:
                mention_list.append(u)

    if exact_nickname_matches:
        return exact_nickname_matches
    if exact_fragment_matches:
        return exact_fragment_matches
    return mention_list