pew
/
phenny
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request #1 from mutantmonkey/master 

updating vtluug's fork
master
Paul Walko 2017-05-27 18:30:04 -04:00 committed by GitHub
parent a10f513dbd 42325d85e8
commit 7dc5ece29b
8 changed files with 204 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.travis.yml
View File

@ -2,10 +2,9 @@ language: python
sudo: false
cache: pip
python:
- 3.2
- 3.3
- 3.4
- 3.5
- 3.6
install:
- pip install -r requirements.txt
script: nosetests

2
README.md
View File

@ -12,7 +12,7 @@ will need to be updated to run on Python3 if they do not already. All of the
core modules have been ported, removed, or replaced.
## Requirements
* Python 3.2+
* Python 3.4+
* [python-requests](http://docs.python-requests.org/en/latest/)
## Installation

45
__init__.py
View File

@ -7,7 +7,12 @@ Licensed under the Eiffel Forum License 2.
http://inamidst.com/phenny/
"""
import sys, os, time, threading, signal
import os
import signal
import sys
import threading
import time
class Watcher(object):
# Cf. http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/496735
@ -18,51 +23,65 @@ class Watcher(object):
self.watch()
def watch(self):
try: os.wait()
try:
os.wait()
except KeyboardInterrupt:
self.kill()
sys.exit()
def kill(self):
try: os.kill(self.child, signal.SIGKILL)
except OSError: pass
try:
os.kill(self.child, signal.SIGKILL)
except OSError:
pass
def sig_term(self, signum, frame):
self.kill()
sys.exit()
def run_phenny(config):
if hasattr(config, 'delay'):
delay = config.delay
else: delay = 20
else:
delay = 20
def connect(config):
import bot
p = bot.Phenny(config)
p.run(config.host, config.port, config.ssl, config.ipv6,
config.ca_certs)
try: Watcher()
ssl_context = p.get_ssl_context(config.ca_certs)
if config.ssl_cert and config.ssl_key:
ssl_context.load_cert_chain(config.ssl_cert, config.ssl_key)
p.run(config.host, config.port, config.ssl, config.ipv6, None,
ssl_context)
try:
Watcher()
except Exception as e:
print('Warning:', e, '(in __init__.py)', file=sys.stderr)
while True:
try: connect(config)
try:
connect(config)
except KeyboardInterrupt:
sys.exit()
sys.exit()
if not isinstance(delay, int):
break
warning = 'Warning: Disconnected. Reconnecting in %s seconds...' % delay
print(warning, file=sys.stderr)
msg = "Warning: Disconnected. Reconnecting in {0} seconds..."
print(msg.format(delay), file=sys.stderr)
time.sleep(delay)
def run(config):
t = threading.Thread(target=run_phenny, args=(config,))
if hasattr(t, 'run'):
t.run()
else: t.start()
else:
t.start()
if __name__ == '__main__':
print(__doc__)

81
irc.py
View File

@ -7,9 +7,13 @@ Licensed under the Eiffel Forum License 2.
http://inamidst.com/phenny/
"""
import sys, re, time, traceback
import socket, asyncore, asynchat
import asynchat
import asyncore
import re
import socket
import ssl
import sys
import time
class Origin(object):
@ -23,7 +27,8 @@ class Origin(object):
if len(args) > 1:
target = args[1]
else: target = None
else:
target = None
mappings = {bot.nick: self.nick, None: None}
self.sender = mappings.get(target, target)
@ -82,45 +87,39 @@ class Bot(asynchat.async_chat):
self.__write(args, text)
except Exception as e:
raise
#pass
def run(self, host, port=6667, ssl=False,
ipv6=False, ca_certs=None):
self.ca_certs = ca_certs
self.initiate_connect(host, port, ssl, ipv6)
def run(self, host, port=6667, ssl=False, ipv6=False, ca_certs=None,
ssl_context=None):
if ssl_context is None:
ssl_context = self.get_ssl_context(ca_certs)
self.initiate_connect(host, port, ssl, ipv6, ssl_context)
def initiate_connect(self, host, port, use_ssl, ipv6):
def get_ssl_context(self, ca_certs):
return ssl.create_default_context(
purpose=ssl.Purpose.SERVER_AUTH,
cafile=ca_certs)
def initiate_connect(self, host, port, use_ssl, ipv6, ssl_context):
if self.verbose:
message = 'Connecting to %s:%s...' % (host, port)
print(message, end=' ', file=sys.stderr)
if ipv6 and socket.has_ipv6:
af = socket.AF_INET6
af = socket.AF_INET6
else:
af = socket.AF_INET
self.create_socket(af, socket.SOCK_STREAM, use_ssl, host)
af = socket.AF_INET
self.create_socket(af, socket.SOCK_STREAM, use_ssl, host, ssl_context)
self.connect((host, port))
try: asyncore.loop()
try:
asyncore.loop()
except KeyboardInterrupt:
sys.exit()
def create_socket(self, family, type, use_ssl=False, hostname=None):
def create_socket(self, family, type, use_ssl=False, hostname=None,
ssl_context=None):
self.family_and_type = family, type
sock = socket.socket(family, type)
if use_ssl:
# this stuff is all new in python 3.4, so fallback if needed
try:
context = ssl.create_default_context(
purpose=ssl.Purpose.SERVER_AUTH,
cafile=self.ca_certs)
sock = context.wrap_socket(sock, server_hostname=hostname)
except:
if self.ca_certs is None:
# default to standard path on most non-EL distros
ca_certs = "/etc/ssl/certs/ca-certificates.crt"
else:
ca_certs = self.ca_certs
sock = ssl.wrap_socket(sock, ssl_version=ssl.PROTOCOL_TLSv1,
cert_reqs=ssl.CERT_REQUIRED, ca_certs=ca_certs)
sock = ssl_context.wrap_socket(sock, server_hostname=hostname)
# FIXME: this doesn't work with SSL enabled
#sock.setblocking(False)
self.set_socket(sock)
@ -153,11 +152,13 @@ class Bot(asynchat.async_chat):
if line.startswith(':'):
source, line = line[1:].split(' ', 1)
else: source = None
else:
source = None
if ' :' in line:
argstr, text = line.split(' :', 1)
else: argstr, text = line, ''
else:
argstr, text = line, ''
args = argstr.split()
origin = Origin(self, source, args)
@ -174,11 +175,13 @@ class Bot(asynchat.async_chat):
# Cf. http://swhack.com/logs/2006-03-01#T19-43-25
if isinstance(text, str):
try: text = text.encode('utf-8')
try:
text = text.encode('utf-8')
except UnicodeEncodeError as e:
text = e.__class__ + ': ' + str(e)
if isinstance(recipient, str):
try: recipient = recipient.encode('utf-8')
try:
recipient = recipient.encode('utf-8')
except UnicodeEncodeError as e:
return
@ -244,20 +247,23 @@ class Bot(asynchat.async_chat):
if line.startswith('File "/'):
report.append(line[0].lower() + line[1:])
break
else: report.append('source unknown')
else:
report.append('source unknown')
self.msg(origin.sender, report[0] + ' (' + report[1] + ')')
except: self.msg(origin.sender, "Got an error.")
except:
self.msg(origin.sender, "Got an error.")
class TestBot(Bot):
def f_ping(self, origin, match, args):
delay = m.group(1)
delay = match.group(1)
if delay is not None:
import time
time.sleep(int(delay))
self.msg(origin.sender, 'pong (%s)' % delay)
else: self.msg(origin.sender, 'pong')
else:
self.msg(origin.sender, 'pong')
f_ping.rule = r'^\.ping(?:[ \t]+(\d+))?$'
@ -266,5 +272,6 @@ def main():
bot.run('irc.freenode.net')
print(__doc__)
if __name__=="__main__":
if __name__ == "__main__":
main()

11
modules/search.py
View File

@ -184,16 +184,5 @@ def search(phenny, input):
phenny.reply(result)
search.commands = ['search']
def suggest(phenny, input):
if not input.group(2):
return phenny.reply("No query term.")
query = input.group(2)
uri = 'http://websitedev.de/temp-bin/suggest.pl?q='
answer = web.get(uri + web.quote(query).replace('+', '%2B'))
if answer:
phenny.say(answer)
else: phenny.reply('Sorry, no result.')
suggest.commands = ['suggest']
if __name__ == '__main__':
print(__doc__.strip())

9
modules/test/test_search.py
View File

@ -8,7 +8,7 @@ import unittest
from mock import MagicMock, Mock
from modules.search import duck_api, google_search, google_count, \
formatnumber, g, gc, gcs, bing_search, bing, duck_search, duck, \
search, suggest
search
class TestSearch(unittest.TestCase):
@ -75,10 +75,3 @@ class TestSearch(unittest.TestCase):
duck(self.phenny, input)
assert self.phenny.reply.called is True
def test_suggest(self):
input = Mock(group=lambda x: 'vtluug')
suggest(self.phenny, input)
assert (self.phenny.reply.called is True or \
self.phenny.say.called is True)

2
modules/test/test_vtluugwiki.py
View File

@ -8,7 +8,7 @@ import unittest
from mock import MagicMock, Mock
from modules import vtluugwiki
@unittest.skip('Skipping until wiki is back up')
class TestVtluugwiki(unittest.TestCase):
def setUp(self):
self.phenny = MagicMock()

41
phenny
View File

@ -11,18 +11,22 @@ Run ./phenny, then edit ~/.phenny/default.py
Then run ./phenny again
"""
import sys, os, imp
import argparse
import imp
import os
import sys
from textwrap import dedent as trim
dotdir = os.path.expanduser('~/.phenny')
def check_python_version():
if sys.version_info < (3, 0):
error = 'Error: Requires Python 3.0 or later, from www.python.org'
if sys.version_info < (3, 4):
error = 'Error: Requires Python 3.4 or later, from www.python.org'
print(error, file=sys.stderr)
sys.exit(1)
def create_default_config(fn):
f = open(fn, 'w')
print(trim("""\
@ -69,6 +73,7 @@ def create_default_config(fn):
"""), file=f)
f.close()
def create_default_config_file(dotdir):
print('Creating a default config file at ~/.phenny/default.py...')
default = os.path.join(dotdir, 'default.py')
@ -77,9 +82,11 @@ def create_default_config_file(dotdir):
print('Done; now you can edit default.py, and run phenny! Enjoy.')
sys.exit(0)
def create_dotdir(dotdir):
print('Creating a config directory at ~/.phenny...')
try: os.mkdir(dotdir)
try:
os.mkdir(dotdir)
except Exception as e:
print('There was a problem creating %s:' % dotdir, file=sys.stderr)
print(e.__class__, str(e), file=sys.stderr)
@ -88,6 +95,7 @@ def create_dotdir(dotdir):
create_default_config_file(dotdir)
def check_dotdir():
default = os.path.join(dotdir, 'default.py')
@ -96,6 +104,7 @@ def check_dotdir():
elif not os.path.isfile(default):
create_default_config_file(dotdir)
def config_names(config):
config = config or 'default'
@ -123,19 +132,20 @@ def config_names(config):
print('What happened to ~/.phenny/default.py?', file=sys.stderr)
sys.exit(1)
def main(argv=None):
# Step One: Parse The Command Line
parser = argparse.ArgumentParser(description="A Python IRC bot.")
parser.add_argument('-c', '--config', metavar='fn',
help='use this configuration file or directory')
help='use this configuration file or directory')
args = parser.parse_args(argv)
# Step Two: Check Dependencies
check_python_version() # require python2.4 or later
check_python_version()
if not args.config:
check_dotdir() # require ~/.phenny, or make it and exit
check_dotdir() # require ~/.phenny, or make it and exit
# Step Three: Load The Configurations
@ -160,6 +170,12 @@ def main(argv=None):
if not hasattr(module, 'ca_certs'):
module.ca_certs = None
if not hasattr(module, 'ssl_cert'):
module.ssl_cert = None
if not hasattr(module, 'ssl_key'):
module.ssl_key = None
if not hasattr(module, 'ipv6'):
module.ipv6 = False
@ -168,7 +184,7 @@ def main(argv=None):
if module.host == 'irc.example.net':
error = ('Error: you must edit the config file first!\n' +
"You're currently using %s" % module.filename)
"You're currently using %s" % module.filename)
print(error, file=sys.stderr)
sys.exit(1)
@ -176,9 +192,11 @@ def main(argv=None):
# Step Four: Load Phenny
try: from __init__ import run
try:
from __init__ import run
except ImportError:
try: from phenny import run
try:
from phenny import run
except ImportError:
print("Error: Couldn't find phenny to import", file=sys.stderr)
sys.exit(1)
@ -187,7 +205,8 @@ def main(argv=None):
# @@ ignore SIGHUP
for config_module in config_modules:
run(config_module) # @@ thread this
run(config_module) # @@ thread this
if __name__ == '__main__':
main()