File size: 1,860 Bytes
c45dff0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | import os
import hashlib
from utils.logger_handler import logger
from langchain_core.documents import Document
from langchain_community.document_loaders import PyPDFLoader, TextLoader
def get_file_md5_hex(filepath: str): # 获取文件的md5的十六进制字符串
if not os.path.exists(filepath):
logger.error(f"[md5计算]文件{filepath}不存在")
return
if not os.path.isfile(filepath):
logger.error(f"[md5计算]路径{filepath}不是文件")
return
md5_obj = hashlib.md5()
chunk_size = 4096 # 4KB分片，避免文件过大爆内存
try:
with open(filepath, "rb") as f: # 必须二进制读取
while chunk := f.read(chunk_size):
md5_obj.update(chunk)
"""
chunk = f.read(chunk_size)
while chunk:
md5_obj.update(chunk)
chunk = f.read(chunk_size)
"""
md5_hex = md5_obj.hexdigest()
return md5_hex
except Exception as e:
logger.error(f"计算文件{filepath}md5失败，{str(e)}")
return None
def listdir_with_allowed_type(path: str, allowed_types: tuple[str]): # 返回文件夹内的文件列表（允许的文件后缀）
files = []
if not os.path.isdir(path):
logger.error(f"[listdir_with_allowed_type]{path}不是文件夹")
return allowed_types
for f in os.listdir(path):
if f.endswith(allowed_types):
files.append(os.path.join(path, f))
return tuple(files)
def pdf_loader(filepath: str, passwd=None) -> list[Document]:
return PyPDFLoader(filepath, passwd).load()
def txt_loader(filepath: str) -> list[Document]:
return TextLoader(filepath, encoding="utf-8").load()
|