[TOC]
目標(biāo)
做為一個程序員耍群,已成中年油膩大叔
了木西?那怎么行畴栖,來趣消除App
、扶我起來學(xué)數(shù)學(xué)App
的成語消消樂
游戲battle下
這里沒有作弊一說八千,只有發(fā)揮自己的特長吗讶。做為已畢業(yè)十幾年的人,肯定不是成語容量恋捆,而是工作專業(yè)技能 -- 上代碼
測試環(huán)境
App: 趣消除App
iOS版本照皆、扶我起來學(xué)數(shù)學(xué)App
版本
工具: mitmproxy、python沸停、Charles
背景知識:mitmproxy膜毁、python、抓包
解決
分析
游戲界面上呈現(xiàn)的:
網(wǎng)絡(luò)呈現(xiàn):
游戲是通過websocket協(xié)議來傳輸json格式的字符串愤钾,舉asking
消息如下:
{
"ask_string": "步不之來疏筆口平去重學(xué)青字斟淺之論暗明尊易伐道不云句來刊酌誅才師",
"type": "asking",
"scores": [
{
"nick": "呂耀輝",
....
},
{
"nick": "xxxx",
......
}
]
}
成語答案:answer
消息
{
"answer": "口誅筆伐",
"answer_index": [
"6",
"29",
"5",
"21"
],
"type": "answer"
}
目的是把上面的"ask_string": "步不之來疏筆口平去重學(xué)青字斟淺之論暗明尊易伐道不云句來刊酌誅才師"
解析為一個個成語
工作原理
寫2個文件:chengyu_mitm.py
[代碼文件]瘟滨、chengyu.text
[數(shù)據(jù)文件]
-
chengyu_mitm.py
從asking
消息里解析出ask_string
到chengyu.text
文件里查找是否包含相應(yīng)的成語 - 如果包含成語,圖形化能颁、格式化顯示結(jié)果
-
chengyu.text
文件剛開始是空的杂瘸;在每局游戲結(jié)束時,游戲都會發(fā)送game_result
消息給我們伙菊,里面有這局游戲的答案成語败玉,把這些成語寫到文件中 - 玩的局?jǐn)?shù)越多,
chengyu.text
文件包含的成語越多镜硕,查找到答案的可能性越大
所以我們只要關(guān)注:asking
消息运翼、game_result
消息
如果要程序回復(fù)答案,可以關(guān)注下answer
消息[客戶端發(fā)給服務(wù)器的]
代碼
chengyu_mitm.py
:
import json
import time
from mitmproxy import ctx
'''
> mitmdump -s chengyu_mitm.py '~u websock_m/websock_message'
'''
Red = '\033[0;31m'
Green = '\033[0;32m'
Yellow = '\033[0;33m'
Blue = '\033[0;34m'
Purple = '\033[0;35m'
Cyan = '\033[0;36m'
White = '\033[0;37m'
colors = {
0:Red,
1:Purple,
2:Yellow,
3:Blue,
4:White,
}
class Chengyu(object):
def __init__(self):
self.dictpath = '/Users/zhoujie/chengyu.text'
self.chengyu = set()
with open(self.dictpath, 'rt') as f:
for line in f.readlines():
self.chengyu.add(line.strip())
self.answers = list()
self.ask_string = ''
# {'和':[1,8], '我':[11]}
self.char_indexs_dict = dict()
# {1:'和', 8:'和', 11:'我'}
self.index_char_dict = dict()
self.count = 0
# 自動提交答案的網(wǎng)絡(luò)發(fā)送次數(shù)
self.auto_send_count = 0
# 找到的的成語中各異字符為2個的答案數(shù)量:如 [真真假假]
self.answer_2chars_count = 0
# {'中流砥柱':[1,9,21,25]}
self.answer_indexs_dict = dict()
#
self.error_answers = []
# 玩了多少局
self.play_times = 0
# General lifecycle
def load(self, loader):
ctx.log.info('\033[1;31mevent: load\033[0m')
def configure(self, updated):
ctx.log.info('\033[1;31mevent: configure\033[0m')
# Websocket lifecycle
def websocket_message(self, flow):
"""
Called when a WebSocket message is received from the client or
server. The most recent message will be flow.messages[-1]. The
message is user-modifiable. Currently there are two types of
messages, corresponding to the BINARY and TEXT frame types.
"""
ctx.log.info('\033[1;31m websocket_message \033[0m')
# get the latest message
message = flow.messages[-1]
# simply print the content of the message
ctx.log.info('')
ctx.log.info(message.content)
ctx.log.info('')
m = json.loads(message.content)
t = m['type']
if m.get('ask_string'):
ask_string = m['ask_string']
self.ask_string = ask_string
# 計(jì)算答案
self.find_answers_v2(ask_string)
self.play_times += 1
if m['type'] == 'answer':
self.answer_indexs_dict[m['answer']] = m['answer_index']
# 刪除已回答正確的答案
if m.get('ack') == 1:
answer = m['answer']
answer_index = self.answer_indexs_dict.get(answer,[])
for i in answer_index:
self.index_char_dict[int(i)] = ' '
try:
self.answers.remove(m['answer'])
except:
pass
# 自動答題
self.auto_answer(flow)
# 顯示答案
self.print_answers()
if m['type'] == 'game_result':
# 把答案增加到內(nèi)存字典中
self.__add_new_worlds_to_memory(m)
self.reset_data_to_init()
def websocket_end(self, flow):
"""
A websocket connection has ended.
"""
ctx.log.info('\033[1;31m websocket_end \033[0m')
self.reset_data_to_init()
if self.play_times % 5 == 0:
with open(self.dictpath, 'wt') as f:
l = list(self.chengyu)
l.sort()
for item in l:
f.write(item)
f.write('\n')
# ---------------------------------------------
def find_answers_v2(self, ask_string):
'''
在內(nèi)存成語字典查找答案
'''
ask_set = set(ask_string)
for i, c in enumerate(ask_string):
self.char_indexs_dict.setdefault(c, []).append(i)
self.index_char_dict = dict( zip(range(len(ask_string)), ask_string))
max_count = len(ask_string) / 4
for item in self.chengyu:
item_set = set(item)
if not (item_set - ask_set):
self.answers.append(item)
if len(self.answers) - self.answer_2chars_count >= max_count :
self.count = len(self.answers)
return
self.count = len(self.answers)
def auto_answer(self, flow):
if len(self.answers):
item = self.answers[0]
answer_index = []
for c in item:
if self.char_indexs_dict[c]:
index = self.char_indexs_dict[c][0]
answer_index.append( str(index) )
del self.char_indexs_dict[c][0]
else:
'''
這個答案是錯誤的
'''
self.error_answers.append(item)
self.answers.remove(item)
return
ask_string = self.ask_string
if len(set(answer_index)) < 4:
ctx.log.error('算法有錯誤:{} 小于4'.format(answer_index))
send_message = {
'answer': item,
'answer_index': answer_index,
'type': 'answer'
}
mm = json.dumps(send_message)
# -----------------------
print(mm)
# -----------------------
self.answer_indexs_dict[item] = answer_index
# 向服務(wù)器發(fā)送消息
if not flow.ended and not flow.error:
self.auto_send_count += 1
self.answers.remove(item)
flow.inject_message(flow.server_conn, mm)
time.sleep(0.5)
def __add_new_worlds_to_memory(self, m):
'''
把答案增加到內(nèi)存字典中
'''
for answer in m['all_answer']:
self.chengyu.add(answer['phrase'])
ctx.log.info('\033[1;31m 共收錄{}個成語 \033[0m'.format(len(self.chengyu)))
def print_answers(self):
'''
圖形化兴枯、色彩化顯示答案
'''
self.print_color('共找到 {}/{} 個成語'.format(self.count, len(self.ask_string)//4))
self.print_color('錯誤成語 {}'.format(self.error_answers))
self.print_color('共自動 {} 次提交'.format(self.auto_send_count))
for item in self.answers:
self.print_color(item)
self.print_matrix(item)
if (not self.answers) and self.index_char_dict:
self.print_matrix()
def print_matrix(self, item = []):
chars_in_line = 6
length = len(self.ask_string)
lines = (length + chars_in_line - 1) // chars_in_line
PADDING = ' '*(lines * chars_in_line - length)
is_need_padding = len(PADDING) != 0
self.print_color('--'*chars_in_line)
global colors, White
for i, c in self.index_char_dict.items():
end = ''
if (i+1) % chars_in_line == 0 or (i+1) == length:
end = '\n'
color = White
if c in item:
color = colors[item.index(c)]
line, first = divmod(i, chars_in_line)
if is_need_padding and first == 0 and (line + 1 == lines):
c = PADDING + c
self.print_color(c, end=end, color=color)
self.print_color('--'*chars_in_line)
def print_color(self, message, end='\n', color=Red):
print('{}{}\033[0m'.format(color, message), end=end)
def reset_data_to_init(self):
self.ask_string = ''
self.answers.clear()
self.index_char_dict.clear()
self.count = 0
self.auto_send_count = 0
self.answer_2chars_count = 0
self.answer_indexs_dict.clear()
self.char_indexs_dict.clear()
self.error_answers.clear()
addons = [
Chengyu()
]
# if __name__ == "__main__":
# c = Chengyu()
# ask_string = '臘見家義降德若功贖仁判悲生升道肘兩身樂極盡立罪春命明回人捉襟性暗'
# c.ask_string = ask_string
# c.find_answers_v2(ask_string)
# c.print_answers()
chengyu.text
一勞永逸
一擲千金
一曝十寒
一石二鳥
一籌莫展
一落千丈
一衣帶水
一語破的
注意:self.dictpath = '/Users/xxx/chengyu.text'
一定要修改成你自己的chengyu.text所在路徑