服务端增加根据上传附件格式 xlsx 类型，将表格第一个sheet数据批量快速导入数据库
服务端
import socketserver
import json
import os
#import pymysql
import cx_Oracle #Oracle 数据库连接
import time
import tqdm
import pandas as pd
import openpyxl
class MyServer(socketserver.BaseRequestHandler):
def handle(self):
self.add_ip = self.client_address[0]
self.add_post = str(self.client_address[1])
while True:
try:
data = self.request.recv(102400)
#如果获取为空就退出
if not data:break
#否则解码处理数据
self.data = json.loads(data.decode('utf-8'))
# data 是获取字典内容，self.client_address 是 ip地址与 端口
print('客户端的消息:',self.data,self.client_address)
#数据库对访问记录存档
self.log_record()
# 定义处理规则
self.visit_response()
#将结果反馈给客户端
self.request.sendall(self.fankui.encode('utf-8'))
except :
#print('连接异常')
break
#定义处理规则
def visit_response(self):
#获取时间
in_time = self.get_current_time()[0:14]
print('self.data1111',self.data)
if self.data['leixing'] == '文件传递':
try:
file_download_statr,path,file_type = self.file_download()
# 如果文件名类型是 xlsx 就把数据导入数据库
# time.sleep(0.5)
try:
print('file_type', file_type)
if str(file_type).upper() == 'XLSX':
wb = openpyxl.load_workbook(path)
#wb.sheetnames[0] 第一个sheet名
ws = wb[wb.sheetnames[0]]
data_import = pd.read_excel(path, sheet_name=wb.sheetnames[0], header=0, dtype='str',keep_default_na=False);
data_import = data_import.where(data_import.notnull(), None);
file_path_state1 = os.path.exists(path)
print('path', type(path), file_path_state1)
ws_type, excel_nr,sheet_name = excel_type(str(path))
print('结果', ws_type, excel_nr,sheet_name)
table_name = 'ceshi_tmp_' + str(self.add_ip).replace('.', '')
excel_in_state = excel_in_new(table_name, ws_type, excel_nr,path,sheet_name)
print('excel_in_state', excel_in_state)
except:
print('数据导入数据库失败')
self.fankui = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s" }' % (
self.data['leixing'], self.add_ip, excel_in_state, file_download_statr)
except:
self.fankui = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s" }' % (
self.data['leixing'], self.add_ip, self.add_post, '传递失败')
elif self.data['leixing'] == '文件下载': #***********************20230906新增
self.user_ip = self.service_user()
print('user_ip', self.user_ip, type(self.user_ip))
#限制ip地址，不在里面的不允许
if self.add_ip in ('134.35.33.250','134.80.99.107','134.80.98.158','134.80.98.173','134.35.10.10') or self.add_ip in self.user_ip:
# self.user_ip 是通过 shzc.yytowz_service_user 这个表维护
# 获取sql语句生成表格
download_start, file_path, file_name, file_size = Oracle_download(self.data['mac'], in_time,str(self.data['ziduan1']).replace("^","'"))
print('download_start', download_start, '大小字节', file_size, '文件名', file_name)
try:
# 将表格返回客户端
file_download_statr = self.file_up(file_path,self.data['ziduan3'])
self.fankui = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s" }' % (
self.data['leixing'], file_path, file_size, file_download_statr)
except:
self.fankui = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s" }' % (
self.data['leixing'], file_path, file_size, '传递失败')
else:
self.fankui = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s" }' % (
self.data['leixing'], 'file_path', '0', '传递失败')
elif self.data['leixing'] == '发起访问':
try:
self.fankui = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s" }' % (
self.data['leixing'], self.add_ip, self.add_post, '访问成功')
except:
self.fankui = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s" }' % (
self.data['leixing'], self.add_ip, self.add_post, '访问失败')
else:
try:
self.fankui = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s" }' % (
self.data['leixing'], self.add_ip, self.add_post, '访问成功')
except:
self.fankui = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s" }' % (
self.data['leixing'], self.add_ip, self.add_post, '访问失败')
#文件下载函数
def file_download(self):
in_time = self.get_current_time()[0:14]
file_path = self.data['ziduan1']
file_size = self.data['ziduan2']
file_hz = file_path.split('/')[-1]
file_type = file_hz.split('.')[-1]
print('接收文件名：',file_hz,' 接收文件大小：',file_size,' 字节')
# 文件传输的缓冲区
BUFFER_SIZE = 4096
# 接受客户端信息
filename, file_size, new_filename = self.data['ziduan1'], self.data['ziduan2'], str(self.data['ziduan3'])[0:6]
#判断月文件夹是否存在，不存在创建一个
file_path_state1 = os.path.exists('./file_server/'+new_filename)
if file_path_state1 == False:
os.mkdir('./file_server/'+new_filename)
# 获取文件的名字
filename = os.path.basename(filename)
path = './file_server/'+new_filename +'/' + self.data['mac'] + '_' + in_time + '_' + file_hz
print('filename',os.path.isfile(filename),filename)
file_size = int(file_size)
if os.path.isfile(path):
f = open(path, "wb")
else:
f = open(path, "wb")
rece_size = 0
while rece_size < file_size:
data = self.request.recv(BUFFER_SIZE)
f.write(data)
rece_size += len(data)
else:
return '传递成功',path,file_type
def file_up(self,fujian_label,in_time):
# 文件传输的缓冲区
BUFFER_SIZE = 4096
# 传递文件到指定目录下
filename = fujian_label.replace('/', '//')
# 文件大小
file_size = os.path.getsize(filename)
# 创建连接
chuandi_tup = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%d","ziduan3":"%s" }' % (
'文件提取', filename, file_size, in_time)
self.request.sendall(chuandi_tup.encode('utf-8'))
# 文件传输
progress = tqdm.tqdm(range(file_size), f"发送{filename}", unit="B", unit_divisor=1024)
with open(filename, "rb") as f:
for _ in progress:
# 读取文件
bytes_read = f.read(BUFFER_SIZE)
if not bytes_read:
break
# sendall确保及时网络忙碌的时候，数据仍然可以传输
self.request.sendall(bytes_read)
progress.update(len(bytes_read))
# 关闭资源
self.request.close()
#数据库登录
def mysql_execute(self, in_sql, leixing):
# 登录数据库
#conn = pymysql.connect(host='127.0.0.1', port=3306, user='szc', password='szcNSP850219', database='szc_sql',charset='utf8')
dsn = "134.80.200.216/pdbzbjs1"
try:
conn = cx_Oracle.connect(user="zbweb", password="zibo_533_03", dsn=dsn, encoding="UTF-8")
except:
time.sleep(10)
conn = cx_Oracle.connect(user="zbweb", password="zibo_533_03", dsn=dsn, encoding="UTF-8")
# 得到一个可以执行SQL语句的光标对象
cursor = conn.cursor()
# 数据库执行导入的语句
if leixing == '数量':
# 反馈数量
count = cursor.execute(in_sql)
elif leixing == '单条':
# 反馈单条
cursor.execute(in_sql)
count = cursor.fetchone()[0]
elif leixing == '多条':
# 反馈多条
cursor.execute(in_sql)
count = cursor.fetchall()
elif leixing == '编辑':
count = cursor.execute(in_sql)
conn.commit()
# 关闭光标对象
cursor.close()
# 关闭数据库连接
conn.close()
# 反馈
return count
# 时间计算
def get_current_time(self):
ct = time.time()
local_time = time.localtime(ct)
data_head = time.strftime("%Y%m%d%H%M%S", local_time)
data_secs = abs(ct - round(ct)) * 1000
time_stamp = "%s%03d" % (data_head, data_secs)
return time_stamp
#日志留存
def log_record(self):
in_time = self.get_current_time()[0:14]
ziduan1 = str(self.data['ziduan1']).replace("'","^")
sql = "insert into shzc.yytowz_service_title (leixing,ziduan1,ziduan2,ziduan3,ip_id,post_id,in_time,mac,hostname,ip) values('%s','%s','%s','%s','%s','%s','%s','%s','%s','%s') " % (
self.data['leixing'], ziduan1, self.data['ziduan2'], self.data['ziduan3'], str(self.client_address[0]),str(self.client_address[1]),in_time, self.data['mac'], self.data['hostname'], self.data['ip'])
num = self.mysql_execute(sql,'编辑')
#print('num',num)
return num
# shzc.yytowz_service_user 获取文件下载ip范围
def service_user(self):
sql = "select distinct a.ip_id from shzc.yytowz_service_user a where STATUS='1' "
num = self.mysql_execute(sql, '多条')
ip_list = []
for ip in num:
for ip_id in ip:
ip_list.append(ip_id)
return ip_list
# 时间计算
def get_current_time(input_date='0'):
# 如果时间传入为空
if input_date == '0':
ct = time.time() # - 24 * 60 * 60 #如果是取昨天日期是减数值
local_time = time.localtime(ct)
data_head = time.strftime("%Y%m%d%H%M%S", local_time)
data_secs = abs(ct - round(ct)) * 1000
time_stamp = "%s%03d" % (data_head, data_secs)
else:
time_stamp = input_date + '120000001'
return time_stamp
def file_transfer(user,file_name):
#获取月份
in_month = get_current_time()[0:6]
# 文件传递给服务器
file_path = file_name
file_statr = user.file_up(file_path, in_month)
# 如果执行结果不成功，再次执行一次，保底
if file_statr['ziduan3'] != '传递成功':
file_statr = file_transfer(user,file_name)
return file_statr
else:
return file_statr
#这里用作程序预备，目前建立必要的文件夹
def server_init():
# 程序执行前先确认 ./file_server/ 是否存在，不存在新建
file_path_state1 = os.path.exists('./file_server')
if file_path_state1 == False:os.mkdir('./file_server')
file_path_state1 = os.path.exists('./file_server/file_out')
if file_path_state1 == False: os.mkdir('./file_server/file_out')
#文件数据生成，分mac与时间，不然没法同步下载
def Oracle_download(mac,in_time,sql='0'):
# 结果数据生成表格准备发送
dsn = "134.80.200.216/pdbzbjs1"
conn = cx_Oracle.connect(user="zbweb", password="zibo_533_03", dsn=dsn, encoding="UTF-8")
df = pd.read_sql("""%s""" % sql, con=conn)
df.to_excel("./file_server/file_out/"+mac+"_"+in_time+"_结果下载.xlsx", index=False)
# 文件大小
file_path = './file_server/file_out/'+mac+'_'+in_time+'_结果下载.xlsx'
file_name = file_path.split('/')[-1]
file_size = os.path.getsize(file_path)
#返回根据语句处理结果与
return "结果下载",file_path,file_name,file_size
#获取表格数据 #.replace(" ","").replace("[","").replace("]","").replace("'","").split(",")
def excel_type(path):
excel_nr ={}
wb = openpyxl.load_workbook(path)
print('sheetnames',wb.sheetnames)
ws = wb[wb.sheetnames[0]]
sheet_name = wb.sheetnames[0]
minr = ws.min_row
minc = ws.min_column
maxr = ws.max_row
maxc = ws.max_column
# 数据库内容
print(minr, minc, maxr, maxc)
ws_type = {'hang': maxr, 'lie': maxc}
rngs = ws.iter_rows(min_row=minr, min_col=minc, max_row=maxr, max_col=maxc)
row_cs = 0
for row in rngs:
#标题做表头
if row_cs == 0:
value = [c.value for c in row]
# print(len(value))
row_bt = []
for valus_id in range(len(value)):
row_bt.append('y' + str(value[valus_id] +'_'+ str(valus_id)))
excel_nr[row_cs] = row_bt
elif row_cs == 1:
value = [c.value for c in row]
# print(len(value))
row_nr = []
for valus_id in range(len(value)):
row_nr.append(str(value[valus_id]))
excel_nr[row_cs] = row_nr
row_cs += 1
try:
excel_nr.pop(None)
except:
pass
return ws_type, excel_nr,sheet_name
#数据库登录
def mysql_execute(in_sql,leixing):
# 登录数据库
#conn = pymysql.connect(host='127.0.0.1', port=3306, user='szc', password='szcNSP850219', database='szc_sql',charset='utf8')
dsn = "134.80.200.216/pdbzbjs1"
try:
conn = cx_Oracle.connect(user="zbweb", password="zibo_533_03", dsn=dsn, encoding="UTF-8")
except:
time.sleep(10)
conn = cx_Oracle.connect(user="zbweb", password="zibo_533_03", dsn=dsn, encoding="UTF-8")
# 得到一个可以执行SQL语句的光标对象
cursor = conn.cursor()
# 数据库执行导入的语句
if leixing == '数量':
# 反馈数量
count = cursor.execute(in_sql)
elif leixing == '单条':
# 反馈单条
cursor.execute(in_sql)
count = cursor.fetchone()[0]
elif leixing == '多条':
# 反馈多条
cursor.execute(in_sql)
count = cursor.fetchall()
elif leixing == '编辑':
count = cursor.execute(in_sql)
conn.commit()
# 关闭光标对象
cursor.close()
# 关闭数据库连接
conn.close()
# 反馈
return count
def excel_in_new(table_name,ws_type,excel_nr,path,sheet_name):
print('结果',table_name, ws_type)
# 先删除这个表
sql = "drop table zhyw." + table_name
try:mysql_execute(sql, '编辑')
except:pass
# 创建临时表
sql = "create table zhyw." + table_name + " ("
for i in range(ws_type['lie']):
# print('i',i,excel_nr[0][i])
sql = sql + excel_nr[0][i] + " varchar2(100),"
sql = sql[:-1] + ")"
print('sql:', sql)
try:mysql_execute(sql, '编辑')
except:pass
dsn = "134.80.200.216/pdbzbjs1";
conn = cx_Oracle.connect(user="zbweb", password="zibo_533_03", dsn=dsn, encoding="UTF-8");
cursor = conn.cursor();
data_import = pd.read_excel(path, sheet_name=sheet_name, header=0, dtype='str', keep_default_na=False);
data_import = data_import.where(data_import.notnull(), None);
df = data_import.apply(lambda x: tuple(x), axis=1).values.tolist() # dataframe 需要改改为包含元组的列表才可以导入数据库
#deal_export ="insert into zhyw.shc_caigou_kucun_mx values(:1,:2,:3,:4,:5,:6,:7,:8,:9,:10,:11,:12,:13,:14,:15,:16,:17,:18,:19)"
deal_export = "insert into zhyw." + table_name +" values("
for i in range(ws_type['lie']):
deal_export = deal_export + ":" + str(i+1) +","
deal_export = deal_export[:-1] + ")"
print('deal_export',deal_export)
cursor.prepare(deal_export); # 编译sql
print(cursor.executemany(None, df));
conn.commit();
return "zhyw."+table_name
if __name__ == '__main__':
#服务器文件夹准备
server_init()
#服务器开始
s = socketserver.ThreadingTCPServer(('134.35.10.10', 8967), MyServer)
#类似实现连接循环
s.serve_forever()
客户端与之前无变化
import socket
import json
import time
import os
import tqdm
import uuid #获取系统mac
class My_Main():
def __init__(self):
# 程序执行前先确认 ./file_server/file_work_order/ 是否存在，不存在新建
file_path_state1 = os.path.exists('./file_main')
if file_path_state1 == False: os.mkdir('./file_main')
# 定义服务端地址
self.ip_num, self.port_num = '134.35.10.10', 8967
# mac
self.mac = uuid.UUID(int=uuid.getnode()).hex[-12:]
# 获取主机名
self.hostname = socket.gethostname()
# 获取IP
self.ip = socket.gethostbyname(self.hostname)
#发起访问调用模块
def socket_dlgc(self,leixing, name_text, pass_text):
# 获取14位长度时间，年月日时分秒的
self.in_time = get_current_time()[0:14]
tcp_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
tcp_client.connect((self.ip_num, self.port_num))
if None == name_text:print('与服务器断开连接')
#发送
msg = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%s","ziduan3":"%s","mac":"%s","hostname":"%s","ip":"%s" }' % (
str(leixing), str(name_text), str(pass_text), self.in_time,self.mac, self.hostname, self.ip)
tcp_client.send(msg.encode("utf-8")) # 说话 #
data = tcp_client.recv(102400) # 听话
js_data = json.loads(data.decode('utf-8'))
tcp_client.close()
return js_data
# 文件下载函数
def file_download(self,sql):
# 获取14位长度时间，年月日时分秒的
self.in_time = get_current_time()[0:14]
# 套接字是对访问的ip地址和端口反馈，需要从开始定好
tcp_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
tcp_client.connect((self.ip_num, self.port_num))
# 申请数据下载
chuandi_tup = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%d","ziduan3":"%s","mac":"%s","hostname":"%s","ip":"%s" }' % (
'文件下载', sql, 1, self.in_time+'.xlsx',self.mac, self.hostname, self.ip)
tcp_client.send(chuandi_tup.encode("utf-8")) # 说话
self.data = tcp_client.recv(102400) # 听话
js_data = json.loads(self.data.decode('utf-8'))
#print(js_data, type(js_data))
if js_data['ziduan1'] not in ('', None):
file_size = int(js_data['ziduan2'])
#print(file_size)
filename = js_data['ziduan3']
rece_size = 0
recv_data = tcp_client.recv(4096)
if recv_data: # 如果获取数据不为空
try:
with open('./file_main/' + filename, "wb")as f:
f.write(recv_data)
while rece_size < file_size:
recv_data = tcp_client.recv(4096)
f.write(recv_data)
rece_size += len(self.data)
js_data = {'leixing': '文件下载', 'ziduan1': js_data['ziduan3'], 'ziduan2': file_size, 'ziduan3': '传递成功'}
except:
js_data = {'leixing': '文件下载', 'ziduan1': js_data['ziduan3'], 'ziduan2': file_size, 'ziduan3': '传递失败'}
# 关闭套接字
tcp_client.close()
return js_data
#客户端文件上传，fujian_label 文件全路径
def file_up(self,fujian_label):
# 获取14位长度时间，年月日时分秒的
self.in_time = get_current_time()[0:14]
# 文件传输的缓冲区
BUFFER_SIZE = 4096
# 创建连接
s = socket.socket()
s.connect((self.ip_num, self.port_num))
# 传递文件到指定目录下
filename = fujian_label.replace('/', '//')
# 文件大小
file_size = os.path.getsize(filename)
chuandi_tup = '{"leixing":"%s","ziduan1":"%s","ziduan2":"%d","ziduan3":"%s","mac":"%s","hostname":"%s","ip":"%s" }' % (
'文件传递', filename, file_size, self.in_time,self.mac, self.hostname, self.ip )
s.send(chuandi_tup.encode())
# 文件传输
progress = tqdm.tqdm(range(file_size), f"发送{filename}", unit="B", unit_divisor=BUFFER_SIZE)
with open(filename, "rb") as f:
for _ in progress:
# 读取文件
bytes_read = f.read(BUFFER_SIZE)
if not bytes_read:
break
try:
# sendall确保及时网络忙碌的时候，数据仍然可以传输
s.sendall(bytes_read)
progress.update(len(bytes_read))
except:
js_data = {'leixing': '文件传递', 'ziduan1': self.ip_num, 'ziduan2': self.port_num, 'ziduan3': '传递失败'}
break
#文件传递完后，看看是否有反馈，有的话函数返回
try:
data = s.recv(102400) # 听话
js_data = json.loads(data.decode('utf-8'))
except:
js_data = {'leixing': '文件传递', 'ziduan1': self.ip_num, 'ziduan2': self.port_num, 'ziduan3': '传递失败'}
# 关闭资源
s.close()
return js_data
# 时间计算
def get_current_time(input_date='0'):
# 如果时间传入为空
if input_date == '0':
ct = time.time() # - 24 * 60 * 60 #如果是取昨天日期是减数值
local_time = time.localtime(ct)
data_head = time.strftime("%Y%m%d%H%M%S", local_time)
data_secs = abs(ct - round(ct)) * 1000
time_stamp = "%s%03d" % (data_head, data_secs)
else:
time_stamp = input_date + '120000001'
return time_stamp
#文件传递给服务器
def file_transfer(user,file_name,num=0):
# 防止有语法错误等原因导致死循环，限制最多处理4次
if num < 5:
# 文件传递给服务器
try:
file_statr = user.file_up(file_name)
num += 1
except:
file_statr = {'leixing': '文件传递', 'ziduan1': file_name, 'ziduan2': '0', 'ziduan3': '传递失败'}
num += 1
# 如果执行结果不成功，再次执行一次，保底
if file_statr['ziduan3'] != '传递成功':
file_statr = file_transfer(user,file_name,num)
return file_statr
else:
return file_statr
else:
file_statr = {'leixing': '文件传递', 'ziduan1': file_name, 'ziduan2': '0', 'ziduan3': '传递失败'}
return file_statr
#文件传递给服务器
def file_gain(user,sql,num=0):
#防止有语法错误等原因导致死循环，限制最多处理4次
if num < 5:
# 文件传递给服务器
try:
file_statr = user.file_download(sql)
num += 1
except:
file_statr = {'leixing': '文件下载', 'ziduan1': sql, 'ziduan2': '0', 'ziduan3': '传递失败'}
num += 1
# 如果执行结果不成功，再次执行一次，保底
if file_statr['ziduan3'] != '传递成功':
file_statr = file_gain(user,sql,num)
return file_statr
else:
return file_statr
else:
file_statr = {'leixing': '文件下载', 'ziduan1': sql, 'ziduan2': '0', 'ziduan3': '传递失败'}
return file_statr
def use_show():
print('请您选择需要处理的类型(请输入选择的编码)：')
print('类型：发起访问 编码：1')
print('类型：文件传递 编码：2')
print('类型：文件下载 编码：3')
def use_choice(use_input):
if use_input == '1':use_type = '发起访问'
elif use_input == '2':use_type = '文件传递'
elif use_input == '3':use_type = '文件下载'
else:use_type = '类型不详'
return use_type
def use_decision(user,use_input):
if use_input == '1':
name_text = input('请您输入需要发起的标题：')
pass_text = input('请您输入需要发起的内容：')
socket_dlgc_statr = user.socket_dlgc('发起访问', name_text, pass_text)
print('系统反馈：', socket_dlgc_statr)
elif use_input == '2':
wav_write = input('请您输入需要发送文件名：')
transfer_statr = file_transfer(user, wav_write)
print('系统反馈：',transfer_statr)
elif use_input == '3':
sql = sql_out()
gain_statr = file_gain(user, sql)
print('系统反馈：', gain_statr)
else:
use_type = '类型不详'
pass
#获取sql语句处理
def sql_out():
# 语句存储
sql_text = ''
# 循环执行
while True:
sql = input('请您输入需要发送的语句：').replace("'", "^")
try:
sql_out = sql.index(';')
#print('sql_out', sql_out, type(sql_out))
except:
sql_out = -1
#print('sql_out', sql_out, type(sql_out))
if sql_out == -1:
sql_text = sql_text + ' ' + sql
else:
sql_text = sql_text + ' ' + sql
break
return sql_text.replace(";", " ")
if __name__ == '__main__':
# 加载类
user = My_Main()
# 循环执行
while True:
use_show()
use_input = input('请您输入选择的编码：')
use_type = use_choice(use_input)
print('您选择需要处理的类型：',use_type)
#类型不详就断开
if use_type=='类型不详':break
use_decision(user, use_input)
