Spaces:
Running
Running
File size: 4,281 Bytes
8576e2e | 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 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """
Path utilities for handling MinerU file paths and prefix matching
"""
import os
import logging
from pathlib import Path
from typing import Optional
logger = logging.getLogger(__name__)
def convert_mineru_path_to_local(mineru_path: str, project_root: Optional[Path] = None) -> Optional[Path]:
"""
将 /files/mineru/{extract_id}/{rel_path} 格式的路径转换为本地文件系统路径
Args:
mineru_path: MinerU URL 路径，格式为 /files/mineru/{extract_id}/{rel_path}
project_root: 项目根目录路径（如果为 None，则自动计算）
Returns:
本地文件系统路径（Path 对象），如果转换失败则返回 None
"""
try:
if not mineru_path.startswith('/files/mineru/'):
return None
# Remove '/files/mineru/' prefix
rel_path = mineru_path.replace('/files/mineru/', '')
# Get project root if not provided
if project_root is None:
# Navigate to project root (assuming this file is in backend/utils/)
current_file = Path(__file__).resolve()
backend_dir = current_file.parent.parent
project_root = backend_dir.parent
# Construct full path: {project_root}/uploads/mineru_files/{rel_path}
local_path = project_root / 'uploads' / 'mineru_files' / rel_path
return local_path
except Exception as e:
logger.warning(f"Failed to convert MinerU path to local: {mineru_path}, error: {str(e)}")
return None
def find_mineru_file_with_prefix(mineru_path: str, project_root: Optional[Path] = None) -> Optional[Path]:
"""
查找 MinerU 文件，支持前缀匹配
首先尝试直接路径匹配，如果失败则尝试前缀匹配。
前缀匹配逻辑：如果文件名看起来像是一个前缀+扩展名（前缀长度 >= 5），
则在目录中查找以该前缀开头的文件。
Args:
mineru_path: MinerU URL 路径，格式为 /files/mineru/{extract_id}/{rel_path}
project_root: 项目根目录路径（如果为 None，则自动计算）
Returns:
找到的文件路径（Path 对象），如果未找到则返回 None
"""
# First try direct path conversion
local_path = convert_mineru_path_to_local(mineru_path, project_root)
if local_path is None:
return None
# Direct file matching
if local_path.exists() and local_path.is_file():
return local_path
# Try prefix match using the generic function
return find_file_with_prefix(local_path)
def find_file_with_prefix(file_path: Path) -> Optional[Path]:
"""
查找文件，支持前缀匹配
首先检查文件是否存在，如果不存在则尝试前缀匹配。
前缀匹配逻辑：如果文件名看起来像是一个前缀+扩展名（前缀长度 >= 5），
则在目录中查找以该前缀开头的文件。
Args:
file_path: 要查找的文件路径（Path 对象）
Returns:
找到的文件路径（Path 对象），如果未找到则返回 None
"""
# Direct file matching
if file_path.exists() and file_path.is_file():
return file_path
# Try prefix match if not found and filename looks like a prefix with extension
filename = file_path.name
dirpath = file_path.parent
if '.' in filename and dirpath.exists() and dirpath.is_dir():
prefix, ext = os.path.splitext(filename)
if len(prefix) >= 5:
try:
for fname in os.listdir(dirpath):
fp, fe = os.path.splitext(fname)
if fp.lower().startswith(prefix.lower()) and fe.lower() == ext.lower():
matched_path = dirpath / fname
if matched_path.is_file():
logger.debug(f"Prefix match found: {file_path} -> {matched_path}")
return matched_path
except OSError as e:
logger.warning(f"Failed to list directory {dirpath}: {str(e)}")
return None
|