Python全棧項目（電子詞典、協程、pdb調試）

什麼是項目？

軟體項目：實作一定完整性功能的代碼

軟體項目的開發流程

需求分析

概要設計

項目規劃

詳細設計

編碼測試

項目測試

調試修改

項目釋出

後期維護更新

需求分析：确定使用者的真實需求

1. 确定使用者的真實需求，即項目的基本功能

2. 對項目的難度和可行性進行分析

3. 完成需求分析文檔，進行确認

概要設計：對項目進行整體分析，初步确定技術方向

1. 整體設計，确定項目架構

2. 确定項目功能子產品劃分

3. 确定大的技術方向

4. 編寫項目概要設計文檔，開發流程圖

項目規劃：确定項目分工，按照項目時限進行規劃

1. 确定開發順序

2. 确定開發的時間軸和裡程碑

3. 人員的配置設定

4. 完成甘特圖和思維導圖指導開發

詳細設計：項目具體的開發設計，完成設計手冊

1.根據開發内容，形成詳細設計文檔

思路，邏輯流程，功能說明，技術點說明，資料結構，代碼說明，注意事項，預期效果，環境限制

編碼測試：按照規劃完成編碼，做基本的測試工作

1.寫代碼

2.代碼基本測試

3.技術攻關

4.代碼整合

項目測試 : 對項目整體功能進行測試

1. 跨平台性，是否符合環境，功能bug，壓力測試

2. 完成測試報告

3. 根據測試結果修改bug

項目釋出

1.将項目送出給使用者，進行釋出使用

2.完成項目使用文檔

後期維護

1.處理使用中出現的問題

2.項目的更新和功能的添加

項目注意事項

1.按時完成項目開發是首要工作

2.項目實施人員的沖突問題

3.無計劃的實施必要帶來後期更大的效率低下

項目開發工具

文檔編寫：

word ppt markdowm LaTex

項目流程圖：

Mindmanager Xmind visio

項目管理：

project

代碼管理工具：

git svn

編輯工具：

pycharm Webstream eclipse

sublime vim vscode atom

電子詞典項目文檔 功能說明

：

1. 使用者可以登入注冊

登入憑借使用者名密碼即可如果輸入不正确可以重複輸入

注冊要求使用者有填寫使用者名密碼，且使用者名不能重複。其他資訊随意

2. 使用者資訊可以長期儲存在伺服器，保證下次登入可以使用

3. 能夠滿足多個使用者端程式同時操作的情況

4. 功能分為用戶端和服務端，用戶端運作後即進入第一界面

第一界面：登入注冊退出

5. 登入成功後進入第二界面

第二界面：查詞檢視曆史記錄退出

6. 功能說明

登入：

選擇登入功能輸入使用者名密碼，如果成功進入第二界面，不成功保持在第一界面，提示失敗原因

注冊：

選擇注冊功能，填寫資訊，成功後可以保持第一界面或者使用新注冊使用者直接完成登入到第二界面，失敗提示失敗原因

第一界面退出：

直接退出用戶端

第二界面退出 :

第二界面退出相當于登出，即退回到第一界面

查詞：

可以循環輸入單詞，顯示出單詞詞義

輸入q表示查詞結束回到第二界面。如果查詢的詞不存在則有相應提示

*單詞本：每一行一個單詞

單詞和解釋之間一定有空格

單詞有序排列

1. 文本查找 2.資料庫查找

曆史記錄：

選擇檢視曆史記錄即列印出使用者的查詢記錄

可以列印所有記錄也可以列印最近10條。

name word time

項目分析

子產品 : socket 套接字

pymysql/pymongo

os multiprocessing threading select

确定服務端和用戶端分為哪些功能，

每個功能要做什麼工作 服務端：

main（）：

建立套接字，父子程序，子程序處理用戶端請求，父程序接受新的連接配接

資料庫比對

傳回結果

判斷是否重複

插入資料庫傳回注冊成功

使用者存在傳回注冊失敗

query 接受使用者單詞

通過資料庫或者檔案查找單詞

将單詞結果傳回給使用者

如果沒有查到傳回相應資訊

如果查詞成功則插入曆史記錄

history 接受客戶請求

查詢資料庫傳回曆史記錄

如果使用者沒有曆史記錄則傳回資訊

用戶端：

main：建立套接字 ---> 連接配接 ---> 列印一級界面

發送給服務端

接受傳回結果，如果成功則跳轉到二級界面

失敗列印結果

接受傳回結果

query ：循環輸入單詞

發送單詞給服務端

接受結果并列印

history ：發送請求 ---> 接受結果列印

确定建立什麼樣的資料表，表的結構，

将表建立起來

user ： id name passwd

create table user (
id int auto_increment primary key,
name varchar(32) not null,
passwd varchar(16) default '000000');

