代碼經過網絡搜索,綜合算是原創吧.py腳本為服務端

項目文件在

https://github.com/jinjie412/service_client_socket

import socketserver

import json

import base64

import os

from te import OPMysql,Caltime

import time

#數據庫操作

opm = OPMysql()

#加密發包內容

def crypt(source,key):

from itertools import cycle

result=''

temp=cycle(key)

for ch in source:

result=result+chr(ord(ch)^ord(next(temp))^ord('1'))

return result

def jiami(f,key):

if os.path.isfile(f):

f=open(f,encoding='gbk')

content=f.read()

# print(content)

f.close()

print(content)

# content=content.encode(encoding='gbk')

content=crypt(content,key)

print(content)

content=crypt(content,key)

print(content)

content=content.encode(encoding='gbk')

print(content)

file_size = len(content)

return file_size,content

else:

return None,False

#數據庫搜索客戶端發送來的信息有效性

def yz(datajs):

sql = "select id,user,dqsj,rjbh from kh where user='%s' and rjbh='%s'" %(datajs['user'],datajs['rjbh'])

res = opm.op_select(sql)

#驗證用戶有效性

if res:

# print(res)

timeArray = time.strptime(res[2], "%Y-%m-%d %H:%M:%S")

# print(timeArray)

#驗證用戶卡是否到期

dt=Caltime(time.localtime(),timeArray)

return dt

else:

return False

class myTCPhandler(socketserver.BaseRequestHandler):

def handle(self):

while True:

self.data = self.request.recv(1024).decode('gbk', 'ignore').strip()

if not self.data : break

self.data=base64.b64decode(self.data)

print(self.data)

datajs = json.loads(self.data)

print(datajs)

if yz(datajs):

#發送數據長度,發送數據內容

dsize,data = jiami('./js/'+datajs["jb"],datajs['key'])

if dsize:

#向客戶端發送數據長度

self.feedback_data = ('%d' % (dsize,)).encode('gbk')

self.request.sendall(self.feedback_data)

#等待客戶端返回

feedback = self.request.recv(100).decode('gbk', 'ignore').strip() #等待客戶端確認

print(feedback)

if int(feedback) == dsize:

#發送所有數據

self.request.sendall(data)

else:

print('open file error')

else:

print('erro')

self.feedback_data = 'erro'.encode('gbk')

self.request.sendall(base64.b64encode(self.feedback_data))

def finish(self):

print("client is disconnect!")

host = '127.0.0.1'

port = 9090

server = socketserver.ThreadingTCPServer((host,port),myTCPhandler)

server.serve_forever()

te.py 文件

使用了數據庫連接池,網絡搜索

import MySQLdb

from DBUtils.PooledDB import PooledDB

import time

import datetime

mysqlInfo = {

"host": 'localhost',

"user": 'root',

"passwd": '000000',

"db": 'w',

"port": 3306,

"charset": 'utf8'

}

class OPMysql(object):

__pool = None

def __init__(self):

# 構造函數，創建數據庫連接、游標

self.coon = OPMysql.getmysqlconn()

self.cur = self.coon.cursor()

# 數據庫連接池連接

@staticmethod

def getmysqlconn():

if OPMysql.__pool is None:

__pool = PooledDB(creator=MySQLdb, mincached=5, maxcached=20, host=mysqlInfo['host'], user=mysqlInfo['user'], passwd=mysqlInfo['passwd'], db=mysqlInfo['db'], port=mysqlInfo['port'], charset=mysqlInfo['charset'])

print(__pool)

return __pool.connection()

# 插入\更新\刪除sql

def op_insert(self, sql):

pass

# 查詢

def op_select(self, sql):

try:

# print('op_select', sql)

self.cur.execute(sql) # 執行sql

select_res = self.cur.fetchone() # 返回結果為字典

# select_res =self.cur.fetchall()

return select_res

except Exception as e:

print(str(e))

return False

#釋放資源

def dispose(self):

self.coon.close()

self.cur.close()

#計算到期時間

def Caltime(date1,date2):

date1=datetime.datetime(date1[0],date1[1],date1[2],date1[3],date1[4],date1[5])

date2=datetime.datetime(date2[0],date2[1],date2[2],date2[3],date2[4],date2[5])

return date2>date1

if __name__ == '__main__':

#申請資源

opm = OPMysql()

sql = "select id,user,dqsj,rjbh from w_kh where user='000' and rjbh='000'"

res = opm.op_select(sql)

print(res)

timeArray = time.strptime(res[2], "%Y-%m-%d %H:%M:%S")

print(timeArray)

dt=Caltime(time.localtime(),timeArray)

print(dt)

最后于 2018-5-5 12:27

被進口堅果編輯

，原因：

總結

以上是生活随笔為你收集整理的python 服务端与c++客户端通讯_[原创]python socket 服务端 与 c++客户端通讯,发包内容加密,支持大文件,并发...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。