本章内容
1.命令模块
2.面向对象 封装 继承 多态
3.反射
4.异常处理
5.socket
1.命令模块
import os
os.system("ls")
a=os.system("ls")
保存命令执行结果
popen
a=os.popen("df").read()
print(a)
subprocess执行复杂命令 | 保存执行结果 | 返回执行结果
(run方法python3.5才有的）
import subprocess
subprocess.run("df -h |grep sda",shell=True)
a=subprocess.run("df -h |grep sda",shell=True)
查看命令执行结果
poll() 查看命令执行中或执行完毕(执行中返回None 执行完返回)
stdout=subprocess.PIPE 查看命令标准输出
stderr=subprocess.PIPE 查看命令标准错误输出
>>> res=subprocess.Popen("sleep 10;echo 'hello':",shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
>>> print(res.poll())
None
>>> print(res.poll())
0
>>>
>>> res=subprocess.Popen("sleep 10;ech 'hello':",shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
>>> print(res.poll())
127
getstatusoutput执行命令返回结果和状态
>>> subprocess.getstatusoutput("df -h |grep sda")
(0, '/dev/sda2 44G 2.3G 40G 6% /')
>>> subprocess.getstatusoutput("df -h |grep zz")
(1, '')
wait()等待命令只想完毕
>>> res=subprocess.Popen("sleep 10;echo 'hello':",shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
>>> res.wait()
0
>>>
terminate()停止令执行
>>> res=subprocess.Popen("sleep 10;echo 'hello':",shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
>>> res.terminate()
>>> res.stdout.read()
b''
>>>
传递密码
subprocess.run("echo "123" | sudo -S yum install vim",shell=True)
2.面向对象
def person(name, age, sex, job):
def walk(p):
print("person %s is walking..." % p['name'])
data = {
'name': name,
'age': age,
'sex': sex,
'job': job,
'walk': walk
}
return data
def dog(name, dog_type):
def bark(d):
print("dog %s:wang.wang..wang..." % d['name'])
data = {
'name': name,
'type': dog_type,
'bark': bark
}
return data
d1 = dog("李磊","京巴")
p1 = person("严帅",36,"F","运维")
p1['walk'](p1)
class dog(object):
def __init__(self,name): #构造方法
self.name=name #属性 成员变量
def sayhi(self): #方法
print("hello word",self.name)
d1=dog("abc") #实例
d1.sayhi()
class dog(object):
def __init__(self,name): #构造方法
self.name=name
def sayhi(self): #类方法
print("hello word",self.name)
def eat(self,food):
print("%s asd %s " %(self.name,food))
d1=dog("abc") #实例
d1.eat('as')
class Role(object):
def __init__(self, name, role, weapon, life_value=100, money=15000):
self.name = name
self.role = role
self.weapon = weapon
self.life_value = life_value
self.money = money
def shot(self):
print("shooting...")
def got_shot(self):
print("ah...,I got shot...",self.weapon)
def buy_gun(self,gun_name):
print("just bought %s" % gun_name)
self.weapon=gun_name
r2 = Role('Jack', 'terrorist','B22')
r2.buy_gun('B51')
print(r2.weapon)
2.1封装(私有属性)
class Role(object):
def __init__(self, name, role, weapon, life_value=100, money=15000):
self.name = name
self.role = role
self.weapon = weapon
self.life_value = life_value
self.money = money
self.__heart= 'normal' #私有属性
def get_heart(self): #访问私有属性只读
return self.__heart
def shot(self):
print("shooting...")
self.__heart='die'
print(self.__heart)
r2 = Role('Jack', 'terrorist','B22')
print(r2.name)
print(self.__heart)
r2.shot()
r2.get_heart() #访问私有属性只读
print(r2._Role__heart) #强制访问私有属性
2.2封装(公有属性)
class Role(object):
nationality = 'JP' #公有属性
def __init__(self, name, role, weapon, life_value=100, money=15000):
self.name = name
self.role = role
self.weapon = weapon
self.life_value = life_value
self.money = money
self.__heart= 'normal'
def shot(self):
print("shooting...")
self.__heart='die'
print(self.__heart)
r1 = Role('Tom', 'terrorist','B51')
r2 = Role('Jack', 'terrorist','B22')
print(r1.nationality)
print(Role.nationality)
#修改
# Role.nationality='US'
# r1.nationality='CN'
# print(r1.nationality)
# print(Role.nationality)
class Role(object):
nationality = 'JP'
def __init__(self, name, role, weapon, life_value=100, money=15000):
self.name = name
self.role = role
self.weapon = weapon
self.life_value = life_value
self.money = money
self.__heart= 'normal'
def shot(self):
print("shooting...")
self.__heart='die'
print(self.__heart)
r1 = Role('Tom', 'terrorist','B51')
r2 = Role('Jack', 'terrorist','B22')
def shot2(self):
print("run my own shot method",self.name) #设置私有方法
r1.shot=shot2
r1.shot(r1)
类的析构方法(做一些程序收尾的工作)
class Role(object):
nationality = 'JP'
def __init__(self, name, role, weapon, life_value=100, money=15000):
self.name = name
self.role = role
self.weapon = weapon
self.life_value = life_value
self.money = money
self.__heart= 'normal'
def shot(self):
print("shooting...")
self.__heart='die'
print(self.__heart)
def __del__(self):
print("del .......run")
r1=Role('hantao','tufei','b22')
2.3继承
class Person(object):
def __init__(self,name,age):
self.name=name
self.age=age
self.sex="noraml"
def talk(self):
print("person is talking...l")
class BlackPerson(Person): #先继承 再构造
def __init__(self,name,age,strength): #构造
Person.__init__(self,name,age) #传递
self.strength = strength #构造
print(self.name,self.age,self.sex)
def talk(self): #构造
Person.talk(self)
print("blck person is blaba **")
b=BlackPerson("wei er smith",30,"strong")
b.talk()
class SchoolMember(object):
'''学校成员基类'''
member = 0
def __init__(self,name,age,sex):
self.name = name
self.age = age
self.sex = sex
self.enroll()
def enroll(self):
'''注册'''
print("just enrolled a new school member [%s]"% self.name)
SchoolMember.member +=1
def tell(self):
print('------info:%s--------'%self.name)
for k,v in self.__dict__.items():
print(k,v)
def __del__(self):
print("开出[%s]..."% self.name)
SchoolMember.member -=1
class Teacher(SchoolMember): #类
'''讲师类'''
def __init__(self,name,age,sex,salary,course): #构造方法
SchoolMember.__init__(self,name,age,sex)
self.salary = salary #属性
self.course = course
def teaching(self): #方法
print("Teacher [%s] is teaching [%s]" % (self.name,self.course))
class Student(SchoolMember):
def __init__(self,name,age,sex,course,tuition):
SchoolMember.__init__(self,name,age,sex)
self.course = course
self.tuition = tuition
self.amount = 0
def pay_tuition(self,amount):
print("student [%s] has just paied [%s]" %(self.name,amount))
self.amount += amount
t1 = Teacher("Wusir",28,"F*M",3000,"Python") #属性 对象
s1 = Student("HaiTao",38,"N/A","PYS15",30000)
s2 = Student("LIChuang",12,"M","PYS15",11000)
print(SchoolMember.member)
t1.tell()
s2.tell()
新式类写法
class SchoolMember(object):
'''学校成员基类'''
member = 0
def __init__(self,name,age,sex):
self.name = name
self.age = age
self.sex = sex
self.enroll()
def enroll(self):
'''注册'''
print("just enrolled a new school member [%s]"% self.name)
SchoolMember.member +=1
def tell(self):
print('------info:%s--------'%self.name)
for k,v in self.__dict__.items():
print(k,v)
def __del__(self):
print("开出[%s]..."% self.name)
SchoolMember.member -=1
class Teacher(SchoolMember):
'''讲师类'''
def __init__(self,name,age,sex,salary,course):
#SchoolMember.__init__(self,name,age,sex)
super(Teacher,self).__init__(name,age,sex) #新式类写法
self.salary = salary
self.course = course
def teaching(self):
print("Teacher [%s] is teaching [%s]" % (self.name,self.course))
t1 = Teacher("Wusir",28,"F*M",3000,"Python")
print(SchoolMember.member)
t1.tell()
多继承
class SchoolMember(object):
'''学校成员基类'''
member = 0
def __init__(self,name,age,sex):
self.name = name
self.age = age
self.sex = sex
self.enroll()
def enroll(self):
'''注册'''
print("just enrolled a new school member [%s]"% self.name)
SchoolMember.member +=1
def tell(self):
print('------info:%s--------'%self.name)
for k,v in self.__dict__.items():
print(k,v)
def __del__(self):
print("开出[%s]..."% self.name)
SchoolMember.member -=1
class Schol(object):
def open_branch(self,addr):
print("test",addr)
class Teacher(SchoolMember,Schol): ######
pass
t1 = Teacher("Wusir",28,"F*M")
t1.open_branch("zhang")
新式类继承vs经典继承
广度查找从左到右---深度查找从下到上
python3 新式类继承和经典集成都是广度查找
python2 经典继承深度查找--新式类继承广度查找
强制重构
class Ainmal(object):
def __init__(self,name):
self.name = name
def talk(self):
raise NotImplementedError("Subclass must implement abstract method")
class Cat(Ainmal):
def talk(self):
print("ok")
class Dog(Ainmal):
def talk(self):
return "Woof! Woof!"
d = Dog('d1')
c = Cat("c1")
c.talk()
d.talk()
2.4多态
#模拟多态
class Ainmal(object):
def __init__(self,name):
self.name = name
def talk(self):
raise NotImplementedError("Subclass must implement abstract method")
class Cat(Ainmal):
def talk(self):
print("ok")
class Dog(Ainmal):
def talk(self):
return "Woof! Woof!"
d = Dog('d1')
c = Cat("c1")
def animal_talk(obj):
print(obj.talk())
animal_talk(d)
animal_talk(c)
2.5面向对象总结
self就是调用当前方法的发放的对象
静态字段使用场景 每个对象保存的东西相同时 可以使用静态字段
class Foo(object):
'''静态字段
共有属性'''
country = '中国'
def __init__(self,name,count):
'''普通字段
普通属性'''
self.name = name
self.count = count
def bar(self):
print("asd",self.name,self.count,self.country)
obj1 = Foo('河南',100000)
obj1.bar()
obj2 = Foo('山东',10000)
obj2.bar()
继承
class F1:
def __init__(self,n):
self.N = n
print("F1")
class F2:
def __init__(self,arg1):
self.a = arg1
print("F2")
class F3:
def __init__(self,arg2):
self.b = arg2
print("F3")
o1 = F1('alex')
o2 = F2(o1)
o3 = F3(o2)
print(o3.b.a.N)
class F1:
def __init__(self):
print("F1")
def a1(self):
print("F1a1")
def a2(self):
print("F1a2")
class F2(F1):
def __init__(self):
print("F2")
def a1(self):
self.a2()
print("F2a1")
def a2(self):
print("F2a2")
class F3(F2):
def __init__(self):
print("F3")
def a2(self):
print("F3a2")
obj = F3()
obj.a1()
静态方法 保存在类中 调用者类 无需创建对象 可以有任意个参数
class F1:
@staticmethod
def a1(a1,a2):
print('alex')
F1.a1(1,2)
什么时候适合用面向对象
-如果多个函数需传入多个共同的参数时
-根据一个模板创建某些东西
-应用场景
class SSH:
def __init__(self,host,port,pwd,username):
self.host =
...
def connection(self):
#去创建链接
self.comm = 和服务器创建的链接对象()
def close(self):
#关闭
self.conn.关闭
def upload(self):
self.conn 使用链接上传文件
def cmd(self):
self.conn() 使用连接执行文件
obj = SSH(....)
obj.connection()
obj.upload()
obj.close()
staticmethod静态方法
静态方法 和普通函数一样 只是属于类下面的一个函数 不可以调用类变量
class Dog(object):
def __init__(self,name):
self.name = name
@staticmethod
def eat(self):
print("%s is eating %s" %(self.name,'aa'))
def talk(self):
print("%s is talking"% self.name)
d = Dog("ChenRonghua")
d.eat(d)
d.talk()
classmethod类方法
类方法 只可以访问类变量 不可以访问实例变量
class Dog(object):
name = "huazai"
def __init__(self,name):
self.name = name
@classmethod
def eat(self):
print("%s is eating " %(self.name))
def talk(self):
print("%s is talking"% self.name)
d = Dog("ChenRonghua")
d.eat()
d.talk()
property属性方法
属性方法 把一个方法变成属性
class Dog(object):
def __init__(self,name):
self.name = name
self.__food = None
@property
def eat(self):
print("%s is eating " %(self.name,self.__food))
@eat.setter
def eat(self,food):
self.__food = food
print("%s is eating %s" % (self.name,food))
@eat.deleter
def eat(self):
del self.__food
print("删除完了")
d = Dog("ChenRonghua")
d.eat='abc'
del d.eat
d.eat
属性方法应用场景
1. 连接航空公司API查询
2. 对查询结果进行解析
3. 返回结果给你的用户
class Flight(object):
def __init__(self,name):
self.flight_name = name
def checking_status(self):
print("check flight %s status "% self.flight_name)
return 1
@property
def flight_status(self):
status = self.checking_status()
if status == 0:
print("flight got cancleed")
elif status == 1:
print("flight is arrived")
elif status == 2:
print("flight has departuerd already")
else:
print("canot confirm the dlight status pleass check later")
@flight_status.setter
def flight_status(self,status):
print("flight %s has changed status to %s "%(self.flight_name,status))
f = Flight("CA980")
f.flight_status
f.flight_status = 2
__doc__打印描述信息
class Foo:
'''描述信息'''
def func(self)
pass
print(Foo.__doc__)
__module__，__class__
__module__ 表示当前操作的对象在那个模块
__class__ 表示当前操作的对象的类是什么
class c:
def __init__(self):
self.name = 'wupeiqi'
from test import c
obj = c()
print(obj.__module__)
print(obj.__class__)
__call__
__call__ `对象后面加括号 执行
class Foo:
def __init__(self):
pass
def __call__(self,*args,**kwargs):
print('__call__')
obj = Foo()
obj()
__dict__
__dict__ 查看类或对象中的所有成员
class Province:
country = 'china'
def __init__(self,name,count):
self.name = name
self.count = count
def func(self,*args,**kwargs):
print('func')
print (Province.__dict__) 获取类的成员，即：静态字段、方法
obj = Province("abc",1000)
print(obj.__dict__) 获取 对象obj 的成员
__str__
如果一个类中定义了__str__方法，那么在打印 对象 时，默认输出该方法的返回值
class Foo:
def __str__(self):
return "alex li"
obj = Foo()
print(obj)
__getitem__、__setitem__、__delitem__
用于索引操作，如字典。以上分别表示获取、设置、删除数据
class Foo(object):
def __getitem__(self,key):
print('__gettiem__',key)
def __setitem__(self,key,value):
print('__setitem__',key,value)
def __delitem__(self,key):
print("__delitem",key)
obj = Foo()
result = obj['k1'] 自动触发执行 __getitem_
obj['k2'] = 'alex' 自动触发执行 __setitem__
del obj['k1']
3.反射
hasattr- getattr
class Dog(object):
def __init__(self,name):
self.name = name
def eat(self,food):
print("%s is eating ..%s"%(self.name,food))
d = Dog("abc")
choice = input(">>:").strip()
if hasattr(d,choice): #判断方法是否存在 返回True或False
func = getattr(d,choice) #存在就执行
func("kk") #传参数
setattr
def bulk(self):
print("%s is telling.."%self.name)
class Dog(object):
def __init__(self,name):
self.name = name
def eat(self,food):
print("%s is eating ..%s"%(self.name,food))
d = Dog("abc")
choice = input(">>:").strip()
if hasattr(d,choice):
func = getattr(d,choice)
func("kk")
setattr(d,choice,"rh") #修改属性的参数
print(d.name)
else:
setattr(d,choice,bulk) #如果输入的方法不存在 全部执行 bulk方法
d.talk(d)
setattr(d,choice,22) #如果输入的属性不存在 就返回22
print(getattr(d,choice))
如果输入的方法不存在 全部执行 bulk方法
if hasattr(d,choice):
getattr(d,choice)
else:
setattr(d,choice,bulk)
func = getattr(d,choice)
func(d)
如果输入的属性不存在 就返回None
if hasattr(d,choice):
getattr(d,choice)
else:
setattr(d,choice,None)
v=getattr(d,choice)
print(v)
delattr
delattr
def bulk(self):
print("%s is telling.."%self.name)
class Dog(object):
def __init__(self,name):
self.name = name
def eat(self,food):
print("%s is eating ..%s"%(self.name,food))
d = Dog("abc")
choice = input(">>:").strip()
if hasattr(d,choice): #输入的方法存在就删出
delattr(d,choice)
4.异常处理
try:
names[3]
except Exception as e:
print("错误",e)
names = []
try:
names[3]
except IndexError as e:
print("没有这个index",e)
try:
open(test.txt) #尝试执行代码
except (KeyError,IndexError) as e: #如果报错打印下面信息
print("没有这个 key",e)
except Exception as e:
print("未知错误",e)
else:
print("一切正常")
finally:
print("不管有没有错，都执行")
自定义错误
class AlexError(Exception):
def __init__(self,msg):
self.message = msg
try:
raise AlexError('自定义错误')
except AlexError as e:
print(e)
1 AttributeError 试图访问一个对象没有的树形，比如foo.x，但是foo没有属性x 2 IOError 输入/输出异常；基本上是无法打开文件 3 ImportError 无法引入模块或包；基本上是路径问题或名称错误 4 IndentationError 语法错误（的子类） ；代码没有正确对齐 5 IndexError 下标索引超出序列边界，比如当x只有三个元素，却试图访问x[5] 6 KeyError 试图访问字典里不存在的键 7 KeyboardInterrupt Ctrl+C被按下 8 NameError 使用一个还未被赋予对象的变量 9 SyntaxError Python代码非法，代码不能编译(个人认为这是语法错误，写错了） 10 TypeError 传入对象类型与要求的不符合 11 UnboundLocalError 试图访问一个还未被设置的局部变量，基本上是由于另有一个同名的全局变量， 12 导致你以为正在访问它 13 ValueError 传入一个调用者不期望的值，即使值的类型是正确的 14 15 常用异常
1 ArithmeticError 2 AssertionError 3 AttributeError 4 BaseException 5 BufferError 6 BytesWarning 7 DeprecationWarning 8 EnvironmentError 9 EOFError 10 Exception 11 FloatingPointError 12 FutureWarning 13 GeneratorExit 14 ImportError 15 ImportWarning 16 IndentationError 17 IndexError 18 IOError 19 KeyboardInterrupt 20 KeyError 21 LookupError 22 MemoryError 23 NameError 24 NotImplementedError 25 OSError 26 OverflowError 27 PendingDeprecationWarning 28 ReferenceError 29 RuntimeError 30 RuntimeWarning 31 StandardError 32 StopIteration 33 SyntaxError 34 SyntaxWarning 35 SystemError 36 SystemExit 37 TabError 38 TypeError 39 UnboundLocalError 40 UnicodeDecodeError 41 UnicodeEncodeError 42 UnicodeError 43 UnicodeTranslateError 44 UnicodeWarning 45 UserWarning 46 ValueError 47 Warning 48 ZeroDivisionError 49 50 更多异常
5.socket
#服务端
import socket
server = socket.socket() #声明socket类型
server.bind(('localhost',6969)) #设置侦听端口
server.listen() #开始侦听
print("开始等待:")
conn,addr = server.accept() #等待数据
#conn是客户端链接过来的一个实例，addr是ip和port
print(conn,addr)
print("等待来了")
data = conn.recv(1024) #设置接受数据大小
print("recv:",data.decode('utf-8'))
conn.send(data.upper()) #再发送给客户端
server.close() #关闭
#客户端
import socket
client = socket.socket() #声明socket类型和socket链接对象(ip.port)
client.connect(('localhost',6969)) #链接端口
client.send("你好".encode('utf-8')) #发送数据,转码utf8
# client.send(b"Hello World!")
data = client.recv(1024) #设置接受数据大小
print("recv",data.decode('utf-8')) #打印数据
client.close() #关闭连接
#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
#服务器端
import socket
server = socket.socket()
server.bind(('localhost',7000))
server.listen()
print("开始等待:")
while True:
conn,addr = server.accept()
print(conn,addr)
print("等待来了")
while True:
data = conn.recv(1024)
print("recv:",data)
if not data:
print("client not data")
break
conn.send(data.upper())
server.close()
------------------
#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
#客户端
import socket
client = socket.socket()
client.connect(('localhost',7000))
while True:
msg = input(">>:").strip()
if len(msg) == 0:continue
client.send(msg.encode("utf-8"))
data = client.recv(1024)
print("recv",data.decode())
client.close()
1 #!/usr/bin/env python 2 # _*_ encoding:utf-8 _*_ 3 4 #客户端 5 import socket 6 7 client = socket.socket() 8 client.connect(('localhost',7000)) 9 10 while True: 11 msg = input(">>:").strip() 12 if len(msg) == 0:continue 13 client.send(msg.encode("utf-8")) 14 data = client.recv(1024) 15 print("recv",data.decode()) 16 17 client.close() 18 19 --------------------------------- 20 21 # _*_ encoding:utf-8 _*_ 22 23 #服务器端 24 25 import socket 26 import os 27 import subprocess 28 29 server = socket.socket() 30 server.bind(('localhost',7777)) 31 server.listen() 32 33 print("开始等待:") 34 while True: 35 conn,addr = server.accept() 36 37 print(conn,addr) 38 print("等待来了") 39 40 while True: 41 data = conn.recv(1024) 42 print(data) 43 if not data: 44 print("client not data") 45 break 46 res=os.popen(data).read() 47 conn.sendall(res) 48 49 server.close()
1 #!/usr/bin/env python 2 # _*_ encoding:utf-8 _*_ 3 4 import socket 5 6 client = socket.socket() #声明socket类型和socket链接对象(ip.port) 7 client.connect(('localhost',7777)) #链接端口 8 9 f = open("ac.av","wb") 10 while True: 11 msg = input(">>:").strip() 12 if len(msg) == 0:continue 13 client.send(msg.encode('utf-8')) #发送数据,转码utf8 14 data = client.recv(102400) #设置接受数据大小 15 f.write(data) 16 f.flush() 17 18 client.close() #关闭连接 19 --------------------- 20 21 #服务器端 22 23 import socket 24 import os 25 26 server = socket.socket() 27 server.bind(('localhost',7777)) 28 server.listen() 29 30 print("开始等待:") 31 while True: 32 conn,addr = server.accept() 33 34 print(conn,addr) 35 print("等待来了") 36 37 while True: 38 data = conn.recv(102400) 39 print(data) 40 if not data: 41 print("client not data") 42 break 43 #res=os.popen(data).read() 44 #conn.sendall(res) 45 f = open('abc.wmv') 46 data = f.read() 47 conn.sendall(data) 48 49 server.close()
简单接受发送
1 #server 2 3 import socket 4 import json 5 6 s = socket.socket() 7 s.bind(('localhost',9000)) 8 9 s.listen(5) 10 11 while True: 12 conn,client_addr = s.accept() 13 print('got a new conn:',client_addr) 14 while True: 15 data = conn.recv(1024) 16 print("recv data:",data) 17 data = json.loads(data.decode()) 18 19 if data.get("action") is not None: 20 if data['action'] == 'put': 21 file_obj = open(data['filename'],'wb') 22 received_size = 0 23 24 while received_size < data['size']: 25 recv_data = conn.recv(4096) 26 file_obj.write(recv_data) 27 received_size += len(recv_data) 28 print(data['size'],received_size) 29 else: 30 print("---successfully received file [%s]--",data['filename']) 31 file_obj.close() 32 33 elif data['action'] == 'get': 34 pass 35 36 37 38 39 40 41 42 #client 43 44 import socket 45 import os 46 import json 47 48 client = socket.socket() 49 client.connect(('localhost',9000)) 50 51 while True: 52 choice = input(">>>").strip() 53 if len(choice) == 0:continue 54 cmd_list = choice.split() 55 if cmd_list[0] == "put": 56 if len(cmd_list) == 1: 57 print("no filename folliws after put cmd") 58 continuelename 59 fi = cmd_list[1] 60 if os.path.isfile(filename): 61 file_obj = open(filename,'rb') 62 base_filename = filename.split("/")[-1] 63 print(base_filename,os.path.getsize(filename)) 64 data_header = { 65 "action":"put", 66 "filename":base_filename, 67 "size":os.path.getsize(filename) 68 } 69 client.send(json.dumps(data_header).encode()) 70 71 for line in file_obj: 72 client.send(line) 73 print("-------file is donw-------") 74 75 76 else: 77 print("file is not valid") 78 continue 79 elif cmd_list[0] == "get": 80 pass
更多功能
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM,0)
参数一：地址簇 socket.AF_INET IPv4（默认） socket.AF_INET6 IPv6 socket.AF_UNIX 只能够用于单一的Unix系统进程间通信 参数二：类型 socket.SOCK_STREAM 流式socket , for TCP （默认） socket.SOCK_DGRAM 数据报式socket , for UDP socket.SOCK_RAW 原始套接字，普通的套接字无法处理ICMP、IGMP等网络报文，而SOCK_RAW可以；其次，SOCK_RAW也可以处理特殊的IPv4报文；此外，利用原始套接字，可以通过IP_HDRINCL套接字选项由用户构造IP头。 socket.SOCK_RDM 是一种可靠的UDP形式，即保证交付数据报但不保证顺序。SOCK_RAM用来提供对原始协议的低级访问，在需要执行某些特殊操作时使用，如发送ICMP报文。SOCK_RAM通常仅限于高级用户或管理员运行的程序使用。 socket.SOCK_SEQPACKET 可靠的连续数据包服务 参数三：协议 0 （默认）与特定的地址家族相关的协议,如果是 0 ，则系统就会根据地址格式和套接类别,自动选择一个合适的协议
sk.bind(address)
s.bind(address) 将套接字绑定到地址。address地址的格式取决于地址族。在AF_INET下，以元组（host,port）的形式表示地址。
sk.listen(backlog)
开始监听传入连接。backlog指定在拒绝连接之前，可以挂起的最大连接数量。
backlog等于5，表示内核已经接到了连接请求，但服务器还没有调用accept进行处理的连接个数最大为5
这个值不能无限大，因为要在内核中维护连接队列
sk.setblocking(bool)
是否阻塞（默认True），如果设置False，那么accept和recv时一旦无数据，则报错。
sk.accept()
接受连接并返回（conn,address）,其中conn是新的套接字对象，可以用来接收和发送数据。address是连接客户端的地址。
接收TCP 客户的连接（阻塞式）等待连接的到来
sk.connect(address)
连接到address处的套接字。一般，address的格式为元组（hostname,port）,如果连接出错，返回socket.error错误。
sk.connect_ex(address)
同上，只不过会有返回值，连接成功时返回 0 ，连接失败时候返回编码，例如：10061
sk.close()
关闭套接字
sk.recv(bufsize[,flag])
接受套接字的数据。数据以字符串形式返回，bufsize指定最多可以接收的数量。flag提供有关消息的其他信息，通常可以忽略。
sk.recvfrom(bufsize[.flag])
与recv()类似，但返回值是（data,address）。其中data是包含接收数据的字符串，address是发送数据的套接字地址。
sk.send(string[,flag])
将string中的数据发送到连接的套接字。返回值是要发送的字节数量，该数量可能小于string的字节大小。即：可能未将指定内容全部发送。
sk.sendall(string[,flag])
将string中的数据发送到连接的套接字，但在返回之前会尝试发送所有数据。成功返回None，失败则抛出异常。
内部通过递归调用send，将所有内容发送出去。
sk.sendto(string[,flag],address)
将数据发送到套接字，address是形式为（ipaddr，port）的元组，指定远程地址。返回值是发送的字节数。该函数主要用于UDP协议。
sk.settimeout(timeout)
设置套接字操作的超时期，timeout是一个浮点数，单位是秒。值为None表示没有超时期。一般，超时期应该在刚创建套接字时设置，因为它们可能用于连接的操作（如 client 连接最多等待5s ）
sk.getpeername()
返回连接套接字的远程地址。返回值通常是元组（ipaddr,port）。
sk.getsockname()
返回套接字自己的地址。通常是一个元组(ipaddr,port)
sk.fileno()
class socketserver.ForkingTCPServer tcp 进程 class socketserver.ForkingUDPServer udp 进程 class socketserver.ThreadingTCPServer tcp 线程 class socketserver.ThreadingUDPServer udp线程
多线程接受数据
#服务端
import socketserver
class MyTCPHandler(socketserver.BaseRequestHandler):
def handle(self):
while True:
self.data = self.request.recv(1024).strip()
print(self.client_address[0])
print(self.data)
self.request.sendall(self.data.upper())
if __name__ == '__main__':
HOST,PORT = "localhost",9000
server = socketserver.ThreadingTCPServer((HOST,PORT),MyTCPHandler)
server.serve_forever()
#客户端
import socket
import os
import json
client = socket.socket()
client.connect(('localhost',9000))
while True:
choice = input("输入>>>:").strip()
if len(choice) == 0:continue
client.send(choice.encode())
recv = client.recv(1024)
print("recviver:",recv.decode())
避免断开报错
客户端断开地址会报错地址在占用 因为服务端没有释放ip 添加此内容就会释放ip
import socket server = socket.socket(socket.AF_INET,socket.SOCK_STREAM) socket.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
简单线程
import threading
import time
def run(n):
time.sleep(1)
print("run....",n)
for i in range(10):
t = threading.Thread(target=run,args=(i,))
t.start()
class_system
bin/start.py
1 import os 2 import sys 3 import platform 4 5 if platform.system() == "Windows": 6 BASE_DIR = "\".join(os.path.abspath(os.path.dirname(__file__)).split("\")[:-1]) 7 8 else: 9 BASE_DIR = "/".join(os.path.abspath(os.path.dirname(__file__)).split("/")[:-1]) 10 11 12 from core import main 13 from conf import settings 14 15 if __name__ == '__main__': 16 obj = main.Manage_center() 17 obj.run()
conf/settings.py
1 #!/usr/bin/python env 2 # _*_ encoding:utf-8 _*_ 3 4 import os 5 import sys 6 import platform 7 8 if platform.system() == "Windows": 9 BASE_DIR = "\".join(os.path.abspath(os.path.dirname(__file__)).split("\")[:-1]) 10 database_path = os.path.join(BASE_DIR,'database') 11 12 else: 13 BASE_DIR = "/".join(os.path.abspath(os.path.dirname(__file__)).split("/")[:-1]) 14 database_path = os.path.join(BASE_DIR,"database") 15 16 school_db_file = os.path.join(database_path,"school")
core/main.py
1 import os 2 import sys 3 import shelve 4 from conf import settings 5 from modules.school import School 6 7 class Manage_center(object): 8 def __init__(self): 9 pass 10 11 def run(self): 12 while True: 13 print(" 欢迎进入CLASS_SYSTEM系统 " 14 "1.学生视图 " 15 "2.教师视图 " 16 "3.学校视图 " 17 "q 推出") 18 user_choice = input(" 33[35m请选择视图: 33[0m") 19 if user_choice == '1': 20 Manage_student() 21 elif user_choice == '2': 22 Manage_teacher() 23 elif user_choice == '3': 24 Manage_school() 25 elif user_choice == 'q': 26 print(" 33[32m你已推出class_system系统 33[0m") 27 break 28 else: 29 print(" 33[31m输入错误！请输入真确的选项! 33[0m") 30 31 32 33 class Manage_school(object): 34 '''学校管理视图''' 35 def __init__(self): 36 if os.path.exists(settings.school_db_file + ".dat"): 37 self.school_db = shelve.open(settings.school_db_file) 38 self.run_manage() 39 self.school_db.close() 40 else: 41 print(" 33[35m初始化数据库 33[0m") 42 self.initialize_school() 43 self.run_manage() 44 self.school_db.close() 45 46 def initialize_school(self): 47 '''实例化两个学校 北京上海''' 48 self.school_db = shelve.open(settings.school_db_file) 49 self.school_db['北京'] = School("北京","中国.北京") 50 self.school_db["上海"] = School("上海", "中国.上海") 51 52 def run_manage(self): 53 '''运行学校管理视图''' 54 while True: 55 for key in self.school_db: 56 print("学校名称:",key) 57 choice_school = input(" 33[35m请选择管理的学校: 33[0m").strip() 58 if choice_school in self.school_db: 59 self.choice_school = choice_school 60 self.school_obj = self.school_db[choice_school] 61 while True: 62 print(" 欢迎来到老男孩校区: " 63 "添加课程 add_course " 64 "增加班级 add_class " 65 "招聘讲师 add_teacher " 66 "查看课程 check_course " 67 "查看班级 check_class " 68 "查看讲师 check_teacher " 69 "推出程序 exit") 70 user_func = input(" 33[35m请选择： 33[0m").strip() 71 if hasattr(self,user_func): 72 getattr(self,user_func)() 73 else: 74 print(" 33[36m 输入错误!请输入正确的学校名: 33[0m") 75 76 def add_course(self): 77 course_name = input(" 33[35m请输入添加的课程的名称: 33[0m").strip() 78 course_price = input(" 33[33m请输入课程的价格: 33[0m").strip() 79 course_time = input(" 33[35m请输入课程的时长: 33[0m").strip() 80 if course_name in self.school_obj.school_course: 81 print(" 33[32m课程已存在 33[0m") 82 self.school_obj.create_course(course_name,course_price,course_time) 83 print(" 33[32m课程更新成功 33[0m") 84 else: 85 self.school_obj.create_course(course_name,course_price,course_time) 86 print(" 33[32m课程添加成功 33[0m") 87 self.school_db.update({self.choice_school:self.school_obj}) 88 89 def add_class(self): 90 class_name = input(" 33[34m请输入添加的班级名称: 33[0m").strip() 91 course_name = input(" 33[34m请输入关联的课程: 33[0m").strip() 92 if class_name not in self.school_obj.school_class: 93 if course_name in self.school_obj.school_course: 94 course_obj = self.school_obj.school_course[course_name] 95 self.school_obj.create_class(class_name,course_obj) 96 self.school_db.update({self.choice_school:self.school_obj}) 97 print(" 33[32m班级创建成功 33[0m") 98 else: 99 print(" 33[31m输入错误： 关联的课程不存在 33[0m") 100 else: 101 print(" 33[31m输入错误：班级已存在 33[0m") 102 103 def add_teacher(self): 104 teacher_name = input(" 33[34m请输入老师的名称: 33[0m").strip() 105 teacher_salary = input(" 33[34m请输入老师的薪资: 33[0m").strip() 106 teacher_class = input(" 33[34m请输入关联的班级: 33[0m").strip() 107 if teacher_class in self.school_obj.school_class: 108 class_obj = self.school_obj.school_class[teacher_class] 109 if teacher_name not in self.school_obj.school_teacher: 110 self.school_obj.create_teacher(teacher_name,teacher_salary,teacher_class,class_obj) 111 print(" 33[32m讲师创建成功 33[0m") 112 else: 113 print(" 33[31m讲师已经存在 信息更新完成 33[0m") 114 115 self.school_db.update({self.choice_school:self.school_obj}) 116 else: 117 print(" 33[31m错误! 关联的班级不存在 33[0m") 118 119 def check_course(self): 120 self.school_obj.show_course() 121 122 def check_class(self): 123 self.school_obj.show_class() 124 125 def check_teacher(self): 126 self.school_obj.show_teacher() 127 128 def exit(self): 129 self.school_db.close() 130 sys.exit(" 33[32m你已推出学员管理系统 33[0m") 131 132 class Manage_student(object): 133 '''学生视图''' 134 def __init__(self): 135 if os.path.exists(settings.school_db_file + ".dat"): 136 self.school_db = shelve.open(settings.school_db_file) 137 self.run_manage() 138 self.school_db.close() 139 else: 140 print(" 33[31m数据文件不存在 请先创建学校 33[0m") 141 exit() 142 143 def run_manage(self): 144 print(" 欢迎进入学员视图:") 145 for key in self.school_db: 146 print("学校名称:",key) 147 choice_school = input(" 33[34m请选择学校名: 33[0m").strip() 148 if choice_school in self.school_db: 149 self.choice_school = choice_school 150 self.school_obj = self.school_db[choice_school] 151 student_name = input(" 33[31m输入学生姓名: 33[0m").strip() 152 student_age = input(" 33[34m输入学生年龄: 33[0m").strip() 153 self.school_obj.show_class_course() 154 class_choice = input(" 33[34m请选择班级: 33[0m").strip() 155 if class_choice in self.school_obj.school_class: 156 self.school_obj.create_student(student_name,student_age,class_choice) 157 self.school_db.update({self.choice_school:self.school_obj}) 158 print(" 33[32m学生注册成功 33[0m") 159 else: 160 print(" 33[31m错误: 输入的班级不存在 33[0m") 161 else: 162 print(" 33[31m错误:输入的学校不存在 33[0m") 163 164 165 class Manage_teacher(object): 166 '''教师视图''' 167 def __init__(self): 168 if os.path.exists(settings.school_db_file + ".dat"): 169 self.school_db = shelve.open(settings.school_db_file) 170 self.run_manage() 171 self.school_db.close() 172 else: 173 print(" 33[31m数据文件不存在，请先创建学校 33[0m") 174 exit() 175 def run_manage(self): 176 for key in self.school_db: 177 print("学校名:",key) 178 chocie_school = input(" 33[34m请输入选择的学校名称: 33[0m").strip() 179 if chocie_school in self.school_db: 180 self.choice_school = chocie_school 181 self.school_obj = self.school_db[chocie_school] 182 teacher_name = input(" 33[34m请输入登录讲师的姓名： 33[0m").strip() 183 while True: 184 if teacher_name in self.school_obj.school_teacher: 185 print(" 你以成功登陆到教师中心 " 186 "查看班级 check_class " 187 "推出程序 exit") 188 user_func = input(" 33[34m请选择： 33[0m") 189 if hasattr(self,user_func): 190 getattr(self,user_func)(teacher_name) 191 else: 192 print(" 33[31m错误讲师不存在 33[0m") 193 break 194 195 def check_class(self,teacher_name): 196 self.school_obj.show_teacher_classinfo(teacher_name) 197 198 def exit(self,*args): 199 self.school_db.close() 200 sys.exit(" 33[32m你已推出学员管理系统 33[0m")
modules/classs.py
1 class Class(object): 2 '''班级类 班级名称 课程 学生''' 3 def __init__(self,class_name,course_obj): 4 self.class_name = class_name 5 self.class_course = course_obj 6 self.class_student = {} #学生字典
modules/course.py
1 class Course(): 2 '''课程类 课程名称 价格 周期''' 3 def __init__(self,course_name,course_price,course_time): 4 self.course_name = course_name 5 self.course_price = course_price 6 self.course_time = course_time
modules/school.py
1 from modules.course import Course 2 from modules.classs import Class 3 from modules.teacher import Teacher 4 from modules.student import Student 5 6 class School(object): 7 '''学校类 学校名称 地址 课程 班级 教师''' 8 def __init__(self,school_name,school_addr): 9 self.school_name = school_name 10 self.school_addr = school_addr 11 self.school_course = {} 12 self.school_class = {} 13 self.school_teacher = {} 14 15 def create_course(self,course_name,course_price,course_time): 16 '''创建课程''' 17 course_obj = Course(course_name,course_price,course_time) 18 self.school_course[course_name] = course_obj 19 20 def show_course(self): 21 '''查看课程信息''' 22 for key in self.school_course: 23 course_obj = self.school_course[key] 24 print(" 33[35m课程:%s 价格:%s 周期:%s 33[0m"%(course_obj.course_name,course_obj.course_price,course_obj.course_time)) 25 26 def create_class(self,class_name,course_obj): 27 '''创建班级''' 28 class_obj = Class(class_name,course_obj) 29 self.school_class[class_name] = class_obj 30 31 def show_class(self): 32 for key in self.school_class: 33 class_obj = self.school_class[key] 34 print(" 33[35m班级:%s 课程:%s 33[0m"%(class_obj.class_name,class_obj.class_course.course_name)) 35 36 def show_class_course(self): 37 for key in self.school_class: 38 class_obj = self.school_class[key] 39 course_obj = class_obj.class_course 40 print(" 33[36m班级:%s 关联课程:%s 价格:%s 周期:%s:月 33[0m"%(class_obj.class_name, course_obj.course_name,course_obj.course_price,course_obj.course_time)) 41 42 def create_teacher(self,teacher_name,teacher_salary,class_name,class_obj): 43 '''创建讲师''' 44 teacher_obj = Teacher(teacher_name,teacher_salary) 45 teacher_obj.teacher_add_class(class_name,class_obj) 46 self.school_teacher[teacher_name] = teacher_obj 47 48 def update_teacher(self,teacher_name,class_name,class_obj): 49 '''更新讲师信息''' 50 teacher_obj = self.school_teacher[teacher_name] 51 teacher_obj.teacher_add_class(class_name,class_obj) 52 53 def show_teacher(self): 54 '''查看讲师信息''' 55 for key in self.school_teacher: 56 teacher_obj = self.school_teacher[key] 57 class_list = [] 58 for i in teacher_obj.teacher_class: 59 class_list.append(i) 60 print(" 33[36m讲师:%s 薪资:%s 关联班级:%s 33[0m"%(teacher_obj.teacher_name,teacher_obj.teacher_salary,class_list)) 61 62 def create_student(self,student_name,student_age,class_choice): 63 '''注册学生''' 64 student_obj = Student(student_name,student_age) 65 class_obj = self.school_class[class_choice] 66 class_obj.class_student[student_name] = student_obj 67 self.school_class[class_choice] = class_obj 68 69 def show_teacher_classinfo(self,teacher_name): 70 teacher_obj = self.school_teacher[teacher_name] 71 for i in teacher_obj.teacher_class: 72 class_obj = self.school_class[i] 73 student_list = [] 74 for k in class_obj.class_student: 75 student_list.append(k) 76 print(" 33[33m班级:%s 关联课程:%s 学员:%s 33[0m" %(class_obj.class_name,class_obj.class_course.course_name,student_list))
modules/student.py
1 class Student(object): 2 '''学生类 学生名称 年龄''' 3 def __init__(self,student_name,student_age): 4 self.student_name = student_name 5 self.student_age = student_age
modules/teacher.py
1 # class Teacher(object): 2 '''讲师类 老师姓名 老师工资 老师关联的班级''' 3 def __init__(self,teacher_name,teacher_salary): 4 self.teacher_name = teacher_name 5 self.teacher_salary = teacher_salary 6 self.teacher_class = {} 7 8 def teacher_add_class(self,class_name,class_obj): 9 self.teacher_class[class_name] = class_obj
ElectiveSys
bin/start.py
1 #_*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 import os,sys 4 from src.services import admin_service 5 from src.services import teacher_service 6 from src.services import student_service 7 from src.services import initialize_service 8 BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 9 sys.path.append(BASE_DIR) 10 11 12 13 14 def show_role(): 15 msg= """ 16 0:初始化 17 1:管理员 18 2:老师 19 3:学生 20 """ 21 print(msg) 22 23 if __name__ == '__main__': 24 role_main={ 25 '0':initialize_service.main, 26 '1':admin_service.login, 27 '2':teacher_service.login, 28 '3':student_service.login, 29 } 30 while True: 31 show_role() 32 choice=input('输入角色: ').strip() 33 if choice not in role_main:continue 34 role_main[choice]()
conf/settings.py
1 #_*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 import os 4 BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 5 6 ADMIN_DB_DIR=os.path.join(BASE_DIR,'db','admin') 7 SCHOOL_DB_DIR=os.path.join(BASE_DIR,'db','school') 8 TEACHER_DB_DIR=os.path.join(BASE_DIR,'db','teacher') 9 COURSE_DB_DIR=os.path.join(BASE_DIR,'db','course') 10 COURSE_TO_TEACHER_DB_DIR=os.path.join(BASE_DIR,'db','course_to_teacher') 11 CLASSES_DB_DIR=os.path.join(BASE_DIR,'db','classes') 12 STUDENT_DB_DIR=os.path.join(BASE_DIR,'db','student')
lib/commons.py
1 #_*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 import uuid,hashlib 4 import time 5 6 def create_uuid(): 7 return str(uuid.uuid1()) 8 9 def create_md5(): 10 m=hashlib.md5() 11 m.update(bytes(str(time.time()),encoding='utf-8')) 12 return m.hexdigest() 13 14 if __name__ == '__main__': 15 x=create_md5() 16 17 print(x) 18 19 y=create_uuid() 20 print(y)
src/identifier.py
1 #_*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 from lib import commons 4 import os,pickle 5 6 class Nid: 7 def __init__(self,role,db_path): 8 role_list=[ 9 'admin','school','teacher','course','course_to_teacher','classes','student' 10 ] 11 if role not in role_list: 12 raise Exception('用户角色错误,选项: %s' % ','.join(role_list)) 13 self.role=role 14 self.uuid=commons.create_uuid() 15 self.db_path=db_path 16 17 def __str__(self): 18 return self.uuid 19 20 def get_obj_by_uuid(self): 21 for filename in os.listdir(self.db_path): 22 if filename == self.uuid: 23 return pickle.load(open(os.path.join(self.db_path,filename),'rb')) 24 return None 25 26 class AdminNid(Nid): 27 def __init__(self,db_path): 28 super(AdminNid,self).__init__('admin',db_path) 29 30 class SchoolNid(Nid): 31 def __init__(self,db_path): 32 super(SchoolNid,self).__init__('school',db_path) 33 34 class TeacherNid(Nid): 35 def __init__(self,db_path): 36 super(TeacherNid,self).__init__('teacher',db_path) 37 38 class CourseNid(Nid): 39 def __init__(self,db_path): 40 super(CourseNid,self).__init__('course',db_path) 41 42 class Course_to_teacherNid(Nid): 43 def __init__(self,db_path): 44 super(Course_to_teacherNid,self).__init__('course_to_teacher',db_path) 45 46 class ClassesNid(Nid): 47 def __init__(self,db_path): 48 super(ClassesNid,self).__init__('classes',db_path) 49 50 class StudentNid(Nid): 51 def __init__(self,db_path): 52 super(StudentNid,self).__init__('student',db_path)
src/models.py
1 #_*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 import time 4 import pickle 5 import os 6 from conf import settings 7 from src import identifier 8 9 10 class BaseModel: 11 def save(self): 12 file_path=os.path.join(self.db_path,str(self.nid)) 13 pickle.dump(self,open(file_path,'wb')) 14 15 @classmethod 16 def get_all_obj_list(cls): 17 ret=[] 18 for filename in os.listdir(cls.db_path): 19 file_path=os.path.join(cls.db_path,filename) 20 ret.append(pickle.load(open(file_path,'rb'))) 21 return ret 22 23 class Admin(BaseModel): 24 db_path=settings.ADMIN_DB_DIR 25 def __init__(self,username,password): 26 self.nid=identifier.AdminNid(self.db_path) 27 self.username=username 28 self.password=password 29 self.create_time=time.strftime('%Y-%m-%d') 30 31 @staticmethod 32 def login(): 33 try: 34 name=input('请输入用户名: ').strip() 35 pas=input('请输入密码: ').strip() 36 for obj in Admin.get_all_obj_list(): 37 if obj.username == name and obj.password == pas: 38 status = True 39 error='' 40 data=' 33[45;1m登录成功 33[0m' 41 break 42 else: 43 raise Exception(' 33[43;1m用户名或密码错误 33[0m' %name) 44 except Exception as e: 45 status=False 46 error=str(e) 47 data='' 48 return {'status':status,'error':error,'data':data} 49 50 51 class School(BaseModel): 52 db_path=settings.SCHOOL_DB_DIR 53 def __init__(self,name,addr): 54 self.nid=identifier.SchoolNid(self.db_path) 55 self.name=name 56 self.addr=addr 57 self.create_time=time.strftime('%Y-%m-%d %X') 58 self.__income=0 59 60 def __str__(self): 61 return self.name 62 63 class Teacher(BaseModel): 64 db_path=settings.TEACHER_DB_DIR 65 def __init__(self,name,level): 66 self.nid=identifier.TeacherNid(self.db_path) 67 self.name=name 68 self.level=level 69 self.__account=0 70 self.create_time=time.strftime('%Y-%m-%d %X') 71 72 class Course(BaseModel): 73 db_path=settings.COURSE_DB_DIR 74 def __init__(self,name,price,period,school_nid): 75 self.nid=identifier.CourseNid(self.db_path) 76 self.name=name 77 self.price=price 78 self.period=period 79 self.school_nid=school_nid 80 81 class Course_to_teacher(BaseModel): 82 db_path=settings.COURSE_TO_TEACHER_DB_DIR 83 def __init__(self,course_nid,school_nid): 84 self.nid=identifier.Course_to_teacherNid(self.db_path) 85 self.course_nid=course_nid 86 self.school_nid=school_nid 87 88 def get_course_to_teacher_list(self): 89 ret=self.get_all_obj_list() 90 if ret: 91 return [ret.course_nid.get_obj_by_uuid(),ret.classes_nid.get_obj_by_uuid()] 92 return [None,None] 93 94 class Classes(BaseModel): 95 db_path=settings.CLASSES_DB_DIR 96 def __init__(self,name,tuition,school_nid,course_to_teacher_list): 97 self.nid=identifier.ClassesNid(self.db_path) 98 self.name=name 99 self.tuition=tuition 100 self.school_nid=school_nid 101 self.course_to_teacher_list=course_to_teacher_list 102 103 class Score: 104 def __init__(self,nid): 105 self.nid=nid 106 self.score_dict={} 107 108 def set(self,course_to_teacher_nid,number): 109 self.score_dict[course_to_teacher_nid]=number 110 111 def get(self,course_to_teacher_nid): 112 return self.score_dict.get(course_to_teacher_nid) 113 114 class Student(BaseModel): 115 db_path=settings.STUDENT_DB_DIR 116 def __init__(self,name,age,qq,classes_nid): 117 self.nid=identifier.StudentNid(self.db_path) 118 self.name=name 119 self.age=age 120 self.qq=qq 121 self.classes_nid=classes_nid 122 self.score=Score(self.nid)
/src/services/admin_service.py
1 #_*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 from src.models import Admin 4 from src.models import School 5 from src.models import Teacher 6 from src.models import Course 7 def create_school(): 8 try: 9 name=input('请输入学校名字: ').strip() 10 addr=input('请输入学校地址: ').strip() 11 school_name_list=[(obj.name,obj.addr) for obj in School.get_all_obj_list()] 12 if (name,addr) in school_name_list: 13 raise Exception(' 33[43;1m[%s] [%s]校区 已经存在,不可重复创建 33[0m' %(name,addr)) 14 obj=School(name,addr) 15 obj.save() 16 status=True 17 error='' 18 data=' 33[33;1m[%s] [%s]校区 创建成功 33[0m' %(obj.name,obj.addr) 19 except Exception as e: 20 status=False 21 error=str(e) 22 data='' 23 return {'status':status,'error':error,'data':data} 24 25 26 def show_school(): 27 for obj in School.get_all_obj_list(): 28 print(' 33[45;1m学校[%s] 地址[%s] 创建日期[%s] 33[0m'.center(60,'-') 29 %(obj.name,obj.addr,obj.create_time)) 30 31 def create_teacher(): 32 try: 33 name=input('请输入老师姓名: ').strip() 34 level=input('请输入老师级别: ').strip() 35 teacher_name_list=[obj.name for obj in Teacher.get_all_obj_list()] 36 if name in teacher_name_list: 37 raise Exception(' 33[43;1m老师[%s] 已经存在,不可重复创建 33[0m' %(name)) 38 obj=Teacher(name,level) 39 obj.save() 40 status=True 41 error='' 42 data=' 33[33;1m老师[%s] 级别[%s] 时间[%s]创建成功 33[0m' %(obj.name,obj.level,obj.create_time) 43 except Exception as e: 44 status=False 45 error=str(e) 46 data='' 47 return {'status':status,'error':error,'data':data} 48 49 50 def show_teacher(): 51 for obj in Teacher.get_all_obj_list(): 52 print(' 33[33;1m老师[%s] 级别[%s] 创建时间[%s] 33[0m'.center(60,'-') 53 %(obj.name,obj.level,obj.create_time)) 54 55 56 def create_course(): 57 try: 58 print('创建课程'.center(60,'=')) 59 school_list=School.get_all_obj_list() 60 for k,obj in enumerate(school_list): 61 print(k,obj,obj.addr) 62 sid=int(input('请选择学校: ')) 63 school_obj=school_list[sid] 64 65 name=input('请输入课程名: ').strip() 66 price=input('请输入课程价格: ').strip() 67 period=input('请输入课程周期: ').strip() 68 69 course_name_list=[(obj.name,obj.school_nid.uuid) for obj in Course.get_all_obj_list()] 70 if (name,school_obj.nid.uuid) in course_name_list: 71 raise Exception(' 33[43;1m课程[%s] 已经存在,不可重复创建 33[0m' %(name)) 72 obj=Course(name,price,period,school_obj.nid) 73 obj.save() 74 status=True 75 error='' 76 data=' 33[33;1m课程[%s] 价格[%s] 周期[%s]创建成功 33[0m' %(obj.name,obj.price,obj.period) 77 except Exception as e: 78 status=False 79 error=str(e) 80 data='' 81 return {'status':status,'error':error,'data':data} 82 83 def show_course(): 84 for obj in Course.get_all_obj_list(): 85 print(' 33[33;1m[%s] [%s]校区 [%s]课程 价格[%s] 周期[%s] 33[0m'.center(60,'-') 86 %(obj.school_nid.get_obj_by_uuid().name,obj.school_nid.get_obj_by_uuid().addr, 87 obj.name,obj.price,obj.period)) 88 89 def create_course_to_teacher(): 90 pass 91 92 def create_classes(): 93 pass 94 95 def show_classes(): 96 pass 97 98 def create_student(): 99 pass 100 101 def show_student(): 102 pass 103 104 105 106 def show(): 107 msg=''' 108 0:选项 109 1:创建学校 110 2:查看学校 111 3:创建老师 112 4:查看老师 113 5:创建课程 114 6:查看课程 115 7:关联老师与课程 116 8:创建班级 117 9:查看班级 118 10:创建学生 119 11:查看学生 120 12:退出 121 ''' 122 print(msg) 123 124 125 def main(): 126 choice_dic={ 127 '0':show, 128 '1':create_school, 129 '2':show_school, 130 '3':create_teacher, 131 '4':show_teacher, 132 '5':create_course, 133 '6':show_course, 134 '7':create_course_to_teacher, 135 '8':create_classes, 136 '9':show_classes, 137 '10':create_student, 138 '11':show_student, 139 '12':exit 140 } 141 show() 142 while True: 143 choice=input('请输入选项: ').strip() 144 if choice not in choice_dic:continue 145 ret=choice_dic[choice]() 146 if ret: 147 if ret['status']: 148 print(ret['data'].center(60,'-')) 149 else: 150 print(ret['error'].center(60,'-')) 151 152 153 def login(): 154 ret=Admin.login() 155 if ret: 156 if ret['status']: 157 print(ret['data'].center(60,'-')) 158 main() 159 else: 160 print(ret['error'].center(60,'-')) 161 162 if __name__ == '__main__': 163 main()
/src/services/initialize_service.py
1 #_*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 import getpass 4 from src.models import Admin 5 6 7 def initialize(): 8 try: 9 user = input('请输入初始化用户名：') 10 # pwd = getpass.getpass('请输入初始化密码：') 11 pwd = input('请输入初始化密码：') 12 obj = Admin(user, pwd) 13 obj.save() 14 return True 15 except Exception as e: 16 print(e) 17 18 19 def main(): 20 show = """ 21 1. 初始化管理员账户 22 """ 23 choice_dict = { 24 '1': initialize 25 } 26 while True: 27 print(show) 28 choice = input('请输入操作选项：') 29 if choice not in choice_dict: 30 print('选项错误，请重新输入！！！') 31 func = choice_dict[choice] 32 ret = func() 33 if ret: 34 print('操作成功') 35 return 36 else: 37 print('操作异常，请重新操作')
/src/services/student_service.py
1 #_*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 from src.models import Student 4 5 class register: 6 """ 7 学生注册 8 """ 9 10 11 class score: 12 """ 13 学生查看个人成绩 14 """ 15 16 17 def main(): 18 pass 19 20 def login(): 21 pass
/src/services/teacher_service.py
1 #_*_coding:utf-8_*_ 2 __author__ = 'Linhaifeng' 3 4 def class_info(): 5 """ 6 查看班级信息 7 """ 8 9 10 def student_info(): 11 """ 12 查看学生信息 13 """ 14 15 16 def set_student_score(): 17 """ 18 设置学生分数 19 """ 20 21 22 def main(): 23 pass 24 25 def login(): 26 pass
FTP
READ.txt
1 程序使用说明： 2 3 1.服务端启动 python ftp_server.py start 4 2.客户端启动 python ftp_client.py -s127.0.0.1 -P9999 -ualex -pabcw323 5 3.下载 get test.avi 6 4.上传 put test.avi 7 5.查看目录 ls 8 6.切换目录 cd test
FtpClient/ftp_client.py
1 import socket 2 import os,json 3 import optparse 4 import getpass 5 import hashlib 6 import sys 7 8 STATUS_CODE = { 9 250:"Invalid cmd format, e.g: {'action':'get','filename':'test.py','size':344}", 10 251:"Lnvalid cmd", 11 252:"Lnvalid auth data", 12 253:"Wrong username or password", 13 254:"Passed authentication", 14 } 15 16 class FTPClient(object): 17 def __init__(self): 18 """ 定义交互输入格式""" 19 parser = optparse.OptionParser() 20 parser.add_option("-s","--server",dest="server",help="ftp server ip_addr") 21 parser.add_option("-P","--port",type="int",dest="port",help="ftp server port") 22 parser.add_option("-u","--username",dest="username",help="username") 23 parser.add_option("-p","--password",dest='password',help="password") 24 self.options,self.args = parser.parse_args() 25 self.verify_args(self.options,self.args) 26 self.make_connection() 27 28 def make_connection(self): 29 """链接服务端""" 30 self.sock = socket.socket() 31 self.sock.connect((self.options.server,self.options.port)) 32 33 def verify_args(self,options,args): 34 '''判断用户输入的是否合法''' 35 if options.username is not None and options.password is not None: 36 pass 37 elif options.username is None and options.password is None: 38 pass 39 else: 40 exit("Error: username and password must be provided together..") 41 42 if options.server and options.port: 43 if options.port >0 and options.port<65535: 44 return True 45 else: 46 exit("Error:host port must in 0-65535") 47 48 49 def authenticate(self): 50 '''判断用户是否输入''' 51 if self.options.username: 52 print(self.options.username,self.options.password) 53 return self.get_auth_result(self.options.username,self.options.password) 54 else: 55 retry_count =0 56 while retry_count <3: 57 username = input("username:").strip() 58 password = input("password:").strip() 59 return self.get_auth_result(username,password) 60 61 def get_auth_result(self,user,password): 62 """把用户输入的密码发送服务端 获取结果""" 63 data = {'action':'auth', 64 'username':user, 65 'password':password} 66 self.sock.send(json.dumps(data).encode()) 67 response = self.get_response() 68 if response.get('status_code') == 254: 69 print("Passwd authentication!") 70 self.user = user 71 return True 72 else: 73 print(response.get("status_msg")) 74 75 def get_response(self): 76 '''接收服务端验证结果''' 77 data = self.sock.recv(1024) 78 data = json.loads(data.decode()) 79 return data 80 81 def interactive(self): 82 """判断用户cmd并且执行""" 83 if self.authenticate(): 84 print("--start interactive iwth u...") 85 while True: 86 choice = input("[%s]:"%self.user).strip() 87 if len(choice) == 0:continue 88 cmd_list = choice.split() 89 if hasattr(self,"_%s"%cmd_list[0]): 90 func = getattr(self,"_%s"%cmd_list[0]) 91 func(cmd_list) 92 else: 93 print("Invalid cmd.") 94 95 def __md5_required(self,cmd_list): 96 '''检测命令是否需要进行MD5验证''' 97 if '--md5' in cmd_list: 98 return True 99 100 def show_progress(self,total): 101 """滚动条""" 102 received_size = 0 103 current_percent = 0 104 while received_size < total: 105 if int((received_size / total) * 100 ) > current_percent: 106 print("#",end="",flush=True) 107 current_percent = int((received_size / total) * 100 ) 108 109 new_size = yield 110 received_size += new_size 111 112 def _get(self,cmd_list): 113 """下载""" 114 print("get--",cmd_list) 115 if len(cmd_list) ==1: 116 print("ni filename follows...") 117 return 118 data_header = { 119 'action':'get', 120 'filename':cmd_list[1] 121 } 122 if self.__md5_required(cmd_list): 123 data_header['md5'] = True 124 125 self.sock.send(json.dumps(data_header).encode()) 126 response = self.get_response() 127 print(response) 128 if response["status_code"] ==257: 129 self.sock.send(b'1') 130 base_filename = cmd_list[1].split('/')[-1] 131 received_size = 0 132 file_obj = open(base_filename,"wb") 133 if self.__md5_required(cmd_list): 134 md5_obj = hashlib.md5() 135 progress = self.show_progress(response['file_size']) 136 progress.__next__() 137 while received_size < response['file_size']: 138 data = self.sock.recv(4096) 139 received_size += len(data) 140 try: 141 progress.send(len(data)) 142 except StopIteration as e: 143 print("100%") 144 file_obj.write(data) 145 md5_obj.update(data) 146 else: 147 print("--->file rece done-----") 148 file_obj.close() 149 md5_val = md5_obj.hexdigest() 150 md5_from_server = self.get_response() 151 if md5_from_server['status_code'] ==258: 152 if md5_from_server['md5'] == md5_val: 153 print("%s 文件一只性校验成功！"%base_filename) 154 else: 155 progress = self.show_progress(response['file_size']) 156 progress.__next__() 157 158 while received_size < response['file_size']: 159 data = self.sock.recv(4096) 160 received_size += len(data) 161 file_obj.write(data) 162 try: 163 progress.send(len(data)) 164 except StopIteration as e: 165 print("100%") 166 else: 167 print("-->file rece done----") 168 file_obj.close() 169 170 def _put(self,cmd_list): 171 if len(cmd_list) ==1: 172 print("错误没有输入上传文件！") 173 return 174 if os.path.isfile(cmd_list[1]): 175 file_size = os.path.getsize(cmd_list[1]) 176 data_header = { 177 'action':'put', 178 'filename':cmd_list[1], 179 'file_size':file_size 180 } 181 182 if self.__md5_required(cmd_list): 183 data_header['md5'] = True 184 md5_obj = hashlib.md5() 185 186 file_obj = open(cmd_list[1], "rb") 187 self.sock.send(json.dumps(data_header).encode()) 188 response = self.get_response() 189 if response.get('status_code') == 259: 190 #print(response['status_msg']) 191 for line in file_obj: 192 self.sock.send(line) 193 md5_obj.update(line) 194 else: 195 file_obj.close() 196 md5_val = md5_obj.hexdigest() 197 data_header['md5_size'] = md5_val 198 if response.get('status_code') == 260: 199 print(response['status_msg']) 200 201 else: 202 file_obj = open(cmd_list[1],"rb") 203 self.sock.send(json.dumps(data_header).encode()) 204 response = self.get_response() 205 if response.get('status_code') == 259: 206 # print(response['status_msg']) 207 for line in file_obj: 208 self.sock.send(line) 209 else: 210 file_obj.close() 211 if response.get('status_code') == 261: 212 print(response['status_msg']) 213 214 215 def _ls(self,cmd_list): 216 """ls命令""" 217 if len(cmd_list) == 0: 218 print("输入错误！") 219 return 220 data_header = {'action':'ls'} 221 if cmd_list[0] == "ls": 222 self.sock.send(json.dumps(data_header).encode()) 223 response = self.get_response() 224 if response.get('status_code') == 262: 225 print(response['cmd'].strip()) 226 return 227 else: 228 print("命令输入错误！") 229 230 def _cd(self,cmd_list): 231 if len(cmd_list) == 1: 232 print("错误cd没有输入路径！") 233 return 234 data_header = {'action':'cd', 235 'dir':cmd_list[1] 236 } 237 self.sock.send(json.dumps(data_header).encode()) 238 response = self.get_response() 239 if response.get('status_code') ==263: 240 print(response['cmd']) 241 return 242 243 244 245 246 247 248 if __name__ == "__main__": 249 ftp = FTPClient() 250 ftp.interactive()
FtpServer/bin/ftp_server.py
1 import os,sys 2 3 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 4 sys.path.append(BASE_DIR) 5 6 from core import main 7 8 if __name__ == "__main__": 9 main.ArvgHandler()
FtpServer/conf/accounts.cfg
1 [DEFAULT] 2 3 [alex] 4 Password = abcw323 5 Quotation = 100 6 7 [jack] 8 Password = 123 9 Quotation = 100
FtpServer/conf/settings.py
1 import os 2 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 3 4 USER_HOME = "%s/home" % BASE_DIR 5 LOG_DIR = "%s/log" % BASE_DIR 6 LOG_LEVEL = 'DEBUG' 7 8 ACCOUT_DIR = "%s/db" %BASE_DIR 9 ACCOUNT_FILE = "%s/conf/accounts.cfg" %BASE_DIR 10 11 HOST = "127.0.0.1" 12 PORT = 9999
FtpServer/core/ftp_server.py
1 import socketserver 2 import configparser 3 from conf import settings 4 import os 5 import hashlib 6 7 8 9 STATUS_CODE = { 10 250 : "Invalid cmd format,e.g:{'action':'get','filename':'test.py','size':344}", 11 251 : "Invalid cmd", 12 252 : "Invalid auth data", 13 253 : "Wrong username or passwrod", 14 254 : "passed authentication", 15 255 : "Filename doesn't provided", 16 256 : "File doesn,t exit on server", 17 257 : "ready to send file", 18 258 : "md5 verification", 19 259 : "接收到上传文件大小", 20 260 : "文件上传MD5校验成功", 21 261 : "文件上传成功", 22 262 : "cmd ls", 23 263 : "cmd cd" 24 } 25 import json 26 class FTPHandler(socketserver.BaseRequestHandler): 27 28 cd_dict = {} 29 30 def handle(self): 31 while True: 32 self.data = self.request.recv(1024).strip() 33 print(self.client_address[0]) 34 print(self.data) 35 if not self.data: 36 print("client closed...") 37 break 38 data = json.loads(self.data.decode()) 39 if data.get('action') is not None: 40 print('--->',hasattr(self,"_auth")) 41 if hasattr(self,"_%s"% data.get('action')): 42 func = getattr(self,"_%s"%data.get('action')) 43 func(data) 44 else: 45 print("invalid cmd") 46 self.send_response(251) 47 else: 48 print("invalid cmd format") 49 self.send_response(250) 50 51 def send_response(self,status_code,data=None): 52 '''向客户端返回数据''' 53 response = {'status_code':status_code,'status_msg':STATUS_CODE[status_code]} 54 if data: 55 response.update(data) 56 self.request.send(json.dumps(response).encode()) 57 58 def _auth(self,*args,**kwargs): 59 data = args[0] 60 if data.get("username") is None or data.get("password") is None: 61 self.send_response(252) 62 63 user = self.authenticate(data.get("username"),data.get("password")) 64 if user is None: 65 self.send_response(253) 66 else: 67 print("passed authentication",user) 68 self.user = user 69 self.send_response(254) 70 def authenticate(self,username,password): 71 '''验证用用户合法性，合法就返回用户数据''' 72 config = configparser.ConfigParser() 73 config.read(settings.ACCOUNT_FILE) 74 if username in config.sections(): 75 _password = config[username]["Password"] 76 if _password == password: 77 print("pass auth",username) 78 config[username]['Username'] = username 79 return config[username] 80 81 82 def _put(self,*args,**kwargs): 83 """上传文件""" 84 data=(args[0]) 85 file_size=int(data['file_size']) 86 self.send_response(259) 87 file_dir = "%s/%s/%s"%(settings.USER_HOME,self.user["Username"],data['filename']) 88 file_obj = open(file_dir,'wb') 89 received_size = 0 90 if data.get('md5'): 91 md5_obj = hashlib.md5() 92 while received_size < file_size: 93 data = self.request.recv(4096) 94 file_obj.write(data) 95 md5_obj.update(data) 96 received_size += len(data) 97 else: 98 file_obj.close() 99 md5_val = md5_obj.hexdigest() 100 if md5_val == data['md5_size']: 101 self.send_response(260) 102 103 else: 104 while received_size < file_size: 105 data = self.request.recv(4096) 106 file_obj.write(data) 107 received_size += len(data) 108 else: 109 file_obj.close() 110 self.send_response(261) 111 112 113 114 115 def _get(self,*args,**kwargs): 116 """下载""" 117 data = args[0] 118 if data.get("filename") is None: 119 self.send_response(255) 120 121 user_home_dir = "%s/%s"%(settings.USER_HOME,self.user["Username"]) 122 file_abs_path = "%s/%s"%(user_home_dir,data.get('filename')) 123 print("file abs path",file_abs_path) 124 125 if os.path.isfile(file_abs_path): 126 file_obj = open(file_abs_path,"rb") 127 file_size = os.path.getsize(file_abs_path) 128 self.send_response(257,data={'file_size':file_size}) 129 self.request.recv(1) 130 131 if data.get('md5'): 132 md5_obj = hashlib.md5() 133 for line in file_obj: 134 self.request.send(line) 135 md5_obj.update(line) 136 else: 137 file_obj.close() 138 md5_val = md5_obj.hexdigest() 139 self.send_response(258,{'md5':md5_val}) 140 print("send file done....") 141 142 else: 143 for line in file_obj: 144 self.request.send(line) 145 else: 146 file_obj.close() 147 print("send file done...") 148 else: 149 self.send_response(256) 150 151 152 def _cd(self, rgs): 153 """cd 切换目录""" 154 data = rgs 155 a = data['dir'] 156 home_dir = "%s/%s/%s" % (settings.USER_HOME, self.user["Username"],a) 157 os.chdir(home_dir) 158 cmd = os.getcwd() 159 self.send_response(263, data={'cmd': cmd}) 160 c_dir = {'cmd': cmd} 161 FTPHandler.cd_dict.update(c_dir) 162 163 164 def _ls(self, *args, **kwargs): 165 """ls查看目录""" 166 if FTPHandler.cd_dict: 167 os.chdir(FTPHandler.cd_dict['cmd']) 168 else: 169 cmd_dir = "%s/%s" % (settings.USER_HOME, self.user["Username"]) 170 os.chdir(cmd_dir) 171 dat = os.getcwd() 172 cmd = "ls %s" % dat 173 data = os.popen(cmd).read() 174 self.send_response(262, data={'cmd': data}) 175 176 if __name__ == "__main__": 177 HOST,PORT = "127.0.0.1",9999
FtpServer/core/main.py
1 import optparse 2 from core.ftp_server import FTPHandler 3 import socketserver 4 from conf import settings 5 6 class ArvgHandler(object): 7 def __init__(self): 8 self.parser = optparse.OptionParser() 9 # parser.add_option("-s","--host",dest="host",help="server binding host address") 10 # parser.add_option("-p",'--port',dest='port',help="server binding port") 11 (options,args) = self.parser.parse_args() 12 self.verify_args(options,args) 13 14 def verify_args(self,options,args): 15 '''校验并调用相应的功能''' 16 if hasattr(self,args[0]): 17 func = getattr(self,args[0]) 18 func() 19 else: 20 self.parser.print_help() 21 22 def start(self): 23 print('----going to start server----') 24 server = socketserver.ThreadingTCPServer((settings.HOST,settings.PORT),FTPHandler) 25 #socket.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) 26 27 server.serve_forever()