hist ： id name word time

create table hist (
id int auto_increment primary key,
name varchar(32) not null,
word varchar(64) not null,
time varchar(64));

words ： id word interpret

create table words (
id int auto_increment primary key,
word varchar(64),
interpret text);

如果要使用資料庫查詞則程式設計

将單詞本内容存入資料庫

import pymysql 
import re 

def main():
    # 擷取詞典檔案流對象
    f = open('dict.txt')
    # 連結資料庫并建立遊标對象
    db = pymysql.connect\
    ('localhost','root','123456','dict')
    cursor = db.cursor()
    # 循環讀取每一行内容
    for line in f:
        try:
            L = re.split("[ ]+",line)
        except Exception:
            pass 
        sql = "insert into words (word,interpret)\
         values ('%s','%s')"%(L[0],' '.join(L[1:]))

        # 處理破壞SQL的特殊字元
        try:
            cursor.execute(sql)
            db.commit()
        except Exception:
            db.rollback()
            
    f.close()


if __name__ == "__main__":
    main()

4. 搭建架構

，實作通信（建立套接字，設定結構，建立并發）

伺服器端：

#!/usr/bin/env python3
#coding=utf-8

'''
name : Paris
date : 2018-8-27
email : [email protected]
modules: python3.5  mysql  pymysql
This is a dict project for AID
'''

from socket import *
import os 
import pymysql
import time
import sys 
import signal 

DICT_TEXT = "./dict.txt"
HOST = '0.0.0.0'
PORT = 8000
ADDR = (HOST,PORT)

#主要制流程
def main():
    #連接配接資料庫
    db = pymysql.connect\
    ('localhost','root','123456','dict')

    #建立流式套接字
    s = socket()
    s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
    s.bind(ADDR)
    s.listen(5)
    #或略子程序退出
    signal.signal(signal.SIGCHLD,signal.SIG_IGN)

    while True:
        try:
            c,addr = s.accept()
            print("Connect from",addr)
        except KeyboardInterrupt:
            s.close()
            sys.exit("伺服器退出")
        except Exception as e:
            print(e)
            continue 
        #建立子程序處理用戶端請求
        pid = os.fork() 
        if pid == 0:
            s.close()
            do_child(c,db)
        else:
            c.close()

def do_child(c,db):
    #循環接收請求
    while True:
        data = c.recv(128).decode()
        print("Request:",data)
        if (not data) or data[0] == 'E':
            c.close()
            sys.exit(0)
        elif data[0] == 'R':
            do_register(c,db,data) 
        elif data[0] == "L":
            do_login(c,db,data)
        elif data[0] == 'Q':
            do_query(c,db,data)
        elif data[0] == 'H':
            do_history(c,db,data)


def do_register(c,db,data):
    pass


def do_login(c,db,data):
    pass

def do_query(c,db,data):
    pass


def do_history(c,db,data):
    pass

          
if __name__ == "__main__":
    main()

用戶端：

#!/usr/bin/env python3
#coding=utf-8

from socket import *
import sys 
import getpass

def main():
    if len(sys.argv) < 3:
        print("argv is error")
        return
    HOST = sys.argv[1]
    PORT = int(sys.argv[2])
    ADDR = (HOST,PORT)

    s = socket()
    s.connect(ADDR)

    while True:
        print('''\n
            ===========Welcome=========
            --1.注冊   2.登入  3.退出--
            ===========================
            ''')
        try:
            cmd = int(input("輸入選項>>"))
        except Exception:
            print("輸入指令錯誤")
            continue  

        if cmd not in [1,2,3]:
            print("對不起，沒有該指令")
            sys.stdin.flush() #清除輸入
            continue 
        elif cmd == 1:
            name = do_register(s)
            if name != 1:
                print("注冊成功,直接登入！")
                login(s,name)
            else:
                print("注冊失敗！")
        elif cmd == 2:
            name = do_login(s)
            if name != 1:
                print("登入成功！")
                login(s,name)
            else:
                print("登入失敗！")
        elif cmd == 3:
            s.send(b"E")
            sys.exit("謝謝使用")

def do_register(s):
    pass


def do_login(s):
    pass


def login(s,name):
    pass


def do_query(s,name):
    pass


def do_history(s,name):
    pass


if __name__ == "__main__":
    main()

5. 實作具體架構優化和具體功能

完整代碼：

#!/usr/bin/env python3
#coding=utf-8

'''
name : Paris
date : 2018-8-27
email : [email protected]
modules: python3.5  mysql  pymysql
This is a dict project for AID
'''

from socket import *
import os 
import pymysql
import time
import sys 
import signal 


DICT_TEXT = "./dict.txt"
HOST = '0.0.0.0'
PORT = 8000
ADDR = (HOST, PORT)


