客户端解析
Posted dominic-ji
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了客户端解析相关的知识，希望对你有一定的参考价值。
客户端
项目目录结构
client目录
tcpClient.py---》客户端连接
conf目录
setting.py：存放配置信息
core目录
admin.py：管理员视图相关功能函数
src.py ：主视图
user.py ：用户视图相关功能函数
download_movie目录
存放下载完的电影
upload_movie目录
存放要上传的电影
lib目录
common.py：存放公共方法
start.py:启动文件
各文件功能代码
client.py
import socket from conf import setting def client_conn(): # 先建立连接 client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client.connect(setting.server_address) return client
setting.py
import os BASE_DIR = os.path.dirname(os.path.dirname(__file__)) BASE_MOVIE_UP = os.path.join(BASE_DIR, ‘upload_movie‘) BASE_MOVIE_DOWN = os.path.join(BASE_DIR, ‘download_movie‘) server_address = (‘127.0.0.1‘, 8087)
admin.py
from client import tcpClient from lib import common from conf import setting import os admin_data = { ‘session‘: None, } send_dic = {‘type‘: None, ‘user_type‘: ‘admin‘, ‘session‘: None} def admin_register(client): print(‘管理员注册‘) while True: name = input(‘请输入手机号>>:‘).strip() if name == ‘q‘: break password = input(‘请输入密码>>‘).strip() conf_password = input(‘请确认密码>>:‘).strip() if password == conf_password: send_dic = {‘type‘: ‘register‘, ‘user_type‘: ‘admin‘, ‘name‘: name, ‘password‘: common.make_md5(password)} back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: print(back_dic[‘msg‘]) break else: print(back_dic[‘msg‘]) else: print(‘两次密码不一致‘) def admin_login(client): print(‘管理员登录‘) while True: name = input(‘用户名>>‘).strip() if name == ‘q‘: break password = input(‘密码>>‘).strip() send_dic = {‘type‘: ‘login‘, ‘user_type‘: ‘admin‘, ‘name‘: name, ‘password‘: common.make_md5(password)} back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: admin_data[‘session‘] = back_dic[‘session‘] print(back_dic[‘msg‘]) break else: print(back_dic[‘msg‘]) def upload_movie(client): if not admin_data[‘session‘]: print(‘请先登录‘) return print(‘上传视频‘) while True: up_list = common.get_allfile_by_path(setting.BASE_MOVIE_UP) if not up_list: print(‘暂无能上传的影片‘) break for i, m in enumerate(up_list): print(‘%s : %s‘ % (i, m)) choose = input(‘请选择要上传的影片‘).strip() if choose == ‘q‘: break if choose.isdigit(): choose = int(choose) # 先把md5值传上去校验一下文件是否存在，在决定要不要上传 if choose >=len(up_list): print(‘请输入范围内的数字‘) continue movie_path = os.path.join(setting.BASE_MOVIE_UP, up_list[choose]) file_md5 = common.get_bigfile_md5(movie_path) send_dic={‘type‘:‘check_movie‘,‘session‘:admin_data[‘session‘],‘file_md5‘:file_md5} back_dic=common.send_data(client,send_dic,None) if back_dic[‘flag‘]: is_free = input(‘是否免费（y/n）>>:‘).strip() if is_free == ‘y‘: is_free = 1 else: is_free = 0 fsize = os.path.getsize(movie_path) send_dic = {‘type‘: ‘upload‘, ‘user_type‘: ‘admin‘, ‘session‘: admin_data[‘session‘], ‘file_name‘: up_list[choose], ‘file_size‘: fsize, ‘file_md5‘:common.get_bigfile_md5(movie_path),‘is_free‘: is_free} back_dic = common.send_data(client, send_dic, movie_path) if back_dic[‘flag‘]: print(back_dic[‘msg‘]) break else: print(back_dic[‘msg‘]) else: print(back_dic[‘msg‘]) else: print(‘请输入数字‘) def delete_movie(client): ‘‘‘ 先拿到视频列表，打印 根据视频前数字，选择要删除的视频 删除成功/失败，打印 ‘‘‘ if not admin_data[‘session‘]: print(‘请先登录‘) return print(‘删除视频‘) while True: send_dic[‘type‘] = ‘get_movie_list‘ send_dic[‘session‘] = admin_data[‘session‘] send_dic[‘movie_type‘] = ‘all‘ back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: for i, mo in enumerate(back_dic[‘movie_list‘]): print(‘%s : %s--%s‘ % (i, mo[0], mo[1])) choose = input(‘请输入要删除的电影（数字）：‘).strip() if choose == ‘q‘: break if choose.isdigit(): choose = int(choose) if choose >= len(back_dic[‘movie_list‘]): print(‘请输入范围内的数字‘) continue send_dic[‘type‘] = ‘delete_movie‘ # 回忆后台返回的数据是什么样的 send_dic[‘movie_id‘] = back_dic[‘movie_list‘][choose][2] back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: print(back_dic[‘msg‘]) break else: print(back_dic[‘msg‘]) else: print(‘请输入数字‘) else: print(back_dic[‘msg‘]) break def release_notice(client): if not admin_data[‘session‘]: print(‘请先登录‘) return print(‘发布公告‘) while True: notice_name = input(‘请输入公告标题：‘).strip() notice_content = input(‘请输入公告内容:‘).strip() if notice_name == ‘q‘: break send_dic[‘type‘] = ‘release_notice‘ send_dic[‘session‘] = admin_data[‘session‘] send_dic[‘notice_name‘] = notice_name send_dic[‘notice_content‘] = notice_content back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: print(back_dic[‘msg‘]) break else: print(back_dic[‘msg‘]) fun_dic = { ‘1‘: admin_register, ‘2‘: admin_login, ‘3‘: upload_movie, ‘4‘: delete_movie, ‘5‘: release_notice, } def admin_view(): client = tcpClient.client_conn() while True: print(‘‘‘ 注册 登录 上传视频 删除视频 发布公告 ‘‘‘) choose = input(‘please choose>>:‘).strip() if ‘q‘ == choose: break if choose not in fun_dic: continue fun_dic[choose](client) client.close()
src.py
from core import admin, user fun_dic = { ‘1‘: admin.admin_view, ‘2‘: user.user_view } def run(): while True: print(‘‘‘ 管理员视图 用户视图 ‘‘‘) choose = input(‘please choose>>:‘).strip() if ‘q‘ == choose: break if choose not in fun_dic: continue fun_dic[choose]()
user.py
from client import tcpClient from lib import common from conf import setting import os import time user_data = { ‘session‘: None, ‘is_vip‘: None } send_dic = {‘type‘: None, ‘user_type‘: ‘user‘, ‘session‘: None} def user_register(client): print(‘用户注册‘) while True: name = input(‘请输入手机号>>:‘).strip() password = input(‘请输入密码>>‘).strip() conf_password = input(‘请确认密码>>:‘).strip() if password == conf_password: send_dic = {‘type‘: ‘register‘, ‘user_type‘: ‘user‘, ‘name‘: name, ‘password‘: common.make_md5(password)} back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: print(back_dic[‘msg‘]) break else: print(back_dic[‘msg‘]) else: print(‘两次密码不一致‘) def user_login(client): print(‘用户登录‘) while True: name = input(‘用户名>>‘).strip() if name == ‘q‘: break password = input(‘密码>>‘).strip() send_dic = {‘type‘: ‘login‘, ‘user_type‘: ‘user‘, ‘name‘: name, ‘password‘: common.make_md5(password)} back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: user_data[‘session‘] = back_dic[‘session‘] user_data[‘is_vip‘] = back_dic[‘is_vip‘] print(back_dic[‘msg‘]) print(back_dic[‘last_notice‘]) break else: print(back_dic[‘msg‘]) def buy_member(client): print(‘购买会员‘) if not user_data[‘session‘]: print(‘请先登录‘) return if user_data[‘is_vip‘] == 1: print(‘您已经是会员了‘) return while True: buy = input(‘是否购买会员（y/n）q 退出‘).strip() if ‘y‘ == buy: send_dic[‘type‘] = ‘buy_member‘ send_dic[‘session‘] = user_data[‘session‘] back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: user_data[‘is_vip‘] = 1 print(back_dic[‘msg‘]) break else: print(back_dic[‘msg‘]) elif ‘q‘ == buy: break else: print(‘您没有购买‘) def get_movie_list(client): if not user_data[‘session‘]: print(‘请先登录‘) return print(‘查看视频列表‘) send_dic[‘type‘] = ‘get_movie_list‘ send_dic[‘movie_type‘] = ‘all‘ send_dic[‘session‘] = user_data[‘session‘] back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: for i, mo in enumerate(back_dic[‘movie_list‘]): print(‘%s : %s-->%s‘ % (i, mo[0], mo[1])) else: print(back_dic[‘msg‘]) def down_free_movie(client): if not user_data[‘session‘]: print(‘请先登录‘) return print(‘下载免费视频‘) ‘‘‘ 先查询免费视频，打印出来， 用户选择后，通过电影id去后台下载 ‘‘‘ send_dic[‘type‘] = ‘get_movie_list‘ send_dic[‘movie_type‘] = ‘free‘ send_dic[‘session‘] = user_data[‘session‘] back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: for i, mo in enumerate(back_dic[‘movie_list‘]): print(‘%s : %s-->%s‘ % (i, mo[0], mo[1])) choose = input(‘请输入要下载的电影（数字）：‘).strip() if choose.isdigit(): choose = int(choose) # 回忆后台返回的数据是什么样的 send_dic[‘type‘] = ‘download_movie‘ send_dic[‘session‘] = user_data[‘session‘] send_dic[‘movie_id‘] = back_dic[‘movie_list‘][choose][2] send_dic[‘movie_type‘] = ‘free‘ back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: if back_dic[‘wait_time‘] > 0: print(‘请等待 %s 秒‘ % back_dic[‘wait_time‘]) time.sleep(back_dic[‘wait_time‘]) recv_size = 0 print(‘----->‘, back_dic[‘filename‘]) path = os.path.join(setting.BASE_MOVIE_DOWN, back_dic[‘filename‘]) with open(path, ‘wb‘) as f: while recv_size < back_dic[‘filesize‘]: recv_data = client.recv(1024) f.write(recv_data) recv_size += len(recv_data) print(‘recvsize:%s filesize:%s‘ % (recv_size, back_dic[‘filesize‘])) print(‘%s :下载成功‘ % back_dic[‘filename‘]) else: print(back_dic[‘msg‘]) else: print(back_dic[‘msg‘]) def down_charge_movie(client): if not user_data[‘session‘]: print(‘请先登录‘) return print(‘下载收费视频‘) if user_data[‘is_vip‘]: charge = input(‘您是会员，收费5元（y 确认）‘).strip() else: charge = input(‘您不是会员，收费10元（y 确认）‘).strip() if not charge == ‘y‘: # 不是y，相当于没付钱，直接返回 return send_dic[‘type‘] = ‘get_movie_list‘ send_dic[‘movie_type‘] = ‘charge‘ send_dic[‘session‘] = user_data[‘session‘] back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: for i, mo in enumerate(back_dic[‘movie_list‘]): print(‘%s : %s-->%s‘ % (i, mo[0], mo[1])) choose = input(‘请输入要下载的电影（数字）：‘).strip() if choose.isdigit(): choose = int(choose) # 回忆后台返回的数据是什么样的 send_dic[‘type‘] = ‘download_movie‘ send_dic[‘session‘] = user_data[‘session‘] send_dic[‘movie_id‘] = back_dic[‘movie_list‘][choose][2] send_dic[‘movie_type‘] = ‘charge‘ back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: recv_size = 0 print(‘----->‘, back_dic[‘filename‘]) path = os.path.join(setting.BASE_MOVIE_DOWN, back_dic[‘filename‘]) with open(path, ‘wb‘) as f: while recv_size < back_dic[‘filesize‘]: recv_data = client.recv(1024) f.write(recv_data) recv_size += len(recv_data) print(‘recvsize:%s filesize:%s‘ % (recv_size, back_dic[‘filesize‘])) print(‘%s :下载成功‘ % back_dic[‘filename‘]) else: print(back_dic[‘msg‘]) else: print(back_dic[‘msg‘]) def check_download_record(client): if not user_data[‘session‘]: print(‘请先登录‘) return print(‘查看观影记录‘) send_dic[‘type‘] = ‘check_download_record‘ send_dic[‘session‘] = user_data[‘session‘] back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: for re in back_dic[‘download_list‘]: print(re) else: print(back_dic[‘msg‘]) def check_notice(client): if not user_data[‘session‘]: print(‘请先登录‘) return print(‘查看公告‘) send_dic[‘type‘] = ‘check_notice‘ send_dic[‘session‘] = user_data[‘session‘] back_dic = common.send_data(client, send_dic, None) if back_dic[‘flag‘]: for value in back_dic[‘notice_list‘]: print(value) else: print(back_dic[‘msg‘]) fun_dic = { ‘1‘: user_register, ‘2‘: user_login, ‘3‘: buy_member, ‘4‘: get_movie_list, ‘5‘: down_free_movie, ‘6‘: down_charge_movie, ‘7‘: check_download_record, ‘8‘: check_notice } def user_view(): client = tcpClient.client_conn() while True: print(‘‘‘ 注册 登录 冲会员 查看视频 下载免费视频 下载收费视频 查看观影记录 查看公告 ‘‘‘) choose = input(‘please choose>>:‘).strip() if ‘q‘ == choose: break if choose not in fun_dic: continue fun_dic[choose](client) client.close()
common.py
import struct import json import os import hashlib def send_data(client, send_dic, file): # 发送部分 head_json_bytes = json.dumps(send_dic).encode(‘utf-8‘) # 先把报头转为bytes格式 client.send(struct.pack(‘i‘, len(head_json_bytes))) # 先发报头的长度 client.send(head_json_bytes) # 再发送报头 if file: # 如果存在文件，再把文件打开一行一行发送 with open(file, ‘rb‘) as f: for line in f: client.send(line) # 接收部分 back_len_bytes = client.recv(4) # 先收报头4个bytes,得到报头长度的字节格式 back_head_len = struct.unpack(‘i‘, back_len_bytes)[0] # 提取报头的长度 head_bytes = client.recv(back_head_len) # 按照报头长度back_head_len,收取报头的bytes格式 header = json.loads(head_bytes.decode(‘utf-8‘)) # 把bytes格式的报头，转换为json格式 return header def get_allfile_by_path(path): file_list = os.listdir(path) return file_list def make_md5(password): md = hashlib.md5() md.update(password.encode(‘utf-8‘)) return md.hexdigest() def get_bigfile_md5(file_path): if os.path.exists(file_path): md = hashlib.md5() filesize = os.path.getsize(file_path) file_list = [0, filesize // 3, (filesize // 3) * 2, filesize - 10] with open(file_path, ‘rb‘) as f: for line in file_list: f.seek(line) md.update(f.read(10)) return md.hexdigest()
start.py
import os, sys path = os.path.dirname(__file__) sys.path.append(path) from core import src if __name__ == ‘__main__‘: src.run()
以上是关于客户端解析的主要内容，如果未能解决你的问题，请参考以下文章
无法解析片段中的 ViewModelProvider 构造？
Android 逆向使用 Python 解析 ELF 文件 ( Capstone 反汇编 ELF 文件中的机器码数据 | 创建反汇编解析器实例对象 | 设置汇编解析器显示细节 )(代码片段