youku服務端

来源:https://www.cnblogs.com/limengjie0104/archive/2018/05/22/9074194.html
-Advertisement-
Play Games

文件結構 config import os IP_PORT = ('127.0.0.1',8080) BACKLOG = 5 BASE_DIR = os.path.dirname(os.path.dirname(__file__)) BASE_MOVIE_LIST = os.path.join(BA ...


文件結構

config

import os


IP_PORT = ('127.0.0.1',8080)
BACKLOG = 5

BASE_DIR = os.path.dirname(os.path.dirname(__file__))
BASE_MOVIE_LIST = os.path.join(BASE_DIR,'movie_list')
setting

db

from orm_pool.orm import Models,StringField,IntegerField


class User(Models):
    table_name = 'userinfo'
    id = IntegerField('id',primary_key=True)
    name = StringField('name')
    password = StringField('password')
    is_vip = IntegerField('is_vip')
    locked = IntegerField('locked')
    user_type = StringField('user_type')


class Notice(Models):
    table_name = 'notice'
    id = IntegerField('id',primary_key=True)
    name = StringField('name')
    content = StringField('content')
    create_time = StringField('create_time')
    user_id = IntegerField('user_id')


class Movie(Models):
    table_name = 'movie'
    id = IntegerField('id',primary_key=True)
    name = StringField('name')
    path = StringField('path')
    is_free = IntegerField('is_free')
    is_delete = IntegerField('is_delete')
    create_time = StringField('create_time')
    user_id = IntegerField('user_id')
    file_md5 = StringField('file_md5')


class Download_Record(Models):
    table_name = 'download_record'
    id = IntegerField('id', primary_key=True)
    user_id = IntegerField('user_id')
    movie_id = IntegerField('movie_id')
models

interface

import os
from db import models
from lib import common
from conf import setting


@common.login_auth
def check_upload_movie_exist(msg_dic,conn):
    movie = models.Movie.select_one(file_md5=msg_dic['file_md5'])
    if movie:
        back_dic = {'flag':False,'msg':'該影片已存在'}
    else:
        back_dic = {'flag':True,'msg':'影片可上傳'}
    common.send_back(back_dic,conn)


@common.login_auth
def upload_movie(msg_dic,conn):
    movie_name = common.get_file_uuid(msg_dic['file_name']) + msg_dic['file_name']
    movie_path = os.path.join(setting.BASE_MOVIE_LIST,movie_name)
    recv_size = 0
    file_size = msg_dic['file_size']
    with open(movie_path,'wb') as f:
        while recv_size < file_size:
            recv_data = conn.recv(1024)
            if not recv_data:
                back_dic = {'flag':False,'msg':'伺服器收到數據為空'}
                common.send_back(back_dic, conn)
                break
            f.write(recv_data)
            recv_size += len(recv_data)
            percent = recv_size / file_size
            common.print_progress(percent)

    movie = models.Movie(
        name=movie_name,
        path=movie_path,
        is_free=msg_dic['is_free'],
        is_delete=0,
        create_time=str(common.get_current_time()),
        user_id=msg_dic['user_id'],
        file_md5=msg_dic['file_md5']
    )
    movie.save()
    back_dic = {'flag':True,'msg':'上傳成功'}
    common.send_back(back_dic,conn)


@common.login_auth
def delete_movie(msg_dic,conn):
    movie = models.Movie.select_one(id=msg_dic['del_movie_id'])
    if movie:
        movie.is_delete = 1
        movie.user_id = msg_dic['user_id']
        movie.update()
        back_dic = {'flag':True,'msg':'刪除成功'}
    else:
        back_dic = {'flag':False,'msg':'無此影片'}
    common.send_back(back_dic,conn)


@common.login_auth
def release_notice(msg_dic,conn):
    name = msg_dic['name']
    content = msg_dic['content']
    user_id = msg_dic['user_id']
    create_time = str(common.get_current_time())
    notice = models.Notice(
        name=name,
        content=content,
        user_id=user_id,
        create_time=create_time
    )
    notice.save()
    back_dic = {'flag':True,'msg':'發佈公告成功'}
    common.send_back(back_dic,conn)
admin_interface 
from db import models
from lib import common
from tcp_server import online_user
from interface import user_interface


def register(msg_dic,conn):
    user = models.User.select_one(name=msg_dic['name'])
    if user:
        back_dic = {'flag':False,'msg':'用戶名已存在'}
    else:
        user = models.User(
            name=msg_dic['name'],
            password=msg_dic['password'],
            is_vip=0,
            locked=0,
            user_type=msg_dic['user_type']
        )
        user.save()
        back_dic = {'flag':True,'msg':'用戶註冊成功'}
    common.send_back(back_dic,conn)


def login(msg_dic,conn):
    user = models.User.select_one(name=msg_dic['name'])
    # 用戶存在
    if user:
        # 密碼一致
        if msg_dic['password'] == user.password:
            # 類型一致
            if msg_dic['user_type'] == user.user_type:
                # 登錄成功，伺服器記錄狀態，併發送session給客戶端cookie賦值
                session = common.get_uuid(msg_dic['name']) + msg_dic['name']
                online_user.mutex.acquire()
                online_user.alive_user[msg_dic['addr']] = [session,user.id]
                online_user.mutex.release()
                back_dic = {'flag':True,'session':session,'msg':'登錄成功'}

                # 若登錄用戶為普通用戶，則在客戶端列印最近一條公告
                if msg_dic['user_type'] == 'user':
                    back_notice = user_interface.check_notice_by_count(1)
                    if back_notice:
                        back_dic = {'flag':True,'back_notice':back_notice,'is_vip':user.is_vip,'session':session,'msg':'登錄成功'}
                    else:
                        back_dic = {'flag': True, 'back_notice': '暫無公告內容', 'is_vip': user.is_vip, 'session': session,
                                    'msg': '登錄成功'}
            else:
                back_dic = {'flag': False, 'msg': '用戶登錄類型不一致'}
        else:
            back_dic = {'flag': False, 'msg': '密碼不一致'}
    else:
        back_dic = {'flag':False,'msg':'用戶不存在'}
    common.send_back(back_dic,conn)


def check_moive_list(msg_dic,conn):
    movie_list = models.Movie.select_many()
    back_movie_list = []
    if movie_list:
        for movie in movie_list:
            # 視頻未刪除
            if not movie.is_delete:
                if msg_dic['movie_type'] == 'all':
                    back_movie_list.append([movie.id,movie.name,'免費' if movie.is_free else '收費'])
                elif msg_dic['movie_type'] == 'free':
                    if movie.is_free:
                        back_movie_list.append([movie.id, movie.name, '免費'])
                else:
                    if not movie.is_free:
                        back_movie_list.append([movie.id, movie.name, '收費'])
        if back_movie_list:
            back_dic = {'flag':True,'msg':'查詢成功','back_movie_list':back_movie_list}
        else:
            back_dic = {'flag': False, 'msg': '查詢失敗，視頻數據為空'}
    else:
        back_dic = {'flag':False,'msg':'伺服器上無視頻數據或視頻已被刪除'}
    common.send_back(back_dic,conn)
common_interface 
import os
from db import models
from lib import common


@common.login_auth
def check_notice(msg_dic,conn):
    notice_list = check_notice_by_count()
    back_dic = {'flag':True,'msg':'歷史公告信息如下：','notice_list':notice_list}
    common.send_back(back_dic,conn)


# 內部藉口使用功能
def check_notice_by_count(count=None):
    notice_list = models.Notice.select_many()
    if notice_list:
        back_notice = []
        if not count:
            for notice in notice_list:
                back_notice.append({notice.name: notice.content})
        else:
            notice_list = sorted(notice_list,key=lambda notice:notice.create_time)
            current_notice = notice_list[len(notice_list)-1]
            back_notice.append({current_notice.name:current_notice.content})
        return back_notice
    else:
        return False


@common.login_auth
def charge_vip(msg_dic,conn):
    user = models.User.select_one(id=msg_dic['user_id'])
    user.is_vip = 1
    user.update()
    back_dic = {'flag':True,'msg':'充值成功'}
    common.send_back(back_dic,conn)


@common.login_auth
def download_free_movie(msg_dic,conn):
    movie = models.Movie.select_one(id=msg_dic['movie_id'])
    if movie:
        user = models.User.select_one(id=msg_dic['user_id'])
        if user.is_vip:
            wait_time = 0
        else:
            wait_time = 5

        send_size = 0
        total_size = os.path.getsize(movie.path)
        back_dic = {'flag': True, 'wait_time': wait_time,'file_size':total_size,'movie_id':movie.id}
        common.send_back(back_dic, conn)

        with open(movie.path,'rb') as f:
            while send_size < total_size:
                send_data = f.readline()
                conn.send(send_data)
                send_size += len(send_data)
                percent = send_size / total_size
                common.print_progress(percent)

        download_record = models.Download_Record(
            user_id=msg_dic['user_id'],
            movie_id=movie.id
        )
        download_record.save()


    else:
        back_dic = {'flag':False,'msg':'影片不存在'}
        common.send_back(back_dic, conn)


@common.login_auth
def download_charge_movie(msg_dic,conn):
    movie = models.Movie.select_one(id=msg_dic['movie_id'])
    if movie:
        user = models.User.select_one(id=msg_dic['user_id'])
        if user.is_vip:
            cost_money = 5
        else:
            cost_money = 10

        send_size = 0
        total_size = os.path.getsize(movie.path)
        back_dic = {'flag': True, 'cost_money': cost_money,'file_size':total_size,'movie_id':movie.id}
        common.send_back(back_dic, conn)

        with open(movie.path,'rb') as f:
            while send_size < total_size:
                send_data = f.readline()
                conn.send(send_data)
                send_size += len(send_data)
                percent = send_size / total_size
                common.print_progress(percent)

        download_record = models.Download_Record(
            user_id=msg_dic['user_id'],
            movie_id=movie.id
        )
        download_record.save()


    else:
        back_dic = {'flag':False,'msg':'影片不存在'}
        common.send_back(back_dic, conn)



@common.login_auth
def check_download_record(msg_dic,conn):
    download_record_list = models.Download_Record.select_many(user_id=msg_dic['user_id'])
    back_download_record_list = []
    if download_record_list:
        for record in download_record_list:
            movie_id = record.movie_id
            movie = models.Movie.select_one(id=movie_id)
            back_download_record_list.append('影片信息：%s,%s' % (movie.name,movie.create_time))
        back_dic = {'flag':True,'msg':'查看下載記錄成功：','back_download_record_list':back_download_record_list}

    else:
        back_dic = {'flag':False,'msg':'暫無下載記錄'}
    common.send_back(back_dic,conn)
user_interface

lib

import json
import struct
import hashlib
import time
from tcp_server import online_user


def login_auth(func):
    def wrapper(*args,**kwargs):
        # args = (msg_dic,conn) 元祖
        for value_list in online_user.alive_user.values():
            # value_list = [session,user.id]
            if args[0]['session'] == value_list[0]:
                user_id = value_list[1]
                args[0]['user_id'] = user_id
                break
        user_id = args[0].get('user_id',None)
        if user_id:
            func(*args,**kwargs)
        else:
            back_dic = {'flag': False, 'msg': '非授權用戶'}
            send_back(back_dic, args[1])
    return wrapper


def send_back(back_dic,conn):
    back_dic_json = json.dumps(back_dic)
    back_dic_bytes = back_dic_json.encode('utf-8')
    back_dic_head = struct.pack('i',len(back_dic_bytes))
    conn.send(back_dic_head)
    conn.send(back_dic_bytes)


# session,file_name中使用
def get_uuid(name):
    md = hashlib.md5()
    md.update(str(time.time()).encode('utf-8'))
    md.update(name.encode('utf-8'))
    return md.hexdigest()



def get_current_time():
    return time.strftime('%Y-%m-%d %X',time.localtime(time.time()))



def print_progress(percent):
    width = 70
    sign = '■'
    if percent > 1:
        percent = 1
    print_str = '[%%-%ds]' % width % (int(percent * width) * sign)
    print('\r%s %d%%' % (print_str,int(percent * 100)),end='')



def get_file_uuid(file_name):
    md = hashlib.md5()
    md.update('密碼加鹽'.encode('utf-8'))
    md.update(file_name.encode('utf-8'))
    return md.hexdigest()

if __name__ == '__main__':
    percent = 0
    while percent < 1:
        percent += 0.1
        print_progress(percent)
common

orm_pool

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,
    maxconnections=6,
    mincached=2,
    maxcached=5,
    maxshared=3,
    maxusage=None,
    setsession=[],
    blocking=True,
    ping=0,
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123456',
    database='youku',
    charset='utf8',
    autocommit=True
)
DBUtils_interface 
import pymysql
from orm_pool import DBUtils_interface

