目录
- 仿快播系统
- 一、项目总结三步走
- 二、项目需求分析
- 三、搭建框架
- 四、ORM框架分析
- 五、功能分析
- 六、数据库表创建
- 七、项目开发--仿快播视频
- 服务端client
-
服务端server
- start.py --- 服务端启动文件
- conf --- setting.py --- 项目配置文件
- orm_control --- mysql_control.py --- 数据库配置
- orm_control --- ORM.py --- 对象关系映射
- socket_server --- tcp_server.py --- 连接服务端、套接字
- db --- models.py --- 所有表类
- db --- session_data.py --- 存方服务端session值
- interface --- admin_interface.py --- 管理员接口
- interface --- user_interface.py
- interface --- common_interface.py --- 公共接口功能
- lib --- common.py --- 公共功能
- movie_files --- 服务端存方上传的电影目录
仿快播系统
一、项目总结三步走
1.项目生命周期为基准线、分析要有层次感、不要想到什么说什么。
2.这条基准线上，负责的是哪一块，做了什么。
3.举例说明项目中遇到的问题及怎么解决的。
二、项目需求分析
管理员
1 注册
2 登录
3 上传视频
4 删除视频
5 发布公告
用户
1 注册
2 登录
3 冲会员
4 查看视频
5 下载免费视频
6 下载收费视频
7 查看观影记录
8 查看公告
三、搭建框架
层级结构：客户端 服务端 数据库
客户端：
基于tcp连接的套接字程序
管理员视图
注册、登录、上传视频、删除视频、发布公告
用户视图
注册、登录、购买vip、查看视频、下载免费视频、下载收费视频、查看下载记录、查看公告
服务端：
tcpserver：基于多线程实现并发的套接字通信 解决粘包问题
interface：admin_interface、user_interface、common_interface
models类和ORM框架：models类中的四张表继承ORM框架中的基类model
数据库：
创建四张表：user、movie、notice、download_record
四、ORM框架分析
# 优点：让一个不懂数据库操作的小白也能够简单快速操作数据库实现相应功能
# 缺点：sql封装固定,不利于sql查询优化
# 对象关系映射
# 类 >>> 数据库的表
# 对象 >>> 表的一条条的记录
# 对象获取属性或方法 >>> 记录的字段对应的值
# 一张表有字段,字段又有字段名,字段类型,字段是否是主键,字段的默认值
class Field(object):
pass
# 为了在定义的时候更加方便 通过继承Field定义具体的字段类型
class StringField(Field):
pass
class IntegerField(Field):
pass
class Models(dict):
pass
def __getattr__(self,item):
return self.get(item)
def __setattr__(self,key,value)
self[key] = value
# 查询
def select(self,**kwargs):
# select * from userinfo
# select * from userinfo where id = 1
# 新增
def save(self):
# insert into userinfo(name,password) values(\'jason\',\'123\')
# 修改:是基于已经存在了的数据进行修改操作
def update(self):
# update userinfo set name=\'jason\',password=\'234\' where id = 1
"""
(******)
hasattr
getattr
setattr
"""
# 元类拦截类的创建过程 使它具备表的特性
class ModelsMetaClass(type):
def __new__(cls,class_name,class_bases,class_attrs):
# 只拦截模型表的创建表
if class_name == \'Models\':
return type.__new__(cls,class_name,calss_bases,class_attrs)
table_name = class_attrs.get(\'table_name\',class_name)
primary_key = None
mappings = {}
for k,v in class_attrs.items():
if isinstance(v,Field):
mappings[k] = v
if v.primary:
if primary_key:
raise TypeError(\'主键重复\')
primary_key = v.name
for k in mappings.keys():
class_attrs.pop(k)
if not primary_key:
raise TypeError(\'必须要有一个主键\')
class_attrs[\'table_name\'] = table_name
class_attrs[\'primary_key\'] = primary_key
class_attrs[\'mappings\'] = mappings
return type.__new__(cls,class_name,calss_bases,class_attrs)
五、功能分析
# 注意!!!!
- 3.接收客户端的第一次请求，判断客户端是哪一个功能
- 4.去调用对应的接口
- 5.服务端在接口层查询用户是否存在
- 6.若用户存在，则返回用户已存在给客户端
- 7.否则，写入数据库，并返回注册成功!
- 登录
- 登录成功:
- 在服务端:
- 1.生成一个 {addr: [session值 + 用户id]}
- 2.将session返回给客户端
- 在客户端:
- 1.判断登录成功后
- 2.在客户端记录session ---》 存放在user_info字典中的cookies值
- cookies与session与token： 面试题，cookies与session与token的区别;
- cookies:是存放在浏览器客户端的用户信息;
- 优点:
可以将数据存在客户端一方;
- 缺点:
不安全
- session:是存在服务端的用户信息;
- 优点:
数据安全
- 缺点:
session值不是唯一的，可以存放多份，导致服务端占用空间过大!
- 上传电影
- 1.客户端先打印所有可上传的电影
- 2.让用户选择需要上传的电影
- 3.获取电影的md5值,并发送给服务端，让服务端校验电影的md5值是否存在，若存在证明电影已存在!
- 4.若电影不存在，则返回消息给客户端
- 5.客户端开始发送上传电影功能字典，并发送电影数据;
- 6.服务端先接收上传电影功能字典，调用对应的接口，并且接收客户端上传的电影数据，还要写入数据库中;
- 删除电影
- 1.客户端去服务端查看是否有可删除的电影，若有服务端则返回所有可删除的电影;
- 2.客户端开始选择可删除的电影，并将请求发送给服务端;
- 3.服务端接收到删除电影的请求后，获取电影对象，并修改电影的is_delete字段为1;
- 发布公告
- 1.客户端直接输入公告标题与公告内容，然后发送给服务端;
- 2.服务端将公告信息插入数据库中的公告表中;
- 登录认证装饰器:
- 1.将被装饰对象中的所有参数(back_dic, conn)，传给了装饰器中的inner;
- 2.先通过back_dic获取用户的addr，根据addr获取服务端中的 [session+user_id]值;
- 3.若[session + id]值存在，获取客户端传过来的cookies(session值) 与 服务端的session做比对;
- 4.若比对成功,则将用户的user_id 添加到back_dic中 ---》 args[0], 被装饰对象功能原路返回;
- 5.若比对不成功, 则组织一个字典返回给客户端;
六、数据库表创建
七、项目开发--仿快播视频
服务端client
start.py ---- 启动文件
import os
import sys
from core import src
# 将启动文件绝对路径加入到环境变量
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
if __name__ == \'__main__\':
src.run()
conf --setting.py-------项目配置
import os
ip = \'127.0.0.1\'
port = 8080
BASE_PATH = os.path.dirname(os.path.dirname(__file__))
UPLOAD_MOVIE_PATH = os.path.join(BASE_PATH, \'upload_movies\')
DOWNLOAD_MOVIE_PATH = os.path.join(BASE_PATH, \'download_movies\')
core --- src.py --- 首页视图
from core import admin
from core import user
func_dict = {
\'1\': admin.admin_view,
\'2\': user.user_view
}
def run():
while True:
print(\'---欢迎辜氏电影家族---\')
print(\'\'\'
优酷系统：
1.管理员视图
2.普通用户视图
q.退出
\'\'\')
choice = input(\'请选择功能：\').strip()
if choice == \'q\':
break
if choice not in func_dict:
print(\'选择的功能有误！请重新选择！\')
continue
func_dict[choice]()
core --- admin.py --- 管理员视图
from socket_cilent import tcp_cilent
from lib import common
import os
from conf import setting
user_info = {\'cookies\': None}
# 管理员注册
def register(client):
while True:
username = input(\'请输入用户名：\').strip()
password = input(\'请输入密码：\').strip()
re_password = input(\'请确认密码：\').strip()
if not password == re_password:
print(\'密码不一致，重新输入！\')
continue
send_dict = {
\'type\': \'register\',
\'username\': username,
\'password\': password,
\'user_type\': \'admin\'
}
recv_dict = common.send_and_recv(send_dict, client)
if recv_dict.get(\'flag\'):
print(recv_dict.get(\'msg\'))
user_info[\'cookies\'] = recv_dict.get(\'session\')
break
else:
print(recv_dict.get(\'msg\'))
# 管理员登录
def login(client):
while True:
username = input(\'请输入用户名：\').strip()
password = input(\'请输入密码：\').strip()
send_dict = {
\'type\': \'login\',
\'username\': username,
\'password\': password,
\'user_type\': \'admin\'
}
recv_dict = common.send_and_recv(send_dict, client)
if recv_dict.get(\'flag\'):
print(recv_dict.get(\'msg\'))
# 登录成功后设置session值
user_info[\'cookies\'] = recv_dict.get(\'session\')
break
else:
print(recv_dict.get(\'msg\'))
# 上传电影
def unload_movie(client):
while True:
# 获取上传电影目录中的所有电影，并选择
movie_list = common.get_movie_list()
# 若没有可上传的电影
if not movie_list:
print(\'没有可以上传的电影，请先后台添加后上传！\')
break
for index, movie_name in enumerate(movie_list):
print(index, movie_name)
choice = input(\'请选择要上传的电影编号(q.退出)：\').strip()
if choice == \'q\':
break
if not choice.isdigit():
print(\'请输入数字！\')
continue
choice = int(choice)
if choice not in range(len(movie_list)):
print(\'输入的不在范围，重新输入！\')
continue
# 电影名字
movie_name = movie_list[choice]
# 电影路径
movie_path = os.path.join(setting.UPLOAD_MOVIE_PATH, movie_name)
# 获取电影大小
movie_size = os.path.getsize(movie_path)
# 获取电影md5值
movie_md5 = common.get_movie_md5(movie_path)
# 校验电影是否存在
send_dict = {
\'type\': \'check_movie\',
\'cookies\': user_info.get(\'cookies\'),
\'movie_md5\': movie_md5,
}
# 获取电影文件校验后的结果
back_dict = common.send_and_recv(send_dict, client)
if not back_dict.get(\'flag\'):
print(back_dict.get(\'msg\'))
continue
# 确认电影是否免费
is_free = input(\'y/n 免费/收费\').strip()
number = 1
if is_free == \'y\':
number = 0
# 电影不存在， 发送上传电影请求
send_dict = {
\'type\': \'upload_movie\',
\'cookies\': user_info.get(\'cookies\'),
\'movie_md5\': movie_md5,
\'movie_name\': movie_name,
\'movie_size\': movie_size,
\'is_free\': number
}
recv_dict = common.send_and_recv(send_dict, client, file=movie_path)
if recv_dict.get(\'flag\'):
print(recv_dict.get(\'msg\'))
break
# 删除电影
def delete_movie(client):
while True:
print(\'---删除电影---\')
# 1.获取服务端可以删除的电影
send_dict = {
\'type\': \'get_movie_list\',
\'cookies\': user_info.get(\'cookies\'),
\'movie_type\': \'all\'
}
recv_dict = common.send_and_recv(send_dict, client)
if not recv_dict.get(\'flag\'):
print(recv_dict.get(\'msg\'))
break
# 2.打印可删除的电影
# movie_list --> [[电影名，ID，收费免费],[],[]]
movie_list = recv_dict.get(\'movie_list\')
for index, movie_name in enumerate(movie_list):
print(index, movie_name)
choice = input(\'请输入你要删除的电影编号（q:退出）：\').strip()
if choice == \'q\':
break
if not choice.isdigit():
print(\'请输入数字！\')
continue
choice = int(choice)
if choice not in range(len(movie_list)):
print(\'输入不在范围！\')
continue
# 获取电影的id，因为名字可能一样，但是数据库中id是唯一的
movie_name_id = movie_list[choice][1]
send_dict = {
\'type\': \'delete_movie\',
\'cookies\': user_info.get(\'cookies\'),
\'movie_id\': movie_name_id
}
recv_dict2 = common.send_and_recv(send_dict, client)
if recv_dict2.get(\'flag\'):
print(recv_dict2.get(\'msg\'))
break
# 发布公告
def send_notice(client):
while True:
title = input(\'请输入公告标题(15字以内):\').strip()
if len(title)>15:
continue
content = input(\'输入公告内容(100字以内)：\')
if len(content)>100:
continue
send_dict = {
\'type\': \'send_notice\',
\'cookies\': user_info.get(\'cookies\'),
\'title\': title,
\'content\': content
}
recv_dict = common.send_and_recv(send_dict, client)
if recv_dict.get(\'flag\'):
print(recv_dict.get(\'msg\'))
break
else:
print(recv_dict.get(\'msg\'))
break
func_dict = {
\'1\': register,
\'2\': login,
\'3\': unload_movie,
\'4\': delete_movie,
\'5\': send_notice,
}
def admin_view():
client = tcp_cilent.get_client()
while True:
print(\'---管理员---\')
print(\'\'\'
1.注册
2.登录
3.上传视频
4.删除视频
5.发布公告
q.退出
\'\'\')
choice = input(\'请选择功能：\').strip()
if choice == \'q\':
break
if choice not in func_dict:
print(\'选择的功能有误！请重新选择！\')
continue
func_dict[choice](client)
core --- user.py --- 用户视图
from socket_cilent import tcp_cilent
from lib import common
user_info = {\'cookies\': None}
# 用户注册
def register(client):
while True:
username = input(\'请输入用户名：\').strip()
password = input(\'请输入密码：\').strip()
re_password = input(\'请确认密码：\').strip()
if not password == re_password:
print(\'密码不一致，重新输入！\')
continue
send_dict = {
\'type\': \'register\',
\'username\': username,
\'password\': password,
\'user_type\': \'user\'
}
recv_dict = common.send_and_recv(send_dict, client)
if recv_dict.get(\'flag\'):
print(recv_dict.get(\'msg\'))
user_info[\'cookies\'] = recv_dict.get(\'session\')
break
else:
print(recv_dict.get(\'msg\'))
# 用户登录
def login(client):
while True:
username = input(\'请输入用户名：\').strip()
password = input(\'请输入密码：\').strip()
send_dict = {
\'type\': \'login\',
\'username\': username,
\'password\': password,
\'user_type\': \'user\'
}
recv_dict = common.send_and_recv(send_dict, client)
if recv_dict.get(\'flag\'):
print(recv_dict.get(\'msg\'))
# 登录成功后设置session值
user_info[\'cookies\'] = recv_dict.get(\'session\')
break
else:
print(recv_dict.get(\'msg\'))
# 查看电影
def check_movies(client):
print(\'---查看电影---\')
send_dict = {\'type\': \'get_movie_list\',
\'cookies\': user_info.get(\'cookies\'),
\'movie_type\': \'all\'
}
recv_dict = common.send_and_recv(send_dict, client)
if recv_dict.get(\'flag\'):
movie_list = recv_dict.get(\'movie_list\')
for index, move in enumerate(movie_list):
print(index, move)
else:
print(recv_dict.get(\'msg\'))
# 充值会员
def charge_vip(client):
while True:
print(\'---充值会员---\')
is_vip = input(\'确认充值VIP(y/n)？:\').strip()
if is_vip == \'n\':
break
elif is_vip == \'y\':
send_dict = {
\'type\': \'charge_vip\',
\'cookies\': user_info.get(\'cookies\'),
}
recv_dict = common.send_and_recv(send_dict, client)
else:
print(\'输入不规范！\')
continue
if recv_dict.get(\'flag\'):
print(recv_dict.get(\'msg\'))
break
# 下载免费电影
def download_free_movie(client):
common.download_movie(client, is_pay=\'free\')
# 下载收费电影
def download_charge_movie(client):
common.download_movie(client, is_pay=\'pay\')
# 查看下载记录
def check_download_record(client):
send_dict = {
\'type\': \'check_record\',
\'cookies\': user_info.get(\'cookies\')
}
recv_dict = common.send_and_recv(send_dict, client)
# 判断有无下载记录
if recv_dict.get(\'flag\'):
record_list = recv_dict.get(\'record_list\')
for line in record_list:
print(line)
else:
print(recv_dict.get(\'msg\'))
# 查看公告
def check_notice(client):
send_dict = {
\'type\': \'check_notice\',
\'cookies\': user_info.get(\'cookies\')
}
recv_dict = common.send_and_recv(send_dict, client)
# 判断有无公告！
if recv_dict.get(\'flag\'):
notice_list = recv_dict.get(\'notice_list\') # -->[[],[],[]]
for line in notice_list:
print(line)
else:
print(recv_dict.get(\'msg\'))
func_dict = {
\'1\': register,
\'2\': login,
\'3\': check_movies,
\'4\': charge_vip,
\'5\': download_free_movie,
\'6\': download_charge_movie,
\'7\': check_download_record,
\'8\': check_notice
}
def user_view():
client = tcp_cilent.get_client()
while True:
print(\'---用户---\')
print(\'\'\'
- 1.注册
- 2.登录
- 3.查看视频
- 4.充值会员
- 5.下载免费电影
- 6.下载收费电影
- 7.查看下载记录
- 8.查看公告
\'\'\')
choice = input(\'请选择功能：\').strip()
if choice == \'q\':
break
if choice not in func_dict:
print(\'选择的功能有误！请重新选择！\')
continue
func_dict[choice](client)
lib --- common.py --- 公共功能
import json
import struct
import os
from conf import setting
import hashlib
from core.user import user_info
# 客户端发送并接收消息
def send_and_recv(send_dict, client, file=None):
\'\'\'
:param send_dict:用户信息字典
:param client:拿到socket客户端对象
:param file: 电影路径，证明有没有上传的文件。比如注册没有上传文件，只有用户字典信息
:return:recv_dict 解码之后的真实字典数据
\'\'\'
# 客户端王服务端发送数据
# 1.字典转成json格式
json_dict = json.dumps(send_dict).encode(\'utf-8\')
# 2.制作报头
hander = struct.pack(\'i\', len(json_dict))
# 3.发送报头
client.send(hander)
# 4.发送json字典
client.send(json_dict)
# 判断是否有电影文件
if file:
with open(file, \'rb\')as f:
# 循环发送
for line in f:
client.send(line)
# 客户端接收服务端返回的数据
# 1.接收报头
hander = client.recv(4)
# 解析，得到字典长度
dict_len = struct.unpack(\'i\', hander)[0]
# 接收json格式的字典
dict_json = client.recv(dict_len)
# 解开json,解码二进制，得到原始真正字典
recv_dict = json.loads(dict_json.decode(\'utf-8\'))
return recv_dict
# 获取电影列表
def get_movie_list():
movie_list = os.listdir(setting.UPLOAD_MOVIE_PATH)
return movie_list
# 获取电影md5值
def get_movie_md5(movie_path):
md5 = hashlib.md5()
# 获取电影大小，用来对电影的数据进行截取
movie_size = os.path.getsize(movie_path)
# 截取部分的位置
# [电影开头，电影1/4，电影1/2，电影结尾]
bytes_list = [0, movie_size//4, movie_size//2, movie_size -10]
with open(movie_path, \'rb\')as f:
for line in bytes_list:
# 光标移动到指定位置
f.seek(line)
# 每个位置获取10个bytes
data = f.read(10)
# 生成md5值，加密
md5.update(data)
return md5.hexdigest()
# 装饰器
def outter(func):
from core.admin import user_info
from core.user import user_info
def inner(*args, **kwargs):
if user_info.get(\'cookies\'):
return func(*args, **kwargs)
else:
from core import admin
from socket_cilent import tcp_cilent
print(\'请先登录！\')
admin.login(tcp_cilent.get_client())
return inner
# 下载免费收费电影
def download_movie(client, is_pay):
while True:
# 获取所有免费电影列表
send_dict = {
\'type\': \'get_movie_list\',
\'cookies\': user_info.get(\'cookies\'),
# 获取电影的类型
\'movie_type\': is_pay
}
recv_dict = send_and_recv(send_dict, client)
free_movie_list = recv_dict.get(\'movie_list\')
# 判断是否有免费电影
if free_movie_list:
for index, movie_list in enumerate(free_movie_list):
print(index, movie_list)
# 用户选择
choice = input(\'输入下载的电影编号(q:退出)：\').strip()
if choice == \'q\':
break
if not choice.isdigit():
print(\'输入不规范！\')
continue
choice = int(choice)
if choice not in range(len(free_movie_list)):
print(\'输入不在范围！\')
continue
# 获取选择的电影id [名字，id,免费]
# free_movie_list = [[\'在线发牌——2019-12-22 20:46:57.821142\', 15, \'免费\']]
movie_list_id = free_movie_list[choice][1]
send_dict = {
\'type\': \'download_movie\',
\'cookies\': user_info.get(\'cookies\'),
\'movie_id\': movie_list_id
}
# 发送
recv_dict = send_and_recv(send_dict, client)
if recv_dict.get(\'flag\'):
# 获取电影名称
movie_name = recv_dict.get(\'movie_name\')
# 获取电影大小
movie_size = recv_dict.get(\'movie_size\')
# 拼接下载电影存放目录
movie_path = os.path.join(setting.DOWNLOAD_MOVIE_PATH, movie_name)
# 开始接收数据
recv_data = 0
with open(movie_path, \'wb\')as f:
while recv_data < int(movie_size):
data = client.recv(1024)
f.write(data)
recv_data += len(data)
print(f\'电影【{movie_name}】下载成功！\')
break
else:
print(\'没有可下载的电影！\')
break
else:
print(\'没有免费电影！\')
break
socket_client --- tcp_cilent.py --- 套接字对象
import socket
from conf import setting
def get_client():
client = socket.socket()
client.connect((setting.ip, setting.port))
return client
upload_movies --- 上传文件目录
里面放要上传的电影文件
download_movies --- 下载文件目录
用户下载的电影自动放在里面
服务端server
start.py --- 服务端启动文件
from socket_server import tcp_server
import os
import sys
sys.path.append(os.path.dirname(__file__))
if __name__ == \'__main__\':
tcp_server.run()
conf --- setting.py --- 项目配置文件
import os
ip = \'127.0.0.1\'
port = 8080
BASE_PATH = os.path.dirname(os.path.dirname(__file__))
MOVIE_FILES_PATH = os.path.join(BASE_PATH, \'movie_files\')
orm_control --- mysql_control.py --- 数据库配置
import pymysql
class MySQL:
__instance = None
# 单例模式
@classmethod
def singleton(cls):
if not cls.__instance:
cls.__instance = cls()
return cls.__instance
def __init__(self):
self.mysql_client = pymysql.connect(
host=\'127.0.0.1\',
port=3306,
user=\'root\',
passwd=\'123\',
db=\'youku\',
charset=\'utf8\',
autocommit=True
)
# 游标对象，以字典形式返回
self.cursor = self.mysql_client.cursor(pymysql.cursors.DictCursor)
# 自定义查询方法
def select(self, sql, args=None):
# 提交sql语句
# select * from table where id = %s
self.cursor.execute(sql, args)
# 获取返回值,[{},{}]字典套列表，每一个字典是数据库每一行记录
res = self.cursor.fetchall()
return res
# 自定义提交sql语句方法
def execute(self, sql, args):
try:
self.cursor.execute(sql, args)
except Exception as e:
print(e)
# 自定义关闭
def close(self):
# 先关闭游标，再关闭数据库连接
self.cursor.close()
self.mysql_client.close()
orm_control --- ORM.py --- 对象关系映射
from orm_control.mysql_control import MySQL
class Field:
def __init__(self, name, column_type, primary_key, default):
self.name = name
self.column_type = column_type
self.primary_key = primary_key
self.default = default
class String(Field):
def __init__(self, name, column_type=\'varchar(64)\', primary_key=False, default=None):
super().__init__(name, column_type, primary_key, default)
class Integer(Field):
def __init__(self, name, column_type=\'int\', primary_key=False, default=None):
super().__init__(name, column_type, primary_key, default)
class OrmMetaClass(type):
\'\'\'
1.控制有且只有一个主键
2.给类的名称空间添加表名，主键名、字段对象
3.一张表必须要有表名
\'\'\'
def __new__(cls, class_name, class_bases, class_dict):
# 过滤Models类，不能控制Models
if class_name == \'Models\':
return type.__new__(cls, class_name, class_bases, class_dict)
table_name = class_dict.get(\'table_name\', class_name)
primary_key = None
mappings = {} # 字段名，与字段对象字典
for key, value in class_dict.items():
# 过滤非字段
if isinstance(value, Field):
mappings[key] = value
# 下面判断一个表有且只有一个主键
if value.primary_key:
if primary_key:
raise TypeError(\'一张表只能有一个主键！\')
primary_key = value.name
if not primary_key:
raise TypeError(\'必须有一个主键！\')
# 过滤掉类名称空间中重复的字段
for key in mappings.keys():
class_dict.pop(key)
# 给类的名称空间添加表名，主键、字段名，与字段对象字典
class_dict[\'table_name\'] = table_name
class_dict[\'primary_key\'] = primary_key
class_dict[\'mappings\'] = mappings
return type.__new__(cls, class_name, class_bases, class_dict)
class Models(dict, metaclass=OrmMetaClass):
# 对象点属性没有的时候触发，让对象可以点到属性
def __getattr__(self, item):
return self.get(item)
# 赋值的时候触发
def __setattr__(self, key, value):
self[key] = value
# 查询方法
@classmethod
def orm_select(cls, **kwargs):
# 调用MySQL生成游标对象
mysql = MySQL()
if not kwargs:
sql = \'select * from %s\'% cls.table_name
res = mysql.select(sql)
else:
# kwargs是一个字典，所有这里的key是第一个字段名，id，value是id对应的值
key = list(kwargs.keys())[0]
value = kwargs.get(key)
# 条件查询
sql = \'select * from %s where %s=?\'% (cls.table_name, key)
sql = sql.replace(\'?\', \'%s\')
# 用游标对象提交--->接收返回值为[{}]字典套列表
res = mysql.select(sql, value)
# 列表生成式
return [cls(**d) for d in res]
# 插入方法
def orm_insert(self):
# 拿到游标对象
mysql = MySQL()
# 除主键字段名
keys = []
# 除主键字段值
values = []
# 存方？号，有几个字段就存几个问号。替换和%s占位
args = []
for key, value in self.mappings.items():
# 过滤掉主键，主键自增
if not value.primary_key:
keys.append(value.name)
# 存主键以外的字段值，若值没有，则使用默认值
values.append(getattr(self, value.name, value.default))
args.append(\'?\')
# insert into table_name (v1,v2,v3)values(?,?,?)
sql = \'insert into %s(%s) values(%s)\'%(
self.table_name,
\',\'.join(keys),
\',\'.join(args)
)
# sql: insert into table_name(v1, v2, v3) values(%s, %s, %s)
sql = sql.replace(\'?\', \'%s\')
mysql.execute(sql, values)
# 更新方法
def orm_update(self):
# 拿到游标对象
mysql = MySQL()
# 字段名
keys = []
# 字段值
values = []
# 主键
primary_key = None
for key, value in self.mappings.items():
if value.primary_key:
primary_key = value.name + \'= %s\' % getattr(self, value.name)
else:
keys.append(value.name + \'=?\')
values.append(getattr(self, value.name))
sql = \'update %s set %s where %s\'%(
self.table_name,
\',\'.join(keys),
primary_key
)
sql = sql.replace(\'?\', \'%s\')
mysql.execute(sql, values)
socket_server --- tcp_server.py --- 连接服务端、套接字
import socket
from conf import setting
from concurrent.futures import ThreadPoolExecutor
import struct
import json
from interface import common_interface, admin_interface, user_interface
server = socket.socket()
server.bind((setting.ip, setting.port))
server.listen(5)
pool = ThreadPoolExecutor(10)
func_dict = {
# 注册接口
\'register\': common_interface.register_interface,
# 登录接口
\'login\': common_interface.login_interface,
# 检测电影是否存在接口
\'check_movie\': common_interface.check_movie_interface,
# 上传接口
\'upload_movie\': admin_interface.upload_movie_interface,
# 获取电影列表接口(未删除全部，未删除免费，未删除收费)
\'get_movie_list\': common_interface.get_movie_list_interface,
# 删除电影接口
\'delete_movie\': admin_interface.delete_movie_interface,
# 发布公告接口
\'send_notice\': admin_interface.send_notice_interface,
# 购买会员接口
\'charge_vip\': user_interface.charge_vip_interface,
# 下载电影接口
\'download_movie\': user_interface.download_movie_interface,
# 查看下载记录
\'check_record\': user_interface.check_record_interface,
# 查看公告接口
\'check_notice\': user_interface.check_notice_interface,
}
def run():
print(\'---启动服务端---\')
while True:
conn, addr = server.accept() # 允许连接
print(\'连接成功！\')
pool.submit(working, conn, addr) # 异步提交
# 异步回调函数
def working(conn, addr):
while True:
try:
# 1.接收报头
hander = conn.recv(4)
# 2.解开字典，获取字典长度
dict_len = struct.unpack(\'i\', hander)[0]
# 3.接收真实json格式字典数据的长度
json_dict = conn.recv(dict_len)
# 4.解开json,解开二进制,获取原始真实字典
recv_dict = json.loads(json_dict.decode(\'utf-8\'))
# 5.把addr添加到字典中
recv_dict[\'addr\'] = str(addr)
# 调用任务分发函数：对任务进行分发
dispacher(recv_dict, conn)
except Exception as e:
print(e)
conn.close()
break
# 任务分发
def dispacher(recv_dict, conn):
type = recv_dict.get(\'type\') # 获取字典传过来的功能函数名字
if type in func_dict:
func_dict.get(type)(recv_dict, conn)
db --- models.py --- 所有表类
from orm_control.ORM import Models, Integer, String
# 用户表类:
class User(Models):
\'\'\'
u_id :用户id
username :用户名
password :用户密码
user_type :用户类型（管理员、普通）
is_vip :是否VIP 0表示普通用户，1表示VIP
register_time :注册时间
\'\'\'
u_id = Integer(name=\'u_id\', primary_key=True)
username = String(name=\'username\')
password = String(name=\'password\')
user_type = String(name=\'user_type\')
is_vip = Integer(name=\'is_vip\')
register_time = String(name=\'register_time\')
# 电影表类:
class Movie(Models):
\'\'\'
m_id :电影id
movie_name ：电影名字
movie_size ：电影大小
movie_path ：电影路径
is_free ：是否免费 0：免费 1 收费
user_id ：上传用户Id
movie_md5 ：电影唯一标识
upload_time ：上传时间
is_delete ：是否删除 0 未删除 1删除
\'\'\'
m_id = Integer(name=\'m_id\', primary_key=True)
movie_name = String(name=\'movie_name\')
movie_size = String(name=\'movie_size\')
movie_path = String(name=\'movie_path\')
is_free = Integer(name=\'is_free\')
user_id = Integer(name=\'user_id\')
movie_md5 = String(name=\'movie_md5\')
upload_time = String(name=\'upload_time\')
is_delete = Integer(name=\'is_delete\')
# 公告表类:
class Notice(Models):
\'\'\'
n_id :id
title :公告标题
content ： 公告内容
create_time ：发布时间
u_id ：发布用户id
\'\'\'
n_id = Integer(name=\'n_id\', primary_key=True)
title = String(name=\'title\')
content = String(name=\'content\')
create_time = String(name=\'create_time\')
user_id = Integer(name=\'user_id\')
# 下载记录表:
class DownloadRecord(Models):
\'\'\'
d_id :id
movie_id :下载的电影id
user_id :下载用户的id
download_time :下载时间
\'\'\'
d_id = Integer(name=\'d_id\', primary_key=True)
movie_id = Integer(name=\'movie_id\')
user_id = Integer(name=\'user_id\')
download_time = String(name=\'download_time\')
db --- session_data.py --- 存方服务端session值
# 给session字典 {地址：[session,用户id]}这样session值就是唯一
# {"(\'127.0.0.1\', 10638)": [\'5554b74623f210e35d2c9fd0de4a05ff\', 19]}
session_dict = {}
interface --- admin_interface.py --- 管理员接口
from conf import setting
import os
import datetime
from db.models import Movie, Notice
from lib import common
# 上传电影接口
@common.login_auth
def upload_movie_interface(recv_dict, conn):
# 服务端获取电影名字
movie_name = recv_dict.get(\'movie_name\')
# time = datetime.datetime.now()
movie_path = os.path.join(
setting.MOVIE_FILES_PATH, movie_name
)
# 电影后面拼接时间,文件名字不能有冒号。命名规范
time = str(datetime.datetime.now()).replace(\':\', \'-\')
movie_name = f\'{movie_name}——{time}\'
# 服务端获取电影的大小
movie_size = recv_dict.get(\'movie_size\')
recv_data = 0
# 开始接收电影文件数据
with open(movie_path, \'wb\') as f:
while recv_data < movie_size:
data = conn.recv(1024)
f.write(data)
recv_data += len(data)
# 保存电影数据到MySQL中
movie_obj = Movie(
movie_name=movie_name,
movie_size=movie_size,
movie_path=movie_path,
is_free=recv_dict.get(\'is_free\'),
user_id=recv_dict.get(\'user_id\'),
# 用来校验电影是否已存在
movie_md5=recv_dict.get(\'movie_md5\'),
upload_time=datetime.datetime.now(),
is_delete=0
)
movie_obj.orm_insert()
send_dic = {\'flag\': True, \'msg\': f\'电影[{movie_name}]上传成功\'}
common.send_dict(send_dic, conn)
# 删除电影接口
@common.login_auth
def delete_movie_interface(recv_dict, conn):
movie_id = recv_dict.get(\'movie_id\')
movie_obj = Movie.orm_select(m_id=movie_id)[0] # [obj]
movie_obj.is_delete = 1 # 0未删除，1删除
# 更新数据
movie_obj.orm_update()
# 发送反馈字典
send_dict = {\'flag\': True,
\'msg\': f\'电影【{movie_obj.movie_name}】删除成功！\'}
common.send_dict(send_dict, conn)
# 发布公告接口
@common.login_auth
def send_notice_interface(recv_dict, conn):
notice_obj = Notice(
title=recv_dict.get(\'title\'),
content=recv_dict.get(\'content\'),
create_time=str(datetime.datetime.now()),
user_id=recv_dict.get(\'user_id\'),
)
# 插入数据库
notice_obj.orm_insert()
# 发送反馈信息给客户端
send_dict = {\'flag\': True, \'msg\': f\'【{notice_obj.title}】公告发布成功！\'}
common.send_dict(send_dict, conn)
interface --- user_interface.py
from lib import common
from db import models
import datetime
# 购买会员接口
@common.login_auth
def charge_vip_interface(recv_dict, conn):
# 对当前用户的Is_vip字段修改为1 0：普通用户，1：会员
user_id = recv_dict.get(\'user_id\')
user_obj = models.User.orm_select(u_id=user_id)[0]
user_obj.is_vip = 1
# 更新
user_obj.orm_update()
send_dict = {\'flag\': True, \'msg\': \'会员充值成功！\'}
# 发送
common.send_dict(send_dict, conn)
# 下载电影接口
@common.login_auth
def download_movie_interface(recv_dict, conn):
movie_id = recv_dict.get(\'movie_id\')
movie_obj = models.Movie.orm_select(m_id=movie_id)[0]
# 获取当前id的电影名称
movie_name = movie_obj.movie_name
# 获取当前电影大小
movie_size = movie_obj.movie_size
# 获取服务器电影路径
movie_path = movie_obj.movie_path
# 把数据做成字典，发送给客户端
send_dict = {
\'flag\': True,
\'movie_name\': movie_name,
\'movie_size\': movie_size
}
common.send_dict(send_dict, conn)
# 开始发送电影,流水发送
with open(movie_path, \'rb\')as f:
for line in f:
conn.send(line)
# 记录下载的电影
record_obj = models.DownloadRecord(
movie_id=movie_id,
user_id=recv_dict.get(\'user_id\'),
download_time=datetime.datetime.now(),
)
# 插入数据
record_obj.orm_insert()
# 查看下载记录接口
@common.login_auth
def check_record_interface(recv_dict, conn):
\'\'\'
只能查看自己的下载记录
1.获取当前用户的id
2.查询当前用户id的记录
\'\'\'
user_id = recv_dict.get(\'user_id\')
# 查询--->[obj,obj,obj]
record_list_obj = models.DownloadRecord.orm_select(user_id=user_id)
# 如果没有记录
if not record_list_obj:
send_dic = {
\'flag\': False,
\'msg\': \'下载记录为空！\'
}
else:
record_list = []
for record_obj in record_list_obj:
# 因为记录表中没有电影名字，所有1.拿到电影id，2.去电影表中拿到名字
# 1.获取id
movie_id = record_obj.movie_id
# 2.根据id去电影表中获取电影对象
movie_obj = models.Movie.orm_select(m_id=movie_id)[0]
# 3.根据对象拿到电影名字
movie_name = movie_obj.movie_name
# 拿到下载记录表中的下载时间
download_time = record_obj.download_time
# 把需要展示给用户的信息添加到列表中
record_list.append(
[movie_name, str(download_time)]
)
send_dic = {
\'flag\': True,
\'record_list\': record_list
}
common.send_dict(send_dic, conn)
# @common.login_auth
def check_notice_interface(recv_dict, conn):
notice_list_obj = models.Notice.orm_select()
notice_list = []
for notice_obj in notice_list_obj:
title = notice_obj.title
content = notice_obj.content
create_time = notice_obj.create_time
notice_list.append(
[title, content, create_time]
)
if notice_list:
send_dict = {\'flag\': True, \'notice_list\': notice_list}
else:
send_dict = {\'flag\': False, \'msg\': \'还没有公告！\'}
common.send_dict(send_dict, conn)
interface --- common_interface.py --- 公共接口功能
from db import models
import datetime
from lib import common
from threading import Lock
from db.session_data import session_dict
# 公共锁
mutex = Lock()
# 注册
def register_interface(recv_dict, conn):
username = recv_dict.get(\'username\')
# 1.判断用户是否存在 ---接收到--->[{},{}] 列表套字典
user_obj_list = models.User.orm_select(username=username)
# 2.有值--->用户存在
if user_obj_list:
send_dict = {\'flag\': False, \'msg\': \'用户已存在\'}
# 3.没有值就注册
else:
user_obj = models.User(
username=username,
password=common.md5(recv_dict.get(\'password\')),
user_type=recv_dict.get(\'user_type\'),
is_vip=0,
register_time=datetime.datetime.now()
)
user_obj.orm_insert()
send_dict = {\'flag\': True, \'msg\': \'注册成功！\'}
print(f\'【{username}】用户注册成功！\')
# 调用公共方法发送数据到客户端
common.send_dict(send_dict, conn)
# 登录
def login_interface(recv_dict, conn):
# 1.判断用户是否存在
username = recv_dict.get(\'username\')
# 1.判断用户是否存在 ---接收到--->[{}] 列表套字典
user_obj_list = models.User.orm_select(username=username)
# 2.没有值--->用户不存在
if not user_obj_list:
send_dict = {\'flag\': False, \'msg\': \'用户不存在\'}
# 1.用户存在的情况
else:
user_obj = user_obj_list[0]
# 2.判断密码是否正确
re_password = recv_dict.get(\'password\') # 用户输入的密码
re_password = common.md5(re_password)
mysql_password = user_obj.password # 数据库存的真正密码
# 密码正确
if re_password == mysql_password:
addr = recv_dict.get(\'addr\')
mutex.acquire() # 加锁
# 用户登录成功后，获取session值
session = common.get_session()
# 给session字典添加值 {地址：[session,用户id]}这样session值就是唯一
session_dict[addr] = [session, user_obj.u_id]
mutex.release() # 释放锁
# 默认不是VIP
is_vip = False
if user_obj.is_vip:
is_vip = True
send_dict = {\'flag\': True,
\'msg\': \'登录成功！\',
\'session\': session,
\'is_vip\': is_vip}
# 密码不正确
else:
send_dict = {\'flag\': False, \'msg\': \'密码错误！\'}
# 调用接口发送反馈信息字典
common.send_dict(send_dict, conn)
# 检测电影是否存在接口
@common.login_auth
def check_movie_interface(recv_dict, conn):
movie_md5 = recv_dict.get(\'movie_md5\')
# 校验数据库中电影是否存在--->判断该md5是否存在
movie_obj_list = models.Movie.orm_select(movie_md5=movie_md5)
# 若电影不存在，则返回可以上传
if not movie_obj_list:
send_dict = {\'flag\': True, \'msg\': \'可以上传\'}
# 若电影存在，则返回不可以上传
else:
send_dict = {\'flag\': False, \'msg\': \'电影已存在！\'}
# 调接口，发送
common.send_dict(send_dict, conn)
# 获取相应电影接口
@common.login_auth
def get_movie_list_interface(recv_dict, conn):
# 查询电影表，获取所有电影对象
movie_obj_list = models.Movie.orm_select()
movie_list = []
if movie_obj_list:
for movie_obj in movie_obj_list:
# 过滤已删除的电影对象
if movie_obj.is_delete == 0:
# 获取所有电影
if recv_dict.get(\'movie_type\') == \'all\':
# 电影名称， 电影ID，电影是否免费
movie_list.append(
[movie_obj.movie_name, movie_obj.m_id,
\'免费\' if movie_obj.is_free == 0 else \'收费\']
)
elif recv_dict.get(\'movie_type\') == \'free\':
if movie_obj.is_free == 0:
movie_list.append(
[movie_obj.movie_name,
movie_obj.m_id,
\'免费\'])
elif recv_dict.get(\'movie_type\') == \'pay\':
if movie_obj.is_free == 1:
movie_list.append(
[movie_obj.movie_name,
movie_obj.m_id,
\'收费\'])
if len(movie_list)>0:
send_dict = {\'flag\': True, \'movie_list\': movie_list}
common.send_dict(send_dict, conn)
if not len(movie_list) > 0:
send_dict = {\'flag\': False, \'msg\': \'还没有免费电影!\'}
common.send_dict(send_dict, conn)
else:
send_dict = {\'flag\': False, \'msg\': \'目前还没有电影！\'}
common.send_dict(send_dict, conn)
lib --- common.py --- 公共功能
import json
import struct
import hashlib
import uuid
# 发送消息的功能
def send_dict(send_dict, conn):
# 1.将字典转成json,二进制格式
json_dict = json.dumps(send_dict).encode(\'utf-8\')
# 2.制作报头
hander_dict = struct.pack(\'i\', len(json_dict))
# 3.发送报头
conn.send(hander_dict)
# 4.发送json字典
conn.send(json_dict)
# 加密
def md5(pwd):
md5 = hashlib.md5()
value = \'辜氏家族终极密码\'
md5.update(value.encode(\'utf-8\'))
md5.update(pwd.encode(\'utf-8\'))
return md5.hexdigest()
# 生成session随机字符串
def get_session():
md5 = hashlib.md5()
value = str(uuid.uuid4())
md5.update(value.encode(\'utf-8\'))
return md5.hexdigest()
# 登录认证装饰器
def login_auth(func):
from db.session_data import session_dict
# args-->func-->upload_movie(recv_dict,coon)
def inner(*args, **kwargs):
# 获取客户端的cookies值
client_cookies = args[0].get(\'cookies\') # None 用户没有登录 有值：已登录
# 从back_dict中获取客户端addr，拿到addr去获取存在服务端的session值
addr = args[0].get(\'addr\')
# 判断客户端cookies有值
if client_cookies:
# 在判断客户端cookies和服务端session值是否相等
server_session = session_dict.get(addr)[0] # [0]存的式session值，[1]存的是用户id
if server_session == client_cookies:
# 将服务端存的用户id添加到客户端发过来的字典中
# recv_dict[\'user_id\'] = [session, user_id][1]
args[0][\'user_id\'] = session_dict.get(addr)[1]
return func(*args, **kwargs) # 原路执行
else:
send_dic = {\'flag\': False, \'msg\': \'同一个用户不能多台电脑同时登录！\'}
# 拿到conn
conn = args[1]
# 发送反馈信息到服务端
send_dict(send_dic, conn)
else:
send_dic = {\'flag\': False, \'msg\': \'请先登录！\'}
# 拿到conn
conn = args[1]
# 发送反馈信息到服务端
send_dict(send_dic, conn)
return inner
movie_files --- 服务端存方上传的电影目录
服务端存方上传的电影目录