jupyter/体质检测管理.ipynb

体质检测管理

体测单位信息导入

In [ ]:

import openpyxl
import json
import pymongo

myclient = pymongo.MongoClient('mongodb://localhost:27017/')
mydb = myclient["tice"]
mycol = mydb["unit"]

name = '中国石油化工股份有限公司安庆炼化分公司'
short_name = '安庆炼化'
myquery = { "name": name}
if mycol.find_one(myquery) is None:    
    wb = openpyxl.load_workbook('data/中国石油化工股份有限公司安庆炼化分公司员工在职人员名单.xlsx')
    sheet = wb.active
    #sheets = wb.sheetnames
    depart = []
    dict1 = {}
    new_code = []
    for n in range(2,sheet.max_row+1):
        m_depart = sheet.cell(n,5).value
        if m_depart not in depart:
            depart.append(sheet.cell(n,5).value)
        #print(sheet.cell(n,5).value)

    dict1['name'] = '中国石油化工股份有限公司安庆炼化分公司'
    dict1['short_name']  = '安庆炼化'
    dict1['depart'] = depart
    mycol.insert_one(dict1)
    print('ok!')
else:
    print('该单位已存在！')    
    

录入体测信息

In [ ]:

import pymongo
from bson.objectid import ObjectId
from dateutil import parser

myclient = pymongo.MongoClient('mongodb://localhost:27017/')
mydb = myclient["tice"]
mycol = mydb["unit"]
mycol1 = mydb["event"]
dict1 = {}

name = '中国石油化工股份有限公司安庆炼化分公司'
myquery = { "name": name}
dict1['code'] = 120
dict1['unit'] = '中国石油化工股份有限公司安庆炼化分公司'
dict1['address'] = '安徽安庆'
dict1['date_begin'] ='2021-09-08'
dict1['date_end'] = '2021-09-30'
if mycol.find_one(myquery) is None:    
    print('体测单位不存在！')
else:
    myquery1 = { "code": dict1['code']}
    if mycol1.find_one(myquery1) is None:
        mycol1.insert_one(dict1)
        print('ok!')
    else:
        print('该体测信息已经存在！')

导出单位员工信息

In [ ]:

import openpyxl
import json
import pymongo
from bson.objectid import ObjectId

myclient = pymongo.MongoClient('mongodb://localhost:27017/')
mydb = myclient["tice"]
mycol = mydb["person"]

wb = openpyxl.load_workbook('data/中国石油化工股份有限公司安庆炼化分公司员工在职人员名单.xlsx')
sheet = wb.active
#sheets = wb.sheetnames
list1 = []
person = {}
new_code = []
for n in range(2,sheet.max_row+1):
    code = sheet.cell(n,1).value
    person.setdefault(code,{})
    dict1 = {}    
    dict1['name'] = sheet.cell(n,2).value
    dict1['sex'] = sheet.cell(n,3).value
    dict1['unit'] = sheet.cell(n,5).value
    if sheet.cell(n,4).value is None:
        dict1['birth'] = sheet.cell(n,6).value
    else:
        dict1['id_num'] = sheet.cell(n,4).value
    person[code] = dict1
filename = '安庆炼化人员名单.json'
with open(filename,'w') as fl:
    json.dump(person, fl,ensure_ascii=False) 
print('ok')

获取测试人员名单导入数据库

In [ ]:

import json
import pymongo
from bson.objectid import ObjectId

myclient = pymongo.MongoClient('mongodb://localhost:27017/')
mydb = myclient["tice"]
mycol = mydb["person"]

db = pymysql.connect("localhost","root","songyi","ydyb" )
cursor = db.cursor()

place_id = 120
exam_person = []
sql = f'SELECT distinct a.avatar_id FROM places_result AS a WHERE a.place_id={place_id} and a.avatar_id < 4999 ORDER BY a.avatar_id '
cursor.execute(sql)
results = cursor.fetchall()
cursor.close
for result in results:
    exam_person.append(str(result[0]).rjust(5,'0'))

filename = '安庆炼化人员名单.json'

list1 = []
list2 = []
with open(filename,'r') as fl:
    all_person = json.load(fl)
with open('result.json','r') as fl1:
    res = json.load(fl1)
for k,v in all_person.items():
    if k in exam_person:
        dict2 = {}
        dict3 = {}
        dict2['name'] = v['name']
        if v['sex'] == '男':
            dict2['sex'] = 'M'
        else:
            dict2['sex'] = 'F'
        if 'birth' in v:
            s = str(v['birth'])
            dict2['birth'] = f'{s[:4]}-{s[4:6]}-{s[6:8]}'
        if 'id_num' in v:
            s = v['id_num']
            dict2['id_num'] = s
            dict2['birth'] = f'{s[6:10]}-{s[10:12]}-{s[12:14]}'
        dict2.setdefault('exam',[])
        dict3['code'] = 120
        dict3['sn'] = k
        dict3['result'] = res[k]
        dict2['exam'].append(dict3)
        list1.append(dict2)
x = mycol.insert_many(list1)
mycol1 = mydb["event"]
myquery = {'code':120}
mycol1.update_one(myquery,{'$push':{'persons':x.inserted_ids}})
print('ok!')

测试项目添加测试人员名单

In [ ]:

import json
from bson.objectid import ObjectId
import pymongo
import pymysql

db = pymysql.connect("localhost","root","songyi","ydyb" )
cursor = db.cursor()

list1 = []
sql = f'SELECT distinct a.avatar_id FROM places_result AS a WHERE a.place_id={place_id} and a.avatar_id < 4999 ORDER BY a.avatar_id '
cursor.execute(sql)
results = cursor.fetchall()
cursor.close
for result in results:
    list1.append(str(result[0]).rjust(5,'0'))

