python-优酷系统管理员视图粗糙版(无详细注释)
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了python-优酷系统管理员视图粗糙版(无详细注释)，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含25277字，纯文字阅读大概需要37分钟。
内容图文
目录
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()
breakinterface/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
# aabd3987f88b2db46566cf6d9ec864e2lib/lock_file.py
mutex = Nonedb/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.clientlib/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')内容总结
以上是互联网集市为您收集整理的python-优酷系统管理员视图粗糙版(无详细注释)全部内容，希望文章能够帮你解决python-优酷系统管理员视图粗糙版(无详细注释)所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】