Skip to content

I
ItChat

BillLi / ItChat
与 Fork 源项目一致

从无法访问的项目Fork

通知 3
Star 0
Fork 0
I
ItChat

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 05dd4684 编写于 作者: L LittleCoder
浏览文件
操作

Delete useless files

上级 84206842
隐藏空白更改
内联 并排
Showing with 56 addition and 1313 deletion
.gitignore
浏览文件 @ 05dd4684
plugin/config/tuling.json
log/*
storage/*
!storage/upload/*
*.pyc
*.swp
ItChat.py 已删除 100644 → 0
浏览文件 @ 84206842
#coding=utf8
import time, sys
if sys.version[:3] != '2.7': print('This project should be run with python2.7');sys.exit()
import itchat.storage, itchat.out, itchat.argparser, itchat.robot
from itchat.client import WeChatClient
from plugin.ChatLikeCMD import ChatLikeCMD
# it's recorded in https://github.com/littlecodersh/ItChat/wiki/Screenshots
ROBOT = True # change to False if you need to use command line wechat
def demo_robot(s, msgList, client): # ONLY FOR DEMO
print('Start auto-replying')
while 1:
if msgList:
msg = msgList.pop()
itchat.robot.deal_with_msg(msg, s, client)
time.sleep(.1)
if __name__ == '__main__':
from PluginTest import plugin_load_succeed
if not plugin_load_succeed(): print('Try to fix the plugins and test them with PluginTest.py');sys.exit()
client_s = itchat.storage.Storage()
if ROBOT:
client = WeChatClient(client_s, robot = True)
else:
client = WeChatClient(client_s)
client.login()
msgList = client.storage()
if ROBOT:
demo_robot(client_s, msgList, client)
else:
front = ChatLikeCMD(header = client_s.find_nickname(client_s.userName), symbol = '>', inPip = msgList)
cmdList = front.get_command_pip()
front.start()
itchat.argparser.startCommandLine(client_s, client, msgList, front, cmdList)
PluginTest.py 已删除 100644 → 0
浏览文件 @ 84206842
#coding=utf8
import json
import sys, traceback, os
succeed = True
def plugin_load_succeed():
return succeed
def sys_print(level, msg):
if level == 'SUCC': msg += ' plugin loaded'
if level == 'WARN': global succeed; succeed = False
print '[%s] %s'%(level[:4], msg)
# load plugin list
try:
pluginListDir = 'pluginlist.json'
with open(pluginListDir) as f: pluginList = json.loads(f.read())
except:
pluginList = {}
sys_print('WARN', 'There is something wrong with the format of your pluginlist.json')
# Test grouptalking.py
if 'grouptalking' in pluginList['systemmodules']:
try:
from plugin.msgdealers.grouptalking import grouptalking
sys_print('INFO', 'Group talking plugin is open now, please be careful about it.')
except:
sys_print('WARN', 'There is something wrong with the grouptalking plugin')
traceback.print_exc()
# Test tuling.py
if 'tuling' in pluginList['systemmodules']:
try:
pluginList['systemmodules'].remove('tuling')
import plugin.tuling as tuling
try:
tuling.get_response('Hi', 'LittleCoder')
pluginList['systemmodules'].append('tuling')
sys_print('SUCC', 'Tuling')
except:
sys_print('INFO', 'Your key for tuling robot can\'t be used now, change one in plugin/config/tuling.json')
sys_print('~~~~', 'You can get it from http://www.tuling123.com/')
except:
sys_print('WARN', 'There is something wrong with the format of your plugin/config/tuling.json')
# Test QRCode.py
if 'QRCode' in pluginList['systemmodules']:
try:
import plugin.QRCode as QRCode
sys_print('SUCC', 'Command line QRCode')
except:
sys_print('INFO', 'Command line QRCode loaded failed, if you want to use this, you need to run `pip install Image`')
# Test msgdealers.autoreply
if 'autoreply' in pluginList['msgdealers']:
try:
pluginList['msgdealers'].remove('autoreply')
from plugin.msgdealers.autoreply import autoreply
pluginList['msgdealers'].append('autoreply')
sys_print('SUCC', 'msgdealers.autoreply')
except Exception, e:
sys_print('WARN', e.message)
# Test msgdealers.vote
if 'vote' in pluginList['msgdealers']:
try:
pluginList['msgdealers'].remove('vote')
from plugin.msgdealers.vote import vote
pluginList['msgdealers'].append('vote')
sys_print('SUCC', 'msgdealers.vote')
sys_print('INFO', 'But whether it can be properly used need to be tested online')
except:
sys_print('WARN', 'Vote plugin loaded failed, this is strange, you need to contact me')
# traceback.print_exc()
def send_msg(msg):
if len(msg) > 5:
if msg[:5] == '@fil@':
try:
with open(msg[5:]): pass
print 'File: %s'%msg[5:]
except:
pass
if msg[:5] == '@img@':
try:
with open(msg[5:]): pass
print 'Picture: %s'%msg[5:]
except:
pass
elif msg[:5] == '@msg@':
print msg[5:]
else:
print msg
else:
print msg[5:]
if __name__ == '__main__':
try:
print 'Loading %s'%('successfully' if plugin_load_succeed() else 'failed')
pluginOrder = [('autoreply', autoreply), ('tuling', tuling.get_response)]
while True:
msg = raw_input('>').decode(sys.stdin.encoding)
if not msg: continue
res = ''
for plugin in pluginOrder:
if plugin[0] in (pluginList['msgdealers'] + pluginList['systemmodules']):
r = plugin[1](msg, None, None)
if r: res = r;break
if not res: r = 'No plugin matched'
send_msg(r)
except:
print 'Exit'
traceback.print_exc()
else:
sys_print('INFO', 'Plugin loading finished')
__init__.py 0 → 100644
浏览文件 @ 05dd4684
__version__ = '0.1b'
from client import itchat
from storage import Storage
itchat/client.py → client.py
浏览文件 @ 05dd4684
......@@ -2,22 +2,13 @@ import os, sys
import requests, time, re
import thread, subprocess
import json, xml.dom.minidom, mimetypes
from urllib import unquote
import config, storage, out, log, tools
try:
from plugin.QRCode import QRCode
CMD_QRCODE = True
except:
CMD_QRCODE = False
import config, storage, out, tools
BASE_URL = config.BASE_URL
DEBUG = False
class WeChatClient:
def __init__(self, storageClass = None, robot = False):
class itchat:
def __init__(self, storageClass = None):
self.storageClass = storageClass if storageClass else storage.Storage()
self.robot = robot
self.msgList = self.storageClass.msgList
self.loginInfo = {}
self.s = requests.Session()
......@@ -27,8 +18,7 @@ class WeChatClient:
out.print_line('Getting uuid', True)
while self.get_QRuuid(): time.sleep(1)
out.print_line('Getting QR Code', True)
if self.get_QR():
break
if self.get_QR(): break
elif get_count >= 9:
out.print_line('Failed to get QR Code, please restart the program')
sys.exit()
......@@ -36,18 +26,9 @@ class WeChatClient:
while self.check_login(): time.sleep(1)
self.web_init()
self.show_mobile_login()
while 1:
voidUserList = self.get_contract()
if not voidUserList: break
out.print_line('These uses need new RemarkNames and are added to a group', True)
chatRoomName = self.create_chatroom(voidUserList, 'RemarkNames')
self.delete_member(chatRoomName, [voidUserList[0]])
self.add_member(chatRoomName, [voidUserList[0]])
while raw_input('Enter "ready" after you rename all of them and DELETE the group: ') != 'ready': pass
out.print_line('Start reload contract list', False)
self.get_contract()
out.print_line('Login successfully as %s\n'%(
self.storageClass.find_nickname(self.storageClass.userName)), False)
log.log('SIGN IN', True)
thread.start_new_thread(self.maintain_loop, ())
def get_QRuuid(self):
url = '%s/jslogin'%BASE_URL
......@@ -63,13 +44,10 @@ class WeChatClient:
def get_QR(self):
try:
url = '%s/qrcode/%s'%(BASE_URL, self.uuid)
r = self.s.get(url, stream = True)# params = payloads, headers = HEADER,
QR_DIR = os.path.join(config.QR_DIR, 'QR.jpg')
r = self.s.get(url, stream = True)
QR_DIR = 'QR.jpg'
with open(QR_DIR, 'wb') as f: f.write(r.content)
if CMD_QRCODE:
q = QRCode(QR_DIR, 37, 3, 'BLACK')
q.print_qr()
elif config.OS == 'Darwin':
if config.OS == 'Darwin':
subprocess.call(['open', QR_DIR])
elif config.OS == 'Linux':
subprocess.call(['xdg-open', QR_DIR])
......@@ -80,7 +58,6 @@ class WeChatClient:
return False
def check_login(self):
url = '%s/cgi-bin/mmwebwx-bin/login'%BASE_URL
# add tip so that we can get many reply, use string payloads to avoid auto-urlencode
payloads = 'tip=1&uuid=%s&_=%s'%(self.uuid, int(time.time()))
r = self.s.get(url, params = payloads)
regx = r'window.code=(\d+)'
......@@ -159,35 +136,7 @@ class WeChatClient:
# deal with emoji
memberList = tools.emoji_dealer(memberList)
# RemarkPYQuanPin & PYQuanPin is used as identifier
voidUserList = []
validUserList = []
unknownCount = 0
for m in memberList:
if m['RemarkPYQuanPin'] != '': m['PYQuanPin'] = m['RemarkPYQuanPin']
if m['PYQuanPin'] == '' and m['Alias'] != '': m['PYQuanPin'] == m['Alias']
if m['UserName'] == self.storageClass.userName:
m['PYQuanPin'] = m['NickName']
elif m['PYQuanPin'] == '':
voidUserList.append(m)
m['PYQuanPin'] = '@Unknown%s'%unknownCount
unknownCount += 1
while 1:
insertResult = self.storageClass.update_user(m['PYQuanPin'], NickName = m['RemarkName'] or m['NickName'],
UserName = m['UserName'])
if insertResult and not m['PYQuanPin'] in validUserList:
validUserList.append(m['PYQuanPin']);break
else:
voidUserList.append(m)
m['PYQuanPin'] = '@Unknown%s'%unknownCount
unknownCount += 1
if DEBUG:
with open('MemberList.txt', 'w') as f: f.write(str(memberList))
if len(voidUserList) == 1:
m = voidUserList[0]
self.storageClass.update_user('', NickName = m['RemarkName'] or m['NickName'], UserName = m['UserName'])
return []
else:
return voidUserList
self.storageClass.updateMemberList(memberList)
def show_mobile_login(self):
url = '%s/webwxstatusnotify'%self.loginInfo['url']
payloads = {
......@@ -216,9 +165,8 @@ class WeChatClient:
count = 0
except Exception, e:
count += 1
log.log('Exception %s:'%count, False, exception = e)
time.sleep(count*3)
log.log('LOG OUT', False)
out.print_line('LOG OUT', False)
def sync_check(self):
url = '%s/synccheck'%self.loginInfo['url']
payloads = {
......@@ -366,7 +314,7 @@ class WeChatClient:
elif m['MsgType'] in srl:
continue
else:
log.log('MsgType Unknown: %s\n%s'%(m['MsgType'], str(m)), False)
out.print_line('MsgType Unknown: %s\n%s'%(m['MsgType'], str(m)), False)
srl.append(m['MsgType'])
continue
rl.append(msg)
......
config.py 0 → 100644
浏览文件 @ 05dd4684
import os, platform
DIR = os.getcwd()
BASE_URL = 'https://login.weixin.qq.com'
OS = platform.system() #Windows, Linux, Darwin
WELCOME_WORDS = 'Hello World!'
itchat/argparser.py 已删除 100644 → 0
浏览文件 @ 84206842
import os, time, thread
import config
arg = {
'talk': {
'help': 'talk [name] :talk to the user, only talk will talk to default user',
},
'send': {
'help': 'send name message :send message to the user',
},
'search': {
'help': 'search name :find the full name of the user',
},
'history': {
'help': 'history name [count] :print the history of the user',
},
'clear': {
'help': 'clear :clear the screen',
},
}
SEND_SYMBOL = '->'
def startCommandLine(client_s, client, msgList, front, cmdList):
frontStatus = 0 # 0 for command, 1 for talks
userTalkingTo = None
def set_header(header = ''):
nickName = client_s.find_nickname(client_s.userName)
front.set_header('%s%s%s'%(nickName if nickName else '', '@' if header else '', header))
while front.isLaunch or front.isPause:
if cmdList:
cmd = cmdList.pop()
if frontStatus == 0:
cmd = cmd.split(' ')
if cmd[0] == 'help':
for i,j in arg.items():
front.print_line(j['help'])
elif cmd[0] == 'talk':
if len(cmd) > 1:
nickName = ' '.join(cmd[1:])
userName = client_s.find_user(nickName)
if not userName: front.print_line('User Not Found');continue
else:
if client_s.lastInputUserName:
userName = client_s.lastInputUserName
else:
userName = client_s.userName
nickName = client_s.find_nickname(userName)
set_header(nickName)
frontStatus = 1
userTalkingTo = userName
front.print_line('Talking to %s, press Esc to exit talking mode'%nickName)
elif cmd[0] == 'send': #UserName BUG
if len(cmd) < 3 : front.print_line('%s params found 3 needed'%len(cmd));continue
for i in range(2, len(cmd)):
if cmd[i] == '--':
userName = client_s.find_user(cmd[1:i])
msg = ' '.join(cmd[i:])
if not userName: front.print_line('User Not Found');continue
if not msg: front.print_line('No message');continue
if msg and msg[1] == msg[-1] == '"': msg = msg[1:-1]
front.print_line('%s%s'%(SEND_SYMBOL, msg))
def send(front, userTalkingTo, cmd):
try:
client.send_msg(userTalkingTo, cmd)
except:
front.print_line('[FAIL: SEND AGAIN]%s'%cmd)
thread.start_new_thread(send, (front, userTalkingTo, msg))
elif cmd[0] == 'search':
if len(cmd) == 1: front.print_line('%s params found 2 needed'%len(cmd));continue
nickName = ' '.join(cmd[1:])
l = client_s.search_nickname(nickName)
if not l: front.print_line('No names found');continue
if len(l) > 21: front.print_line('%s names found, please keep it more specific'%len(l));continue
s = '%s found:'%len(l)
for i in range(len(l) / 3 + 1):
s += '\n'
for j in range(3):
if i * 3 + j < len(l): s += (l[i * 3 + j] + ' ')
front.print_line(s)
elif cmd[0] == 'history': #UserName BUG
if len(cmd) < 2 : front.print_line('%s params found 2 at least needed'%len(cmd));continue
if len(cmd) < 3 or not cmd[-1].isdigit():
userName = client_s.find_user(' '.join(cmd[1:]))
count = 10
else:
userName = client_s.find_user(' '.join(cmd[1:-1]))
count = int(cmd[-1])
if not userName: front.print_line('User Not Found');continue
for m in client_s.find_msg_list(userName, count):
msg = m[1]
if m[-1] == 'to':
msg = SEND_SYMBOL + msg
else:
msg = '%s: %s'%(m[2], msg)
front.print_line(msg)
elif cmd[0] == 'clear':
front.clear()
else:
front.print_line('Error command, type in \'help\' for help')
else:
if chr(27) in cmd:
frontStatus = 0
set_header('')
front.print_line('Talk with %s finished'%client_s.find_nickname(userTalkingTo))
else:
front.print_line('%s%s'%(SEND_SYMBOL, cmd))
def send(front, userTalkingTo, cmd):
try:
client.send_msg(userTalkingTo, cmd)
except:
front.print_line('[FAIL: SEND AGAIN]%s'%cmd)
thread.start_new_thread(send, (front, userTalkingTo, cmd))
cmd = None
time.sleep(.01)
itchat/config.py 已删除 100644 → 0
浏览文件 @ 84206842
import os, platform
DIR = os.getcwd()
QR_DIR = 'log'
LOG_DIR = 'log'
PIC_DIR = os.path.join('storage', 'picture')
VID_DIR = os.path.join('storage', 'video')
REC_DIR = os.path.join('storage', 'recording')
ATT_DIR = os.path.join('storage', 'attachment')
ACC_DIR = os.path.join('storage', 'account')
for d in [QR_DIR, LOG_DIR, PIC_DIR, VID_DIR, REC_DIR, ATT_DIR, ACC_DIR]:
if not os.path.exists(os.path.join(DIR, d)): os.makedirs(os.path.join(DIR, d))
BASE_URL = 'https://login.weixin.qq.com'
OS = platform.system() #Windows, Linux, Darwin
WELCOME_WORDS = 'Hello World!'
itchat/log.py 已删除 100644 → 0
浏览文件 @ 84206842
import time, os
import traceback
import config
LOG_DIR = config.LOG_DIR
def OpenLog(fn, *args, **kwargs):
def wrapped(*args, **kwargs):
with open(os.path.join(LOG_DIR, 'run.log'), 'a') as f:
result = fn(f, *args, **kwargs)
return result
return wrapped
@OpenLog
def log(f, msg, succeed = True, exception = None):
f.write('[%s]%s: %s\n'%(('SUCC' if succeed else 'FAIL'), time.ctime(), msg))
if exception: f.write(' %s\n'%(exception))
itchat/robot.py 已删除 100644 → 0
浏览文件 @ 84206842
#coding=utf8
import itchat.out as out
import itchat.log as log
from PluginTest import *
try:
import plugin.tuling as tuling
tuling.get_response('Hi', 'LittleCoder')
TULING = True
except:
TULING = False
PRINT_ON_CMD = True
def send_msg(client, toUserName, msg):
if len(msg) > 5:
if msg[:5] == '@fil@':
try:
with open(msg[5:]): pass
client.send_file(msg[5:], toUserName)
except:
log.log('Send file %s failed'%msg[5:], False)
elif msg[:5] == '@img@':
try:
with open(msg[5:]): pass
client.send_image(msg[5:], toUserName)
except:
log.log('Send image %s failed'%msg[5:], False)
elif msg[:5] == '@msg@':
client.send_msg(toUserName, msg[:5])
else:
client.send_msg(toUserName, msg)
else:
client.send_msg(toUserName, msg)
def get_reply(msg, s, client, isGroupChat = False, cmdPrint = PRINT_ON_CMD):
reply = ''
if msg['MsgType'] == 'Text':
if not isGroupChat and cmdPrint:
out.print_line('%s: %s'%(s.find_nickname(msg['FromUserName']), msg['Content']))
content = msg['Content']
# Plugins should be added in order as ('name', function)
pluginOrder = [('vote', vote), ('autoreply', autoreply), ('tuling', tuling.get_response)]
if isGroupChat: pluginOrder = [('autoreply', autoreply), ('tuling', tuling.get_response)]
getReply = False
for plugin in pluginOrder:
if plugin[0] in (pluginList['msgdealers'] + pluginList['systemmodules']):
r = plugin[1](content, client.storageClass, msg['FromUserName'])
if r:
reply = r
getReply = True
break
if not getReply: reply = 'I received: %s'%content
elif msg['MsgType'] == 'Map':
return 'You are there!'
if not isGroupChat and cmdPrint:
out.print_line('%s is at %s'%(s.find_nickname(msg['FromUserName']), msg['Content']))
elif msg['MsgType'] == 'Picture':
return 'Picture received!'
if not isGroupChat and cmdPrint:
out.print_line('%s sent a picture [%s]'%(s.find_nickname(msg['FromUserName']), msg['Content']))
elif msg['MsgType'] == 'Recording':
return 'Nice Voice!'
if not isGroupChat and cmdPrint:
out.print_line('%s sent a recording'%(s.find_nickname(msg['FromUserName'])))
elif msg['MsgType'] == 'Card':
return 'Greeting, %s!'%msg['Content']
if not isGroupChat and cmdPrint:
out.print_line('%s sent a business card of [%s]'%(s.find_nickname(msg['FromUserName']), msg['Content']))
elif msg['MsgType'] == 'Sharing':
return '"%s" is good!'%msg['Content']
if not isGroupChat and cmdPrint:
out.print_line('%s sent a web about [%s]'%(s.find_nickname(msg['FromUserName']), msg['Content']))
elif msg['MsgType'] == 'Attachment':
return '"%s" received!'%msg['Content']
if not isGroupChat and cmdPrint:
out.print_line('%s sent an attachment: [%s]'%(s.find_nickname(msg['FromUserName']), msg['Location']))
elif msg['MsgType'] == 'Video':
return 'I received a video'
if not isGroupChat and cmdPrint:
out.print_line('%s sent a video [%s]'%(s.find_nickname(msg['FromUserName']), msg['Content']))
elif msg['MsgType'] == 'Note':
if not isGroupChat and cmdPrint: out.print_line('Notification: %s'%(msg['Content']))
else:
pass#out.print_line(str(msg))
return reply
def deal_with_msg(msg, s, client):
if msg.has_key('FromUserName') and msg['FromUserName'][:2] == '@@':
try:
r = grouptalking(msg, s, client, get_reply)
send_msg(client, msg['FromUserName'], r)
except:
log.log('Send group chat failed', False)
else:
r = get_reply(msg, s, client)
send_msg(client, msg['FromUserName'], r)
itchat/storage.py 已删除 100644 → 0
浏览文件 @ 84206842
import os, sqlite3, time
import config
from plugin.Sqlite3Client import Sqlite3Client
class Storage:
def __init__(self):
self.userName = None
self.nickName = None
self.msgList = []
self.lastInputUserName = None
def load_sql_storage(self):
self.sqlDir = os.path.join(config.ACC_DIR, '%s.db'%self.nickName)
with Sqlite3Client(self.sqlDir) as s3c:
s3c.execute('create table if not exists message (time integer, message text, nickname text, fromto text)')
s3c.execute('create table if not exists memberList (PYQuanPin text, NickName text, UserName text, Other text)')
def find_msg_list(self, userName, count):
with Sqlite3Client(self.sqlDir) as s3c:
r = s3c.query('select * from message where nickname=? order by time desc limit ?',
(self.find_nickname(userName), count))
return r
def store_msg(self, userName, msg, fromto):
with Sqlite3Client(self.sqlDir) as s3c:
s3c.insert_data('message',
[int(time.time()), msg, self.find_nickname(userName), fromto])
if fromto == 'from': self.lastInputUserName = userName
def update_user(self, PYQuanPin, **kwargs):
with Sqlite3Client(self.sqlDir) as s3c:
dataInStorage = s3c.query('select count(*) from memberList where PYQuanPin = ?', (PYQuanPin,))[0][0]
if dataInStorage == 0:
dataDict = {'NickName': '', 'UserName': '', 'Other': ''}
else:
dataTuple = s3c.query('select * from memberList where PYQuanPin = ?', (PYQuanPin,))[0]
dataDict = {'NickName': dataTuple[1], 'UserName': dataTuple[2], 'Other': dataTuple[3]}
s3c.execute('delete from memberList where PYQuanPin = ?', (PYQuanPin,))
for key, value in kwargs.items(): dataDict[key] = value
try:
s3c.insert_data('memberList', items = [PYQuanPin, dataDict['NickName'], dataDict['UserName'], dataDict['Other']])
except:
return dataDict
return True
def find_PYQuanPin(self, userName):
with Sqlite3Client(self.sqlDir) as s3c:
members = s3c.query('select * from memberList where UserName = ?', (userName,))
for member in members: return member[0]
def find_user(self, n):
with Sqlite3Client(self.sqlDir) as s3c:
members = s3c.query('select * from memberList where NickName = ?', (n,))
for member in members: return member[2]
def find_nickname(self, u):
with Sqlite3Client(self.sqlDir) as s3c:
members = s3c.query('select * from memberList where UserName = ?', (u,))
for member in members: return member[1]
def search_nickname(self, n):
r = []
with Sqlite3Client(self.sqlDir) as s3c:
members = s3c.query('select * from memberList where NickName like ?', ('%' + n + '%',))
for member in members: r.append(member[1])
return r
def get_other(self, userName):
with Sqlite3Client(self.sqlDir) as s3c:
members = s3c.query('select * from memberList where UserName = ?', (userName,))
for member in members: return member[3]
def get_dict_of_other(self, s):
d = {}
for item in s.split(';'):
if len(item.split(',')) == 2: d[item.split(',')[0]] = item.split(',')[1]
return d
def get_str_of_other(self, d):
return ';'.join(['%s,%s'%(key, value) for key, value in d.items()])
plugin/ChatLikeCMD.py 已删除 100644 → 0
浏览文件 @ 84206842
#coding=utf8
import thread, time, sys, os, platform
try:
import termios, tty
termios.tcgetattr, termios.tcsetattr
import threading
OS = 'Linux'
except (ImportError, AttributeError):
try:
import msvcrt
OS = 'Windows'
except ImportError:
raise Exception('Mac is currently not supported')
OS = 'Mac'
else:
getch = msvcrt.getwch
else:
def fn():
try:
fd = sys.stdin.fileno()
old_settings = termios.tcgetattr(fd)
tty.setraw(fd)
ch = sys.stdin.read(1)
except:
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
raise Exception
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
return ch
getch = fn
CMD_HISTORY = 30
class ChatLikeCMD():
def __init__(self, header = 'LittleCoder', symbol = '>', inPip = None, inputMaintain = False):
self.strBuff = []
self.cmdBuff = []
self.historyCmd = -1
self.cursor = 0
self.inPip = [] if inPip == None else inPip
self.outPip = []
self.isLaunch = False
self.isPause = False
self.header = header
self.symbol = symbol
self.inputMaintain = inputMaintain
def reprint_input(self):
sys.stdout.write(self.header + self.symbol)
if self.strBuff:
for i in self.strBuff: sys.stdout.write(i)
sys.stdout.flush()
def getch(self):
c = getch()
return c if c != '\r' else '\n'
def get_history_command(self, direction):
if direction == 'UP':
if self.historyCmd < CMD_HISTORY - 1 and self.historyCmd < len(self.cmdBuff) - 1: self.historyCmd += 1
else:
if self.historyCmd == 0: return ''
if self.historyCmd > 0: self.historyCmd -= 1
if -1 < self.historyCmd < len(self.cmdBuff): return self.cmdBuff[self.historyCmd]
def output_command(self, s):
self.outPip.append(s if isinstance(s, unicode) else s.decode(sys.stdin.encoding))
if len(self.cmdBuff) >= CMD_HISTORY: self.cmdBuff = self.cmdBuff[::-1].pop()[::-1]
self.cmdBuff.append(s)
def print_thread(self):
while self.isLaunch:
if self.inPip:
sys.stdout.write('\r' + ' ' * 50 + '\r')
sys.stdout.flush()
print self.inPip.pop()
# linux special
sys.stdout.write('\r')
sys.stdout.flush()
self.reprint_input()
time.sleep(0.01)
def fast_input_test(self):
timer = threading.Timer(0.001, thread.interrupt_main)
c = None
try:
timer.start()
c = getch()
except:
pass
timer.cancel()
return c
def process_direction_char(self, c):
if OS == 'Windows':
if ord(c) == 72:
c = 'A'
elif ord(c) == 80:
c = 'B'
elif ord(c) == 77:
c = 'C'
elif ord(c) == 75:
c = 'D'
if ord(c) == 68: # LEFT
self.process_char('\b')
return
# cursor bugs
if self.cursor > 0:
if OS == 'Windows':
sys.stdout.write(chr(224) + chr(75))
else:
sys.stdout.write(chr(27) + '[C')
self.cursor -= 1
elif ord(c) == 67: # RIGHT
return
# cursor bugs
if self.cursor < len(self.strBuff):
if OS == 'Windows':
sys.stdout.write(chr(224) + chr(77))
else:
sys.stdout.write(chr(27) + '[D')
self.cursor += 1
elif ord(c) == 65: # UP
hc = self.get_history_command('UP')
if not hc is None:
self.strBuff = [i for i in hc]
self.cursor = len(hc)
sys.stdout.write('\r' + ' ' * 50 + '\r')
self.reprint_input()
elif ord(c) == 66: # DOWN
hc = self.get_history_command('DOWN')
if not hc is None:
self.strBuff = [i for i in hc]
self.cursor = len(hc)
sys.stdout.write('\r' + ' ' * 50 + '\r')
self.reprint_input()
else:
raise Exception(c)
def process_char(self, c):
if ord(c) == 27: # Esc
if OS == 'Linux':
fitc1 = self.fast_input_test()
if ord(fitc1) == 91:
fitc2 = self.fast_input_test()
if 65 <= ord(fitc2) <= 68:
self.process_direction_char(fitc2)
return
sys.stdout.write('\r' + ' ' * 50 + '\r')
sys.stdout.flush()
self.reprint_input()
self.outPip.append(c)
time.sleep(0.02)
if 'fitc1' in dir():
self.process_char(fitc1)
self.cursor += 1
if 'fitc2' in dir():
self.process_char(fitc2)
self.cursor += 1
elif ord(c) == 3: # Ctrl+C
self.stop()
self.isPause = True
if raw_input('Exit?(y) ') == 'y':
sys.stdout.write('Command Line Exit')
else:
self.start()
self.isPause = False
elif ord(c) in (8, 127): # Backspace
if self.strBuff:
if ord(self.strBuff[-1]) < 128:
sys.stdout.write('\b \b')
else:
sys.stdout.write('\b\b \b')
if OS == 'Linux':
self.strBuff.pop()
self.strBuff.pop()
self.strBuff.pop()
self.cursor -= 1
elif c == '\n':
if self.strBuff:
if self.inputMaintain:
sys.stdout.write(c)
else:
sys.stdout.write('\r' + ' ' * 50 + '\r')
sys.stdout.flush()
self.reprint_input()
self.output_command(''.join(self.strBuff))
self.strBuff = []
self.historyCmd = -1
elif ord(c) == 224: # Windows direction
if OS == 'Windows':
direction = self.getch()
self.process_direction_char(direction)
else:
sys.stdout.write(c)
sys.stdout.flush()
self.strBuff.append(c)
self.cursor += 1
def command_thread(self):
c = None
while self.isLaunch:
c = self.getch()
self.process_char(c)
time.sleep(0.01)
def start(self):
self.isLaunch = True
thread.start_new_thread(self.print_thread, ())
self.reprint_input()
thread.start_new_thread(self.command_thread, ())
def stop(self):
sys.stdout.write('\r' + ' ' * 50 + '\r')
sys.stdout.flush()
self.isLaunch = False
def print_line(self, msg = None):
self.inPip.append(msg)
def clear(self):
os.system('cls' if platform.system() == 'Windows' else 'clear')
self.reprint_input()
def get_command_pip(self):
return self.outPip
def set_header(self, header):
self.header = header
if __name__ == '__main__':
c = ChatLikeCMD()
s = c.get_command_pip()
c.start()
def loopinput(c):
while True:
c.print_line('LOOP INPUT......')
time.sleep(3)
thread.start_new_thread(loopinput, (c,))
while c.isLaunch or c.isPause:
if s:
c.print_line(s.pop())
time.sleep(0.01)
plugin/QRCode.py 已删除 100644 → 0
浏览文件 @ 84206842
from PIL import Image
import sys, os, platform
BLOCK = '\xA1\xF6' if platform.system() == 'Windows' else 'MM'
class QRCode():
def __init__(self, fileName, size, padding = 0, background = 'BLACK'):
self.size = size
self.padding = padding
self.img = Image.open(fileName)
self.times = self.img.size[0]/(size + padding * 2)
self.rgb = self.img.convert('RGB')
self.white = BLOCK if background == 'BLACK' else ' '
self.black = ' ' if background == 'BLACK' else BLOCK
def print_qr(self):
sys.stdout.write(' '*50 + '\r')
sys.stdout.flush()
print self.white * (self.size + 2)
startPoint = self.padding + 0.5
for y in range(self.size):
sys.stdout.write(self.white)
for x in range(self.size):
r,g,b = self.rgb.getpixel(((x + startPoint) * self.times, (y + startPoint) * self.times))
sys.stdout.write(self.white if r > 127 else self.black)
print self.white
print self.white * (self.size + 2)
if __name__ == '__main__':
q = QRCode(os.path.join(os.path.pardir, 'log', 'QR.jpg'), 37, 3, 'BLACK')
q.print_qr()
plugin/Sqlite3Client.py 已删除 100644 → 0
浏览文件 @ 84206842
import sqlite3, re, thread
import json, sys
MAX_NUM = int(2e4)
class Sqlite3Client:
def __enter__(self):
return self
def __init__(self, sqlDir):
self.sqlDir = sqlDir
self._connection = sqlite3.connect(sqlDir)
self._cursor = self._connection.cursor()
self._storedDataSource = None
def execute(self, sql, data = []):
self._cursor.execute(sql, data)
self._connection.commit()
def query(self, sql, data = []):
self._cursor.execute(sql, data)
return self._cursor.fetchall()
def insert_data(self, tableName, items = []):
self._cursor.execute('insert into %s values(%s)'%(tableName,
', '.join(['?' for i in items])), items)
self._connection.commit()
# the following 4 functions don't use safe execute and may cause problems
# please make sure the input is safe before use them
def restruct_table(self, tableName, orderBy, restructedTableName = None):
if restructedTableName is None: restructedTableName = 'restructed_' + tableName
orderByString = ', '.join(['%s %s'%(key[0], key[1]) for key in orderBy])
self.query(self.query('show create table %s'%tableName)[0][1].replace(tableName, restructedTableName, 1))
sys.stdout.write('Restructuring the data storage...\r')
totalNum = self.query('select count(*) from %s'%tableName)[0][0]
s = self.data_source('select * from %s order by %s'%(tableName, orderByString))
count = 0
totalCount = 0
process = -1
for data in s:
self.insert_data(restructedTableName, data)
count += 1
totalCount += 1
if process < totalCount * 100 / totalNum:
process = totalCount * 100 / totalNum
sys.stdout.flush()
sys.stdout.write('Restructuring the data storage: %s%s\r'%(process, '%'))
if count >= MAX_NUM:
count = 0
self._connection.commit()
self._connection.commit()
print 'Restructuring Finished'
return restructedTableName
def simple_data_source(self, sql): # return a function to provide one data at a time
c = self._connection.cursor()
c.execute(sql)
for item in c.fetchall(): yield item
def parallel_get_source_of_data_source(self, sql, beginNumber = 0):
regex = re.compile('from (\S+)')
tableName = re.findall(regex, sql)[0]
totalNum = self.query('select count(*) from %s'%tableName)[0][0]
unitNumber = totalNum / MAX_NUM + 1
self._storedDataSource = self.simple_data_source('%s limit %s, %s'%(sql, beginNumber, MAX_NUM))
def get(i):
self._storedDataSource = self.simple_data_source('%s limit %s, %s'%(sql, i * MAX_NUM, MAX_NUM))
if unitNumber == beginNumber / MAX_NUM: yield self._storedDataSource
for i in range(beginNumber / MAX_NUM + 1, unitNumber + 1):
while self._storedDataSource is None: print 'Thread sucks'
r = self._storedDataSource
self._storedDataSource = None
thread.start_new_thread(get, (i,))
yield r
def data_source(self, sql): # limit is now useless here
regex = re.compile('(select .*? from .*?)(?: limit (\S+),.*)?$')
r = re.findall(regex, sql)[0]
sourceOfDataSource = self.parallel_get_source_of_data_source(r[0], int(r[1]) if r[1] else 0)
for dataSource in sourceOfDataSource:
for data in dataSource: yield data
def __exit__(self, *args):
if self._cursor: self._cursor.close()
if self._connection: self._connection.close()
if __name__ == '__main__':
with Sqlite3Client('wcStorage.db') as s3c:
s3c.insert_data('message', items = [str({'a':'a'}),'a:""','a','a'])
r = s3c.data_source('select * from message')
for item in r:
print item
plugin/config/fields.py 已删除 100644 → 0
浏览文件 @ 84206842
from __future__ import absolute_import
import email.utils
import mimetypes
from .packages import six
def guess_content_type(filename, default='application/octet-stream'):
"""
Guess the "Content-Type" of a file.
:param filename:
The filename to guess the "Content-Type" of using :mod:`mimetypes`.
:param default:
If no "Content-Type" can be guessed, default to `default`.
"""
if filename:
return mimetypes.guess_type(filename)[0] or default
return default
def format_header_param(name, value):
"""
Helper function to format and quote a single header parameter.
Particularly useful for header parameters which might contain
non-ASCII values, like file names. This follows RFC 2231, as
suggested by RFC 2388 Section 4.4.
:param name:
The name of the parameter, a string expected to be ASCII only.
:param value:
The value of the parameter, provided as a unicode string.
"""
if not any(ch in value for ch in '"\\\r\n'):
result = '%s="%s"' % (name, value)
try:
result.encode('ascii')
except UnicodeEncodeError:
pass
else:
return result
if not six.PY3: # Python 2:
value = value.encode('utf-8')
value = email.utils.encode_rfc2231(value, 'utf-8')
# value = '%s*=%s' % (name, value)
value = '%s="%s"' % (name, value.decode('utf8'))
return value
class RequestField(object):
"""
A data container for request body parameters.
:param name:
The name of this request field.
:param data:
The data/value body.
:param filename:
An optional filename of the request field.
:param headers:
An optional dict-like object of headers to initially use for the field.
"""
def __init__(self, name, data, filename=None, headers=None):
self._name = name
self._filename = filename
self.data = data
self.headers = {}
if headers:
self.headers = dict(headers)
@classmethod
def from_tuples(cls, fieldname, value):
"""
A :class:`~urllib3.fields.RequestField` factory from old-style tuple parameters.
Supports constructing :class:`~urllib3.fields.RequestField` from
parameter of key/value strings AND key/filetuple. A filetuple is a
(filename, data, MIME type) tuple where the MIME type is optional.
For example::
'foo': 'bar',
'fakefile': ('foofile.txt', 'contents of foofile'),
'realfile': ('barfile.txt', open('realfile').read()),
'typedfile': ('bazfile.bin', open('bazfile').read(), 'image/jpeg'),
'nonamefile': 'contents of nonamefile field',
Field names and filenames must be unicode.
"""
if isinstance(value, tuple):
if len(value) == 3:
filename, data, content_type = value
else:
filename, data = value
content_type = guess_content_type(filename)
else:
filename = None
content_type = None
data = value
request_param = cls(fieldname, data, filename=filename)
request_param.make_multipart(content_type=content_type)
return request_param
def _render_part(self, name, value):
"""
Overridable helper function to format a single header parameter.
:param name:
The name of the parameter, a string expected to be ASCII only.
:param value:
The value of the parameter, provided as a unicode string.
"""
return format_header_param(name, value)
def _render_parts(self, header_parts):
"""
Helper function to format and quote a single header.
Useful for single headers that are composed of multiple items. E.g.,
'Content-Disposition' fields.
:param header_parts:
A sequence of (k, v) typles or a :class:`dict` of (k, v) to format
as `k1="v1"; k2="v2"; ...`.
"""
parts = []
iterable = header_parts
if isinstance(header_parts, dict):
iterable = header_parts.items()
for name, value in iterable:
if value:
parts.append(self._render_part(name, value))
return '; '.join(parts)
def render_headers(self):
"""
Renders the headers for this request field.
"""
lines = []
sort_keys = ['Content-Disposition', 'Content-Type', 'Content-Location']
for sort_key in sort_keys:
if self.headers.get(sort_key, False):
lines.append('%s: %s' % (sort_key, self.headers[sort_key]))
for header_name, header_value in self.headers.items():
if header_name not in sort_keys:
if header_value:
lines.append('%s: %s' % (header_name, header_value))
lines.append('\r\n')
return '\r\n'.join(lines)
def make_multipart(self, content_disposition=None, content_type=None,
content_location=None):
"""
Makes this request field into a multipart request field.
This method overrides "Content-Disposition", "Content-Type" and
"Content-Location" headers to the request parameter.
:param content_type:
The 'Content-Type' of the request body.
:param content_location:
The 'Content-Location' of the request body.
"""
self.headers['Content-Disposition'] = content_disposition or 'form-data'
self.headers['Content-Disposition'] += '; '.join([
'', self._render_parts(
(('name', self._name), ('filename', self._filename))
)
])
self.headers['Content-Type'] = content_type
self.headers['Content-Location'] = content_location
plugin/config/tuling.json 已删除 100644 → 0
浏览文件 @ 84206842
{
"key": "e0daf2ded41fdc6e471377**********"
}
plugin/config/vote.json 已删除 100644 → 0
浏览文件 @ 84206842
[
{
"name":
"默认投票",
"candidates":
["candidate1", "LittleCoder", "nobody"]
},
{
"name":
"能投三票的投票",
"candidates":
["another candidate1", "only one LittleCoder", "still nobody"],
"maxvote":
3
}
]
plugin/msgdealers/autoreply.py 已删除 100644 → 0
浏览文件 @ 84206842
#coding=utf8
import re, json, time, os
import traceback
from plugin.Sqlite3Client import Sqlite3Client
SQLITE_DIR = os.path.join('plugin', 'config')
FILE_DIR = os.path.join('storage', 'upload')
def compileRegex(tableName, regexList):
regex = ''
try:
with Sqlite3Client(os.path.join(SQLITE_DIR, 'autoreply.db')) as s3c:
for qa in s3c.data_source('select * from %s'%tableName):
regex = qa[0]
regexList.append((re.compile(qa[0]), qa[1]))
except:
raise Exception('Error occured when loading regex table %s: %s is not a correct regex'%(
tableName, regex))
def detectFiles(tableName):
if not os.path.exists(FILE_DIR): os.makedirs(FILE_DIR)
fileName = ''
try:
with Sqlite3Client(os.path.join(SQLITE_DIR, 'autoreply.db')) as s3c:
for qa in s3c.data_source('select * from %s'%tableName):
if qa[1][:5] == '@fil@':
with open(os.path.join(FILE_DIR, qa[1][5:])): pass
elif qa[1][:5] == '@img@':
with open(os.path.join(FILE_DIR, qa[1][5:])): pass
except:
raise Exception('Error occured when loading "%s" in table %s, it should be in storage/upload'%(fileName, tableName))
def getreply():
regexAnsList = []
tableNameList = ['default_reply']
for tableName in tableNameList:
detectFiles(tableName)
compileRegex(tableName, regexAnsList)
while 1:
msg = (yield)
r = False
for regexAns in regexAnsList:
if not re.search(regexAns[0], msg) is None: r = regexAns[1]; break
yield r
getreplyiter = getreply()
getreplyiter.next()
def autoreply(msg, storageClass = None, userName = None):
r = getreplyiter.send(msg)
if r and r[:5] in ('@fil@', '@img@'): r = '%s%s'%(r[:5], os.path.join(FILE_DIR, r[5:]))
getreplyiter.next()
return r
plugin/msgdealers/grouptalking.py 已删除 100644 → 0
浏览文件 @ 84206842
import os, re, time
from chardet import detect
from plugin.Sqlite3Client import Sqlite3Client
SQL_DIR = os.path.join('storage', 'grouptalk')
SQL_NAME = 'grouptalk.db'
if not os.path.exists(SQL_DIR): os.mkdir(SQL_DIR)
with Sqlite3Client(os.path.join(SQL_DIR, SQL_NAME)) as s3c:
s3c.execute('create table if not exists messages (groupname text, time integer, message text, username text, fromto text)')
s3c.execute('create table if not exists memberlist (groupname text, username text, nickname text)')
class GroupRobot:
def __init__(self):
pass
def clear_table(self):
with Sqlite3Client(os.path.join(SQL_DIR, SQL_NAME)) as s3c:
s3c.execute('delete from memberlist')
def touch_user(self, groupName, userName):
with Sqlite3Client(os.path.join(SQL_DIR, SQL_NAME)) as s3c:
r = s3c.query('select count(*) from memberlist where groupname=? and username=?',
[groupName, userName])[0][0]
return True if r else False
def update_group(self, client, groupName):
memberList = client.get_batch_contract(groupName)
memberDict = {member['UserName']: member for member in memberList}
with Sqlite3Client(os.path.join(SQL_DIR, SQL_NAME)) as s3c:
q = s3c.query('select * from memberlist where groupname=?', (groupName,))
for member in q: del memberDict[member[1]]
for userName, member in memberDict.items():
s3c.insert_data('memberlist', (groupName, member['UserName'], member['NickName']))
def store_msg(self, content, groupName, userName, fromto):
with Sqlite3Client(os.path.join(SQL_DIR, SQL_NAME)) as s3c:
s3c.insert_data('messages', [groupName, int(time.time()), content, userName, fromto])
def change_msg_format(self, groupName, userName, msg):
if msg[:5] in ['@fil@', '@img@']: return msg
with Sqlite3Client(os.path.join(SQL_DIR, SQL_NAME)) as s3c:
nickName = s3c.query('select * from memberlist where groupname=? and username=?',
[groupName, userName])[0][2]
# return ('%s:<br/>@%s\342\200\205%s'%(userName.encode('utf8'),
# nickName.encode('utf8'), msg.encode('utf8'))).decode('utf8')
return ('@%s\342\200\205%s'%(nickName.encode('utf8'), msg.encode('utf8'))).decode('utf8')
def get_msg_from_raw(msg):
regex = re.compile('(@[0-9a-z]*?):<br/>(.*)$')
r = re.findall(regex, msg)
if r:
return r[0][0], r[0][1]
else:
return '', ''
gr = GroupRobot()
gr.clear_table()
def grouptalking(msg, s, client, get_reply, replyToAll = False):
# here needs an emoji test
groupName = msg['FromUserName']
userName, msg['Content'] = get_msg_from_raw(msg['Content'])
if not(client.storageClass.nickName in msg['Content'] or replyToAll): return
msg['Content'] = msg['Content'].replace('@%s'%client.storageClass.nickName, '')
reply = get_reply(msg, s, client, isGroupChat = True)
if not gr.touch_user(groupName, userName): gr.update_group(client, groupName)
gr.store_msg(msg['Content'], groupName, userName, 'from')
gr.store_msg(reply, groupName, userName, 'to')
return gr.change_msg_format(groupName, userName, reply)
plugin/msgdealers/vote.py 已删除 100644 → 0
浏览文件 @ 84206842
#coding=utf8
import re, json, time, os
from plugin.Sqlite3Client import Sqlite3Client
VOTE_KEYWORD = u'投票'
def load_vote_list():
with open(os.path.join('plugin', 'config', 'vote.json')) as f: voteList = json.loads(f.read())
if not voteList: raise Exception
for voteItem in voteList: # test json format
if not (voteItem.has_key('name') and voteItem.has_key('candidates')): raise Exception
if not (isinstance(voteItem['candidates'], list) and voteItem['candidates']): raise Exception
return voteList
def get_status(storageClass, userName):
r = storageClass.get_other(userName)
return storageClass.get_dict_of_other(r)
def store_status(storageClass, userName, other):
PYQuanPin = storageClass.find_PYQuanPin(userName)
storageClass.update_user(PYQuanPin, Other = storageClass.get_str_of_other(other))
def vote(msg, storageClass, userName):
if not VOTE_KEYWORD in msg: return False
# key in sqlite3->other: vote
# value in sqlite3->other: 0 for voted, 1 for not voted
# values should be strings
voteList = load_vote_list()
if voteList is None: return False
status = get_status(storageClass, userName)
for voteItem in voteList:
# load vote details
if not voteItem['name'] in msg: continue
if voteItem.has_key('maxvote'):
maxVote = voteItem['maxvote']
else:
maxVote = 1
# load user details
if status.has_key('vote@' + voteItem['name']):
currentVote = int(status['vote@' + voteItem['name']])
if currentVote <= 0:
return u'你已经在“%s”中完成了投票'%voteItem['name']
else:
currentVote = maxVote
# check vote
candidate = None
for c in voteItem['candidates']:
# bug will occur if one candidates's name contains another's
if c in msg: candidate = c;break
if candidate is None: return u'投票选项包括: ' + ', '.join(voteItem['candidates'])
# update vote database
with Sqlite3Client(storageClass.sqlDir) as s3c:
s3c.execute('create table if not exists vote (name text, candidate text, PYQuanPin text, time text)')
PYQuanPin = storageClass.find_PYQuanPin(userName)
s3c.insert_data('vote', items = [voteItem['name'], candidate, PYQuanPin, int(time.time())])
# update user status
status['vote@' + voteItem['name']] = str(currentVote - 1)
store_status(storageClass, userName, status)
return u'已经成功投票给“%s”%s'%(candidate, '' if currentVote == 1 else u'，你还可以投%s票'%(currentVote - 1))
return u'目前进行的投票有: ' + ', '.join([voteItem['name'] for voteItem in voteList])
if __name__ != '__main__':
load_vote_list()
plugin/tuling.py 已删除 100644 → 0
浏览文件 @ 84206842
#coding=utf8
import sys, os
import requests, json
from ChatLikeCMD import ChatLikeCMD
try:
with open(os.path.join('plugin', 'config', 'tuling.json')) as f: key = json.loads(f.read())['key']
except:
raise Exception('There is something wrong with the format of you plugin/config/tuling.json')
def get_response(msg, storageClass = None, userName = None, userid = 'ItChat'):
url = 'http://www.tuling123.com/openapi/api'
payloads = {
'key': key,
'info': msg,
'userid': userid,
}
r = json.loads(requests.post(url, data = payloads).text)
if not r['code'] in (100000, 200000, 302000, 308000, 313000, 314000):
if r['code'] == 400004: return None
raise Exception('code: %s'%r['code'])
elif r['code'] == 100000: # 文本类
return '\n'.join([r['text'].replace('<br>','\n')])
elif r['code'] == 200000: # 链接类
return '\n'.join([r['text'].replace('<br>','\n'), r['url']])
elif r['code'] == 302000: # 新闻类
l = [r['text'].replace('<br>','\n')]
for n in r['list']: l.append('%s - %s'%(n['article'], n['detailurl']))
return '\n'.join(l)
elif r['code'] == 308000: # 菜谱类
l = [r['text'].replace('<br>','\n')]
for n in r['list']: l.append('%s - %s'%(n['name'], n['detailurl']))
return '\n'.join(l)
elif r['code'] == 313000: # 儿歌类
return '\n'.join([r['text'].replace('<br>','\n')])
elif r['code'] == 314000: # 诗词类
return '\n'.join([r['text'].replace('<br>','\n')])
if __name__ == '__main__':
while True:
a = raw_input('>').decode(sys.stdin.encoding).encode('utf8')
for m in get_response(a, 'ItChat'): print m
pluginlist.json 已删除 100644 → 0
浏览文件 @ 84206842
{
"systemmodules": [
"QRCode",
"tuling",
"grouptalking"
],
"msgdealers": [
"autoreply",
"vote"
]
}
storage.py 0 → 100644
浏览文件 @ 05dd4684
import os, sqlite3, time
import config
from plugin.Sqlite3Client import Sqlite3Client
class Storage:
def __init__(self):
self.userName = None
self.nickName = None
self.historyMsg = []
self.memberList = {}
self.msgList = []
self.lastInputUserName = None
def find_msg_list(self, userName, count):
r = []
for msg in self.historyMsg:
if msg['UserName'] == userName:
r.append(msg)
if len(r) >= count: break
return r
def updateMemberList(memberList):
for member in memberList: self.update_user(member)
def store_msg(self, msg, fromto = 'from'):
msg['fromto'] = fromto
self.historyMsg.append(msg)
def update_user(self, user):
self.memberList[user['UserName']] = user
def find_username(self, n):
r = []
for userName, member in memberList.items():
if member['NickName'] == n: r.append(member['UserName'])
return r
def find_nickname(self, u):
r = []
for userName, member in memberList.items():
if member['UserName'] == u: return member['NickName']
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或