基本
無駄がなくて綺麗なコード。やっぱり Python って素敵。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import socket, string
# ソケット作成、接続、ログイン
irc = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
irc.connect( ( 'irc.example.com', 6667 ) )
irc.send( '''
USER arianne * * arianne
NICK arianne
JOIN #hogefuga
''' )
while True:
# recv() で、受信するまでブロッキング
buf = unicode( irc.recv( 1024 ).strip(), 'utf-8' )
# コマンドプロンプトの場合は、'cp932' を指定
print buf.encode( 'utf-8' )
# IRC サーバから PING が来たら PONG を自動応答する
if string.split( buf )[0] == "PING":
irc.send( "PONG " + string.split( buf )[1] + "\n" )
# 文字列に反応してみたり
if buf.find( u"こんにちは" ) > -1:
irc.send( u"PRIVMSG #hogefuga :こんにちわぁ\n".encode( 'utf-8' ) )
モジュール化
車輪の再発明だと思いつつ、自分の使い易い形のクラスにしてみた。サーバー応答対応やエラーと例外処理が未実装だけれど「スレッド化した受信監視→外部から指定されたイベントハンドラ実行」は、これで・・・いけるはず・・・。もうちょっとスマートに書けるようになったらいいな。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import socket, string, threading
import logging
class ircc ( object ):
"""The IRC Client Class"""
def __init__( self, hostname, port, server_encoding='utf-8', terminal_encoding='utf-8' ):
# ソケット作成
self.server = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
self.server.connect( ( hostname, port ) )
self.server_encoding = server_encoding
self.terminal_encoding = terminal_encoding
script_encoding = 'utf-8'
logging.basicConfig( level=logging.DEBUG, format='%(asctime)s %(levelname)s %(message)s' )
def __str__( self ):
return 'ircc'
def send( self, message ):
# 渡された文字列をサーバに送信するだけ
message = unicode( message, self.script_encoding )
logging.debug( ">>> " + message.encode( self.terminal_encoding ) )
self.server.send( message.encode( self.server_encoding ) + "\n")
def recv( self, receive_size = 1024 ):
# サーバから文字列を受信するまで待機
buf = unicode( self.server.recv( receive_size ).strip(), self.server_encoding )
for line in string.split( buf, '\n' ):
logging.debug( "<<< " + line.encode( self.terminal_encoding ) )
return buf
# こういう細々とした IRC コマンドは、はたして個別に実装する必要があるのかニャー
def quit( self, message = '' ):
self.send( 'QUIT :' + message )
def nick( self, nickname ):
self.send( 'NICK ' + nickname )
def join( self, channel ):
self.send( 'JOIN ' + channel )
def privmsg( self, to, message ):
self.send( 'PRIVMSG ' + to + ' ' + ' :' + message )
# ログイン・フェイズはエラー処理が重要になってくるので、メソッド化しておく。
def login( self, loginname, nickname, description, wait_for_motd = True ):
self.send( 'USER %s * * :%s' % ( loginname, description ) )
self.nick( nickname )
#if self.recv().find( '433' ):
# self.quit()
# return 1
while wait_for_motd:
if self.recv().find( ' 376 ' ):
break
# イベントハンドラ、send()、recv() それぞれのアドレスを渡してスレッド化
def set_event( self, event_handler ):
_irc_recv_thread( event_handler, self.send, self.recv )
class _irc_recv_thread( threading.Thread ):
""" メッセージイベントを検知し、外部から定義された
イベントハンドラを呼ぶスレッドオブジェクト """
def __init__( self, event_handler, send_handler, receive_handler ):
self.__event = event_handler
self.__send = send_handler
self.__receive = receive_handler
self.__msg = ''
threading.Thread.__init__( self )
self.setDaemon( True )
self.start()
def run( self ):
while True:
# メインループ
self.__msg = self.__receive()
if len( self.__msg ) > 0:
for line in string.split( self.__msg, '\n' ):
if string.split( line )[0] == 'PING':
self.__send( 'PONG ' + string.split( line )[1] )
self.__event( line )
else:
break
使うのは、以下のようなイメージで。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import ircc
# イベントハンドラ
def check_message( msg ):
if msg.find( u"こん" ) > -1:
irc.privmsg( '#hogefuga', 'こん' )
irc = ircc.ircc( 'irc.example.com', 6667, 'utf-8' )
irc.login( 'arianne', 'arianne', 'Arianne the BOT', False )
irc.join( '#hogefuga' )
# イベントハンドラを登録
irc.set_event( check_message )
while True:
pass # 何か他の事でもやる
んー、クラス化するメリットが少ないかも。日々使ってないとすぐ忘れるタチなので、wxPython/PyGame なんかと一緒に IRC プロトコルを使った通信将棋やチェスでも書きたいなぁ。
※ 2020/07/02 度重なるブログ移転・ブログシステムのアップデートにより崩れた記事を校正。Python なのにインデントがすべて吹き飛んでいるので間違いがあるかも…。