class Mysql_pool_interface:
    def __init__(self):
        self.conn = DBUtils_interface.POOL.connection()
        self.cursor = self.conn.cursor(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)
            affected = self.cursor.rowcount
        except Exception as e:
            print(e)
        return affected

if __name__ == '__main__':
    ms = Mysql_pool_interface()
    res = ms.select('select * from userinfo where id = %s',1)
    print(res)
mysql_pool 
from orm_pool import mysql_pool

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 StringField(Field):
    def __init__(self, name=None, column_type='varchar(255)', primary_key=False, default=None):
        super().__init__(name, column_type, primary_key, default)


class IntegerField(Field):
    def __init__(self, name=None, column_type='int', primary_key=False, default=None):
        super().__init__(name, column_type, primary_key, default)


class ModelsMetaclass(type):
    def __new__(cls, name, bases, attrs):
        if name == 'Models':
            return type.__new__(cls, name, bases, attrs)
        table_name = attrs.get('table_name', None)
        if not table_name:
            table_name = name
        primary_key = None
        mappings = dict()
        for k, v in attrs.items():
            if isinstance(v, Field):
                mappings[k] = v
                if v.primary_key:
                    if primary_key:
                        raise KeyError('重覆主鍵：%s' % k)
                    primary_key = k
        for k in mappings.keys():
            attrs.pop(k)
        if not primary_key:
            raise KeyError('無主鍵')
        attrs['table_name'] = table_name
        attrs['primary_key'] = primary_key
        attrs['mappings'] = mappings
        return type.__new__(cls,name,bases,attrs)