myclient = pymongo.MongoClient('mongodb://localhost:27017/')
mydb = myclient["tice"]
mycol = mydb["event"]
myquery = {'code':120}
mycol.update_one(myquery,{'$push':{'persons':list1}})
    
print('ok')
    
    

获取人员测试成绩

In [ ]:

import pymysql
import json
import time

filename = 'item.json'
item = {}
unit = {}
with open(filename,'r') as fl:
    dict1 = json.load(fl) 
for k,v in dict1.items():
    item[k] = v
user_id = 5
place_id =120
#user = '包永'
db = pymysql.connect("localhost","root","songyi","ydyb" )
cursor = db.cursor()

sql = f'SELECT a.item_id,a.performance,a.score,DATE_FORMAT(a.date,"%Y-%m-%d") as date1,a.avatar_id FROM places_result AS a WHERE a.place_id={place_id} AND a.avatar_id < 4999 ORDER BY a.avatar_id,date1'
cursor.execute(sql)
re_ta = {}
dict1 = {}
print("\n运动项目信息：")
results = cursor.fetchall()
for result in results:    
    user = str(result[4]).rjust(5,'0')
    m_item = str(result[0])
    
    re_ta.setdefault(user,{})
    re_ta[user]['date'] =result[3]
    item_name = item[m_item]['name']
    re_ta[user].setdefault(item_name,{})
    
    #dict1.setdefault(item[m_item]['name'],{})
    score = result[1]/item[m_item]['divisor']
    #dict1[item[m_item]['name']]['成绩'] =f'{score} {item[m_item]["unit"]}'
    #dict1[item[m_item]['name']]['得分'] =result[2]
    re_ta[user][item_name]['成绩'] = f'{score} {item[m_item]["unit"]}'
    re_ta[user][item_name]['得分'] =result[2]
    
    #re_ta[user] = dict1
    #print(re_ta)
filename = 'result.json'
with open(filename,'w') as fl:
    json.dump(re_ta, fl,ensure_ascii=False) 
print('ok')

读取体测报告信息

In [ ]:

import pdfplumber

# 读取 PDF 文档
pdf = pdfplumber.open("5.pdf")

# 获取页数
print("总页数：",len(pdf.pages))
print("-----------------------------------------")

# 读取第 4 页；索引从 1 开始
page = pdf.pages[1] 
print("本页：",page.page_number + 1)
print("-----------------------------------------")
text = page.extract_text()
#for s in text:
#    print(s)
print(text)
print(text.split('\n'))

In [ ]:

import pdfplumber
import pymysql
"""获取人员测试成绩"""
user_id = 5
place_id =120
db = pymysql.connect("localhost","root","songyi","ydyb" )
cursor = db.cursor()
sql = f'SELECT a.item_id,a.performance,a.score,a.date FROM places_result AS a WHERE a.place_id={place_id} AND a.avatar_id={user_id}'
cursor.execute(sql)
re_ta = {}
print("\n运动项目信息：")
results = cursor.fetchall()
for result in results:
#    print ("%d--%s" %(result[0],result[1]))
    re_ta[result[0]] = result[1]

dict1 = {}

# 读取 PDF 文档
pdf = pdfplumber.open("5.pdf")

# 获取页数
num = len(pdf.pages)
page = pdf.pages[0] 
text = page.extract_text()
dict1['name'] = text.split('\n')[2]
dict1['sn'] = text.split('\n')[4]


print(dict1)

体测项目信息

In [ ]:

import pymysql
import json

db = pymysql.connect("localhost","root","songyi","ydyb" )
cursor = db.cursor()
sql = "SELECT a.id,a.name, b.name,b.divisor FROM items AS a,items_unit AS b WHERE a.unit_id = b.id"
cursor.execute(sql)
dict1 = {}

results = cursor.fetchall()
for result in results:
#    print ("%d--%s" %(result[0],result[1]))
    item = {}
    id = result[0]
    dict1.setdefault(id,{})
    item['name'] = result[1]
    item['unit'] = result[2]
    item['divisor'] = result[3]
    dict1[id] = item
filename = 'item.json'
with open(filename,'w') as fl:
    json.dump(dict1, fl,ensure_ascii=False) 
with open(filename,'r') as fl:
    dict1 = json.load(fl) 
for k,v in dict1.items():
    print(k,v)

In [ ]:

import json
filename = 'item.json'
item = {}
with open(filename,'r') as fl:
    dict1 = json.load(fl) 
for k,v in dict1.items():
    item[k] = v
print(item['1'])

压缩体检报告

In [ ]:

from pdf2image import convert_from_path, convert_from_bytes
import img2pdf  
import glob
import os,sys
import tempfile
from pdf2image.exceptions import (
    PDFInfoNotInstalledError,
    PDFPageCountError,
    PDFSyntaxError
)
#images = convert_from_path('1.pdf',dpi=200,fmt='jpg', output_folder='./sample_data')
fl_path = 'file/210924'
new_path ='file/new'
old = glob.glob(f'{fl_path}/*.pdf')
old.sort()
for old_file in old:
    s = os.path.basename(old_file).split('.')[0].rjust(5,'0')
    new_file = f'{new_path}/{s}.pdf'
    
    with tempfile.TemporaryDirectory() as path:
        images_from_path = convert_from_path(old_file, dpi=100,fmt='jpg', output_folder=path)
        fl=glob.glob(f'{path}/*.jpg')
        fl.sort()
        a4inpt = (img2pdf.mm_to_pt(210),img2pdf.mm_to_pt(297))
        layout_fun = img2pdf.get_layout_fun(a4inpt)
        with open(new_file,"wb") as f:
            f.write(img2pdf.convert(fl,layout_fun=layout_fun))


    #print(path)

    print(new_file,'ok')
#print(old)

In [ ]: