CSIC_716_20191224【python基础结课作业--仿优酷练习】
2024-09-07 11:45:38
需 求：
********管理员界面********
1 注册
2 登录
3 上传视频
4 删除视频
5 发布公告
********普通用户界面********
1 注册
2 登录
3 冲会员
4 查看视频
5 下载免费视频
6 下载收费视频
7 查看下载记录
8 查看公告
q.退出 软件开发目录：
服务端Server
# _*_ coding: gbk _*_
# @Author: Wonder
import os
BASE_PATH = os.path.dirname(os.path.dirname(__file__))
MOVIE_DIR_PATH = os.path.join(BASE_PATH,'movie') IP = '127.0.0.1'
PORT = 9527
settings.py
# _*_ coding: gbk _*_
# @Author: Wonder
from orm_control.orm import Models, StringField, IntegerField
import datetime
'''
每个字段的四个属性name, field_type, is_primary, default
其中field_type、is_primary、default都有默认值
'''
'''
# 用户表类: id、用户名、密码、用户类型、是否为VIP、注册时间
# 电影表类: id、电影名字、电影大小、电影md5值、电影是否免费、电影是否删除、上传时间、上传用户的id
# 公告表类: id、公告标题， 公告内容、发布时间、发布用户id
# 下载记录表: id、下载电影的id、下载用户的id、下载时间
''' class User(Models):
'''
# 用户表类: id、用户名、密码、用户类型、是否为VIP、注册时间
'''
u_id = IntegerField(name='u_id', is_primary=True)
username = StringField(name='username')
password = StringField(name='password')
user_type = IntegerField(name='user_type',) # 此处 0为普通用户，1 为管理员，默认值为
is_vip = IntegerField(name='is_vip') # 此处 0 为普通用户，1 为VIP
reg_time = StringField(name='reg_time') # 每个字段的默认值，可以在这些个性化的类中进行设置 class Movie(Models):
'''
# 电影表类: id、电影名字、电影大小、电影md5值、电影是否免费、电影是否删除、上传时间、上传用户的id
'''
m_id = IntegerField(name='m_id', is_primary=True)
name = StringField(name='name')
size = IntegerField(name='size')
md5 = StringField(name='md5')
is_free = IntegerField(name='is_free') #默认0是免费，1 是收费
is_del = IntegerField(name='is_del')
up_time = StringField(name='movie_name')
u_id = IntegerField(name='movie_name') class Notice(Models):
'''
id、公告标题， 公告内容、发布时间、发布用户id
'''
n_id = IntegerField(name='n_id', is_primary=True)
title = StringField(name='title')
content = StringField(name='content')
release_time = StringField(name='release_time')
u_id = IntegerField(name='u_id') class Record(Models):
'''
# 下载记录表: id、下载电影的id、下载用户的id、下载时间
'''
r_id = IntegerField(name='r_id', is_primary=True)
m_id = IntegerField(name='m_id')
u_id = IntegerField(name='u_id')
down_time = StringField(name='down_time') if __name__ == '__main__':
import time
# user_obj = User(
# username='张五', # id是不用传值的，因为在数据库中实现了自增
# password ='1234',
# user_type=1,
# is_vip=0,
# reg_time=time.strftime('%Y-%m-%d %X')
# )
# user_obj.insert_data()
# res = User.select_data() # [ 字典、]
# print(res) user_obj = User.select_data(u_id=1)[0] # 这里注意，是个字典套对象，要注意取值
user_obj.username = '张九'
user_obj.update_data() res = User.select_data() # [ 字典（对象）、] 这里的对象就是字典，一个特殊的对象，类dict的对象。
print(res)
models.py
# _*_ coding: gbk _*_
# @Author: Wonder
import os
import uuid
from conf import settings
from lib import common
import time
from db.models import Notice, Movie @common.login_auth
def check_movie_interface(dic, conn, addr):
md5 = dic.get('movie_md5')
movie_list = Movie.select_data(md5=md5)
if movie_list:
back_dic = {
'flag': False,
'msg': '电影已存在'
}
else:
back_dic = {
'flag': True,
}
common.send_data(back_dic, conn) @common.login_auth
def upload_movie_interface(dic, conn, addr):
movie_size = dic.get('movie_size')
movie_name = str(uuid.uuid4()).replace('-', '_') + dic.get('movie_name') # 为了保证名字不同的电影重名时，能保存
movie_path = os.path.join(settings.MOVIE_DIR_PATH, movie_name)
temp_size = 0
with open(movie_path, 'wb') as f:
while temp_size < movie_size:
data = conn.recv(1024)
f.write(data)
temp_size += len(data) back_dic = {
'msg': '传完了'
} mv_obj = Movie(
name=movie_name,
size=movie_size,
md5=dic.get('movie_md5'),
is_free=dic.get('is_free'),
up_time=time.strftime('%Y-%m-%d %X'),
u_id=dic.get('u_id'),
)
print(1)
mv_obj.insert_data()
print(2)
common.send_data(back_dic, conn) @common.login_auth
def del_mv_interface(dic, conn, addr):
movie_name = dic.get('mv_name')
mv_obj = Movie.select_data(name=movie_name)[0]
mv_obj.is_del = 1
mv_obj.update_data()
back_dic = {
'msg': f'{mv_obj.name}已删除'
}
print('看一眼')
common.send_data(back_dic, conn) @common.login_auth
def relaese_notice_interface(dic, conn, addr):
notice_obj = Notice(
title=dic.get('title'),
content=dic.get('content'),
release_time=time.strftime('%Y-%m-%d %X'),
u_id=dic.get('u_id') )
print(notice_obj)
notice_obj.insert_data()
back_dic = {'msg': '公告发布成功'}
common.send_data(back_dic, conn)
admin_interface
# _*_ coding: gbk _*_
# @Author: Wonder
from db.models import User
from user_data import session
from lib import common
import time
from db.models import User, Movie
# from tcp_server.server import mutex
from threading import Lock
# mutex = Lock() # 注册接口
def register_interface(dic, conn, addr):
username = dic.get('username')
user_type = dic.get('user_type')
user_list = User.select_data(username=username)
if user_list:
send_dic = {
'flag': False, 'msg': '用户已存在'
}
else:
user_obj = User(
username=username,
user_type=user_type,
password=common.get_md5(dic.get('password')),
reg_time=time.strftime('%Y-%m-%d %X'),
)
user_obj.insert_data()
send_dic = {
'flag': True, 'msg': '注册成功'
}
common.send_data(send_dic, conn) # 登录接口 核对登录密码，更新session文件
def login_interface(dic, conn, addr):
username = dic.get('username')
user_list = User.select_data(username=username)
if not user_list:
dic = {
'flag': False, 'msg': '用户不存在'
}
common.send_data(dic, conn)
else:
input_pwd = common.get_md5(dic.get('password'))
input_type = dic.get('user_type')
record_pwd = user_list[0].get('password')
record_type = user_list[0].get('user_type')
if not input_pwd == record_pwd:
dic = {
'flag': False, 'msg': '用户名或密码不正确'
}
common.send_data(dic, conn)
elif not input_type == record_type:
print('传来的',input_type)
print('系统的',record_type) dic = {
'flag': False, 'msg': '非法用户'
}
common.send_data(dic, conn) else:
u_id = user_list[0].get('u_id')
is_vip = user_list[0].get('is_vip')
new_session = common.get_session()
# mutex.locked()
session.user_info[addr] = [new_session, u_id]
# mutex.release() dic = {
'flag': True,
'msg': '登录成功',
'session': new_session,
'u_id': u_id,
'is_vip': is_vip
}
# print('服务端的session',dic)
common.send_data(dic, conn) # 查看没有被删除的所有电影
@common.login_auth
def check_nodel_movie_interface(dic, conn, addr):
# 查出来可以删除的电影 is_del=0
mv_obj_list = Movie.select_data() if not mv_obj_list:
dic = {
'flag': False,
'msg': '没有电影'
}
else:
mv_name_list = [i for i in mv_obj_list if i.is_del == 0] # 找出所有没有删除的
free_mv_list = [i.name for i in mv_name_list if i.is_free == 0] # 找出没删除的 免费 的 名字
# print('免费的电影', free_mv_list)
pay_mv_list = [i.name for i in mv_name_list if i.is_free == 1] # 找出没删除 付费的 的名字
# print('收费的电影', pay_mv_list)
if dic.get('select_type') == 'all':
mv_list = free_mv_list
mv_list.extend(pay_mv_list)
elif dic.get('select_type') == 'free':
mv_list = free_mv_list
else:
mv_list = pay_mv_list if not mv_list:
dic = {
'flag': False,
'msg': '没有电影'
}
else:
dic = {
'flag': True,
'mv_list': mv_list
}
common.send_data(dic, conn)
common_interface
# _*_ coding: gbk _*_
# @Author: Wonder
import os
import time
from db.models import User, Record, Movie, Notice
from conf.settings import MOVIE_DIR_PATH
from lib import common # 充值会员接口
@common.login_auth
def paymember_interface(dic, conn, addr):
print('开始充钱啦')
user_obj = User.select_data(
u_id=dic.get('u_id')
)[0]
print(user_obj)
user_obj.is_vip = 1
user_obj.update_data()
back_dic = {
'msg': '充值成功'
}
common.send_data(back_dic, conn) # 下载电影接口
@common.login_auth
def download_mv_interface(dic, conn, addr):
mv_name = dic.get('mv_name')
# 打开服务器端供下载的文件夹
file_path = os.path.join(MOVIE_DIR_PATH, mv_name)
mv_size = os.path.getsize(file_path) back_dic = {
'mv_size': mv_size
} # 下载完了之后，还要将下载记录插入到下载表中
mv_obj = Movie.select_data(name=mv_name) # 在此处通过名字取值，会增加数据库负担
m_id = mv_obj[0].m_id #单个一定要指定0，否则不执行
rec_obj = Record(
m_id=m_id,
u_id=dic.get('u_id'), # 装饰器给加进去的
down_time=time.strftime('%Y-%m-%d %X')
)
rec_obj.insert_data()
common.send_data(back_dic, conn, file_path) # 将报头和内容进行整合 #查看下载记录
@common.login_auth
def check_record_interface(dic, conn, addr):
u_id = dic.get('u_id')
record_obj = Record.select_data(u_id=u_id)
if not record_obj:
back_dic = {
'flag': False,
'msg': '没有下载记录'
}
else:
movie_id_list = [i.m_id for i in record_obj] # 将电影的id取出来
mv_obj = Movie.select_data()
movie_list = [i.name for i in mv_obj if i.m_id in movie_id_list] # 取出名字
back_dic = {
'flag': True,
'mv_list': movie_list
}
common.send_data(back_dic, conn) # 查看下载记录
@common.login_auth
def check_notice_interface(dic, conn, addr):
notice_obj = Notice.select_data()
if not notice_obj:
back_dic = {
'flag': False,
'msg': '没有公告'
}
else:
notice_list = [f'标题{i.title},内容{i.content}' for i in notice_obj]
back_dic = {
'flag': True,
'notice_list': notice_list
}
common.send_data(back_dic, conn)
user_interface
# _*_ coding: gbk _*_
# @Author: Wonder
import json
import struct
from hashlib import md5
import uuid
from user_data import session
# from tcp_server.server import mutex
from threading import Lock
# mutex = Lock() # 发数据
def send_data(dic, conn, path=None):
send_dic = json.dumps(dic).encode('utf-8')
send_head = struct.pack('i', len(send_dic))
conn.send(send_head)
conn.send(send_dic)
if path:
with open(path, 'rb') as f:
for line in f:
conn.send(line) # md5进行加密
def get_md5(pwd):
md5_obj = md5()
md5_obj.update(pwd.encode('utf-8'))
sal = 'youku_sys'
md5_obj.update(sal.encode('utf-8'))
return md5_obj.hexdigest() # 获取一个session随机字符串
def get_session():
rand_str = str(uuid.uuid4())
return get_md5(rand_str) # 登录 验证装饰器
def login_auth(func): # 将用户id添加到字典中
def inner(*args, **kwargs): # *args (dic,conn ,addr)
print('进入装饰器啦------------')
recv_dic = args[0] conn = args[1] client_session = recv_dic.get('cookies')
addr = args[2]
# print('打印一下文件中是否有内容',session.user_info)
# mutex.locked()
server_session = session.user_info.get(addr)
# mutex.release() if server_session:
if client_session == server_session[0]:
recv_dic['u_id'] = server_session[1]
res = func(*args, **kwargs)
return res
else:
dic = {
'flag': False,
'msg': '重新登陆'
}
send_data(dic, conn)
else:
dic = {
'flag': False,
'msg': '******请先登陆******'
}
send_data(dic, conn) return inner if __name__ == '__main__':
msg = get_md5('www')
print(msg)
common
# _*_ coding: gbk _*_
# @Author: Wonder from DBUtils.PooledDB import PooledDB
import pymysql Pool = PooledDB(
# creator,
# mincached=0,
# maxcached=0,
# maxshared=0,
# maxconnections=0,
# blocking=False,
# maxusage=None,
# setsession=None,
# reset=True,
# failures=None,
# ping=1,
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='1234,5678a',
database='orm_mysql_demo',
charset='utf8',
autocommit=True )
mysql_pool
# _*_ coding: gbk _*_
# @Author: Wonder import pymysql
from orm_control.mysql_pool import Pool class MySql(): # 数据库连接用单例模式 def __new__(cls, *args, **kwargs):
if not hasattr(cls, '_instance', ):
setattr(cls, '_instance', object.__new__(cls)) return cls._instance def __init__(self):
self.mysql_client = Pool.connection() self.cursor = self.mysql_client.cursor(
pymysql.cursors.DictCursor) def select(self, sql, args=None):
self.cursor.execute(sql, args)
res = self.cursor.fetchall()
return res 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() # # if __name__ == '__main__':
# obj1 =MySql()
# obj2 =MySql()
# obj3 =MySql()
# print(id(obj1))
# print(id(obj2))
# print(id(obj3))
mysql_pro
# _*_ coding: gbk _*_
# @Author: Wonder
from orm_control.mysql_pro import MySql class Field:
def __init__(self, name, field_type, is_primary, default):
self.name = name
self.field_type = field_type
self.is_primary = is_primary
self.default = default class StringField(Field):
def __init__(self, name, field_type='varchar(255)', is_primary=False, default=None):
super().__init__(name, field_type, is_primary, default) class IntegerField(Field):
def __init__(self, name, field_type='int', is_primary=False, default=0):
super().__init__(name, field_type, is_primary, default) class OrmCtrl(type):
def __new__(cls, class_name, class_bases, class_dict):
if class_name == 'Models':
return type.__new__(cls, class_name, class_bases, class_dict) # 此时筛选出来的就是继承Field产生的类，需要控制他们类的创建过程 primary_key = None
# table_name = getattr(cls, 'table_name', class_name) # 取User类中的table_name属性（如果传就用table_name）
table_name = class_dict.get('table_name', class_name) # 看有没有创建并传一个叫table_name的字段信息
mappings = {} # 字典里存放User类的字段 for k, v in class_dict.items():
if isinstance(v, Field): # 筛选出Field的创建的对象，即User等之类的表 mappings[k] = v # 将合格的k、v转义到mappings中 if v.is_primary: # 表示这个字段是主键 V是一个由Field生成的对象
if primary_key:
raise TypeError('只能有一个主键')
primary_key = v.name if not primary_key:
raise TypeError('必须要有主键') for k in mappings:
class_dict.pop(k) 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) # 都要交给元类type进行类的创建 class Models(dict, metaclass=OrmCtrl):
'''
控制实际表类对象的生成，干涉其取值方式。因为继承了字典，所以通过该类生成的对象都是字典
''' def __getattr__(self, item):
return self.get(item) def __setattr__(self, key, value):
self[key] = value @classmethod
def select_data(cls, **kwargs):
mysql_obj = MySql()
if not kwargs: # 说明想查cls这张表里面的所有值
sql = f'select * from {cls.table_name}'
res = mysql_obj.select(sql) else:
# kwargs-->** kwargs a=b --> {a:b}
# 开始取字典中的key dict1.keys()是一个对象
key = list(kwargs.keys())[0]
value = kwargs.get(key) sql = f'select * from {cls.table_name} where {key} = %s'
res = mysql_obj.select(sql, value) # 此时获取的两种res 都是列表套字典
return [cls(**i) for i in res] # 此时将列表中的字典，转化成了对象，就有了相关的属性，比如name,field_type def insert_data(self): # obj = User(name = 'ccc') # 此功能仅实现一次插入一条记录
sql_obj = MySql()
list_k = [] # 添加传来的key
list_value = [] # 添加传来的值
list_tag = []
for k, v in self.mappings.items(): # kv只能在items()中取值
list_k.append(k) # mappings的k都是字符串,因为是user的属性名
list_tag.append('%s') list_value.append(self.get(k, v.default)) # insert into t1 (id,name) values(1,ww),(2,xx)
sql = f'insert into {self.table_name} (%s) values (%s)' %(
','.join(list_k),
','.join(list_tag)
)
sql_obj.execute(sql, list_value) def update_data(self): # 更新的时候，敏感数据是更新的值，要单独传
sql_obj = MySql()
update_list = [] # 将要准备更新的数据，以字符串的形式 存在这里面
update_value = []
pk = None
pk_value = None for k, v in self.mappings.items():
if not v.is_primary:
update_list.append(f'{k}=%s')
update_value.append(self.get(k, v.default)) pk = self.primary_key
pk_value = self.get(pk) # 主键一定是有的 sql = f'update {self.table_name} set %s where %s = %s ' %(
','.join(update_list),
pk,
pk_value
)
print('更新的语句', sql)
sql_obj.execute(sql, update_value)
orm
# _*_ coding: gbk _*_
# @Author: Wonder
import socket
import struct
import json
from user_data.session import user_info
from concurrent.futures import ThreadPoolExecutor
from interface import common_interface
from interface import admin_interface
from interface import user_interface
from threading import Lock
# mutex = Lock() func_dict = {
'register': common_interface.register_interface,
'login': common_interface.login_interface,
'check_movie': admin_interface.check_movie_interface,
'upload_movie': admin_interface.upload_movie_interface,
'check_nodel_movie': common_interface.check_nodel_movie_interface, # 未删除的所有电影
'del_mv': admin_interface.del_mv_interface,
'release_notice': admin_interface.relaese_notice_interface,
'pay_member': user_interface.paymember_interface,
'download_mv': user_interface.download_mv_interface,
'check_record': user_interface.check_record_interface,
'check_notice': user_interface.check_notice_interface } pool = ThreadPoolExecutor(5) def connect(ip, port):
server = socket.socket()
server.bind((ip, port))
server.listen(5)
print('server start.....')
while True:
conn, addr = server.accept()
pool.submit(task, conn, addr) # 并发 def task(conn, addr):
while True:
try:
recv_data = conn.recv(4) dic_size = struct.unpack('i', recv_data)[0]
json_dic = conn.recv(dic_size).decode('utf-8')
recv_dic = json.loads(json_dic)
func_type = recv_dic.get('func_type')
if func_type in func_dict:
# print(recv_dic,'收到的字典')
func_dict.get(func_type)(recv_dic, conn, addr)
except Exception as e:
# print(e)
# mutex.acquire()
user_info.pop(str(addr)) # {str(addr) : [session, u_id ] }
# mutex.release() break
conn.close()
server
# _*_ coding: gbk _*_
# @Author: Wonder user_info = {}
# {(ip:port) : [session, u_id}
session
# _*_ coding: gbk _*_
# @Author: Wonder
import sys
from conf import settings
from tcp_server import server sys.path.append(settings.BASE_PATH) if __name__ == '__main__':
server.connect(settings.IP, settings.PORT)
start
客户端Client
# _*_ coding: gbk _*_
# @Author: Wonder
import os BASE_PATH = os.path.dirname(os.path.dirname(__file__))
DOWNLOAD_PATH = os.path.join(BASE_PATH, 'download_movie')
UPLOAD_PATH = os.path.join(BASE_PATH, 'upload_movie') IP = '127.0.0.1'
PORT = 9527
settings
# _*_ coding: gbk _*_
# @Author: Wonder
from lib import common
import os
import time
from conf.settings import UPLOAD_PATH current_user = {
'cookies': None
} # 注册
def register(client):
while True:
username = input('>>>请输入用户名').strip()
password = input('>>>请输入密码').strip()
re_password = input('>>>请再次输入密码').strip()
if not password == re_password:
print('密码不一致')
continue
dic = {
'func_type': 'register',
'username': username,
'password': password,
'user_type': 1
} back_dic = common.send_and_back(dic, client)
if back_dic.get('flag'):
print(back_dic.get('msg'))
break
else:
print(back_dic.get('msg')) choice = input('>>>是否继续注册，q退出').strip()
if choice == 'q':
break
else:
continue # 登录
def login(client):
while True:
username = input('>>>请输入用户名').strip()
password = input('>>>请输入密码').strip()
send_dic = {
'func_type': 'login',
'username': username,
'password': password,
'user_type': 1 }
back_dic = common.send_and_back(send_dic, client)
print(back_dic, '客户端的')
if back_dic.get('flag'):
cookies = back_dic.get('session')
current_user['cookies'] = cookies
msg = back_dic.get('msg')
print(msg)
break
else:
print(back_dic.get('msg')) choice = input('>>>是否继续登录，q退出').strip()
if choice == 'q':
break
else:
continue # 上传电影
def upload_mv(client):
# 上传电影
# 电影目录
while True: movie_list = os.listdir(UPLOAD_PATH)
if not movie_list:
print('无可上传电影')
break
else:
flag, msg = common.show_and_choice(movie_list)
if not flag:
print(msg)
continue
else:
movie_path = os.path.join(UPLOAD_PATH, msg) movie_size = os.path.getsize(movie_path)
# 先判断电影在不在(先取所有的内容，后面优化定点取值)
movie_md5 = common.get_movie_md5(movie_path) dic1 = {
'func_type': 'check_movie',
'movie_md5': movie_md5,
'cookies': current_user.get('cookies')
} print('组织的字典', dic1) back_dic = common.send_and_back(dic1, client)
print('收到的字典', back_dic)
if not back_dic.get('flag'):
print(back_dic.get('msg'))
break
else:
# 开始上传啦
choice = input('>>>选择是否收费，y收费，任意键免费').strip()
if choice == 'y':
is_free = 1
else:
is_free = 0
dic = {
'func_type': 'upload_movie',
'movie_name': msg,
'movie_size': movie_size,
'movie_md5': movie_md5,
'is_free': is_free,
'cookies': current_user.get('cookies')
}
back_dic = common.send_and_back(dic, client, path=movie_path)
print(back_dic.get('msg'))
break # 删除电影
def delete_mv(client):
# 获取电影列表
while True:
dic = {
'func_type': 'check_nodel_movie',
'cookies': current_user.get('cookies'),
'select_type': 'all'
}
back_dic = common.send_and_back(dic, client)
print('返回的报头', back_dic)
if not back_dic.get('flag'):
print(back_dic.get('msg'))
break
else:
mv_list = back_dic.get('mv_list')
print('可删除的列表', mv_list)
flag, msg = common.show_and_choice(mv_list)
if not flag:
print(msg)
break
else:
del_mv_name = msg dic2 = {
'func_type': 'del_mv',
'cookies': current_user.get('cookies'),
'mv_name': del_mv_name
}
print(dic2)
back_dic = common.send_and_back(dic2, client)
print(back_dic.get('msg'))
break # 发布公告
def announce(client):
title = input('>>>请输入公告标题').strip()
content = input('>>请输入公告内容').strip()
dic = {
'func_type': 'release_notice',
'title': title,
'content': content,
'cookies': current_user.get('cookies')
}
back_dict = common.send_and_back(dic, client)
print(back_dict.get('msg')) def run(client):
func_dict = {
'': register,
'': login,
'': upload_mv,
'': delete_mv,
'': announce,
}
while True:
print('''
********管理员界面********
1 注册
2 登录
3 上传视频
4 删除视频
5 发布公告 任意键退出
''')
choice = input('>>>请选择功能').strip()
if choice in func_dict:
func_dict.get(choice)(client)
else:
print('告辞')
break
admin_view
# _*_ coding: gbk _*_
# @Author: Wonder
from core import admin_view, user_view
from tcp_client.client import connect def run():
func_dict = {
'': admin_view,
'': user_view
}
while True:
client = connect()
print('''
********优酷系统********
1: 管理员
2: 普通用户 任意键退出
''')
choice = input('>>>请选择功能').strip()
if choice in func_dict:
func_dict.get(choice).run(client)
else:
print('告辞')
break if __name__ == '__main__':
run()
src
# _*_ coding: gbk _*_
# @Author: Wonder
from lib import common
import os
import time
from conf.settings import DOWNLOAD_PATH # 注册
current_user = {} def register(client):
while True:
username = input('请输入用户名').strip()
password = input('请输入密码').strip()
re_password = input('请再次输入密码').strip()
if not password == re_password:
print('密码不一致')
continue
dic = {
'func_type': 'register',
'username': username,
'password': password,
'user_type': 0
}
back_dic = common.send_and_back(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()
dic = {
'func_type': 'login',
'username': username,
'password': password,
'user_type': 0
}
back_dic = common.send_and_back(dic, client)
if back_dic.get('flag'):
current_user['cookies'] = back_dic.get('session') # 登陆成功之后，server会将session返回。
current_user['u_id'] = back_dic.get('u_id')
current_user['is_vip'] = back_dic.get('is_vip')
print(back_dic.get('msg'))
break
else:
print(back_dic.get('msg')) # 充会员
def pay_member(client):
# 此处是永久会员，修改is_vip
pay = input('请确认开通vip,按y确认').strip()
if not pay == 'y':
print('再见，未开通')
else:
dic = {
'func_type': 'pay_member',
'is_vip': 1,
'cookies': current_user.get('cookies')
}
back_dic = common.send_and_back(dic, client)
print(back_dic.get('msg')) # 查看电影目录
def check_movie(client):
# 查看所有未删除的电影
dic = {
'func_type': 'check_nodel_movie',
'cookies': current_user.get('cookies'),
'select_type': 'all'
}
back_dic = common.send_and_back(dic, client)
if back_dic.get('flag'):
print(back_dic.get('mv_list'))
else:
print(back_dic.get('msg')) # 下载免费电影
def download_free_mv(client):
# 先查免费电影的列表以供选择
while True:
dic1 = {
'func_type': 'check_nodel_movie',
'cookies': current_user.get('cookies'),
'select_type': 'free'
} back_dic = common.send_and_back(dic1, client)
if not back_dic.get('flag'):
print(back_dic.get('msg'))
break
# 封装一个方法，用于来选择。
free_mv_list = back_dic.get('mv_list')
flag, msg = common.show_and_choice(free_mv_list)
if not flag: # 输入了无效指令
continue
else:
# 接下来发送下载的请求
mv_name = msg
dic2 = {
'func_type': 'download_mv',
'cookies': current_user.get('cookies'),
'mv_name': mv_name
} back_dic = common.send_and_back(dic2, client) # 给出文件的大小
mv_size = back_dic.get('mv_size')
mv_file = os.path.join(DOWNLOAD_PATH, mv_name)
temp_size = 0
with open(mv_file, 'wb') as f:
while temp_size < mv_size:
data = client.recv(1024)
f.write(data)
temp_size += len(data)
print('下载完毕......')
break # 下载付费电影(要么充值会员，要么付费观影)
def download_paid_mv(client):
while True:
# 付费电影，对普通用户有限制 先获取当前用户
if not current_user.get('is_vip'):
print('当前用户不可以下载付费电影')
pay_choice = input('press A去充会员，press B付费下载，任意键退出')
if pay_choice == 'A':
pay_member(client)
login(client)
break
elif pay_choice == 'B':
money = input('>>>充值的钱').strip()
print('广告时间...5秒')
time.sleep(5)
else:
break # 此处可以扩展付费模块，钱包功能。 dic = {
'func_type': 'check_nodel_movie',
'select_type': 'pay',
'cookies': current_user.get('cookies')
}
back_dic = common.send_and_back(dic, client)
if not back_dic.get('flag'):
print(back_dic.get('msg'))
break
else:
pay_mv_list = back_dic.get('mv_list') flag, msg = common.show_and_choice(pay_mv_list)
if not flag:
print(msg)
break
else:
movie_name = msg
dic2 = {
'func_type': 'download_mv',
'cookies': current_user.get('cookies'),
'mv_name': movie_name
}
back_dic = common.send_and_back(dic2, client)
mv_size = back_dic.get('mv_size')
temp_size = 0
file_path = os.path.join(DOWNLOAD_PATH, movie_name)
with open(file_path, 'wb') as f:
while temp_size < mv_size:
data = client.recv(1024)
f.write(data)
temp_size += len(data)
print('下完了') # 这句话是否要从服务端传过来
break # 查看下载记录
def check_record(client):
dic = {
'func_type': 'check_record',
'cookies': current_user.get('cookies'),
}
back_dic = common.send_and_back(dic, client)
if not back_dic.get('flag'):
print(back_dic.get('msg'))
else:
print(back_dic.get('mv_list')) # 查看公告
def check_notice(client):
dic = {
'func_type': 'check_notice',
'cookies': current_user.get('cookies')
}
back_dic = common.send_and_back(dic, client)
if not back_dic.get('flag'):
print(back_dic.get('msg'))
else:
print(back_dic.get('notice_list')) def run(client):
func_dic = {
'': register,
'': login,
'': pay_member,
'': check_movie,
'': download_free_mv,
'': download_paid_mv,
'': check_record,
'': check_notice,
} while True:
print('''
********普通用户界面********
1 注册
2 登录
3 冲会员
4 查看视频
5 下载免费视频
6 下载收费视频
7 查看下载记录
8 查看公告
q.退出
''')
choice = input('>>>请选择功能对应的编号').strip()
if choice == 'q':
print('再见')
break
if choice in func_dic:
func_dic.get(choice)(client)
else:
print('无效指令')
user_view
# _*_ coding: gbk _*_
# @Author: Wonder import json
import struct
import os
from hashlib import md5
from tcp_client.client import connect # 发数据
def send_and_back(dic, client, path=None): # 一次请求一次响应
send_dic = json.dumps(dic).encode('utf-8')
send_head = struct.pack('i', len(send_dic))
client.send(send_head)
client.send(send_dic)
if path:
with open(path, 'rb') as f:
for line in f:
client.send(line) head2 = client.recv(4)
back_size = struct.unpack('i', head2)[0]
back_dic = json.loads(client.recv(back_size).decode('utf-8'))
return back_dic # 查看列表数据并进行选择
def show_and_choice(li):
for index, content in enumerate(li):
print(index, content)
choice = input('请输入选择的电影序号').strip()
if not choice.isdigit():
return False, '无效指令'
choice = int(choice)
if choice not in range(len(li)):
return False, '无效指令'
return True, li[choice] def get_movie_md5(path):
movie_size = os.path.getsize(path)
list_md5 = [0, movie_size // 2, movie_size - 10]
with open(path, 'rb') as f:
for index in list_md5:
md5_obj = md5()
f.seek(index)
part = f.read(10)
md5_obj.update(part)
return md5_obj.hexdigest()
common
# _*_ coding: gbk _*_
# @Author: Wonder import socket from conf.settings import IP, PORT
def connect():
client = socket.socket()
client.connect((IP, PORT))
return client
client
# _*_ coding: gbk _*_
# @Author: Wonder import sys
from conf import settings
from core import src
sys.path.append(settings.BASE_PATH) if __name__ == '__main__':
src.run()
start
最新文章
- Nodejs与ES6系列3:generator对象
- 原生js实现jquery库中部分事件的功能（jquery库封装二）
- Spring配置文件中使用表达式
- 关于angularjS与jQuery框架的那些事
- 【吉光片羽】ie6兼容性的几个点
- Apache安装配置步骤
- 基于selenium的pyse自动化测试框架
- 179. Largest Number -- 数字字符串比较大小
- 邻接矩阵实现Dijkstra算法以及BFS与DFS算法
- 深入理解ClassLoader(四)—类的父委托加载机制
- centos 下搭建 php环境（1）
- [Delphi]检查URL是否有效的函数
- asp.net MVC实现文章的“上一篇下一篇”
- 获取URL中的参数值
- linux下执行scrapy的爬虫定时任务
- webpack2学习日志
- 201521123033《Java程序设计》第13周学习总结
- js中字节B转化成KB，MB，GB
- jq demo 轮播图，图片可调用，向左，自动+鼠标点击切换
- wordpress添加文章固定字段