# 主要制流程
def main():
    # 連接配接資料庫
    db = pymysql.connect\
    ('localhost', 'root', '123456', 'dict')

    # 建立流式套接字
    s = socket()
    s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    s.bind(ADDR)
    s.listen(5)
    # 或略子程序退出
    signal.signal(signal.SIGCHLD, signal.SIG_IGN)

    while True:
        try:
            c,addr = s.accept()
            print("Connect from", addr)
        except KeyboardInterrupt:
            s.close()
            sys.exit("伺服器退出")
        except Exception as e:
            print(e)
            continue 
        # 建立子程序處理用戶端請求
        pid = os.fork() 
        if pid == 0:
            s.close()
            do_child(c,db)
        else:
            c.close()


def do_child(c,db):
    # 循環接收請求
    while True:
        data = c.recv(128).decode()
        print("Request:", data)
        if (not data) or data[0] == 'E':
            c.close()
            sys.exit(0)
        elif data[0] == 'R':
            do_register(c, db, data) 
        elif data[0] == "L":
            do_login(c, db, data)
        elif data[0] == 'Q':
            do_query(c, db, data)
        elif data[0] == 'H':
            do_history(c, db, data)


def do_register(c, db, data):
    l = data.split(' ')
    name = l[1]
    passwd = l[2]

    cursor = db.cursor()
    sql = \
    "select * from user where name='%s'" % name
    cursor.execute(sql)
    r = cursor.fetchone()
    if r != None:
        c.send(b'EXISTS')
        return 

    sql = "insert into  user (name, passwd)\
     values ('%s', '%s')" % (name, passwd)
    try:
        cursor.execute(sql)
        db.commit()
        c.send(b'OK')
    except:
        db.rollback()
        c.send(b'FALL')
        return
    else:
        print("%s注冊成功" % name)


def do_login(c, db, data):
    l = data.split(' ')
    name = l[1]
    passwd = l[2]
    cursor = db.cursor()

    sql = "select * from user where \
    name='%s' and passwd='%s'" % (name, passwd)

    cursor.execute(sql)
    r = cursor.fetchone()
    if r == None:
        c.send('使用者名或密碼不正确'.encode())
    else:
        c.send(b'OK')

def do_query(c, db, data):
    l = data.split(' ')
    name = l[1]
    word = l[2]
    cursor = db.cursor()

    def insert_history():
        tm = time.ctime()
        sql = "insert into hist (name, word, time)\
         values ('%s', '%s', '%s')" % (name, word, tm)
        try:
            cursor.execute(sql)
            db.commit()
        except:
            db.rollback()
            return


    try:
        f = open(DICT_TEXT, 'rb')
    except:
        c.send("500 服務端異常".encode())
        return
    while True:
        line = f.readline().decode()
        w = line.split(' ')[0]
        if (not line) or w > word:
            c.send("沒找到該單詞".encode())
            break 
        elif w == word:
            c.send(b'OK')
            time.sleep(0.1)
            c.send(line.encode())
            insert_history()
            break
    f.close()


def do_history(c, db, data):
    name = data.split(' ')[1]
    cursor = db.cursor()

    try:
        sql = "select * from hist \
        where name='%s'" % name
        cursor.execute(sql)
        r = cursor.fetchall()
        if not r:
            c.send('沒有曆史記錄'.encode())
            return
        else:
            c.send(b'OK')
    except:
        c.send("資料庫查詢錯誤".encode())
        return
    n = 0
    for i in r:
        n += 1
        #最多顯示10條
        if n > 10:
            break
        time.sleep(0.1)
        msg = "%s   %s   %s" % (i[1], i[2], i[3])
        c.send(msg.encode())
    time.sleep(0.1)
    c.send(b'##')

          
if __name__ == "__main__":
    main()

#!/usr/bin/env python3
#coding=utf-8

from socket import *
import sys
import getpass


def main():
    if len(sys.argv) < 3:
        print("argv is error")
        return
    HOST = sys.argv[1]
    PORT = int(sys.argv[2])
    ADDR = (HOST, PORT)

    s = socket()
    s.connect(ADDR)

    while True:
        print('''\n
            ===========Welcome=========
            --1.注冊   2.登入   3.退出--
            ===========================
            ''')
        try:
            cmd = int(input("輸入選項>>"))
        except Exception:
            print("輸入指令錯誤")
            continue  

        if cmd not in [1, 2, 3]:
            print("對不起，沒有該指令")
            sys.stdin.flush()  # 清除輸入
            continue
        elif cmd == 1:
            name = do_register(s)
            if name != 1:
                print("注冊成功,直接登入！")
                login(s, name)
            else:
                print("注冊失敗！")
        elif cmd == 2:
            name = do_login(s)
            if name != 1:
                print("登入成功！")
                login(s, name)
            else:
                print("登入失敗！")
        elif cmd == 3:
            s.send(b"E")
            sys.exit("謝謝使用")