class Models(dict,metaclass=ModelsMetaclass):
    def __init__(self,**kwargs):
        super().__init__(**kwargs)

    def __getattr__(self, item):
        try:
            return self[item]
        except KeyError:
            raise AttributeError('無此屬性')

    def __setattr__(self, key, value):
        self[key] = value

    @classmethod
    def<
您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • Python數據類型之list和tuple
    list是一種有序的集合，可以隨時添加和刪除其中的元素。 用len()函數可以獲得list元素的個數。 用索引來訪問list中每一個位置的元素，索引是從0開始的。如果要取最後一個元素，除了計算索引位置外，還可以用-1作索引，直接獲取最後一個元素。以此類推，可以獲取倒數第2個、倒數第3個。 list是 ...
  • 第 6 章 C控制語句：迴圈
    6.16.3 使用嵌套迴圈，按下麵格式列印字母： F FE FED FEDC FEDCB FEDCBA 1 #include <stdio.h> 2 3 int main() 4 { 5 const int ROWS = 6; 6 7 for (int row(0); row != ROWS; ++ ...
  • BZOJ 3053: The Closest M Points(K-D Tree)
    Description The course of Software Design and Development Practice is objectionable. ZLC is facing a serious problem .There are many points in K-dimen ...
  • 數組逆序排列
    方法一： public void reverse(String arr[]){ for(int i=arr.length-1;i>=0;i--){ System.out.println(arr[i]); } } 方法二： public static void reverse2(String arr[ ...
  • Spring_配置Tiles時報錯：org.apache.tiles.template.NoSuchAttributeException：Attribute 'header' not found.
    org.apache.tiles.template.NoSuchAttributeException: Attribute 'header' not found. ...
  • C++學習（八）之 圖形庫
    有關圖形庫的學習筆記 1.安裝 ww.easys.cn 2.創建win32控制台應用程式 .cpp文件（圖形庫必須創建cpp文件） *重點 3.安裝好後 重啟一下vs 圖形庫 是一些函數的集合 作用是做一些界面和滑鼠操作 函數-->幫助文檔 1.基本概念 顏色 RGB值 -->RGB值表示一種顏色 ...
  • Python字元串和編碼
    在最早的時候只有127個字元被編碼到電腦里，也就是大小寫英文字母、數字和一些符號，這個編碼被成為ASCII編碼。 但是要處理中文顯然一個位元組是不夠的，至少需要兩個位元組，而且還不能和ASCII編碼衝突，所以中國制定了GB2312編碼，用來把中文編進去。 世界上有很多語言，各國有各國的標準，就會不可避 ...
  • 生成器和迭代器
    生成器和迭代器 列表生成式 現在有這麼一個需求，要將 這個列表中的每個元素都加1，那麼需要怎麼實現呢？你可能會想到這幾種方式： 其實還有一種寫法，如下： 通過列表生成式，我們可以直接創建一個列表。但是受到記憶體限制，列表容量肯定是有限制的，就像是遞歸，最大遞歸深度python就對其作了限制。而且，創建 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類