SELECT版FTP:
使用SELECT或SELECTORS模块实现并发简单版FTP
允许多用户并发上传下载文件
必须使用select or selectors模块支持多并发，禁止使用多线程或多进程
REDMAE
用户登陆 1、查看共享目录文件
2、上传文件,
3、下载方件
4、退出 程序结构:
socket_server_client/#程序目录
|- - -clients/#client程序主目录
| |- - -__init__.py
| |- - -bin/#启用目录
| | |- - - __init__.py
| | |- - -socket_client.py#客户端启动
| |
| |- - -cfg/#配置文件目录
| | |- - - __init__.py
| | |- - -config.py#配置文件
| |
| |- - -core/#主要程序目录
| | |- - - __init__.py
| | |- - -client_func.py#主要函数
| |
| |- - -home/#客户端下载文件目录
|
|- - -servers/#server程序主目录
| |- - -__init__.py
| |- - -bin/#启用目录
| | |- - - __init__.py
| | |- - -registration.py#用户注册
| | |- - -server.py#服务端启动(selectors版)
| | |- - -socket_server.py#服务端启动(select版) | |
| |- - -cfg/#配置文件目录
| | |- - - __init__.py
| | |- - -config.py#配置文件
| |
| |- - -core/#主要程序目录
| | |- - - __init__.py
| | |- - -server_classc.py#主要函数
| |
| |- - -db/#用户上传文件主目录
| |- - -user_file/#用户上传目录(共享)
| |- - -user_names#注册用户文件
|
程序结构:
socket_server_client/#程序目录
|- - -clients/#client程序主目录
| |- - -__init__.py
| |- - -bin/#启用目录
| | |- - - __init__.py
| | |- - -socket_client.py#客户端启动
(2016/06/26修改,去收发大文件异常)
#！usr/bin/env python
#-*-coding:utf-8-*-
# Author calmyan import socket,os,json,sys
BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#获取相对路径转为绝对路径赋于变量
sys.path.append(BASE_DIR)#增加环境变量
from core.client_func import user_pwd
#from core.client_func import show_process
from cfg import config #进度条
def show_process(lens):
received_size=0#定义大小
current_percent=0#当前大小百分比
while received_size<lens:
if int((received_size/lens)*100)>current_percent:
print('#',end='',flush=True)
current_percent=int((received_size/lens)*100)
new_size=yield
received_size+=new_size server_addr=('localhost',9500)#设置绑定的 IP 端口
#server_addr=('192.168.11.50',9500)#设置绑定的 IP 端口
client=socket.socket()
client.connect(server_addr)
while True:
data_d=user_pwd(client)
if data_d['tag']:#运行#用户名登陆成功
while True:
print('''=====指令提示====
查看目录文件: ls
下载文件: get 文件名 或 文件编号 如: get test.txt 或 get 1
上传方件: put 路径/文件名 如 put e:/test.txt
退出：exit
''')
cho=input('指令 >>:').strip()
if len(cho)==0:continue
if cho=='exit':exit()#退出指令
cmd_list=cho.split()
if cmd_list[0]=='put':#如果等于下载指令
if len(cmd_list)==1:
print('没有输入相关文件名')
continue
filename=cmd_list[1]
file_dir=config.USER_DIR+'/'+filename
if os.path.isfile(file_dir):#如果文件存在
file_obj=open(file_dir,"rb")#打开文件
name=file_obj.name.split('/')[-1]#文件名
#name=filename.split("\\")[-1]#文件名
sez=os.path.getsize(file_dir)#获取文件大小
if sez<1:
print('\033[41;1m文件为空!,不能上传\033[0m')
continue
progress = show_process(sez) #进度条 传入文件大小
progress.__next__()
rat=0
file_obj.seek(rat)#移动到位置
data_header={
"action":"put",
"filename":name,
"size":sez
} client.send(json.dumps(data_header).encode())#用json 序列化后,发送相关 信息
attr=client.recv(100)
print("文件[%s]发送中...."%data_header["filename"])
while rat<sez:
line=file_obj.read(4096)
client.send(line)
try:
progress.send(len(line))#传入当前数据大小
except StopIteration as e:
print("100%")
break
print("文件[%s]发送完毕!"%data_header["filename"])
else:
print('\033[41;1m该文件不存在或为目录\033[0m')
continue
elif cmd_list[0]=='get':#如果等于get 上传指令
if len(cmd_list)==1:
print('没有输入相关文件名')
continue
filename=cmd_list[1]
print(filename)
data_header={
"action":"get",
"filename":filename,
"size":''
}
client.send(json.dumps(data_header).encode())#用json 序列化后,发送相关 信息 datas=client.recv(4096)#接收数据 指令
data_l= json.loads(datas.decode())#反序列
# print(data_l)
# print(data_l['size'])
if data_l['filename']==False:
print('\033[41;1m文件不存在或者出错\033[0m')
continue
client.send('为了防粘包'.encode('utf-8'))#防粘包
prten=show_process(data_l["size"])
prten.__next__()
file_dir=config.USER_DIR+'/'+data_l["filename"]
file_obj=open(file_dir,'wb')#打开新建 这个文件
rece_size=0#定义 文件大小值
while rece_size<data_l["size"]:#小于接收的文件大小时,
recv_data=client.recv(4096)
file_obj.write(recv_data)#写入文件
rece_size+=len(recv_data)#增加文件大小计算
try:
prten.send(len(recv_data))
except StopIteration as e:
print('100%') else:
print("文件[%s]接收完毕!"%data_l["filename"])
file_obj.flush()
file_obj.close()#关闭文件
elif cmd_list[0]=='ls':#查看目录文件
data_header={
"action":"ls",
"filename":'',
"size":''
}
client.send(json.dumps(data_header).encode())#用json 序列化后,发送相关 信息
client.recv(100)#防粘包
#client.send('为了防粘包'.encode('utf-8'))#防粘包
datas=client.recv(4096)#接收数据 指令
data_l= json.loads(datas.decode())#反序列
for k,v in enumerate(data_l):
print('编号: %s 文件名:%s'%(k,v)) else:
print(data_d['mag'])
| |- - -cfg/#配置文件目录
| | |- - - __init__.py
| | |- - -config.py#配置文件
#！usr/bin/env python
#-*-coding:utf-8-*-
# Author calmyan import os ,sys
BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#获取相对路径转为绝对路径赋于变量
sys.path.append(BASE_DIR)#增加环境变量 USER_DIR=BASE_DIR+'/home'#定义用户目录文件路径变量
IP='192.168.11.50'
PORST=9500
| |- - -core/#主要程序目录
| | |- - - __init__.py
| | |- - -client_func.py#主要函数
#！usr/bin/env python
#-*-coding:utf-8-*-
# Author calmyan
import socket,os,json,sys
#用户名登陆函数
def user_pwd(client):
user_=input('请输入用户名:').strip()
pwd_=input('请输入密码:').strip()
data_header={
"action":"user",
"name":user_,
"pwd":pwd_
}
client.send(json.dumps(data_header).encode())#用json 序列化后,发送相关 信息
data=client.recv(4096)#接收数据 指令
data_s=json.loads(data.decode('utf-8'))#反序列
return data_s
|- - -servers/#server程序主目录
| |- - -__init__.py
| |- - -bin/#启用目录
| | |- - - __init__.py
| | |- - -registration.py#用户注册
#！usr/bin/env python
#-*-coding:utf-8-*-
# Author calmyan
import socket,os,json,sys,pickle BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#获取相对路径转为绝对路径赋于变量
sys.path.append(BASE_DIR)#增加环境变量
from cfg import config
print('用户注册'.center(60,'='))
while True:
user_=input('请输入您要注册的用户名:').strip()
user_dir=os.path.join(config.USER_DIR,user_)#拼接用户目录路径
if os.path.isdir(user_dir):# 判断一个目录是否存在
print('用户已经存在请重输!')
continue
else:
pwd_=input('请输入密码:').strip()
pwd_two=input('请确认密码:').strip()
if pwd_==pwd_two: if not os.path.isfile(config.USER_FILE):
with open(config.USER_FILE,'w',encoding='utf-8') as f:
f.write('{}')
with open(config.USER_FILE,'r+',encoding='utf-8') as f:
data=eval(f.readline())
data[user_]=pwd_
f.seek(0)
f.write(str(data))
print('用户[%s]注册成功!'%user_)
exit()
| | |- - -server.py#服务端启动(selectors版)(2016/06/26修改,去收发大文件异常)
#!usr/bin/env python
#-*-coding:utf-8-*-
# Author calmyan
#python
#2017/6/24 19:34
#__author__='Administrator'
import select,socket,sys ,queue,json,os
BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#获取相对路径转为绝对路径赋于变量
sys.path.append(BASE_DIR)#增加环境变量 import core
from core.server_class import socket_server s=socket.socket()#实例化一个连接对象
s.setblocking(0)#设置成非阻塞
server_addr=('localhost',9500)#设置绑定的 IP 端口
s.bind(server_addr)#连接对象绑定IP 端口
s.listen(100)#队列 可连接数量
inputs=[s,]#首先要监测本身 outputs=[]#发送列表 meg_queues={} #发送 连接对象的队列集合 字典 while True:
print('监听中......')
readable,writeable,exeptional=select.select(inputs,outputs,inputs)#生成select 对象,返回三个列表 连接,发关,错误 for i in readable: #i为一个socket
if i is s:#如果i 是s 表示有新 连接 进来
conn,client_addr=i.accept()#建立一个新连接
print('接入一个新连接...',client_addr)
conn.setblocking(1)#设成 阻塞
inputs.append(conn)#加入select,的连接列表,避免出现阻塞
meg_queues[conn]=queue.Queue()#创建一个队列 添加到字典
else:
try:
data=i.recv(1024)#如果不是新连接就收数据
except Exception as e:
print(e)
if data: #如果数据不为空
print('[%s] 发来的数据 [%s]'%(i.getpeername,data))
meg_queues[i].put(data)#当前连接的消息队列加入数据
if i not in outputs:#如果当前连接没有在发送列表内,就加入发送列表
outputs.append(i)
else:
print('客户端已经断开了....')#开始清理工作
if i in outputs:#在发送列表
outputs.remove(i)#在发送列表内删除
inputs.remove(i)#在连接列表内删除
del meg_queues[i]#在队列字典内删除 for w in writeable:#循环发送列表
try:
msg=meg_queues[w].get_nowait()#取出队列中的数据,判断
except queue.Empty:#如果数据为空
outputs.remove(w)##从发送列表内删除
else:
data = json.loads(msg.decode())#反序列
serv=socket_server(data,w)
if data['action']=='user':#如果是用户名,进行认证\
#serv=socket_server(data,conn)
ret=serv.ret_l()
if ret['tag']:
pass
else:
break
#print('echoing', repr(data), 'to', conn)
#data=json.loads(data)
if data['action']=="put":#如果接收的字典中是put,就是进行接收
#serv=socket_server(data,conn)
serv.put_file(serv.open_f())#调对象方法
elif data['action']=='get':#下载
#serv=socket_server(data,conn)#实例化
serv.send_file(serv.open_f())#调 用方法
elif data['action']=='ls':#查看
#serv=socket_server(data,conn)
serv.ls_file(serv.open_f())
break #w.send(msg)#发送 for e in exeptional:#循环错误列表
print('连接[%s]出错!'%e.getpeername)
inputs.remove(e)##从发送列表内删除
if e in outputs:#在发送列表
outputs.remove(e)#在发送列表内删除
e.close()
del meg_queues[e]#在队列字典内删除
| | |- - -socket_server.py#服务端启动(select版)(2016/06/26修改,去收发大文件异常)
#！usr/bin/env python
#-*-coding:utf-8-*-
# Author calmyan
import socket,os,json
import sys
import selectors BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#获取相对路径转为绝对路径赋于变量
sys.path.append(BASE_DIR)#增加环境变量 from core.server_class import socket_server
from core.server_class import open_file_list def accept(sock, mask):
conn, addr = sock.accept() # 建立新连接
print('accepted', conn, 'from', addr)
conn.setblocking(1)#设成非阻塞
sel.register(conn, selectors.EVENT_READ, read)#注册 连接,回调函数 read def read(conn,mask):
#gevent.spawn(handle_request, cli)#创建一个新协程来
data = conn.recv(1024) # 接收数据
if data:#不为空
print('接收的数据:',data)
#print(mask)
if len(data)==0:
return
data = json.loads(data.decode())#反序列
serv=socket_server(data,conn)
if data['action']=='user':#如果是用户名,进行认证\
#serv=socket_server(data,conn)
ret=serv.ret_l()
if ret['tag']:
pass
else:
return
elif data['action']=="put":#如果接收的字典中是put,就是进行接收
#serv=socket_server(data,conn)
serv.put_file(serv.open_f())#调对象方法
return
elif data['action']=='get':#下载
#serv=socket_server(data,conn)#实例化
serv.send_file(serv.open_f())#调 用方法
return
elif data['action']=='ls':#查看
#serv=socket_server(data,conn)
serv.ls_file(serv.open_f())
return
else:
return
else:#如果为空
print('closing', conn)
sel.unregister(conn)#取消注册
conn.close()#关闭连接 server_addr=('0.0.0.0',9500)#设置绑定的 IP 端口
s=socket.socket()#定义
s.bind(server_addr)#绑定IP 端口
s.listen(5)#对列5
s.setblocking(False)#非阻塞
print('正在监听中') sel = selectors.DefaultSelector()#生成一个创建一个selectors对象
sel.register(s, selectors.EVENT_READ, accept)#注册连接 返调函数为accepts while True:
events = sel.select()#默认为阻塞模式
for key, mask in events:#如果有连接,接入
callback = key.data#新建连接句柄
callback(key.fileobj, mask)
| |- - -cfg/#配置文件目录
| | |- - - __init__.py
| | |- - -config.py#配置文件
#！usr/bin/env python
#-*-coding:utf-8-*-
# Author calmyan
import os ,sys
BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#获取相对路径转为绝对路径赋于变量
sys.path.append(BASE_DIR)#增加环境变量 USER_DIR=BASE_DIR+'/db/user_file/'#定义用户目录文件路径变量 USER_FILE=BASE_DIR+'/db/user_names'#定义用户名密码文件路径变量
IP='localhost'
PORST=9501
| |- - -core/#主要程序目录
| | |- - - __init__.py
| | |- - -server_classc.py#主要函数(2016/06/26修改,去收发大文件异常)
#！usr/bin/env python
#-*-coding:utf-8-*-
# Author calmyan
import socket,os,json,sys,pickle
import selectors BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#获取相对路径转为绝对路径赋于变量
sys.path.append(BASE_DIR)#增加环境变量 from cfg import config #用户名检测函数 def open_file_list(name,pas):#传入当前类
with open(config.USER_FILE,'r',encoding='utf-8') as f:
data=eval(f.readline())
print(data)
if name in data and pas==data[name]:
return True
else:
return False #连接类
class socket_server(object):
'''连接类'''
file_path=config.USER_DIR#用户路经变量
def __init__(self,data,conn):#传入
self.DATA=data
self.conn=conn def ret_l(self):
self.ret=self.login(self.DATA["name"],self.DATA['pwd'],self.conn)#用户名检测
return self.ret
def open_f(self):#打开目录 file_dir=os.path.join(config.USER_DIR)#用户目录
print(file_dir)
file_name=os.listdir(file_dir)#目录文件列表
f=file_dir+'/'+self.DATA['filename']##上传的文件名
return file_dir,file_name,f#返回 def ls_file(self,data):#查看文件
self.conn.send('为了防粘包'.encode('utf-8'))#防粘包
#self.conn.recv(100)
self.conn.send(json.dumps(data[1]).encode())
return def send_file(self,data): if self.DATA['filename'] in data[1]:#如果是输入文件名
f=data[0]+'/'+self.DATA['filename']
file_obj=open(f,"rb")#打开文件
name=file_obj.name.split('/')[-1]#文件名
sez=os.path.getsize(f)#获取文件大小 if sez<1:
print('文件错误!')
self.DATA['filename']=False
self.conn.send(json.dumps(self.DATA).encode())
return
print(''.center(30,'='))
print(sez)
print(''.center(30,'='))
data_header={
"action":"put",
"filename":name,
"size":sez
}
#self.conn.send(b'1')
self.conn.send(json.dumps(data_header).encode())#用json 序列化后,发送相关 信息
self.conn.recv(100)
file_siez=0
while file_siez<sez:
#for line in file_obj:
line=file_obj.read(1024)
file_siez+=len(line)
self.conn.send(line)#发送数据 elif self.DATA['filename'].isdigit():#如果是输入编号
num=int(self.DATA['filename'])#转为数字
try:
f=data[0]+'/'+data[1][num]#
file_obj=open(f,"rb")#打开文件
name=file_obj.name.split('/')[-1]#文件名
sez=os.path.getsize(f)#获取文件大小
if sez<1:
print('文件错误!')
self.DATA={'filename':False}
self.conn.send(json.dumps(self.DATA).encode())
return
print(sez)
data_header={
"action":"put",
"filename":name,
"size":sez
}
self.conn.send(json.dumps(data_header).encode())#用json 序列化后,发送相关 信息
self.conn.recv(100)
for line in file_obj:
self.conn.send(line)#发送数据
#self.conn.send(json.dumps(f).encode())#发送文件
except Exception as e:
self.DATA={'filename':False}
self.conn.send(json.dumps(self.DATA).encode())
return
else:
data={'filename':False}
self.conn.send(json.dumps(data).encode())
return
def put_file(self,data):#上传文件
file_obj=open(data[2],'wb')#打开新建 这个文件
rece_size=0#定义 文件大小值
self.conn.send('为了防粘包'.encode('utf-8'))#防粘包
while rece_size<self.DATA["size"]:#小于接收的文件大小时,
recv_data=self.conn.recv(4096)
file_obj.write(recv_data)#写入文件
rece_size+=len(recv_data)#增加文件大小计算
else:
print("文件[%s]接收完毕!"%self.DATA["filename"])
file_obj.flush()
file_obj.close()#关闭文件
return
#@staticmethod
def login(self,name,pas,conn):#用户检测 函数
try:
if open_file_list(name,pas):
tag=True
error=''
datas={'user':name}
data={'mag':'用户认证通过','tag':True}
print(json.dumps(data).encode())
conn.send(json.dumps(data).encode())
else:
raise Exception('\033[41;1m用户名或密码错误\033[0m' %name)
except Exception as e:
tag=False
error=str(e)
datas=''
data={'mag':'用户或密码错误','tag':False}
print('发送数据%s'%data)
conn.send(json.dumps(data).encode())
return {'tag':tag,'error':error,'data':datas}