|
|
|
""" |
|
Created on Sun Jun 4 15:55:12 2023 |
|
|
|
@author: wooji |
|
""" |
|
import re |
|
import streamlit as st |
|
import numpy as np |
|
import pandas as pd |
|
import streamlit as st |
|
import pandas as pd |
|
from io import StringIO |
|
import pdfplumber |
|
import xlrd |
|
import re |
|
import pdfplumber |
|
import pandas as pd |
|
import time |
|
import os |
|
import numpy as np |
|
|
|
|
|
import docx2pdf |
|
import docx |
|
|
|
|
|
import pytesseract |
|
from PIL import Image |
|
import os |
|
from pdf2image import convert_from_path,convert_from_bytes |
|
from io import BytesIO |
|
|
|
import openpyxl |
|
import base64 |
|
st.title("MSDS报告CAS号提取程序") |
|
|
|
from docx import Document |
|
|
|
def get_tables(docx_path): |
|
docStr = Document(docx_path) |
|
numTables = docStr.tables |
|
my_list = [] |
|
for table in numTables: |
|
row_count = len(table.rows) |
|
col_count = len(table.columns) |
|
for i in range(row_count): |
|
row = table.rows[i].cells |
|
for j in range(col_count): |
|
content = row[j].text |
|
my_list.append(content) |
|
my_list = ';'.join(my_list).strip('') |
|
return my_list |
|
|
|
|
|
def get_paragraphs(docx_path): |
|
|
|
document = Document(docx_path) |
|
|
|
all_paragraphs = document.paragraphs |
|
paragraph_texts = [] |
|
|
|
for paragraph in all_paragraphs: |
|
paragraph_texts.append(paragraph.text) |
|
paragraph_texts = ';'.join(paragraph_texts).strip('') |
|
return paragraph_texts |
|
|
|
|
|
|
|
|
|
def openpdf(path): |
|
with pdfplumber.open(path) as pdf: |
|
|
|
item = [] |
|
for page in pdf.pages: |
|
text = page.extract_text() |
|
item.append(text) |
|
|
|
item = ';'.join(item).strip('') |
|
return item |
|
|
|
|
|
def extract(text,cas): |
|
pattern = re.compile(cas,re.S) |
|
r_list = pattern.findall(text) |
|
return r_list |
|
|
|
|
|
st.write('使用说明') |
|
st.caption('支持解析的格式:.pdf(扫描版或非扫描版均支持)和.docx。可将MSDS文件夹直接拖拽到下方上传区域') |
|
st.write('excel输出内容详解') |
|
st.caption('第一列为文件名称,所有上传的文件均会显示在第一列,即便该文件格式不支持提取') |
|
st.caption('第二列为文件中提取的CAS号,若为空则表明未提取到') |
|
st.caption('第三列为化学物质名称,仅支持显示与清单匹配成功的化学物质的名称') |
|
st.caption('第四列为匹配结果,共3种结果:3960种、优评优控、重点管控') |
|
st.caption('第五列为备注,共3种结果:1、不支持该格式文件,请手动查看:说明此类文件不支持解析,请手动查看;2、图片pdf,建议人工复核:说明该pdf为图片,提取正确率较低,视情况可进行人工复核;3、未检测到CAS,请手动检查:说明在该文件中未检测到CAS,请人工确认') |
|
|
|
st.caption('提取速度:提取一个电子pdf大约耗时4s,一个扫描版pdf大约耗时10~20s。具体速度由pdf的页数决定') |
|
st.divider() |
|
uploaded_file = st.file_uploader("请上传MSDS报告,可直接往里拖拽文件夹",accept_multiple_files=True) |
|
@st.cache_data |
|
def main(uploaded_file): |
|
data = pd.DataFrame(columns=['CAS','名称','匹配结果','备注']) |
|
|
|
begin = time.time() |
|
|
|
cas = r'[0-9]+-[0-9][0-9]-[0-9][^0-9]' |
|
|
|
for file in range(len(uploaded_file)): |
|
if uploaded_file[file].name[-4:] == 'docx': |
|
text = get_paragraphs(uploaded_file[file]) |
|
|
|
|
|
elif uploaded_file[file].name[-3:] == 'pdf' or uploaded_file[file].name[-3:] == 'PDF': |
|
text = openpdf(uploaded_file[file]) |
|
else: |
|
cas_set = pd.DataFrame({'备注':{uploaded_file[file].name:'不支持该格式文件,请手动查看'}}) |
|
data = pd.concat([data,cas_set],axis=0) |
|
continue |
|
cas_extract = extract(text,cas) |
|
if cas_extract != []: |
|
for item in range(len(cas_extract)): |
|
cas_iso = cas_extract[item] |
|
cas_iso = cas_iso[0:len(cas_iso)-1] |
|
cas_set = pd.DataFrame({'CAS':{uploaded_file[file].name:cas_iso}}) |
|
data = pd.concat([data,cas_set],axis=0) |
|
|
|
elif uploaded_file[file].name[-4:] == 'docx': |
|
text = get_tables(uploaded_file[file]) |
|
|
|
cas_extract = extract(text,cas) |
|
if cas_extract != []: |
|
for item in range(len(cas_extract)): |
|
cas_iso = cas_extract[item] |
|
cas_iso = cas_iso[0:len(cas_iso)-1] |
|
cas_set = pd.DataFrame({'CAS':{uploaded_file[file].name:cas_iso}}) |
|
data = pd.concat([data,cas_set],axis=0) |
|
else: |
|
cas_set = pd.DataFrame({'备注':{uploaded_file[file].name:'未检测到CAS,请手动检查'}}) |
|
data = pd.concat([data,cas_set],axis=0) |
|
else: |
|
pages = convert_from_bytes(uploaded_file[file].getvalue()) |
|
text = [] |
|
for i,page in enumerate(pages): |
|
buf = BytesIO() |
|
page.save(buf,format="JPEG") |
|
buf.seek(0) |
|
img_page=Image.open(buf) |
|
|
|
txt=pytesseract.image_to_string(img_page,lang='chi_sim') |
|
text.append(txt) |
|
text = ';'.join(text).strip('') |
|
cas_extract = extract(text,cas) |
|
if cas_extract != []: |
|
cas_extract = extract(text,cas) |
|
for item in range(len(cas_extract)): |
|
cas_iso = cas_extract[item] |
|
cas_iso = cas_iso[0:len(cas_iso)-1] |
|
print(cas_iso) |
|
|
|
|
|
cas_set = pd.DataFrame({'CAS':{uploaded_file[file].name:cas_iso},'备注':{uploaded_file[file].name:'图片pdf,建议人工复核'}}) |
|
data = pd.concat([data,cas_set],axis=0) |
|
else: |
|
cas_set = pd.DataFrame({'备注':{uploaded_file[file].name:'未检测到CAS,请手动检查'}}) |
|
data = pd.concat([data,cas_set],axis=0) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
data_reset_index = data.reset_index(drop=False) |
|
|
|
data_rename = data_reset_index.rename(columns={'index':'MSDS文件名称'}) |
|
|
|
data_output = data_rename.drop_duplicates() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
target_data = pd.read_excel('物质清单.xlsx',sheet_name='总表',index_col=0) |
|
target_cas = target_data[['CAS','名称','清单']] |
|
target_cas = target_cas.reset_index(drop=True) |
|
|
|
|
|
|
|
for row in data_output.index: |
|
|
|
for b in target_cas.index: |
|
if data_output.loc[row]['CAS'] == target_cas.loc[b]['CAS']: |
|
data_output.loc[row]['匹配结果'] =target_cas.loc[b]['清单'] |
|
data_output.loc[row]['名称'] = target_cas.loc[b]['名称'] |
|
|
|
|
|
data_final = data_output |
|
|
|
end = time.time() |
|
run_time = end - begin |
|
st.write('运行耗时:'+ str(round(run_time,2))+'秒') |
|
return data_final |
|
|
|
|
|
if uploaded_file == []: |
|
st.stop() |
|
else: |
|
data_final = main(uploaded_file) |
|
data_final |
|
data_final.to_excel('resuls.xlsx') |
|
wb2 = openpyxl.load_workbook('resuls.xlsx') |
|
wb2.save('results.xlsx') |
|
data=open('results.xlsx','rb').read() |
|
b64 = base64.b64encode(data).decode('UTF-8') |
|
excel_name = st.text_input(':blue[请输入本次导入的文件所属企业名称,若为空则导出的excel默认取名为myresult.xlsx]') |
|
st.warning('建议示例:广西xx企业-原辅料 or 广西xx企业-产品 ------- 输入完请按回车 ', icon="🚨") |
|
if excel_name: |
|
excel_name = excel_name + '.xlsx' |
|
href = f'<a href="data:file/data;base64,{b64}" download={excel_name}>导出excel</a>' |
|
st.markdown(href, unsafe_allow_html=True) |
|
wb2.close() |
|
else: |
|
href = f'<a href="data:file/data;base64,{b64}" download=myresult.xlsx>导出excel</a>' |
|
st.markdown(href, unsafe_allow_html=True) |
|
wb2.close() |
|
|
|
|
|
st.subheader('!!!单次使用完请刷新页面后再上传新的文件') |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|