-
Notifications
You must be signed in to change notification settings - Fork 47
/
human.py
93 lines (75 loc) · 2.63 KB
/
human.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import os
import irc3
import stat
import codecs
import random
import subprocess
__doc__ = '''
==========================================
:mod:`irc3.plugins.human` Human plugin
==========================================
Store public message addressed to the bot in a file and reply a random message
extracted from this file.
..
>>> from irc3.testing import IrcBot
>>> with open('/tmp/human.db', 'wb') as fd:
... s = fd.write(b'Yo!\\nYo!\\nYo!\\nYo!\\n')
Register the plugin::
>>> bot = IrcBot(human='/tmp/human.db', nick='nono')
>>> bot.include('irc3.plugins.human')
And it should work::
>>> bot.test(':foo!m@h PRIVMSG nono :nono: Yo!')
PRIVMSG foo :Yo!
>>> bot.test(':foo!m@h PRIVMSG #chan :nono: Yo!')
PRIVMSG #chan :foo: Yo!
..
>>> bot.test(':foo!m@h PRIVMSG #chan :!ping')
'''
@irc3.plugin
class Human(object):
requires = [
__name__.replace('human', 'core'),
]
def __init__(self, bot):
self.bot = bot
self.db = os.path.expanduser(
bot.config.get('human', '~/.irc3/human.db'))
self.delay = (2, 5)
try:
os.makedirs(os.path.dirname(self.db))
except OSError:
pass
if not os.path.isfile(self.db): # pragma: no cover
self.initialize(15)
def initialize(self, amount): # pragma: no cover
cmd = (
'wget -q -t 1 -O- '
'"http://www.iheartquotes.com/api/v1/random?max_lines=1" '
'| head -n 1 | grep -v "&" >> {}').format(self.db)
processes = [subprocess.Popen(cmd, shell=True) for i in range(amount)]
for p in processes:
p.wait()
@irc3.event(irc3.rfc.MY_PRIVMSG)
def on_message(self, mask=None, event=None, target=None, data=None):
with codecs.open(self.db, 'ab+', encoding=self.bot.encoding) as fd:
fd.write(data + '\n')
pos = random.randint(0, os.stat(self.db)[stat.ST_SIZE])
with codecs.open(self.db, encoding=self.bot.encoding) as fd:
fd.seek(pos)
fd.readline()
try:
message = fd.readline().strip()
except: # pragma: no cover
pass
message = message or 'Yo!'
if target.is_channel:
message = '{0}: {1}'.format(mask.nick, message)
else:
target = mask.nick
self.call_with_human_delay(self.bot.privmsg, target, message)
@irc3.extend
def call_with_human_delay(self, func, *args, **kwargs):
delay = random.randint(*self.delay)
self.bot.loop.call_later(delay, func, *args, **kwargs)