python-优酷系统管理员视图粗糙版(无详细注释)
Tank-YouKu(仅管理员功能粗糙版)
优酷系统管理员视图功能
- 注册
- 登录
- 上传视频
- 删除视频
- 发布公告
前期准备
先创建好数据库以及各数据表
安装 pymysql 模块 、安装 DBUtils 模块
配置好服务端的 db_pool 中的数据库信息
创库创表语句
手动创建数据库 youku_demo， 需配置数据库编码 utf8 (安装时配置好了命令行中就不用管)
创建数据库：create database youku_demo;
选择该数据库：use youku_demo，然后执行下面的一堆sql代码，或者手动导入
创表代码及测试数据
测试用户：tank 密码： 123
/*
Navicat MySQL Data Transfer
Source Server : localhost-E
Source Server Type : MySQL
Source Server Version : 50645
Source Host : localhost:3306
Source Schema : youku_demo
Target Server Type : MySQL
Target Server Version : 50645
File Encoding : 65001
Date: 28/08/2019 21:22:47
*/
SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
-- ----------------------------
-- Table structure for download_record
-- ----------------------------
DROP TABLE IF EXISTS `download_record`;
CREATE TABLE `download_record` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`user_id` int(11) NULL DEFAULT NULL,
`movie_id` int(11) NULL DEFAULT NULL,
`download_time` datetime(0) NULL DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;
-- ----------------------------
-- Table structure for movie
-- ----------------------------
DROP TABLE IF EXISTS `movie`;
CREATE TABLE `movie` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`path` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`is_free` int(11) NULL DEFAULT NULL,
`file_md5` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`user_id` int(11) NULL DEFAULT NULL,
`is_delete` int(11) NULL DEFAULT NULL,
`upload_time` datetime(0) NULL DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 3 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;
-- ----------------------------
-- Table structure for notice
-- ----------------------------
DROP TABLE IF EXISTS `notice`;
CREATE TABLE `notice` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`title` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`content` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`user_id` int(11) NULL DEFAULT NULL,
`create_time` datetime(0) NULL DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 2 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;
-- ----------------------------
-- Records of notice
-- ----------------------------
INSERT INTO `notice` VALUES (1, \'test1\', \'测试发布公告是否正常\', 1, \'2019-08-28 21:18:38\');
-- ----------------------------
-- Table structure for user
-- ----------------------------
DROP TABLE IF EXISTS `user`;
CREATE TABLE `user` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`pwd` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`is_vip` int(11) NULL DEFAULT NULL,
`is_locked` int(11) NULL DEFAULT NULL,
`user_type` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`register_time` datetime(0) NULL DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 2 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;
-- ----------------------------
-- Records of user
-- ----------------------------
INSERT INTO `user` VALUES (1, \'tank\', \'e23087636dfcd7addf39e32f89e25d44\', 0, 0, \'admin\', \'2019-08-28 21:18:10\');
SET FOREIGN_KEY_CHECKS = 1;
数据库结构
安装pymysql模块
参见博客： MySQL-注释-Navicat基本使用-复杂查询练习题-解题思路-pymysql操作数据库-SQL注入-05 的安装部分
安装DBUtils模块
在命令行输入pip3 install DBUtils 开始安装
配置 db_pool
根据自己电脑的mysql 情况配置
项目架构与数据流向
目录结构
server目录结构
client目录结构
服务端代码
start.py
import os
import sys
from tcp_server.socket_server import SocketServer
sys.path.append(os.path.dirname(__file__))
if __name__ == \'__main__\':
server = SocketServer()
server.run()
tcp_server/socket_server.py
import socket
import struct
import json
from interface import common_interface
from interface import admin_interface
from concurrent.futures import ThreadPoolExecutor
from threading import Lock
from lib import lock_file
lock = Lock()
lock_file.mutex = lock
func_dic = {
\'register\': common_interface.register_interface,
\'login\': common_interface.login_interface,
\'check_movie\': admin_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,
\'put_notice\': admin_interface.put_notice_interface
}
class SocketServer:
def __init__(self):
self.server = socket.socket()
self.server.bind((\'127.0.0.1\', 9527))
self.server.listen(5)
self.pool = ThreadPoolExecutor(50)
def run(self):
print(\'启动服务端...\')
while True:
conn, addr = self.server.accept()
self.pool.submit(self.working, conn, addr)
# 任务分发
def dispatcher(self, client_back_dic, conn):
# # 判断功能的类型
# if client_back_dic.get(\'type\') == \'register\':
# common_interface.register_interface(client_back_dic, conn)
#
# elif client_back_dic.get(\'type\') == \'login\':
# common_interface.login_interface(client_back_dic, conn)
_type = client_back_dic.get(\'type\')
if _type in func_dic: # register
func_dic.get(_type)(client_back_dic, conn)
# 用于执行客户端连接任务
def working(self, conn, addr):
while True:
try:
# 每一个客户端访问服务端都会经过此处
# 此处用于接收客户端传入的数据
headers = conn.recv(4)
data_len = struct.unpack(\'i\', headers)[0]
data_bytes = conn.recv(data_len)
client_back_dic = json.loads(data_bytes.decode(\'utf-8\'))
client_back_dic[\'addr\'] = str(addr)
self.dispatcher(client_back_dic, conn)
except Exception as e:
print(e)
conn.close()
break
interface/common_interface.py
from db import models
from lib import common, lock_file
from db import user_data
def register_interface(client_back_dic, conn):
# 写业务逻辑
# 1.判断用户名是否存在
username = client_back_dic.get(\'username\')
# 通过用户名当作条件查询
user_obj_list = models.User.select(name=username)
# 若存在，给客户端返回数据, 告诉用户，用户已存在!
if user_obj_list:
send_dic = {\'flag\': False, \'msg\': \'用户已存在!\'}
# 若不存在，保存数据到MySQL数据库中， 返回注册成功给客户端
else:
password = client_back_dic.get(\'password\')
user_obj = models.User(
name=username,
# pwd, is_vip, is_locked, user_type, register_time
pwd=common.get_md5_pwd(password),
is_vip=0, # 0表示不是VIP， 1表示VIP
is_locked=0, # 0表示不锁定， 1表示锁定
user_type=client_back_dic.get(\'user_type\'),
register_time=common.get_time())
user_obj.save()
send_dic = {\'flag\': True, \'msg\': \'注册成功\'}
common.send_data(send_dic, conn)
def login_interface(client_back_dic, conn):
username = client_back_dic.get(\'username\')
user_list = models.User.select(name=username)
if not user_list:
send_dic = {\'flag\': False, \'msg\': \'用户不存在\'}
else:
user_obj = user_list[0]
password = client_back_dic.get(\'password\')
# 1.判断客户端传入的密码与数据库中的密码是否相等
if user_obj.pwd == common.get_md5_pwd(password):
# 产生一个随机字符串，作为session值
session = common.get_random_code()
addr = client_back_dic.get(\'addr\')
# 保存session值到服务端，session + user_id一同保存到服务端本地
# 使用锁写入数据
lock_file.mutex.acquire()
user_data.user_online[addr] = [session, user_obj.id]
lock_file.mutex.release()
send_dic = {\'flag\': True, \'msg\': \'登录成功!\', \'session\': session}
else:
send_dic = {\'flag\': False, \'msg\': \'密码错误!\'}
common.send_data(send_dic, conn)
# 获取电影接口
@common.login_auth
def get_movie_list_interface(client_back_dic, conn):
# 获取所有电影对象
movie_obj_list = models.Movie.select()
back_movie_list = []
if movie_obj_list:
# 过滤已删除的电影
for movie_obj in movie_obj_list:
# 没有删除则返回
if not movie_obj.is_delete:
back_movie_list.append(
# [电影名称、是否免费、电影ID]
[movie_obj.name, \'免费\' if movie_obj.is_free else "收费", movie_obj.id]
)
if back_movie_list:
send_dic = {\'flag\': True, \'back_movie_list\': back_movie_list}
else:
send_dic = {\'flag\': False, \'msg\': \'没有可删除的电影!\'}
else:
send_dic = {\'flag\': False, \'msg\': \'没有电影!\'}
common.send_data(send_dic, conn)
db/models.py
from orm.orm import Models, StringField, IntegerField
# 用户表
class User(Models):
# 表名
table_name = \'user\'
# 字段
id = IntegerField(name=\'id\', primary_key=True)
name = StringField(name=\'name\')
# pwd, is_vip, is_locked, user_type, register_time
pwd = StringField(name=\'pwd\')
is_vip = IntegerField(name=\'is_vip\')
is_locked = IntegerField(name=\'is_locked\')
user_type = StringField(name=\'user_type\')
register_time = StringField(name=\'register_time\')
# 电影表
class Movie(Models):
# 表名
table_name = \'movie\'
# 字段
id = IntegerField(name=\'id\', primary_key=True)
name = StringField(name=\'name\')
# path, is_free, file_md5, user_id, is_delete, upload_time
path = StringField(name=\'path\')
is_free = IntegerField(name=\'is_free\') # 1 0
file_md5 = StringField(name=\'file_md5\')
user_id = IntegerField(name=\'user_id\')
is_delete = IntegerField(name=\'is_delete\')
upload_time = StringField(name=\'upload_time\')
# 公告表
class Notice(Models):
table_name = \'notice\'
# 字段
id = IntegerField(name=\'id\', primary_key=True)
title = StringField(name=\'title\')
content = StringField(name=\'content\')
user_id = IntegerField(name=\'user_id\')
create_time = StringField(name=\'create_time\')
# 下载记录表
class DownloadRecord(Models):
table_name = \'download_record\'
# 字段
id = IntegerField(name=\'id\', primary_key=True)
user_id = IntegerField(name=\'user_id\')
movie_id = IntegerField(name=\'movie_id\')
download_time = StringField(name=\'download_time\')
orm/orm.py
\'\'\'
定义字段类
\'\'\'
from orm.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
# varchar
class StringField(Field):
def __init__(self, name, column_type=\'varchar(255)\', primary_key=False, default=None):
super().__init__(name, column_type, primary_key, default)
# int
class IntegerField(Field):
def __init__(self, name, column_type=\'int\', primary_key=False, default=0):
super().__init__(name, column_type, primary_key, default)
# 元类控制表模型类的创建
class OrmMetaClass(type):
# 类名, 类的基类, 类的名称空间
def __new__(cls, class_name, class_bases, class_attr):
# print(class_name, class_bases, class_attr)
# 1.过滤Models类
if class_name == \'Models\':
return type.__new__(cls, class_name, class_bases, class_attr)
# 2.控制模型表中: 表名, 主键, 表的字段
# 如果模型表类中没有定义table_name,把类名当做表名
# 获取表名
table_name = class_attr.get(\'table_name\', class_name) # user_info, User
# 3.判断是否只有一个主键
primary_key = None
# 用来存放所有的表字段, 存不是目的,目的是为了取值方便
mappings = {}
\'\'\'
__main__: xxxx
\'id\': <__main__.IntegerField object at 0x000001E067D48B00>,
\'name\': <__main__.StringField object at 0x000001E067D48AC8>}
\'\'\'
for key, value in class_attr.items():
# 判断value是否是字段类的对象
if isinstance(value, Field):
# 把所有字段都添加到mappings中
mappings[key] = value
if value.primary_key:
if primary_key:
raise TypeError(\'主键只能有一个\')
# 获取主键
primary_key = value.name
# 删除class_attr中与mappings重复的属性, 节省资源
for key in mappings.keys():
class_attr.pop(key)
# 判断是否有主键
if not primary_key:
raise TypeError(\'必须要有一个主键\')
class_attr[\'table_name\'] = table_name
class_attr[\'primary_key\'] = primary_key
class_attr[\'mappings\'] = mappings
\'\'\'
\'table_name\': table_name
\'primary_key\': primary_key
\'mappings\': {\'id\': <__main__.IntegerField object at 0x000001E067D48B00>,
\'name\': <__main__.StringField object at 0x000001E067D48AC8>}
}
\'\'\'
return type.__new__(cls, class_name, class_bases, class_attr)
# 继承字典类,
class Models(dict, metaclass=OrmMetaClass):
def __init__(self, **kwargs):
# print(kwargs) # 接收关键字参数
super().__init__(**kwargs)
# 在对象.属性没有的时候触发
def __getattr__(self, item):
# print(item)
return self.get(item, \'没有这个key\')
# 在对象.属性 = 属性值 时触发
def __setattr__(self, key, value):
# 字典赋值操作
self[key] = value
# 查
@classmethod
def select(cls, **kwargs):
# 获取数据库链接对象
ms = Mysql()
# 若没有kwargs代表没有条件查询
if not kwargs:
# select * from table;
sql = \'select * from %s\' % cls.table_name
res = ms.my_select(sql)
# 若有kwargs代表有条件
else:
# print(kwargs) # {id:1}
key = list(kwargs.keys())[0] # id
value = kwargs.get(key) # 1
# select * from table where id=1;
sql = \'select * from %s where %s=?\' % (cls.table_name, key)
sql = sql.replace(\'?\', \'%s\')
res = ms.my_select(sql, value)
if res:
# [{},{}, {}] ----> [obj1, obj2, obj3]
# 把mysql返回来的 列表套 字典 ---> 列表套 对象
# l1 = []
# # 遍历mysql返回所有的字典
# for d in res:
# # 把每一个字典传给cls实例化成一个个的r1对象
# r1 = cls(**d)
# # 追加到l1列表中
# l1.append(r1)
return [cls(**result) for result in res]
# 插入
def save(self):
ms = Mysql()
# insert into table(x,x,x) values(x,x,x);
# 字段名
fields = []
# 字段的值
values = []
# 存放对应字段的?号
args = []
for k, v in self.mappings.items():
# 把主键过滤掉
if not v.primary_key:
fields.append(v.name)
values.append(getattr(self, v.name, v.default))
args.append(\'?\')
# insert into table(x,x,x) values(?, ?, ?);
sql = \'insert into %s(%s) values(%s)\' % (
self.table_name, \',\'.join(fields), \',\'.join(args)
)
sql = sql.replace(\'?\', \'%s\')
ms.my_execute(sql, values)
# 更新
def sql_update(self):
ms = Mysql()
fields = []
primary_key = None
values = []
for k, v in self.mappings.items():
# 获取主键的值
if v.primary_key:
primary_key = getattr(self, v.name, v.default)
else:
# 获取 字段名=?, 字段名=?,字段名=?
fields.append(v.name + \'=?\')
# 获取所有字段的值
values.append(getattr(self, v.name, v.default))
# update table set %s=?,... where id=1; 把主键当做where条件
sql = \'update %s set %s where %s=%s\' % (
self.table_name, \',\'.join(fields), self.primary_key, primary_key
)
# print(sql) # update User set name=? where id=3
sql = sql.replace(\'?\', \'%s\')
ms.my_execute(sql, values)
orm/mysql_control.py
import pymysql
from orm.db_pool import POOL
class Mysql:
def __init__(self):
# 建立链接
self.conn = POOL.connection()
# 获取游标
self.cursor = self.conn.cursor(pymysql.cursors.DictCursor)
# 关闭游标\链接方法
def close_db(self):
self.cursor.close()
self.conn.close()
# 查看
def my_select(self, sql, args=None):
self.cursor.execute(sql, args)
res = self.cursor.fetchall()
# [{}, {}, {}]
# print(res)
return res
# 提交
def my_execute(self, sql, args):
try:
# 把insert , update...一系列sql提交到mysql中
self.cursor.execute(sql, args)
except Exception as e:
print(e)
orm/db_pool.py
from DBUtils.PooledDB import PooledDB
import pymysql
# pip3 install DBUtils
POOL = PooledDB(
creator=pymysql, # 使用链接数据库的模块
maxconnections=6, # 连接池允许的最大连接数，0和None表示不限制连接数
mincached=2, # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
maxcached=5, # 链接池中最多闲置的链接，0和None不限制
maxshared=3,
# 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
maxusage=None, # 一个链接最多被重复使用的次数，None表示无限制
setsession=[], # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
ping=0,
# ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
host=\'127.0.0.1\',
port=3306,
user=\'root\',
password=\'000000\',
database=\'youku_demo\',
charset=\'utf8\',
autocommit=\'True\'
)
lib/common.py
import time
import hashlib
import json
import struct
import uuid
from functools import wraps
from db import user_data
def get_time():
now_time = time.strftime(\'%Y-%m-%d %X\')
return now_time
def get_md5_pwd(pwd):
md = hashlib.md5()
md.update(pwd.encode(\'utf-8\'))
md.update(\'虹桥炮王，Jason是也!\'.encode(\'utf-8\'))
return md.hexdigest()
def send_data(send_dic, conn):
data_bytes = json.dumps(send_dic).encode(\'utf-8\')
headers = struct.pack(\'i\', len(data_bytes))
conn.send(headers)
conn.send(data_bytes)
def get_random_code():
# uuid可以产生一个世界上唯一的字符串
md5 = hashlib.md5()
md5.update(str(uuid.uuid4()).encode(\'utf-8\'))
return md5.hexdigest()
# 登录认证装饰器
def login_auth(func):
@wraps(func)
# client_back_dic, conn = args
def inner(*args, **kwargs):
# if args[0].get(\'session\') == 服务端存放的session值：
# # [session, user_id] = values
addr = args[0].get(\'addr\')
# addr: [session, user_id]
user_session = user_data.user_online.get(addr)
if args[0].get(\'session\') == user_session[0]:
args[0][\'user_id\'] = user_session[1]
#
# for values in user_data.user_online.values():
# if args[0].get(\'session\') == values[0]:
# # 添加到client_back_dic
# args[0][\'user_id\'] = values[1] # user_id
# 判断user_id是否存在
if args[0].get(\'user_id\'):
func(*args, **kwargs)
else:
send_dic = {\'flag\': False, \'msg\': \'未登录，请去登录!\'}
# send_data(send_dic, conn)
send_data(send_dic, args[1])
return inner
# if __name__ == \'__main__\':
#
# # print(get_time())
# print(get_random_code())
# 05248e1b1a10ac08872f8dd5d9dbd814
# 161df6d362dc52b0037d938a0717963e
# aabd3987f88b2db46566cf6d9ec864e2
lib/lock_file.py
mutex = None
db/user_data.py
user_online = {
# addr: [session, user_id]
}
interface/admin_interface.py
from lib import common
from db import models
import os
from conf import settings
@common.login_auth
def upload_movie_interface(client_back_dic, conn):
print(\'炮王来交货啦!\')
# 确保电影名称是唯一的 随机字符串 + 电影名称
movie_name = common.get_random_code() + client_back_dic.get(\'movie_name\') # .mp4
movie_size = client_back_dic.get(\'file_size\')
movie_path = os.path.join(settings.DOWNLOAD_PATH, movie_name)
# 1.接受上传的文件
data_recv = 0
with open(movie_path, \'wb\') as f:
while data_recv < movie_size:
data = conn.recv(1024)
f.write(data)
data_recv += len(data)
# 2.把电影数据保存到mysql中
movie_obj = models.Movie(
name=movie_name, file_md5=client_back_dic.get(\'file_md5\'),
is_free=client_back_dic.get(\'is_free\'), is_delete=0,
path=movie_path, user_id=client_back_dic.get(\'user_id\'),
upload_time=common.get_time()
)
movie_obj.save()
send_dic = {
\'flag\': True, \'msg\': f\'{client_back_dic.get("movie_name")}电影上传成功!\'
}
common.send_data(send_dic, conn)
@common.login_auth
def check_movie_interface(client_back_dic, conn):
file_md5 = client_back_dic.get(\'file_md5\')
movie_list = models.Movie.select(file_md5=file_md5)
if movie_list:
print(1111)
send_dic = {
\'flag\': False,
\'msg\': \'电影已存在!\'
}
else:
print(222)
send_dic = {
\'flag\': True,
\'msg\': \'电影可以上传\'
}
common.send_data(send_dic, conn)
def delete_movie_interface(client_back_dic, conn):
# 直接删除
movie_obj = models.Movie.select(id=client_back_dic.get(\'movie_id\'))[0]
movie_obj.is_delete = 1
# 调用更新方法
movie_obj.sql_update()
send_dic = {
\'flag\': True,
\'msg\': \'电影删除成功!\'
}
common.send_data(send_dic, conn)
@common.login_auth
def put_notice_interface(client_back_dic, conn):
title = client_back_dic.get(\'title\')
content = client_back_dic.get(\'content\')
user_id = client_back_dic.get(\'user_id\')
notice_obj = models.Notice(title=title, content=content, user_id=user_id,
create_time=common.get_time())
notice_obj.save()
send_dic = {
\'msg\': \'公告发布成功!\'
}
common.send_data(send_dic, conn)
conf/settings.py
import os
BASE_PATH = os.path.dirname(os.path.dirname(__file__))
DOWNLOAD_PATH = os.path.join(BASE_PATH, \'download_files\')
客户端代码
start.py
import os
import sys
from core import src
sys.path.append(os.path.dirname(__file__))
if __name__ == \'__main__\':
src.run()
core/src.py
from core import admin, user
func_dic = {
\'1\': admin.admin_view,
\'2\': user.user_view,
}
def run():
while True:
print(\'\'\'
1.管理员功能
2.用户功能
q.退出
\'\'\')
choice = input(\'请选择功能编号: \').strip()
if choice == \'q\':
break
if choice not in func_dic:
continue
func_dic.get(choice)()
core/admin.py
from tcp_client import socket_client
from lib import common
import os
from conf import settings
user_info = {
\'cookies\': None
}
def register(client):
while True:
username = input(\'请输入用户名:\').strip()
password = input(\'请输入密码:\').strip()
re_password = input(\'请确认密码:\').strip()
if password == re_password:
send_dic = {\'username\': username,
\'password\': password,
\'type\': \'register\',
\'user_type\': \'admin\'}
# {\'flag\': False, \'msg\': \'用户已存在!\'}
# {\'flag\': True, \'msg\': \'注册成功\'}
back_dic = common.send_msg_back_dic(send_dic, client)
if back_dic.get(\'flag\'):
print(back_dic.get(\'msg\'))
break
else:
print(back_dic.get(\'msg\'))
def login(client):
while True:
username = input(\'请输入用户名: \').strip()
password = input(\'请输入密码:\').strip()
send_dic = {
\'type\': \'login\',
\'username\': username,
\'password\': password,
\'user_type\': \'admin\'
}
back_dic = common.send_msg_back_dic(send_dic, client)
if back_dic.get(\'flag\'):
session = back_dic.get(\'session\')
user_info[\'cookies\'] = session
print(back_dic.get(\'msg\'))
break
else:
print(back_dic.get(\'msg\'))
# 上传电影
def upload_movie(client):
while True:
# 1.打印电影列表
movie_list = common.get_movie_list()
for index, movie in enumerate(movie_list):
print(index, movie)
choice = input(\'请输入上传的电影编号:\').strip()
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(settings.UPLOAD_FILES, movie_name)
# 2.去服务端校验电影是否存在
file_md5 = common.get_movie_md5(movie_path)
send_dic = {
\'type\': \'check_movie\',
\'session\': user_info.get(\'cookies\'),
\'file_md5\': file_md5
}
back_dic = common.send_msg_back_dic(send_dic, client)
if back_dic.get(\'flag\'):
print(back_dic.get(\'msg\'))
send_dic = {
\'type\': \'upload_movie\',
\'file_md5\': file_md5,
\'file_size\': os.path.getsize(movie_path),
\'movie_name\': movie_name,
\'session\': user_info.get(\'cookies\')
}
is_free = input(\'上传电影是否免费: y/n\').strip()
if is_free == \'y\':
send_dic[\'is_free\'] = 1
else:
send_dic[\'is_free\'] = 0
back_dic = common.send_msg_back_dic(send_dic, client, file=movie_path)
if back_dic.get(\'flag\'):
print(back_dic.get(\'msg\'))
break
else:
print(back_dic.get(\'msg\'))
#
#
# send_dic = {\'type\': \'upload_movie\',\'session\': user_info.get(\'cookies\')}
# back_dic = common.send_msg_back_dic(send_dic, client)
# print(back_dic)
# 删除电影
def delete_movie(client):
while True:
# 1.从服务端获取电影列表
send_dic = {
\'type\': \'get_movie_list\',
\'session\': user_info.get(\'cookies\')
}
# 发送获取电影请求
back_dic = common.send_msg_back_dic(
send_dic, client)
if back_dic.get(\'flag\'):
back_movie_list = back_dic.get(\'back_movie_list\')
# 打印选择的电影
for index, movie_list in enumerate(back_movie_list):
print(index, movie_list)
# 2.选择需要删除的电影
choice = input(\'请输入需要删除的电影编号：\').strip()
if not choice.isdigit():
continue
choice = int(choice)
if choice not in range(len(back_movie_list)):
continue
movie_id = back_movie_list[choice][2]
send_dic = {
\'type\': \'delete_movie\',
\'movie_id\': movie_id,
\'session\': user_info.get(\'cookies\')
}
# 发送删除电影请求
back_dic = common.send_msg_back_dic(send_dic, client)
if back_dic.get(\'flag\'):
print(back_dic.get(\'msg\'))
break
else:
print(back_dic.get(\'msg\'))
break
# 发布公告
def put_notice(client):
title = input(\'请输入公告标题:\').strip()
content = input(\'请输入公告内容:\').strip()
send_dic = {
\'type\': \'put_notice\',
\'session\': user_info.get(\'cookies\'),
\'title\': title,
\'content\': content
}
back_dic = common.send_msg_back_dic(send_dic, client)
print(back_dic.get(\'msg\'))
func_dic = {
\'1\': register,
\'2\': login,
\'3\': upload_movie,
\'4\': delete_movie,
\'5\': put_notice,
}
def admin_view():
sk_client = socket_client.SocketClient()
client = sk_client.get_client()
while True:
print(\'\'\'
1.注册
2.登录
3.上传视频
4.删除视频
5.发布公告
q.退出
\'\'\')
choice = input(\'请选择功能编号:\').strip()
if choice == \'q\':
break
if choice not in func_dic:
continue
func_dic.get(choice)(client)
tcp_client/socket_client.py
import socket
class SocketClient:
def __init__(self):
self.client = socket.socket()
self.client.connect((\'127.0.0.1\', 9527))
def get_client(self):
return self.client
lib/common.py
import json
import struct
from conf import settings
import os
import hashlib
def send_msg_back_dic(send_dic, client, file=None):
data_bytes = json.dumps(send_dic).encode(\'utf-8\')
headers = struct.pack(\'i\', len(data_bytes))
client.send(headers)
client.send(data_bytes)
# 上传电影
if file:
with open(file, \'rb\') as f:
for line in f:
# print(line)
client.send(line)
headers = client.recv(4)
data_len = struct.unpack(\'i\', headers)[0]
data_bytes = client.recv(data_len)
back_dic = json.loads(data_bytes.decode(\'utf-8\'))
return back_dic
def get_movie_list():
if os.path.exists(settings.UPLOAD_FILES):
movie_list = os.listdir(settings.UPLOAD_FILES)
if movie_list:
return movie_list
# 获取电影的md5值
def get_movie_md5(movie_path):
md5 = hashlib.md5()
# 截取电影的4个位置的md5值
movie_size = os.path.getsize(movie_path)
# 从电影的4个位置个截取10个bytes数据
current_index = [0, movie_size // 3, (movie_size // 3) * 2, movie_size - 10]
with open(movie_path, \'rb\') as f:
for index in current_index:
f.seek(index)
data = f.read(10)
md5.update(data)
return md5.hexdigest()
conf/settings.py
import os
BASE_PATH = os.path.dirname(os.path.dirname(__file__))
UPLOAD_FILES = os.path.join(BASE_PATH, \'upload_files\')
版权声明：本文为suwanbin原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。