def do_register(s):
    while True:
        name = input("使用者名:")
        passwd = getpass.getpass("密 碼:")
        passwd1 = getpass.getpass("确認密碼:")

        if (' ' in name) or (' ' in passwd):
            print("使用者名密碼不允許空格")
            continue
        if passwd != passwd1:
            print("兩次密碼不一緻")
            continue

        msg = "R {} {}".format(name, passwd)
        # 發送請求
        s.send(msg.encode())
        # 接收回複
        data = s.recv(128).decode()

        if data == "OK":
            return name
        elif data == 'EXISTS':
            print("該使用者已存在")
            return 1
        else:
            return 1


def do_login(s):
    name = input("使用者名:")
    passwd = getpass.getpass("密 碼:")
    msg = "L {} {}".format(name, passwd)
    s.send(msg.encode())
    data = s.recv(128).decode()

    if data == 'OK':
        return name
    else:
        print(data)
        return 1


def login(s, name):
    while True:
        print('''\n
            ===========查詢界面============
            1.查詞     2.曆史記錄   3.登出
            =============================
            ''')
        try:
            cmd = int(input("輸入選項>>"))
        except Exception:
            print("指令錯誤")
            continue
        if cmd not in [1,2,3]:
            print("對不起，沒有該指令")
            sys.stdin.flush()  # 清除輸入
            continue 
        elif cmd == 1:
            do_query(s, name)
        elif cmd == 2:
            do_history(s, name)
        elif cmd == 3:
            return


def do_query(s, name):
    while True:
        word = input("單詞:")
        if word == "q":
            break 
        msg = "Q {} {}".format(name, word)
        s.send(msg.encode())
        data = s.recv(128).decode()
        if data == 'OK':
            data = s.recv(2048).decode()
            print(data)
        else:
            print(data)


def do_history(s, name):
    msg = "H {}".format(name)
    s.send(msg.encode())
    data = s.recv(128).decode()
    if data == 'OK':
        while True:
            data = s.recv(1024).decode()
            if data == "##":
                break
            print(data)
    else:
        print(data)


if __name__ == "__main__":
    main()

import getpass

passwd = getpass.getpass()

功能：隐藏密碼輸入

pdb調試：

通過pdb子產品完成調試功能

功能：斷點設定，單步運作，函數檢視，代碼段檢視，變量值檢視等...

break ， b 設定斷點

continue ， c 繼續執行

list ， l 檢視目前代碼段

next， n 單步執行

step， s 進入函數單步執行

pp 列印變量值

help 幫助

pdb.set_trace()

功能：設定初始斷點，開始進入pdb調試模式

以pdb調試模式運作

python3 -m pdb dict_client.py 協程：

定義：纖程，微線程。協程本質隻是一個單線程程式

工作原理：

通過應用層層程式，記錄上下文的執行棧。實作程式在執行過程中的跳躍執行，

選擇可以不能阻塞的部分執行，這樣就可以大大提高IO執行的效率。

yield 是 python實作協程的基本關鍵字

安裝第三方子產品：

sudo pip3 install greenlet

sudo pip3 install gevent

greenlet

子產品：

greenlet.greenlet() 生成協程對象

gr.switch() 選擇要執行的協程事件

gevent

1. 将要執行的事件封裝為函數

2. 生成協程對象

gevent.spawn(func,argv)

功能：

将事件變為協程

參數：

func 綁定的協程函數

argv 給函數傳遞參數

傳回值：

協程對象

3. 回收協程

gevent.joinall([obj1,obj2.....])（obj：協程對象）

4. 協程阻塞

gevent.sleep(n)

Python全棧項目（電子詞典、協程、pdb調試）

繼續閱讀

neo4j之cypher使用文檔

Cloud Studio初體驗

使用 ctypes 進行 Python 和 C 的混合程式設計

【python】【資料處理】畫多元資料分布圖

NOSQL安全攻擊

mybatis_入門程式Mybatis入門

samba伺服器的功能

登入plsql 報錯 the account is locked --使用者被鎖

sqlServer根據經緯查距離

【Linux】UDP廣播封包接收速率問題

SequoiaDB巨杉資料庫C++驅動概述

【python】netconf協定對接管理裝置

「Python 網絡自動化」NETCONF —— Python 使用 NETCONF 管理配置 H3C 網絡裝置

Linux裝置模型（中）之上層容器

PowerPC平台 Linux移植三

在python中建立excel并寫入

Python全棧 項目（電子詞典、協程、pdb調試）

繼續閱讀

Python全棧項目（電子詞典、協程、pdb調試）