在“个人网盘”应用中的路径选择框（移动文件和复制文件的功能需要用到），出现无法展开处于ISO镜像文件中的IMG镜像文件内目录的问题
问题涉及代码
该问题涉及了以下几种函数类型及该类型下的函数（带*为大型函数，因函数内容过多，位于最后）：
- 路径解析与虚拟路径处理函数
| 函数名 | 功能 |
| _get_archive_ext() | 获取归档文件扩展名 |
| split_virtual_path() | 将虚拟路径拆分为归档相对路径和内部路径 |
| _is_archive_or_image() | 检查是否为支持的归档/镜像文件 |
| is_virtual_path() | 检查路径是否包含 :: 分隔符 |
| _nested_cache_key() | 生成嵌套归档缓存目录名 |
| _nested_cache_dir() | 创建嵌套缓存目录 |
| _get_or_extract_nested_inner() | 从外层归档提取内层归档到缓存 | - 格式检测函数
| 函数名 | 功能 |
| _detect_image_format() | 按二进制签名检测镜像真实格式 |
| _is_iso9660_file() | 判断文件是否真的是 ISO9660（可交给 pycdlib）| - 提取函数
| 函数名 | 功能 |
| _extract_nested_archive() | 从外层归档提取内层归档 | *
| _get_nested_archive_path() | 处理多层嵌套虚拟路径 |
| _extract_virtual_path_item() | 从归档（支持多层嵌套）中提取目标内容 | *
| _extract_archive_item() | 从外层压缩包/镜像中提取指定路径的文件到临时目录 | *
| list_nested_virtual_directory() | 处理嵌套压缩包/镜像：提取内部归档并列出其内容 | - 目录列表函数
| 函数名 | 功能 |
| list_virtual_directory() | 列出虚拟目录（压缩包/镜像内部）的文件和文件夹 | * - API 路由
| 函数名 | 功能 |
| api_dir_tree() | 普通目录树 API |
| api_dir_tree_with_path_route() | 路径选择树 API 路由 |
| _api_dir_tree_nested() | 处理多层嵌套虚拟路径 | *
| api_dir_tree_with_path() | 路径选择树 API 核心逻辑 | * - ISO/IMG 专用提取函数
| 函数名 | 功能 |
| _extract_iso_item() | 从 ISO/镜像文件中提取指定项 | *
| _extract_iso_to_dir() | 使用 pycdlib 或 pytsk3 将 ISO/镜像文件内容提取到临时目录 | *
问题表现
- 在进入程序的前端后（html），通过点击文件的移动和复制按钮打开路径选择框，在通过小箭头展开ISO镜像文件中的IMG镜像文件时，发现展开后的内容空白
- 环境因素排除：所有涉及的库已正确安装，ISO镜像文件和IMG镜像文件完好无损坏，可以被直接正常访问
- IMG镜像文件目录展开问题补充：在物理目录中的IMG镜像文件可以被正常展开目录，不存在该问题
- 问题于2天前被测试发现，本人已经实在找不出可能的原因，报错库是pycdlib，推测可能的3种原因：
- 路径的处理逻辑问题
- 从ISO镜像文件中提取IMG镜像文件时出现问题
- IMG镜像文件在有虚拟目录的路径中时有原来的目录展开逻辑失效
问题报错
程序报错日志(因编辑器原因导致显示不完全一致，但核心日志都显示正确)
�失败: Valid ISO9660 filesystems must have at least one PVD
INFO
:
127.0.0.1 - - [28/Aug/2026 16:41:11] “GET /api/dir_tree?path=BOOT/DOS71INS.IMG::MS-DOS%207.10.iso::BOOT HTTP/1.1” 200 -
INFO:main:Request from :127.0.0.1
INFO:main:[开发模式] 脚本位置: d:\个人系统数据\桌面\天星网盘\app.py
INFO:main:运行时目录: d:\个人系统数据\桌面\天星网盘
DEBUG:main:api_dir_tree_with_path called: path=BOOT/DOS71INS.IMG::MS-DOS 7.10.iso::BOOT/DOS71INS.IMG::, user=第三名尔阳
INFO:main:[缓存状态] nested_f072c8b771bc414c86d58754bf07172a 标记为使用中
DEBUG:main:extract_archive_item: archive_path=d:\个人系统数据\桌面\天星网盘\uploads\第三名尔阳\BOOT/DOS71INS.IMG, inner_path=MS-DOS 7.10.iso, temp_dir=C:\Users\HUAWEI\AppData\Local\Temp\天星网盘cache\nested_f072c8b771bc414c86d58754bf07172a, ext=.img, target=C:\Users\HUAWEI\AppData\Local\Temp\天星网盘_cache\nested_f072c8b771bc414c86d58754bf07172a\MS-DOS 7.10.iso
DEBUG:__main:pycdlib 打开失败: Valid ISO9660 filesystems must have at least one PVD
INFO:main:[缓存状态] nested_801db0ea92f0491db22fa08b29dab950 标记为使用中
DEBUG:main
archive_path=C:\Users\HUAWEI\AppData\Local\Temp\天星网盘_cache\nested_f072c8b771bc414c86d58754bf07172a\MS-DOS 7.10.iso, inner_path=BOOT/DOS71INS.IMG, temp_dir=C:\Users\HUAWEI\AppData\Local\Temp\天星网盘_cache\nested_801db0ea92f0491db22fa08b29dab950, ext=.iso, target=C:\Users\HUAWEI\AppData\Local\Temp\天星网盘_cache\nested_801db0ea92f0491db22fa08b29dab950\DOS71INS.IMG
INFO:127.0.0.1 - - [28/Aug/2026 16:41:13] “[35m[1mGET /api/dir_tree?path=BOOT/DOS71INS.IMG::MS-DOS%207.10.iso::BOOT/DOS71INS.IMG:: HTTP/1.1[0m” 500 -浏览器network报错
请求网址：http://127.0.0.1:8000/api/dir_tree?path=1%2FMS-DOS%207.10.iso%3A%3ABOOT%2FDOS71INS.IMG%3A%3A
响应：{“error”: “\u89e3\u6790 ISO \u5931\u8d25\uff1aValid ISO9660 filesystems must have at least one PVD”}
涉及所有函数内容
路径解析与虚拟路径处理函数
def _get_archive_ext(archive_path):
"""获取归档文件的扩展名，特殊处理 .tar.gz/.tar.bz2 等复合扩展名和 ISO 9660 ;1 后缀"""
lower = archive_path.lower()
if lower.endswith('.tar.gz') or lower.endswith('.tgz'):
return '.tar.gz'
if lower.endswith('.tar.bz2') or lower.endswith('.tbz'):
return '.tbz'
# ISO 9660 版本后缀 ;1 会被 splitext 误识别为扩展名
if lower.endswith(';1'):
base = archive_path[:-2] # 去掉 ;1
return os.path.splitext(base)[1].lower()
return os.path.splitext(archive_path)[1].lower()def split_virtual_path(path):
"""将虚拟路径拆分为归档/镜像文件相对路径和内部路径。
支持嵌套归档/镜像（例如 outer.zip::inner.tar::file.txt 解析为 ['outer.zip', 'inner.tar', 'file.txt']）
返回 (archive_rel, inner) 的形式，其中 inner 是路径。
如果是多层嵌套，内部归档会通过 _extract_nested_archive 提取后处理。
"""
if not path:
return '', ''
normalized = path.strip('/')
# 只保留末尾最后一组 :: 表示"归档根目录"标记
# 例如 'middle.zip::innermost.zip::' -> 'middle.zip::innermost.zip::' (保留)
# 例如 'middle.zip::innermost.zip::subdir/' -> 'middle.zip::innermost.zip::subdir/' (保留)
original_stripped = normalized.rstrip('/')
has_trailing_colon = original_stripped.endswith('::')
if has_trailing_colon:
normalized = original_stripped[:-2]
if not normalized:
# 处理纯 '::' 或 '' 的情况
if path.strip('/').endswith('::'):
return '', ''
return '', ''
# 只按第一个 :: 拆分
if '::' in normalized:
idx = normalized.index('::')
archive_rel = normalized[:idx].strip('/')
inner = normalized[idx+2:].strip('/')
# 如果原始路径末尾有 ::，说明 inner 后面还有一个归档根目录标记
if has_trailing_colon and inner:
inner += '::'
elif has_trailing_colon and not inner:
# 只有单个归档名 + :: 的情况，如 'archive.zip::'
return archive_rel, ''
return archive_rel, inner
# 没有 :: 的情况
if has_trailing_colon:
# 如 'archive.zip::' 已经被处理为 normalized = 'archive.zip'
return normalized, ''
return normalized, ''def _is_archive_or_image(filename):
"""检查是否为支持的归档/镜像文件"""
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
ext = os.path.splitext(filename)[1].lower()
return ext in archive_exts or ext in image_extsdef is_virtual_path(path):
"""检查路径是否为虚拟路径（包含 :: 分隔符表示进入归档/镜像内部）"""
return '::' in (path or '')def _nested_cache_key(archive_full, inner_path):
"""根据外层归档路径 + mtime + size + 内部路径 + 代码版本号生成 16 位 sha1 哈希作为缓存子目录名"""
try:
mtime = os.path.getmtime(archive_full)
size = os.path.getsize(archive_full)
except OSError:
mtime = 0
size = 0
key_str = f"{_NESTED_CACHE_VERSION}|{archive_full}|{mtime}|{size}|{inner_path}"
return hashlib.sha1(key_str.encode('utf-8')).hexdigest()[:16]def _nested_cache_dir(user_dir):
"""在 user_dir/.nested_cache/ 下创建隐藏缓存目录，注册进缓存管理系统，
由 GC 线程按 300 秒无活动规则自动回收"""
cache_dir = os.path.join(user_dir, '.nested_cache')
os.makedirs(cache_dir, exist_ok=True)
mark_cache_in_use(cache_dir)
return cache_dirdef _get_or_extract_nested_inner(user_dir, archive_full, archive_rel, inner_check, inner_filename):
"""从外层归档中提取内层归档到 user_dir/.nested_cache/<hash>/ 下。
首次访问时提取；后续访问直接复用缓存，不再反复解压。
返回 (extracted_path, cache_subdir)。失败时抛出异常。"""
cache_base = _nested_cache_dir(user_dir)
cache_key = _nested_cache_key(archive_full, inner_check)
cache_subdir = os.path.join(cache_base, cache_key)
os.makedirs(cache_subdir, exist_ok=True)
mark_cache_in_use(cache_subdir)
# 检查缓存是否已存在
extracted = _find_extracted_file(cache_subdir, inner_filename)
if not extracted:
extracted = _find_extracted_file(cache_subdir, inner_check)
if extracted:
refresh_cache_activity(cache_subdir)
return extracted, cache_subdir
# 首次访问：从外层归档提取
try:
_extract_archive_item(archive_full, inner_check, cache_subdir)
except Exception as e:
logger.warning(f"提取虚拟目录中的归档失败: {inner_check}, error: {e}")
raise
extracted = _find_extracted_file(cache_subdir, inner_filename)
if not extracted:
extracted = _find_extracted_file(cache_subdir, inner_check)
if not extracted:
raise FileNotFoundError(f"嵌套归档不存在: {inner_filename}")
return extracted, cache_subdir格式检测函数
def _detect_image_format(file_path):
"""按二进制签名检测镜像真实格式，而不是信任扩展名。
判定依据（两种格式二进制结构完全不兼容）：
- ISO9660: 卷描述符从逻辑扇区16开始（2048B扇区时偏移0x8000），
PVD 类型字节(0x01)在 0x8000，标识符 "CD001" 在 0x8001~0x8005。
- FAT/MBR: 引导扇区第0字节为跳转指令 0xEB/0xE9，偏移510-511为 0x55AA。
返回:
'iso9660' -> 可安全交给 pycdlib
'fat' -> 磁盘镜像，必须交给 pytsk3
None -> 无法识别，同样交给 pytsk3 尝试
把 FAT 镜像交给 pycdlib 必然抛出
"Valid ISO9660 filesystems must have at least one PVD"。
"""
try:
if not file_path or not os.path.isfile(file_path):
return None
size = os.path.getsize(file_path)
if size < 512:
return None
with open(file_path, 'rb') as f:
# 1) ISO9660：类型字节必须为 0x01（PVD），否则 pycdlib 仍会报缺 PVD
if size >= 0x8006:
f.seek(0x8000)
vd = f.read(6)
if vd[0] == 0x01 and vd[1:6] == b'CD001':
return 'iso9660'
# 2) FAT / MBR：跳转指令 + 0x55AA 签名
f.seek(0)
head = f.read(512)
if len(head) >= 512 and head[510:512] == b'\x55\xAA' \
and head[0] in (0xEB, 0xE9):
return 'fat'
return None
except Exception:
return Nonedef _is_iso9660_file(file_path):
"""轻量判断：该文件是否真的是 ISO9660（可安全交给 pycdlib 打开）"""
return _detect_image_format(file_path) == 'iso9660'
def _find_extracted_file(temp_dir, expected_name):
"""
在临时目录中查找提取后的文件，支持：
- 精确匹配
- 空格/+号互换匹配
- 大小关感不敏感匹配
- ISO 9660 ;1 后缀忽略匹配
- 递归搜索子目录（支持嵌套归档查找）
"""
if not expected_name:
return None
base = os.path.basename(expected_name)
# 1. 精确匹配（直接路径）
path = os.path.join(temp_dir, expected_name.replace('/', os.sep))
if os.path.exists(path):
return path
# 2. 精确匹配（basename）
path = os.path.join(temp_dir, base)
if os.path.exists(path):
return path
# 3. 空格/+号互换匹配
for variant in [base.replace('+', ' '), base.replace(' ', '+')]:
path = os.path.join(temp_dir, variant)
if os.path.exists(path):
return path
# 4. 大小写不敏感匹配（根目录）
base_lower = base.lower()
try:
for f in os.listdir(temp_dir):
if f.lower() == base_lower:
return os.path.join(temp_dir, f)
except Exception:
pass
# 5. ISO ;1 后缀匹配 (忽略大小写及 ;1 后缀)
base_normalized = base_lower.replace(';1', '')
try:
for f in os.listdir(temp_dir):
f_normalized = f.lower().replace(';1', '')
if f_normalized == base_normalized:
return os.path.join(temp_dir, f)
except Exception:
pass
# 6. 递归搜索子目录（修复4层+虚拟目录和移动/复制到嵌套归档的bug）
try:
for root, dirs, files in os.walk(temp_dir):
for f in files:
if f == base or f.lower() == base_lower:
return os.path.join(root, f)
f_normalized = f.lower().replace(';1', '')
if f_normalized == base_normalized:
return os.path.join(root, f)
# 也检查目录名（因为 _find_extracted_file 有时用于查找目录路径中的归档）
for d in dirs:
d_path = os.path.join(root, d)
if expected_name.replace('/', os.sep) == d_path or expected_name.replace('/', os.sep) == d_path.replace('\\', os.sep):
return d_path
except Exception:
pass
return None提取函数
def _extract_nested_archive(user_dir, archive_rel, inner_name):
"""
从外层归档/镜像中提取内层归档/镜像文件到临时目录。
返回临时文件的绝对路径。
"""
archive_full = os.path.join(user_dir, archive_rel)
ext = _get_archive_ext(archive_full)
temp_dir = create_cache_dir(prefix='nested_archive_')
temp_file = os.path.join(temp_dir, os.path.basename(inner_name))
try:
if ext == '.zip':
with zipfile.ZipFile(archive_full, 'r') as zf:
use_gbk = False
has_fpx_utf8 = False
for info in zf.infolist():
if info.flag_bits & 0x0800:
has_fpx_utf8 = True
break
if not has_fpx_utf8:
for info in zf.infolist():
fn = info.filename
if isinstance(fn, str):
if any(ord(c) > 127 and ord(c) < 256 for c in fn):
try:
fn.encode('latin-1').decode('gbk')
use_gbk = True
break
except Exception:
pass
elif isinstance(fn, bytes):
use_gbk = True
break
matched = False
for info in zf.infolist():
fn = info.filename
if use_gbk:
try:
fn = fn.encode('latin-1').decode('gbk')
except Exception:
try:
fn = fn.encode('cp437').decode('gbk')
except Exception:
pass
if fn == inner_name or fn.rstrip('/') == inner_name:
with zf.open(info.filename) as src, open(temp_file, 'wb') as dst:
shutil.copyfileobj(src, dst)
matched = True
break
if not matched:
raise FileNotFoundError(f'内层归档 {inner_name} 未找到')
elif ext in ('.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.tbz'):
with tarfile.open(archive_full, 'r') as tf:
for member in tf.getmembers():
if member.name == inner_name or member.name.rstrip('/') == inner_name:
if member.isfile():
src = tf.extractfile(member)
if src:
with open(temp_file, 'wb') as dst:
shutil.copyfileobj(src, dst)
break
else:
raise FileNotFoundError(f'内层归档 {inner_name} 未找到')
elif ext == '.7z' and HAS_PY7ZR:
with py7zr.SevenZipFile(archive_full, mode='r') as zf:
zf.extract(targets=[inner_name], path=temp_dir)
extracted = os.path.join(temp_dir, inner_name)
if os.path.exists(extracted):
if extracted != temp_file:
os.rename(extracted, temp_file)
temp_file = extracted
else:
raise FileNotFoundError(f'内层归档 {inner_name} 提取失败')
elif ext == '.rar' and HAS_RARFILE:
with rarfile.RarFile(archive_full) as rf:
rf.extract(inner_name, temp_dir)
extracted = os.path.join(temp_dir, inner_name)
if os.path.exists(extracted):
if extracted != temp_file:
os.rename(extracted, temp_file)
temp_file = extracted
else:
raise FileNotFoundError(f'内层归档 {inner_name} 提取失败')
elif ext in ('.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'):
_nested_extracted = False
# 尝试 pycdlib（仅对 .iso 启用；.img 是 FAT/NTFS，pycdlib 打不开会抛 PVD 错误）
if HAS_PYCDLIB and _is_iso9660_file(archive_full):
try:
iso = pycdlib.PyCdlib()
iso.open(archive_full)
inner_lower = inner_name.lower()
inner_lower_no_ver = inner_lower.replace(';1', '')
actual_path = None
# 精确大小写不敏感匹配
for dir_path, dirs, files in iso.walk(iso_path='/'):
for f in files:
full_path = (dir_path.rstrip('/') + '/' + f).lstrip('/')
if full_path.lower() == inner_lower:
actual_path = full_path
break
if actual_path:
break
# 去掉 ;1 版本后缀后匹配
if not actual_path:
for dir_path, dirs, files in iso.walk(iso_path='/'):
for f in files:
full_path = (dir_path.rstrip('/') + '/' + f).lstrip('/')
full_lower = full_path.lower().replace(';1', '')
if full_lower == inner_lower_no_ver or full_lower == inner_lower:
actual_path = full_path
break
if actual_path:
break
# Rock Ridge
if not actual_path and iso.has_rock_ridge():
for dir_path, dirs, files in iso.walk(rr_path='/'):
for f in files:
full_path = (dir_path.rstrip('/') + '/' + f).lstrip('/')
if full_path.lower() == inner_lower or full_path.lower().replace(';1', '') == inner_lower_no_ver:
actual_path = f"rr://{full_path}"
break
if actual_path:
break
# Joliet
if not actual_path and iso.has_joliet():
for dir_path, dirs, files in iso.walk(joliet_path='/'):
for f in files:
full_path = (dir_path.rstrip('/') + '/' + f).lstrip('/')
if full_path.lower() == inner_lower or full_path.lower().replace(';1', '') == inner_lower_no_ver:
actual_path = f"joliet://{full_path}"
break
if actual_path:
break
if not actual_path:
raise FileNotFoundError(f"ISO中未找到: {inner_name}")
os.makedirs(os.path.dirname(temp_file), exist_ok=True)
with open(temp_file, 'wb') as dst_fp:
if actual_path.startswith('rr://'):
iso_fp = actual_path[5:]
iso.get_file_from_iso_fp(dst_fp, rr_path=iso_fp)
elif actual_path.startswith('joliet://'):
iso_fp = actual_path[9:]
iso.get_file_from_iso_fp(dst_fp, joliet_path=iso_fp)
else:
iso_path_arg = '/' + actual_path if not actual_path.startswith('/') else actual_path
iso.get_file_from_iso_fp(dst_fp, iso_path=iso_path_arg)
_nested_extracted = True
iso.close()
except FileNotFoundError:
raise
except Exception:
pass
# 统一落到 pytsk3 回退（pytsk3 可读 FAT/NTFS/ISO9660/分区表镜像）
if not _nested_extracted and not HAS_PYTSK3:
raise FileNotFoundError(f'无法从镜像 {archive_full} 中提取 {inner_name}')
if not _nested_extracted and HAS_PYTSK3:
try:
img = pytsk3.Img_Info(archive_full)
try:
fs = pytsk3.FS_Info(img)
except Exception:
# 整盘镜像（带 MBR/分区表），尝试第一个分区
fs = None
try:
vs = pytsk3.Volume_Info(img)
for part in vs:
# 跳过扩展分区/空分区
if part.len <= 0:
continue
desc = (part.desc.decode('utf-8', errors='replace') if isinstance(part.desc, bytes) else str(part.desc)) if part.desc else ''
if desc.lower().startswith(('extended', 'unallocated', '')) and part.start == 0:
# 空描述且 start==0 时可能是整个盘，但 VS 存在通常是分区表；继续找第一个可分区
pass
try:
fs = pytsk3.FS_Info(img, offset=part.start * 512)
break
except Exception:
fs = None
continue
except Exception:
fs = None
if fs is not None:
_open_matched = None
target_lower = inner_name.lower().rstrip(';1')
# 根目录精确打开
for try_path in (
'/' + inner_name,
'/' + inner_name.upper(),
'/' + inner_name.lower(),
'/' + target_lower,
):
try:
_open_matched = fs.open(try_path)
break
except Exception:
_open_matched = None
# 找不到就遍历根目录（兼容 FAT 短文件名/LFN）
if _open_matched is None:
try:
root = fs.open_dir(path='/')
for entry in root:
try:
en = entry.info.name.name
ename = en.decode('utf-8', errors='replace') if isinstance(en, bytes) else str(en)
except Exception:
continue
if not ename or ename in ('.', '..'):
continue
if ename.lower().rstrip(';1') == target_lower:
try:
_open_matched = entry
break
except Exception:
_open_matched = None
except Exception:
_open_matched = None
if _open_matched is not None and getattr(_open_matched.info, 'meta', None):
ftype = _open_matched.info.meta.type
if ftype == pytsk3.TSK_FS_META_TYPE_DIR:
raise IsADirectoryError(inner_name)
size = _open_matched.info.meta.size or 0
os.makedirs(os.path.dirname(temp_file), exist_ok=True)
with open(temp_file, 'wb') as dst:
offset = 0
while offset < size:
chunk = _open_matched.read_random(offset, min(65536, size - offset))
if not chunk:
break
dst.write(chunk)
offset += len(chunk)
_nested_extracted = True
try:
_open_matched.close()
except Exception:
pass
try:
fs.close()
except Exception:
pass
try:
img.close()
except Exception:
pass
except IsADirectoryError:
raise
except Exception:
pass
if not _nested_extracted:
raise FileNotFoundError(f'无法从镜像 {archive_full} 中提取 {inner_name}')
else:
raise NotImplementedError('不支持的归档格式')
if not os.path.exists(temp_file):
raise FileNotFoundError(f'内层归档提取后文件不存在')
return temp_file
except Exception:
# 清理临时目录
try:
shutil.rmtree(temp_dir, ignore_errors=True)
except Exception:
pass
raisedef _get_nested_archive_path(user_dir, virtual_path):
"""
处理嵌套归档/镜像路径，返回 (temp_archive_path, inner_path, temp_dir)。
temp_archive_path: 最内层归档/镜像的临时文件路径（或原始路径如果不嵌套）
inner_path: 最内层归档中的内部路径
temp_dir: 临时目录（调用者负责清理，或者 None 表示不需要清理）
如果不需要嵌套处理，返回 (archive_full, inner, None)
"""
parts = [p.strip('/\\') for p in virtual_path.split('::')]
parts = [p for p in parts if p]
if len(parts) <= 2:
# 简单情况：只有一个归档
archive_rel, inner = split_virtual_path(virtual_path)
return os.path.join(user_dir, archive_rel), inner, None
# 多层嵌套：从外向内逐层提取
temp_dir = None
current_archive_rel = parts[0]
current_archive_path = os.path.join(user_dir, current_archive_rel)
for i in range(1, len(parts) - 1):
inner_name = parts[i]
extracted_path = _extract_nested_archive(user_dir, current_archive_rel, inner_name)
# 此时 extracted_path 是一个临时文件，需要更新 current_archive_rel
# 但 _extract_nested_archive 依赖 archive_rel 相对于 user_dir 的路径
# 所以需要用更通用的方式处理后续层级
if temp_dir is None:
temp_dir = os.path.dirname(extracted_path)
# 对于后续层级，需要将当前归档作为"新的user_dir"处理
# 简化处理：直接提取最外层，然后逐层处理
current_archive_rel = extracted_path
current_archive_path = extracted_path
inner = parts[-1] if len(parts) > 2 else ''
return current_archive_path, inner, temp_dirdef _extract_virtual_path_item(archive_full, inner_path, temp_dir):
"""
从归档（支持多层嵌套）中提取目标内容到临时目录。
返回提取后的文件/目录的绝对路径。
支持多重虚拟路径：archive_full 是最外层归档，inner_path 可能包含 :: 表示嵌套。
"""
# 递归处理嵌套：如果 inner_path 包含 ::，先提取内层归档再递归
if '::' in inner_path:
nested_archive_path, final_path = split_virtual_path(inner_path)
nested_temp = create_cache_dir(prefix='nested_extract_')
try:
_extract_archive_item(archive_full, nested_archive_path, nested_temp)
nested_archive_file = _find_extracted_file(nested_temp, nested_archive_path)
if not nested_archive_file:
raise FileNotFoundError(f'嵌套归档不存在: {nested_archive_path}')
result = _extract_virtual_path_item(nested_archive_file, final_path, temp_dir)
# 先返回结果，再清理临时目录（确保返回值中的文件已被复制到temp_dir）
return result
finally:
# 从缓存管理中移除，并立即清理
try:
remove_cache_entry(nested_temp)
shutil.rmtree(nested_temp, ignore_errors=True)
except Exception:
pass
# 单层提取
ext = _get_archive_ext(archive_full)
try:
if ext == '.zip':
with zipfile.ZipFile(archive_full, 'r') as zf:
for name in zf.namelist():
if name == inner_path or name.startswith(inner_path + '/'):
zf.extract(name, temp_dir)
elif ext in ('.tar', '.tgz', '.tar.gz'):
with tarfile.open(archive_full, 'r') as tf:
for member in tf.getmembers():
if member.name == inner_path or member.name.startswith(inner_path + '/'):
tf.extract(member, temp_dir)
elif ext in ('.bz2', '.tbz'):
with tarfile.open(archive_full, 'r') as tf:
for member in tf.getmembers():
if member.name == inner_path or member.name.startswith(inner_path + '/'):
tf.extract(member, temp_dir)
elif ext in ('.iso', '.img') and HAS_PYCDLIB and _is_iso9660_file(archive_full):
_extract_iso_item(archive_full, inner_path, temp_dir)
elif ext == '.7z' and HAS_PY7ZR:
with py7zr.SevenZipFile(archive_full, mode='r') as zf:
for name in zf.getnames():
if name == inner_path or name.startswith(inner_path + '/'):
zf.extract(targets=[name], path=temp_dir)
elif ext == '.rar' and HAS_RARFILE:
with rarfile.RarFile(archive_full) as rf:
for name in rf.namelist():
if name == inner_path or name.startswith(inner_path + '/'):
rf.extract(name, temp_dir)
elif ext in ('.dmg', '.wim', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.iso') and HAS_PYTSK3:
# 使用 pytsk3 提取镜像文件
img = pytsk3.Img_Info(archive_full)
fs = pytsk3.FS_Info(img)
open_path = inner_path if inner_path.startswith('/') else '/' + inner_path
base_target = os.path.join(temp_dir, inner_path) if inner_path else temp_dir
try:
def _tsk_entry_name(e):
"""获取条目的文件名，优先长文件名"""
try:
n = e.info.name.name
short_name = n.decode('utf-8') if isinstance(n, bytes) else n
except Exception:
return None
# FAT 文件系统：通过遍历 name 链查找长文件名
if short_name and short_name in ('.', '..'):
return short_name
try:
# 尝试从 meta 中获取长文件名
if hasattr(e, 'meta') and e.meta:
meta = e.meta
# pytsk3 的 FAT 实现中，长文件名通过 name 链的后继条目存储
# 我们尝试遍历 find 或 name_info 寻找长文件名
# 注意：pytsk3 不直接暴露 LFN，但 open_dir 返回的 entry 已经有完整的 name
# 返回短文件名（可能已经是扩展属性中的长文件名）
return short_name
except Exception:
return short_name
def _tsk_extract_recursive(tsk_path, local_dir):
try:
dirobj = fs.open_dir(path=tsk_path)
entries = list(dirobj) # Force iteration
except Exception:
return
for entry in entries:
name = _tsk_entry_name(entry)
if not name or name in ('.', '..'):
continue
if name.startswith('$') and name.upper() in ('$FAT1', '$FAT2', '$FAT12', '$MBR', '$ORPHANFILES'):
continue
try:
is_dir = entry.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR
except Exception:
continue
if is_dir:
child_tsk = tsk_path.rstrip('/') + '/' + name
child_local = os.path.join(local_dir, name)
os.makedirs(child_local, exist_ok=True)
_tsk_extract_recursive(child_tsk, child_local)
else:
file_tsk = tsk_path.rstrip('/') + '/' + name
local_file = os.path.join(local_dir, name)
try:
src = fs.open(file_tsk)
file_size = src.info.meta.size
with open(local_file, 'wb') as dst:
total_read = 0
while total_read < file_size:
data = src.read(65536)
if not data:
break
dst.write(data)
total_read += len(data)
src.close()
if total_read != file_size:
try:
os.remove(local_file)
except Exception:
pass
except Exception:
try:
if os.path.exists(local_file):
os.remove(local_file)
except Exception:
pass
# 关键修复：先检查是否为目录，避免 fs.open() 对目录路径失败
is_directory = False
try:
test_dir = fs.open_dir(path=open_path)
is_directory = True
# 关闭测试打开的目录
try:
test_dir.close()
except Exception:
pass
except Exception:
is_directory = False
if is_directory:
# 目录：直接递归提取
os.makedirs(base_target, exist_ok=True)
_tsk_extract_recursive(open_path, base_target)
else:
# 文件：使用 fs.open() 提取
try:
f = fs.open(open_path)
os.makedirs(base_target, exist_ok=True)
dst_path = os.path.join(base_target, os.path.basename(inner_path) if inner_path else 'file')
with open(dst_path, 'wb') as dst:
while True:
data = f.read(65536)
if not data:
break
dst.write(data)
try:
f.close()
except Exception:
pass
except Exception as e:
logger.warning(f"[pytsk3] 提取文件失败：{open_path} - {e}")
# 尝试作为目录再次提取（fallback）
try:
dirobj = fs.open_dir(path=open_path)
os.makedirs(base_target, exist_ok=True)
_tsk_extract_recursive(open_path, base_target)
try:
dirobj.close()
except Exception:
pass
except Exception:
pass
except (FileNotFoundError, IsADirectoryError):
raise
finally:
try:
fs.close()
except Exception:
pass
try:
img.close()
except Exception:
pass
else:
raise NotImplementedError(f'不支持的归档格式: {ext}')
except Exception as e:
logger.warning(f"_extract_virtual_path_item 提取失败: {e}")
raise
# 确保返回的是实际文件系统路径，不包含::
extracted = os.path.join(temp_dir, inner_path) if inner_path else temp_dir
# 移除可能的::字符（安全保证）
extracted = extracted.replace('::', '')
if not os.path.exists(extracted):
base = os.path.basename(inner_path) if inner_path else ''
alt = os.path.join(temp_dir, base)
if os.path.exists(alt):
extracted = alt
# 最后的检查：确保提取路径不包含非法字符
if '::' in extracted:
extracted = os.path.join(temp_dir, os.path.basename(inner_path))
return extracteddef _extract_archive_item(archive_path, inner_path, temp_dir):
"""从外层压缩包/镜像中提取指定路径的文件到临时目录"""
ext = _get_archive_ext(archive_path)
# 尝试查找已提取的文件，如果不存在则使用默认目标路径
target = _find_extracted_file(temp_dir, inner_path)
if not target:
target = os.path.join(temp_dir, os.path.basename(inner_path))
try:
logger.debug(f"_extract_archive_item: archive_path={archive_path}, inner_path={inner_path}, temp_dir={temp_dir}, ext={ext}, target={target}")
except Exception:
pass
if ext == '.zip':
with zipfile.ZipFile(archive_path, 'r') as zf:
# 尝试用原始路径和GBK解码后的路径来匹配
found = False
all_infos = zf.infolist()
for info in all_infos:
raw_fn = info.filename
decoded_fn = decode_archive_name(raw_fn)
if raw_fn == inner_path or decoded_fn == inner_path or raw_fn.rstrip('/') == inner_path or decoded_fn.rstrip('/') == inner_path:
with zf.open(info.filename) as src, open(target, 'wb') as dst:
shutil.copyfileobj(src, dst)
found = True
break
if not found:
# fallback1: 尝试原始 extract
try:
zf.extract(inner_path, temp_dir)
found = True
except KeyError:
pass
if not found:
# fallback2: 不区分大小写匹配
inner_lower = inner_path.lower()
inner_base_lower = os.path.basename(inner_path).lower()
for info in all_infos:
raw_fn = info.filename
decoded_fn = decode_archive_name(raw_fn)
if raw_fn.lower() == inner_lower or decoded_fn.lower() == inner_lower:
with zf.open(info.filename) as src, open(target, 'wb') as dst:
shutil.copyfileobj(src, dst)
found = True
break
if not found:
# fallback3: 仅按basename匹配（用于子目录路径不匹配的情况）
for info in all_infos:
raw_fn = info.filename
decoded_fn = decode_archive_name(raw_fn)
if os.path.basename(raw_fn).lower() == inner_base_lower or os.path.basename(decoded_fn).lower() == inner_base_lower:
with zf.open(info.filename) as src, open(target, 'wb') as dst:
shutil.copyfileobj(src, dst)
found = True
break
elif ext in ('.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.tbz'):
with tarfile.open(archive_path, 'r') as tf:
# 尝试匹配编码后的文件名
found = False
all_members = tf.getmembers()
for member in all_members:
decoded_name = decode_archive_name(member.name)
if member.name == inner_path or decoded_name == inner_path or member.name.rstrip('/') == inner_path or decoded_name.rstrip('/') == inner_path:
tf.extract(member, temp_dir)
found = True
break
if not found:
# fallback1
try:
member = tf.getmember(inner_path)
tf.extract(member, temp_dir)
found = True
except KeyError:
pass
if not found:
# fallback2: 不区分大小写匹配
inner_lower = inner_path.lower()
inner_base_lower = os.path.basename(inner_path).lower()
for member in all_members:
decoded_name = decode_archive_name(member.name)
if member.name.lower() == inner_lower or decoded_name.lower() == inner_lower:
tf.extract(member, temp_dir)
found = True
break
if not found:
# fallback3: 仅按basename匹配
for member in all_members:
decoded_name = decode_archive_name(member.name)
if os.path.basename(member.name).lower() == inner_base_lower or os.path.basename(decoded_name).lower() == inner_base_lower:
tf.extract(member, temp_dir)
found = True
break
elif ext == '.7z' and HAS_PY7ZR:
with py7zr.SevenZipFile(archive_path, mode='r') as zf:
zf.extractall(path=temp_dir, filters=[{'id': py7zr.FILTER_LZMA2, 'preset': 9}])
# 只保留目标文件
if os.path.exists(os.path.join(temp_dir, inner_path)):
shutil.move(os.path.join(temp_dir, inner_path), target)
elif ext == '.rar' and HAS_RARFILE:
with rarfile.RarFile(archive_path) as rf:
rf.extract(inner_path, temp_dir)
elif ext == '.iso':
# ISO 9660 镜像：pycdlib 优先，失败时 pytsk3 回退（混合镜像场景）
_mirror_extracted = False
# 尝试1：使用 pycdlib（仅对 .iso 启用；.img 不是 ISO9660，pycdlib 会抛 PVD 错误）
if HAS_PYCDLIB and not _mirror_extracted and _is_iso9660_file(archive_path):
try:
_iso = pycdlib.PyCdlib()
_iso.open(archive_path)
try:
_inner_lower = inner_path.lower()
_inner_lower_no_ver = _inner_lower.replace(';1', '')
_actual_path = None
for _dir_path, _dirs, _files in _iso.walk(iso_path='/'):
for _f in _files:
_full_iso_path = (_dir_path.rstrip('/') + '/' + _f).lstrip('/')
if _full_iso_path.lower() == _inner_lower:
_actual_path = _full_iso_path
break
if _actual_path:
break
if not _actual_path:
for _dir_path, _dirs, _files in _iso.walk(iso_path='/'):
for _f in _files:
_full_iso_path = (_dir_path.rstrip('/') + '/' + _f).lstrip('/')
_full_lower = _full_iso_path.lower().replace(';1', '')
if _full_lower == _inner_lower_no_ver or _full_lower == _inner_lower:
_actual_path = _full_iso_path
break
if _actual_path:
break
if not _actual_path and _iso.has_rock_ridge():
for _dir_path, _dirs, _files in _iso.walk(rr_path='/'):
for _f in _files:
_full_rr_path = (_dir_path.rstrip('/') + '/' + _f).lstrip('/')
if _full_rr_path.lower() == _inner_lower or _full_rr_path.lower().replace(';1', '') == _inner_lower_no_ver:
_actual_path = f"rr://{_full_rr_path}"
break
if _actual_path:
break
if not _actual_path and _iso.has_joliet():
for _dir_path, _dirs, _files in _iso.walk(joliet_path='/'):
for _f in _files:
_full_j_path = (_dir_path.rstrip('/') + '/' + _f).lstrip('/')
if _full_j_path.lower() == _inner_lower or _full_j_path.lower().replace(';1', '') == _inner_lower_no_ver:
_actual_path = f"joliet://{_full_j_path}"
break
if _actual_path:
break
if not _actual_path:
raise FileNotFoundError(f"ISO中未找到: {inner_path}")
os.makedirs(os.path.dirname(target), exist_ok=True)
with open(target, 'wb') as _dst_fp:
if _actual_path.startswith('rr://'):
_iso_fp = _actual_path[5:]
_iso.get_file_from_iso_fp(_dst_fp, rr_path=_iso_fp)
elif _actual_path.startswith('joliet://'):
_iso_fp = _actual_path[9:]
_iso.get_file_from_iso_fp(_dst_fp, joliet_path=_iso_fp)
else:
_iso_path_arg = '/' + _actual_path if not _actual_path.startswith('/') else _actual_path
_iso.get_file_from_iso_fp(_dst_fp, iso_path=_iso_path_arg)
_mirror_extracted = True
except FileNotFoundError:
raise
except Exception as _pycdlib_exc:
logger.debug(f"pycdlib 提取失败: {_pycdlib_exc}")
finally:
_iso.close()
except FileNotFoundError:
raise
except Exception as _pycdlib_exc:
logger.debug(f"pycdlib 打开失败: {_pycdlib_exc}")
# 尝试1b：pycdlib 失败时 pytsk3 回退（混合/混合ISO镜像）
if not _mirror_extracted and HAS_PYTSK3:
try:
_img = pytsk3.Img_Info(archive_path)
_fs = None
try:
_fs = pytsk3.FS_Info(_img)
except Exception:
_fs = None
if _fs is None:
try:
_vs = pytsk3.Volume_Info(_img)
for _part in _vs:
if _part.len <= 0:
continue
try:
_fs = pytsk3.FS_Info(_img, offset=_part.start * 512)
break
except Exception:
_fs = None
except Exception:
_fs = None
if _fs is not None:
_target_lower = inner_path.lower().rstrip(';1')
_f = None
# 尝试直接路径打开
for _try in ('/' + inner_path, '/' + inner_path.upper(), '/' + inner_path.lower(), '/' + _target_lower):
try:
_f = _fs.open(_try)
break
except Exception:
_f = None
# 如果直接路径失败，尝试递归搜索子目录
if _f is None:
def _find_file_recursive_iso(fs_obj, dir_path, target_path_lower, current_path='/'):
"""递归搜索目录树查找目标文件，通过路径后缀匹配"""
try:
dir_obj = fs_obj.open_dir(path=dir_path)
except Exception:
return None
for entry in dir_obj:
try:
en = entry.info.name.name
enm = en.decode('utf-8', errors='replace') if isinstance(en, bytes) else str(en)
except Exception:
continue
if not enm or enm in ('.', '..'):
continue
entry_path_lower = (current_path.rstrip('/') + '/' + enm).lower().lstrip('/')
# 检查路径后缀匹配
if entry_path_lower == target_path_lower or entry_path_lower.endswith('/' + target_path_lower):
if hasattr(entry.info, 'meta') and entry.info.meta:
if entry.info.meta.type != pytsk3.TSK_FS_META_TYPE_DIR:
return entry
# 如果是目录，递归搜索
try:
if hasattr(entry.info, 'meta') and entry.info.meta:
if entry.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR:
sub_path = dir_path.rstrip('/') + '/' + enm
found = _find_file_recursive_iso(fs_obj, sub_path, target_path_lower, current_path.rstrip('/') + '/' + enm)
if found:
return found
except Exception:
pass
return None
# 从根目录开始递归搜索
_f = _find_file_recursive_iso(_fs, '/', _target_lower)
if _f is not None and getattr(_f.info, 'meta', None):
if _f.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR:
raise IsADirectoryError(inner_path)
os.makedirs(os.path.dirname(target), exist_ok=True)
with open(target, 'wb') as _dst:
_offset = 0
_size = _f.info.meta.size or 0
while _offset < _size:
_data = _f.read_random(_offset, min(65536, _size - _offset))
if not _data:
break
_dst.write(_data)
_offset += len(_data)
_mirror_extracted = True
try:
_f.close()
except Exception:
pass
try:
_fs.close()
except Exception:
pass
try:
_img.close()
except Exception:
pass
except (FileNotFoundError, IsADirectoryError):
raise
except Exception as _pytsk3_exc:
logger.debug(f"pytsk3 打开 ISO 失败: {_pytsk3_exc}")
if not _mirror_extracted:
raise FileNotFoundError(f'无法从镜像 {archive_path} 中提取 {inner_path}')
elif ext in ('.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'):
# 磁盘镜像：先尝试 pycdlib（如果文件实际是 ISO9660），失败再回退 pytsk3
_mirror_extracted = False
# 尝试 1：pycdlib（仅当文件实际是 ISO9660 时）
if HAS_PYCDLIB and not _mirror_extracted and _is_iso9660_file(archive_path):
try:
_iso = pycdlib.PyCdlib()
_iso.open(archive_path)
try:
_inner_lower = inner_path.lower()
_inner_lower_no_ver = _inner_lower.replace(';1', '')
_actual_path = None
for _dir_path, _dirs, _files in _iso.walk(iso_path='/'):
for _f in _files:
_full_iso_path = (_dir_path.rstrip('/') + '/' + _f).lstrip('/')
if _full_iso_path.lower() == _inner_lower:
_actual_path = _full_iso_path
break
if _actual_path:
break
if not _actual_path:
for _dir_path, _dirs, _files in _iso.walk(iso_path='/'):
for _f in _files:
_full_iso_path = (_dir_path.rstrip('/') + '/' + _f).lstrip('/')
_full_lower = _full_iso_path.lower().replace(';1', '')
if _full_lower == _inner_lower_no_ver or _full_lower == _inner_lower:
_actual_path = _full_iso_path
break
if _actual_path:
break
if not _actual_path and _iso.has_rock_ridge():
for _dir_path, _dirs, _files in _iso.walk(rr_path='/'):
for _f in _files:
_full_rr_path = (_dir_path.rstrip('/') + '/' + _f).lstrip('/')
if _full_rr_path.lower() == _inner_lower or _full_rr_path.lower().replace(';1', '') == _inner_lower_no_ver:
_actual_path = f"rr://{_full_rr_path}"
break
if _actual_path:
break
if not _actual_path and _iso.has_joliet():
for _dir_path, _dirs, _files in _iso.walk(joliet_path='/'):
for _f in _files:
_full_j_path = (_dir_path.rstrip('/') + '/' + _f).lstrip('/')
if _full_j_path.lower() == _inner_lower or _full_j_path.lower().replace(';1', '') == _inner_lower_no_ver:
_actual_path = f"joliet://{_full_j_path}"
break
if _actual_path:
break
if not _actual_path:
raise FileNotFoundError(f"镜像中未找到：{inner_path}")
os.makedirs(os.path.dirname(target), exist_ok=True)
with open(target, 'wb') as _dst_fp:
if _actual_path.startswith('rr://'):
_iso.get_file_from_iso_fp(_dst_fp, rr_path=_actual_path[5:])
elif _actual_path.startswith('joliet://'):
_iso.get_file_from_iso_fp(_dst_fp, joliet_path=_actual_path[9:])
else:
_iso_path_arg = '/' + _actual_path if not _actual_path.startswith('/') else _actual_path
_iso.get_file_from_iso_fp(_dst_fp, iso_path=_iso_path_arg)
_mirror_extracted = True
except FileNotFoundError:
raise
except Exception as _pycdlib_exc:
logger.debug(f"pycdlib 提取失败：{_pycdlib_exc}")
finally:
_iso.close()
except FileNotFoundError:
raise
except Exception as _pycdlib_exc:
logger.debug(f"pycdlib 打开失败：{_pycdlib_exc}")
# 尝试 2：pytsk3 回退
if not _mirror_extracted and HAS_PYTSK3:
try:
_img = pytsk3.Img_Info(archive_path)
_fs = None
try:
_fs = pytsk3.FS_Info(_img)
except Exception:
_fs = None
if _fs is None:
# 整盘镜像，尝试分区表
try:
_vs = pytsk3.Volume_Info(_img)
for _part in _vs:
if _part.len <= 0:
continue
try:
_fs = pytsk3.FS_Info(_img, offset=_part.start * 512)
break
except Exception:
_fs = None
except Exception:
_fs = None
if _fs is not None:
_target_lower = inner_path.lower().rstrip(';1')
_f = None
for _try in ('/' + inner_path, '/' + inner_path.upper(), '/' + inner_path.lower(), '/' + _target_lower):
try:
_f = _fs.open(_try)
break
except Exception:
_f = None
if _f is None:
# 先遍历根目录兼容 FAT 短文件名/LFN
try:
_root = _fs.open_dir(path='/')
for _e in _root:
try:
_en = _e.info.name.name
_enm = _en.decode('utf-8', errors='replace') if isinstance(_en, bytes) else str(_en)
except Exception:
continue
if not _enm or _enm in ('.', '..'):
continue
if _enm.lower().rstrip(';1') == _target_lower:
_f = _e
break
except Exception:
_f = None
if _f is None:
# 根目录未找到，递归搜索整个目录树
def _find_file_recursive_img(fs_obj, dir_path, target_path_lower, current_path='/'):
"""递归搜索目录树查找目标文件，通过路径后缀匹配"""
try:
dir_obj = fs_obj.open_dir(path=dir_path)
except Exception:
return None
for entry in dir_obj:
try:
en = entry.info.name.name
enm = en.decode('utf-8', errors='replace') if isinstance(en, bytes) else str(en)
except Exception:
continue
if not enm or enm in ('.', '..'):
continue
entry_path_lower = (current_path.rstrip('/') + '/' + enm).lower().lstrip('/')
# 检查路径后缀匹配
if entry_path_lower == target_path_lower or entry_path_lower.endswith('/' + target_path_lower):
if hasattr(entry.info, 'meta') and entry.info.meta:
if entry.info.meta.type != pytsk3.TSK_FS_META_TYPE_DIR:
return entry
# 如果是目录，递归搜索
try:
if hasattr(entry.info, 'meta') and entry.info.meta:
if entry.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR:
sub_path = dir_path.rstrip('/') + '/' + enm
found = _find_file_recursive_img(fs_obj, sub_path, target_path_lower, current_path.rstrip('/') + '/' + enm)
if found:
return found
except Exception:
pass
return None
_f = _find_file_recursive_img(_fs, '/', _target_lower)
if _f is not None and getattr(_f.info, 'meta', None):
if _f.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR:
raise IsADirectoryError(inner_path)
os.makedirs(os.path.dirname(target), exist_ok=True)
with open(target, 'wb') as _dst:
_offset = 0
_size = _f.info.meta.size or 0
while _offset < _size:
_data = _f.read_random(_offset, min(65536, _size - _offset))
if not _data:
break
_dst.write(_data)
_offset += len(_data)
_mirror_extracted = True
try:
_f.close()
except Exception:
pass
try:
_fs.close()
except Exception:
pass
try:
_img.close()
except Exception:
pass
except (FileNotFoundError, IsADirectoryError):
raise
except Exception as _pytsk3_exc:
logger.debug(f"pytsk3 提取文件失败: {_pytsk3_exc}")
if not _mirror_extracted:
raise FileNotFoundError(f'无法从镜像 {archive_path} 中提取 {inner_path}')
if not os.path.exists(target):
# 尝试不带前缀路径
base = os.path.basename(inner_path)
if os.path.exists(os.path.join(temp_dir, inner_path)):
shutil.move(os.path.join(temp_dir, inner_path), target)
elif os.path.exists(os.path.join(temp_dir, base)):
pass # 已经存在
else:
try:
logger.debug(f"_extract_archive_item failed to find target: archive_path={archive_path}, inner_path={inner_path}, temp_dir={temp_dir}")
logger.debug(f"temp_dir contents: {os.listdir(temp_dir) if os.path.exists(temp_dir) else 'missing'}")
except Exception:
pass
raise FileNotFoundError(f'无法提取嵌套归档: {inner_path}')def list_nested_virtual_directory(user_dir, outer_archive_path, nested_path, original_virtual_path=None, remaining_inner=''):
"""处理嵌套压缩包/镜像：提取内部归档并列出其内容
Args:
user_dir: 用户目录
outer_archive_path: 外层归档完整路径
nested_path: 内部归档在outer中的路径（如 "subdir/inner.tar"）
original_virtual_path: 最原始的完整虚拟路径（用于生成正确的item路径，不能被覆盖）
remaining_inner: 嵌套归档内部还有剩余路径时传入（如 "subdir2"）
"""
nested_ext = os.path.splitext(nested_path)[1].lower()
temp_dir = create_cache_dir(prefix='nested_')
try:
_extract_archive_item(outer_archive_path, nested_path, temp_dir)
extracted = os.path.join(temp_dir, os.path.basename(nested_path))
logger.debug(f"nested extraction: outer={outer_archive_path}, nested_path={nested_path}, temp_dir={temp_dir}, expected_extracted={extracted}")
if not os.path.exists(extracted):
# 尝试使用 _find_extracted_file 查找（支持子目录递归搜索）
extracted = _find_extracted_file(temp_dir, nested_path)
if not extracted:
try:
logger.debug(f"temp_dir contents: {os.listdir(temp_dir)}")
except Exception:
pass
raise FileNotFoundError(f'嵌套归档提取失败: {nested_path}')
# path_prefix 构建：必须保持从最原始路径到当前嵌套归档的完整路径链
# original_virtual_path 是最顶层的完整路径 (如 'A::B::C::D::')
# remaining_inner 是从当前层开始还有多少层未处理 (如 '')
# 通过 remaining_inner 计算已处理的层数，从而确定 path_prefix 应该包含多少段
if original_virtual_path:
parts = [p.strip('/') for p in original_virtual_path.split('::') if p]
remaining_parts = [p.strip('/') for p in remaining_inner.split('::') if p] if remaining_inner else []
# 已处理层数 = 总层数 - 剩余层数
processed_count = len(parts) - len(remaining_parts)
if processed_count >= 1 and processed_count <= len(parts):
# path_prefix 包含到当前层止的所有段 (processed_count-1 就是已处理层数减1)
path_prefix = '::'.join(parts[:processed_count])
else:
path_prefix = parts[0] if parts else None
else:
path_prefix = None
archive_name = os.path.basename(nested_path)
# 关键修复：判断 archive_name 是否为归档/镜像文件
# 如果是普通文件夹，不应该加 :: 后缀
nested_ext = os.path.splitext(archive_name)[1].lower()
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
is_nested_archive = nested_ext in archive_exts or nested_ext in image_exts
# 用 temp_dir 作为 user_dir，构建 virtual_path
# 只有当 archive_name 是归档/镜像文件时才加 :: 后缀
if is_nested_archive:
inner_vp = archive_name + '::' + remaining_inner if remaining_inner else archive_name + '::'
else:
# 普通文件夹：使用 / 分隔符连接路径
inner_vp = (archive_name + '/' + remaining_inner) if remaining_inner else archive_name
result = list_virtual_directory(temp_dir, inner_vp, temp_dir=temp_dir, path_prefix=path_prefix, original_virtual_path=original_virtual_path)
return result
except Exception:
raise
finally:
try:
if temp_dir and os.path.exists(temp_dir):
remove_cache_entry(temp_dir)
shutil.rmtree(temp_dir, ignore_errors=True)
except Exception:
pass目录列表函数
def list_virtual_directory(user_dir, virtual_path, temp_dir=None, path_prefix=None, original_virtual_path=None):
"""列出虚拟目录（压缩包/镜像内部）的文件和文件夹
支持嵌套压缩包/镜像：如果 inner 路径指向的是压缩包/镜像文件，则提取并打开
Args:
user_dir: 用户目录
virtual_path: 虚拟路径
temp_dir: 如果是嵌套归档，这是临时目录路径（用于后续清理）
path_prefix: 路径前缀，用于嵌套归档生成正确的item路径
original_virtual_path: 最顶层的原始虚拟路径（用于深度嵌套归档保持正确的path_prefix）
"""
# 保存原始路径（第一层调用的原始路径）
if original_virtual_path is None:
original_virtual_path = virtual_path
archive_rel, inner = split_virtual_path(virtual_path)
archive_full = os.path.join(user_dir, archive_rel)
# 调试信息：记录正在打开的归档/镜像及其内部路径（便于诊断无法访问或解析失败的问题）
try:
logger.debug(f"list_virtual_directory: user_dir={user_dir}, archive_rel={archive_rel}, archive_full={archive_full}, inner={inner}, temp_dir={temp_dir}, path_prefix={path_prefix}")
except Exception:
pass
if not is_safe_path(user_dir, archive_full):
raise ValueError('路径不安全')
if not os.path.exists(archive_full) or not os.path.isfile(archive_full):
raise FileNotFoundError('文件不存在')
# 检查 inner 路径是否指向嵌套的压缩包/镜像
# inner 可能包含 ::（如 "subdir/inner.tar::subdir2"），需要先取第一个 :: 之前的部分
nested_archive_path = None
inner_check = inner
if '::' in inner:
inner_check = inner.split('::', 1)[0]
if inner_check:
inner_ext = os.path.splitext(inner_check)[1].lower()
nested_archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
nested_image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
if inner_ext in nested_archive_exts or inner_ext in nested_image_exts:
nested_archive_path = inner_check
if nested_archive_path:
# 嵌套压缩包/镜像：先提取到临时目录，再打开
remaining = ''
if '::' in inner:
remaining = inner.split('::', 1)[1]
return list_nested_virtual_directory(user_dir, archive_full, nested_archive_path, original_virtual_path=original_virtual_path, remaining_inner=remaining)
ext = os.path.splitext(archive_full)[1].lower()
items = []
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
# 构建完整的虚拟路径前缀
virtual_prefix = (path_prefix + '::') if path_prefix else (archive_rel + '::')
if ext == '.zip':
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
try:
with zipfile.ZipFile(archive_full, 'r') as zf:
prefix = (inner + '/') if inner else ''
prefix_gbk = prefix # 尝试用GBK编码后的prefix匹配
try:
prefix_gbk = prefix.encode('utf-8').decode('gbk') if prefix else ''
except Exception:
pass
# 获取所有文件信息用于后续匹配
all_infos = zf.infolist()
all_names = zf.namelist()
# 过滤时同时尝试原始prefix和GBK编码的prefix
names = []
used_prefix = prefix # 记录实际匹配的prefix
for n in all_names:
if n == prefix:
continue
if n.startswith(prefix):
names.append((n, prefix))
elif prefix_gbk and n.startswith(prefix_gbk):
names.append((n, prefix_gbk))
seen = set()
# 构建目录 mtime 和 size 映射，用于显示目录信息
dir_mtime_map = {}
dir_size_map = {}
for info in all_infos:
raw_fn = info.filename
decoded_fn = decode_archive_name(raw_fn)
if '/' in decoded_fn:
dir_part = '/'.join(decoded_fn.split('/')[:-1])
dir_prefix_decoded = decode_archive_name(inner) if inner else ''
if dir_part.startswith(dir_prefix_decoded):
if dir_prefix_decoded and dir_part == dir_prefix_decoded:
continue
rel_dir = dir_part[len(dir_prefix_decoded):].lstrip('/') if dir_prefix_decoded else dir_part
if not rel_dir:
continue
# 递归累加到所有上级目录
parts_list = rel_dir.split('/')
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
if ancestor and ancestor not in dir_mtime_map:
try:
dir_mtime_map[ancestor] = datetime(*info.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[ancestor] = '-'
if not raw_fn.endswith('/'):
dir_size_map[ancestor] = dir_size_map.get(ancestor, 0) + info.file_size
logger.info(f"[list_virtual_directory ZIP] dir_size_map: {dir_size_map}")
logger.info(f"[list_virtual_directory ZIP] dir_mtime_map: {dir_mtime_map}")
logger.info(f"[list_virtual_directory ZIP] inner='{inner}', prefix='{prefix}'")
for n_item in names:
if isinstance(n_item, tuple):
n, matched_prefix = n_item
else:
n, matched_prefix = n_item, prefix
rest = n[len(matched_prefix):] if matched_prefix else n
parts = rest.split('/')
name = parts[0]
# 始终使用统一解码策略修复可能的编码问题
name = decode_archive_name(name)
if name in seen:
continue
seen.add(name)
is_dir = (len(parts) > 1) or n.endswith('/')
item_path = virtual_prefix + ('/'.join([inner, name]).lstrip('/') if inner else name)
# 获取大小和修改时间
size = '-'
mtime = '-'
can_expand = is_dir
# 目录也尝试获取mtime/size（从该目录下的文件推断）
if is_dir:
if name in dir_mtime_map:
mtime = dir_mtime_map[name]
if name in dir_size_map:
size = format_file_size(dir_size_map[name])
elif size == '-':
size = '0 B'
logger.info(f"[list_virtual_directory ZIP] dir item '{name}': size={size}, mtime={mtime}")
# 目录也可能是可展开的（如果是压缩包名作为目录）
file_ext = os.path.splitext(name.rstrip('/'))[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
else:
# 查找匹配的文件信息获取大小和修改时间
matched = False
for info in all_infos:
full_name = info.filename
fn_matched_prefix = prefix
if prefix_gbk and full_name.startswith(prefix_gbk):
fn_matched_prefix = prefix_gbk
fn_rest = full_name[len(fn_matched_prefix):] if fn_matched_prefix else full_name
fn_display = decode_archive_name(fn_rest)
# 始终尝试原始名和解码名匹配
if fn_rest == rest or fn_display == rest or fn_rest == name or fn_display == name:
size = format_file_size(info.file_size)
try:
mtime = datetime(*info.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
matched = True
break
if not matched:
for full_name in all_names:
fn_matched_prefix = prefix
if prefix_gbk and full_name.startswith(prefix_gbk):
fn_matched_prefix = prefix_gbk
fn_rest = full_name[len(fn_matched_prefix):] if fn_matched_prefix else full_name
fn_display = decode_archive_name(fn_rest)
if fn_rest == rest or fn_display == rest or fn_rest == name or fn_display == name:
try:
info = zf.getinfo(full_name)
size = format_file_size(info.file_size)
try:
mtime = datetime(*info.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
except Exception:
pass
break
file_ext = os.path.splitext(name)[1].lower()
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': True})
except Exception as e:
raise Exception(f'读取ZIP失败: {str(e)}')
elif ext in ('.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.tbz'):
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts_list = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
try:
with tarfile.open(archive_full, 'r') as tf:
prefix = (inner + '/') if inner else ''
prefix_gbk = prefix
try:
prefix_gbk = prefix.encode('utf-8').decode('gbk') if prefix else ''
except Exception:
pass
# 过滤时同时尝试原始prefix和GBK编码的prefix
members_list = []
for m in tf.getmembers():
if m.name == prefix:
continue
if m.name.startswith(prefix):
members_list.append((m, prefix))
elif prefix_gbk and m.name.startswith(prefix_gbk):
members_list.append((m, prefix_gbk))
seen = set()
# 构建目录mtime/size映射（使用解码后的文件名）
dir_mtime_map = {}
dir_size_map = {}
for m, mprefix in members_list:
raw_rest = m.name[len(mprefix):] if mprefix else m.name
decoded_rest = decode_archive_name(raw_rest)
if '/' in decoded_rest:
dir_part = '/'.join(decoded_rest.split('/')[:-1])
top_dir = dir_part.split('/')[0]
if top_dir not in dir_mtime_map:
try:
dir_mtime_map[top_dir] = datetime.fromtimestamp(m.mtime).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[top_dir] = '-'
if m.isfile():
dir_size_map[top_dir] = dir_size_map.get(top_dir, 0) + m.size
for m, mprefix in members_list:
raw_rest = m.name[len(mprefix):] if mprefix else m.name
raw_parts = raw_rest.split('/')
name = decode_archive_name(raw_parts[0])
decoded_parts = decode_archive_name(raw_rest).split('/')
if name in seen:
continue
seen.add(name)
is_dir = (len(decoded_parts) > 1) or m.isdir()
item_path = virtual_prefix + ('/'.join([inner, name]).lstrip('/') if inner else name)
size = format_file_size(m.size) if m.isfile() else '-'
try:
mtime = datetime.fromtimestamp(m.mtime).strftime('%Y-%m-%d %H:%M') if hasattr(m, 'mtime') else '-'
except Exception:
if is_dir and name in dir_mtime_map:
mtime = dir_mtime_map[name]
else:
mtime = '-'
if is_dir and name in dir_size_map:
size = format_file_size(dir_size_map[name])
# 检测嵌套压缩包/镜像
can_expand = is_dir
if not is_dir:
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts_list:
can_expand = True
if is_dir:
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts_list:
can_expand = True
file_ext = os.path.splitext(name)[1].lower()
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts_list
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': True})
except Exception as e:
raise Exception(f'读取TAR失败: {str(e)}')
elif ext == '.7z' and HAS_PY7ZR:
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
try:
with py7zr.SevenZipFile(archive_full, mode='r') as zf:
prefix = (inner + '/') if inner else ''
prefix_gbk = prefix
try:
prefix_gbk = prefix.encode('utf-8').decode('gbk') if prefix else ''
except Exception:
pass
# 过滤时同时尝试原始prefix和GBK编码的prefix
names = []
for n in zf.getnames():
if n == prefix:
continue
if n.startswith(prefix):
names.append((n, prefix))
elif prefix_gbk and n.startswith(prefix_gbk):
names.append((n, prefix_gbk))
seen = set()
# 构建文件信息查找表
file_info_map = {}
try:
archive_info = zf.archive_info
if hasattr(archive_info, 'file_info_list'):
for fi in archive_info.file_info_list:
fn = fi.get('filename', '')
file_info_map[fn] = fi
except Exception:
pass
# 也尝试从 headers 构建
if not file_info_map:
try:
headers = zf.headers
if hasattr(headers, 'files_info'):
for fh in headers.files_info:
if hasattr(fh, 'filename'):
fn = fh.filename
info = {'uncompressed': fh.uncompressed, 'm_time': fh.m_time}
if hasattr(fh, 'm_time') and hasattr(fh, 'uncompressed'):
file_info_map[fn] = info
except Exception:
pass
# 构建目录mtime/size映射（从子文件推断）
dir_mtime_map = {}
dir_size_map = {}
for full_fn, fi in file_info_map.items():
matched_p = prefix
if prefix_gbk and full_fn.startswith(prefix_gbk):
matched_p = prefix_gbk
if full_fn.startswith(matched_p) and full_fn != matched_p:
rest = full_fn[len(matched_p):] if matched_p else full_fn
if '/' in rest and not full_fn.endswith('/'):
dir_part = '/'.join(rest.split('/')[:-1])
top_dir = dir_part.split('/')[0]
if top_dir not in dir_mtime_map:
m_time = fi.get('m_time', None) if isinstance(fi, dict) else None
if m_time:
try:
dir_mtime_map[top_dir] = datetime.fromtimestamp(m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[top_dir] = '-'
else:
dir_mtime_map[top_dir] = '-'
uncompressed_val = None
if isinstance(fi, dict):
uncompressed_val = fi.get('uncompressed', None)
else:
uncompressed_val = getattr(fi, 'uncompressed', None)
if isinstance(uncompressed_val, (list, tuple)):
size_value = uncompressed_val[0] if uncompressed_val else 0
else:
size_value = uncompressed_val or 0
if size_value:
dir_size_map[top_dir] = dir_size_map.get(top_dir, 0) + size_value
# 如果没有 file_info_map，尝试从 names 列表中获取大小
if not file_info_map:
for n_item in names:
if isinstance(n_item, tuple):
full_fn, mprefix = n_item
else:
full_fn, mprefix = n_item, prefix
if full_fn.startswith(prefix) and full_fn != prefix and not full_fn.endswith('/'):
rel = full_fn[len(mprefix):] if mprefix else full_fn
if '/' in rel:
top_dir = rel.split('/')[0]
try:
info = zf.getinfo(full_fn)
dir_size_map[top_dir] = dir_size_map.get(top_dir, 0) + info.file_size
except Exception:
pass
for n_item in names:
if isinstance(n_item, tuple):
n, matched_prefix = n_item
else:
n, matched_prefix = n_item, prefix
rest = n[len(matched_prefix):] if matched_prefix else n
parts = rest.split('/')
name = decode_archive_name(parts[0])
if name in seen:
continue
seen.add(name)
is_dir = (len(parts) > 1) or n.endswith('/')
item_path = virtual_prefix + ('/'.join([inner, name]).lstrip('/') if inner else name)
size = '-'
mtime = '-'
can_expand = is_dir
if is_dir:
# 目录尝试从映射表获取mtime/size
if name in dir_mtime_map:
mtime = dir_mtime_map[name]
if name in dir_size_map:
size = format_file_size(dir_size_map[name])
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
else:
# 尝试从查找表获取文件信息——用 rest（去prefix后的完整路径）匹配
matched_info = None
for full_fn, fi in file_info_map.items():
fn_rest = full_fn[len(prefix):] if prefix else full_fn
if fn_rest == rest or fn_rest == name:
matched_info = fi
break
if matched_info:
fi = matched_info
uncompressed = fi.get('uncompressed', fi.get('uncompressed', None)) if isinstance(fi, dict) else None
if uncompressed and uncompressed[0]:
size = format_file_size(uncompressed[0])
m_time = fi.get('m_time', None) if isinstance(fi, dict) else None
if m_time:
try:
mtime = datetime.fromtimestamp(m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
# 如果查找表没有，尝试从 headers 直接获取
if size == '-':
try:
headers = zf.headers
for fh in headers.files_info:
if hasattr(fh, 'filename'):
fh_rest = fh.filename[len(prefix):] if prefix else fh.filename
if fh_rest == rest or fh_rest == name:
if hasattr(fh, 'uncompressed') and fh.uncompressed:
size = format_file_size(fh.uncompressed[0]) if fh.uncompressed[0] else '-'
if hasattr(fh, 'm_time') and fh.m_time:
try:
mtime = datetime.fromtimestamp(fh.m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
break
except Exception:
pass
# 检测是否为可展开的压缩包/镜像
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
file_ext = os.path.splitext(name)[1].lower()
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': False})
except Exception as e:
raise Exception(f'读取7z失败: {str(e)}')
elif ext == '.rar' and HAS_RARFILE:
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
try:
with rarfile.RarFile(archive_full) as rf:
prefix = (inner + '/') if inner else ''
prefix_gbk = prefix
try:
prefix_gbk = prefix.encode('utf-8').decode('gbk') if prefix else ''
except Exception:
pass
infolist = rf.infolist()
# 过滤时同时尝试原始prefix和GBK编码的prefix
names = []
for n in rf.namelist():
if n == prefix:
continue
if n.startswith(prefix):
names.append((n, prefix))
elif prefix_gbk and n.startswith(prefix_gbk):
names.append((n, prefix_gbk))
seen = set()
# 构建目录 mtime/size 映射
dir_mtime_map = {}
dir_size_map = {}
for rarinfo in infolist:
rar_fn = rarinfo.filename
decoded_fn = decode_archive_name(rar_fn)
if '/' in decoded_fn:
dir_prefix_decoded = decode_archive_name(inner) if inner else ''
decoded_rest = decoded_fn[len(dir_prefix_decoded):].lstrip('/') if dir_prefix_decoded else decoded_fn
dir_part = '/'.join(decoded_rest.split('/')[:-1])
top_dir = dir_part.split('/')[0] if dir_part else ''
if top_dir:
if top_dir not in dir_mtime_map:
try:
dir_mtime_map[top_dir] = datetime(*rarinfo.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[top_dir] = '-'
if not rarinfo.is_dir():
dir_size_map[top_dir] = dir_size_map.get(top_dir, 0) + rarinfo.file_size
for n_item in names:
if isinstance(n_item, tuple):
n, mprefix = n_item
else:
n, mprefix = n_item, prefix
rest = n[len(mprefix):] if mprefix else n
parts = rest.split('/')
name = parts[0]
# 始终使用统一解码策略
name = decode_archive_name(name)
if name in seen:
continue
seen.add(name)
is_dir = (len(parts) > 1) or n.endswith('/')
item_path = virtual_prefix + ('/'.join([inner, name]).lstrip('/') if inner else name)
size = '-'
mtime = '-'
can_expand = is_dir
if is_dir:
if name in dir_mtime_map:
mtime = dir_mtime_map[name]
if name in dir_size_map:
size = format_file_size(dir_size_map[name])
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
else:
try:
for rarinfo in infolist:
rar_fn = rarinfo.filename
rar_rest = rar_fn[len(prefix):] if prefix else rar_fn
display_name = decode_archive_name(rar_rest)
if (rar_rest == rest or display_name == rest) and not rarinfo.is_dir():
size = format_file_size(rarinfo.file_size)
try:
mtime = datetime(*rarinfo.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
break
except Exception:
pass
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
file_ext = os.path.splitext(name)[1].lower()
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': False})
except Exception as e:
raise Exception(f'读取RAR失败: {str(e)}')
elif ext in image_exts:
img_writeable = has_image_write_tools()
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
# pycdlib 打开失败时自动回退到 pytsk3
_img_pycdlib_parsed = False
if HAS_PYCDLIB and _is_iso9660_file(archive_full):
try:
iso = pycdlib.PyCdlib()
iso.open(archive_full)
try:
use_rr = iso.has_rock_ridge()
use_joliet = iso.has_joliet()
if use_rr:
walk_kwargs = {'rr_path': '/' + inner if inner else '/'}
elif use_joliet:
walk_kwargs = {'joliet_path': '/' + inner if inner else '/'}
else:
walk_kwargs = {'iso_path': '/' + inner if inner else '/'}
# 可用的路径系统列表（按优先级）
walk_path_key = list(walk_kwargs.keys())[0] if walk_kwargs else 'iso_path'
path_types = []
if use_rr:
path_types.append('rr_path')
if use_joliet:
path_types.append('joliet_path')
path_types.append('iso_path')
if walk_path_key in path_types:
path_types.remove(walk_path_key)
path_types.insert(0, walk_path_key)
def iso_stat(path_str, is_directory=False):
"""尝试用多种方式获取 ISO 文件 stat，返回 (size, mtime) 元组"""
path_variants = set()
path_variants.add(path_str)
if not path_str.startswith('/'):
path_variants.add('/' + path_str)
if is_directory:
dir_total = 0
dir_newest_mtime = None
for ptype in path_types:
for pval in path_variants:
try:
for wd, wdirs, wfiles in iso.walk(**{ptype: pval}):
for wf in wfiles:
child = (wd.rstrip('/') + '/' + wf).lstrip('/')
try:
bio = io.BytesIO()
iso.get_file_from_iso_fp(bio, **{ptype: child})
fs = bio.tell()
if fs > 0:
dir_total += fs
except Exception:
pass
except Exception:
pass
if dir_total > 0:
break
return (dir_total if dir_total > 0 else 0, dir_newest_mtime)
for ptype in path_types:
for pval in path_variants:
try:
stat_res = iso.stat(**{ptype: pval})
s = getattr(stat_res, 'size', None) or getattr(stat_res, 'st_size', None)
t = getattr(stat_res, 'mtime', None) or getattr(stat_res, 'st_mtime', None)
if s is not None and s > 0:
return (s, t)
return (s, None)
except Exception:
pass
if not is_directory:
for ptype in path_types:
for pval in path_variants:
try:
bio = io.BytesIO()
iso.get_file_from_iso_fp(bio, **{ptype: pval})
size = bio.tell()
if size >= 0:
return (size, None)
except Exception:
continue
return None
iso_stats_cache = {}
iso_fs_mtime = os.path.getmtime(archive_full)
def get_iso_item_stat(path_str, is_directory=False):
if path_str in iso_stats_cache:
return iso_stats_cache[path_str]
result = iso_stat(path_str, is_directory)
iso_stats_cache[path_str] = result
return result
try:
for walk_p, walk_dirs, walk_files in iso.walk(**walk_kwargs):
wp_clean = walk_p.rstrip('/')
for wf in walk_files:
full_path = ('/' + walk_p.lstrip('/') + '/' + wf) if walk_p != '/' else '/' + wf
get_iso_item_stat(full_path, is_directory=False)
for wd in walk_dirs:
full_path = ('/' + walk_p.lstrip('/') + '/' + wd) if walk_p != '/' else '/' + wd
get_iso_item_stat(full_path, is_directory=True)
break
except Exception:
pass
for p, dirs, files in iso.walk(**walk_kwargs):
start = '/' + inner if inner else '/'
if p.rstrip('/') == start.rstrip('/'):
for d in dirs:
dir_iso_fp = p.rstrip('/') + '/' + d
dir_s_stat, dir_t_stat = 0, None
dir_size = '-'
dir_mtime = '-'
try:
ds = get_iso_item_stat(dir_iso_fp, is_directory=True)
if ds:
dir_s_stat, dir_t_stat = ds[0] or 0, ds[1]
except Exception:
pass
if dir_t_stat and dir_t_stat > 0:
try:
dir_mtime = datetime.fromtimestamp(dir_t_stat).strftime('%Y-%m-%d %H:%M')
except Exception:
pass
if dir_mtime == '-':
dir_mtime = datetime.fromtimestamp(iso_fs_mtime).strftime('%Y-%m-%d %H:%M')
if dir_s_stat == 0:
try:
for ptype in path_types:
try:
for sub_p, sub_d, sub_ff in iso.walk(**{ptype: dir_iso_fp}):
for sf in sub_ff:
try:
bio_calc = io.BytesIO()
iso.get_file_from_iso_fp(bio_calc, **{ptype: (sub_p.rstrip('/') + '/' + sf).lstrip('/')})
dir_s_stat += bio_calc.tell()
except Exception:
pass
if dir_s_stat > 0:
break
except Exception:
continue
except Exception:
pass
if dir_s_stat > 0:
dir_size = format_file_size(dir_s_stat)
item_path = virtual_prefix + ('/'.join([inner, d]).lstrip('/') if inner else d)
file_ext = os.path.splitext(d)[1].lower()
can_expand_dir = file_ext in archive_exts or file_ext in image_exts
if not can_expand_dir:
can_expand_dir = True
is_archive_d = file_ext in archive_exts
is_image_d = file_ext in image_exts
items.append({'name': d, 'path': item_path, 'is_dir': True, 'size': dir_size, 'mtime': dir_mtime, 'is_archive': is_archive_d, 'is_image': is_image_d, 'can_expand': can_expand_dir, 'can_write': img_writeable})
for f in files:
item_path = virtual_prefix + ('/'.join([inner, f]).lstrip('/') if inner else f)
size = '-'
f_mtime = '-'
f_s = 0
f_t = None
try:
iso_fp = p.rstrip('/') + '/' + f
stat = get_iso_item_stat(iso_fp, is_directory=False)
if stat:
f_s = stat[0] if stat[0] else 0
if f_s:
size = format_file_size(f_s)
f_t = stat[1]
except Exception as e:
logger.warning(f"ISO get_iso_item_stat failed for {f}: {e}")
if f_t and f_t > 0:
try:
f_mtime = datetime.fromtimestamp(f_t).strftime('%Y-%m-%d %H:%M')
except Exception:
pass
if f_mtime == '-':
try:
f_mtime = datetime.fromtimestamp(iso_fs_mtime).strftime('%Y-%m-%d %H:%M')
except Exception as e:
logger.warning(f"ISO mtime fallback failed for {f}: {e}, iso_fs_mtime={iso_fs_mtime}")
pass
file_ext = os.path.splitext(f)[1].lower()
can_expand = file_ext in archive_exts or file_ext in image_exts
is_archive_f = file_ext in archive_exts
is_image_f = file_ext in image_exts
items.append({'name': f, 'path': item_path, 'is_dir': False, 'size': size, 'mtime': f_mtime, 'is_archive': is_archive_f, 'is_image': is_image_f, 'can_expand': can_expand, 'can_write': img_writeable})
break
_img_pycdlib_parsed = True
except Exception as _e:
logger.warning(f"pycdlib failed, falling back to pytsk3: {_e}")
finally:
try:
iso.close()
except Exception:
pass
except Exception as _e:
logger.warning(f"pycdlib open failed, falling back to pytsk3: {_e}")
# pytsk3 回退分支：当 pycdlib 不适用或失败时启用
if not _img_pycdlib_parsed:
parsed = False
walk_path = '/' + inner if inner else '/' # 定义 walk_path 作用域
if HAS_PYTSK3:
try:
iso_fs_mtime = os.path.getmtime(archive_full)
logger.info(f"[pytsk3] Attempting to open image: {archive_full} ({os.path.getsize(archive_full)} bytes)")
img_size = os.path.getsize(archive_full)
if img_size == 0:
raise ValueError(f"Image file is empty ({archive_full})")
img = pytsk3.Img_Info(archive_full)
logger.info(f"[pytsk3] Img_Info created successfully")
fs = pytsk3.FS_Info(img)
logger.info(f"[pytsk3] FS_Info created successfully, walk_path={walk_path}, fs_type={getattr(fs, 'fs_type', 'unknown')}")
def calc_tsk_dir_size(path_str):
total = 0
try:
subdir = fs.open_dir(path=path_str)
for subentry in subdir:
try:
sub_name = subentry.info.name.name.decode('utf-8') if isinstance(subentry.info.name.name, bytes) else subentry.info.name.name
except Exception:
continue
if sub_name in ['.', '..']:
continue
try:
if subentry.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR:
sub_path = path_str.rstrip('/') + '/' + sub_name
total += calc_tsk_dir_size(sub_path)
elif hasattr(subentry.info.meta, 'size') and subentry.info.meta.size is not None:
total += subentry.info.meta.size
except Exception:
pass
except Exception:
pass
return total
try:
dirobj = fs.open_dir(path=walk_path)
except Exception as e:
logger.warning(f"[pytsk3] open_dir failed for walk_path='{walk_path}': {e}, trying alternative formats")
dirobj = None
# 尝试备用路径格式
alt_paths = []
if inner:
# 尝试不带前导/的路径
alt_paths.append(inner)
# 尝试双斜杠
alt_paths.append('/' + inner.lstrip('/'))
# 尝试空路径（根目录）
alt_paths.append('/')
for alt_path in alt_paths:
try:
dirobj = fs.open_dir(path=alt_path)
logger.info(f"[pytsk3] open_dir succeeded with alt_path='{alt_path}'")
break
except Exception as e2:
logger.debug(f"[pytsk3] open_dir failed for alt_path='{alt_path}': {e2}")
dirobj = None
if dirobj is not None:
def _get_entry_name(entry):
try:
n = entry.info.name.name
short_name = n.decode('utf-8') if isinstance(n, bytes) else str(n)
if hasattr(entry.info.meta, 'name'):
meta = entry.info.meta
try:
name_iter = getattr(meta, 'name', None)
while name_iter:
try:
name_bytes = getattr(name_iter, 'name', None)
if name_bytes:
decoded = name_bytes.decode('utf-8') if isinstance(name_bytes, bytes) else str(name_bytes)
if decoded and decoded not in ('.', '..'):
parts = decoded.split('.', 1)
base = parts[0]
ext = parts[1] if len(parts) > 1 else ''
if len(base) > 8 or len(ext) > 3 or base != base.upper():
return decoded
name_iter = getattr(name_iter, 'next', None)
except Exception:
break
except Exception:
pass
return short_name
except Exception:
return None
for entry in dirobj:
name = _get_entry_name(entry)
if not name or name in ['.', '..']:
continue
if name.startswith('$') and name.upper() in ('$FAT1', '$FAT2', '$FAT12', '$MBR', '$ORPHANFILES'):
continue
try:
is_dir = entry.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR
except Exception:
is_dir = False
# 尝试获取文件大小和修改时间
size = '-'
mtime = '-'
try:
meta = entry.info.meta
meta_size = getattr(meta, 'size', None)
if meta_size is not None and not is_dir:
try:
size = format_file_size(int(meta_size))
except (TypeError, ValueError):
pass
elif is_dir:
dir_path = walk_path.rstrip('/') + '/' + name if walk_path != '/' else '/' + name
total_size = calc_tsk_dir_size(dir_path)
if total_size:
size = format_file_size(total_size)
except Exception:
pass
try:
meta = entry.info.meta
def get_timestamp(ts):
if ts is None:
return None
if hasattr(ts, 'secs'):
return ts.secs if ts.secs > 0 else None
if isinstance(ts, (int, float)) and ts > 0:
return int(ts)
return None
ts = get_timestamp(getattr(meta, 'mtime', None))
if ts:
mtime = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
if mtime == '-':
ts = get_timestamp(getattr(meta, 'crtime', None))
if ts:
mtime = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
if mtime == '-':
ts = get_timestamp(getattr(meta, 'ctime', None))
if ts:
mtime = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
# 如果仍然无法获取，使用镜像文件自身的修改时间
if mtime == '-':
try:
mtime = datetime.fromtimestamp(iso_fs_mtime).strftime('%Y-%m-%d %H:%M')
except Exception:
pass
except Exception:
pass
item_path = virtual_prefix + ('/'.join([inner, name]).lstrip('/') if inner else name)
can_expand = is_dir
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': img_writeable})
parsed = True
except Exception as e:
logger.warning(f"pytsk3 failed to open image {archive_full}: {e}\n{_tb.format_exc()}")
try:
if HAS_PYTSK3:
test_img = pytsk3.Img_Info(archive_full)
try:
test_fs = pytsk3.FS_Info(test_img)
walk_path_test = '/' + inner if inner else '/'
error_detail = f'pytsk3 can open image but {walk_path_test} not found in filesystem (fs_type={test_fs.info.ftype})'
test_fs.close()
except Exception as e2:
error_detail = f'pytsk3 cannot open filesystem on image: {e2}'
test_img.close()
return jsonify({'items': [], 'unsupported': True, 'error_detail': error_detail}), 500
except Exception:
pass
return jsonify({'items': [], 'unsupported': True}), 500
if not parsed:
try:
if HAS_PYTSK3:
test_img = pytsk3.Img_Info(archive_full)
test_fs = pytsk3.FS_Info(test_img)
walk_path_test = '/' + inner if inner else '/'
error_detail = f'pytsk3 can open image but {walk_path_test} not found in filesystem (fs_type={test_fs.fs_typ_str})'
try:
test_fs.close()
except Exception:
pass
try:
test_img.close()
except Exception:
pass
else:
error_detail = 'pytsk3 not installed'
except Exception as diag_e:
error_detail = f'pytsk3 error: {diag_e}'
logger.warning(f"无法打开镜像 {archive_full}: {error_detail}")
raise NotImplementedError(f'无法打开此镜像文件: {error_detail}')
else:
raise NotImplementedError('不支持的归档/镜像格式')
return itemsAPI 路由
def api_dir_tree():
return api_dir_tree_with_path(parse_request_path())def api_dir_tree_with_path_route():
"""路径选择树API - 与普通目录列表类似，但过滤掉普通文件只保留可展开项"""
return api_dir_tree_with_path(parse_request_path())def _api_dir_tree_nested(path, user_dir, pre_extracted_path=None, pre_extracted_temp_dir=None, virtual_prefix_override=None, from_userdir_cache=False, _extracted_set=None, _pre_orig_archive_rel=None):
"""处理多层嵌套虚拟路径（如 outer.zip::inner.tar.gz::subdir）
pre_extracted_path: 如果提供，跳过 parts[0] 的提取，直接使用该路径作为当前归档。
pre_extracted_temp_dir: 如果提供了 pre_extracted_path，传入创建该文件的临时目录，以便统一清理。
virtual_prefix_override: 如果提供，使用此值作为虚拟路径前缀（而非从 parts 推导），配合 pre_extracted_path 使用。
_extracted_set: 已提取的文件路径集合，用于避免重复提取（归档链解析时已提取的文件不再重复提取）。
_pre_orig_archive_rel: 如果提供，当 pre_extracted_path 存在时，用作 _orig_archive_rel（而非从 parts[0] 推导的缓存路径）。
"""
parts = [p.strip('/\\') for p in path.split('::')]
parts = [p for p in parts if p]
# 关键修复：拆分形如 BOOT/DOS71INS.IMG 的父目录+归档组合段
all_nested_exts = {'.zip','.tar','.tgz','.tar.gz','.gz','.bz2','.7z','.rar',
'.iso','.img','.vmdk','.vhd','.vhdx','.qcow2','.dmg','.wim'}
_np = [parts[0]] if parts else []
for _p in parts[1:]:
_pe = _get_archive_ext(_p)
if '/' in _p and _pe in all_nested_exts:
_ls = _p.rfind('/'); _pa = _p[:_ls]; _fn = _p[_ls+1:]
if _pa: _np.append(_pa)
_np.append(_fn)
else:
_np.append(_p)
parts = _np
if pre_extracted_path:
_pre_base = os.path.basename(pre_extracted_path)
for _pi, _pseg in enumerate(parts):
if os.path.basename(_pseg) == _pre_base:
parts = parts[_pi + 1:]
break
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
temp_dirs = []
try:
if pre_extracted_path:
# 外层归档已由调用者提取到临时目录，直接使用
current_archive_path = pre_extracted_path
else:
current_archive_path = os.path.join(user_dir, parts[0])
# 将预提取的临时目录加入清理列表
if pre_extracted_temp_dir and pre_extracted_temp_dir not in temp_dirs:
temp_dirs.append(pre_extracted_temp_dir)
# 确定有多少层归档需要提取
# parts[0] 是最外层归档，parts[1..N] 依次是嵌套归档或内部路径
# 关键：判断 parts[-1] 是否是归档文件 - 如果是，说明需要提取所有parts作为归档层
last_part_ext = _get_archive_ext(parts[-1])
last_is_archive = last_part_ext in archive_exts or last_part_ext in image_exts
# 统一归档链提取：从 i=1 向右扫，归档段提取为下一层，父目录段拼进 inner_name 并追加入 vp 后缀
_i = 0 if pre_extracted_path else 1
_vp_suffix_parts = []
while _i < len(parts):
_pex = _get_archive_ext(parts[_i])
if _pex in archive_exts or _pex in image_exts:
_pp = []; _j = _i - 1
while _j >= 1 and not (_get_archive_ext(parts[_j]) in archive_exts or _get_archive_ext(parts[_j]) in image_exts):
_pp.insert(0, parts[_j]); _j -= 1
_inner_name = '/'.join(_pp + [parts[_i]]) if _pp else parts[_i]
# vp 后缀只追加"父目录/归档名::"——父目录/目录段已经通过前面的 _more 分支追加过了
# 这里 _pp 是这个归档名前面的父目录段，需要用 '/' 连起来
_vp_path_to_archive = '/'.join(_pp + [parts[_i]]) if _pp else parts[_i]
_vp_suffix_parts.append(_vp_path_to_archive + '::')
# 检查是否已在归档链解析中提取过（避免重复提取）
_already_extract = None
if _extracted_set:
for _ep in _extracted_set:
if os.path.basename(_ep) == parts[_i] or os.path.basename(_ep) == _inner_name:
_already_extract = _ep
break
if _already_extract and os.path.exists(_already_extract):
current_archive_path = _already_extract
else:
_td = create_cache_dir(prefix='nested_')
temp_dirs.append(_td)
try:
_extract_archive_item(current_archive_path, _inner_name, _td)
except Exception:
return jsonify({'error': f'提取嵌套归档失败: {_inner_name}'}), 500
_ex = _find_extracted_file(_td, parts[_i])
if not _ex: _ex = _find_extracted_file(_td, _inner_name)
if not _ex:
return jsonify({'error': f'嵌套归档不存在: {_inner_name}'}), 404
current_archive_path = _ex
_i += 1
else:
_more = False
for _k in range(_i+1, len(parts)):
_kex = _get_archive_ext(parts[_k])
if _kex in archive_exts or _kex in image_exts:
# 注意：这里不要把 parts[_i] 追加到 _vp_suffix_parts，
# 因为上面归档分支已经把"父目录/归档名"作为整体追加过了。
# 只需要 _i 前进到下一段，循环继续即可。
_more = True; break
if _more:
_i += 1
else:
break
inner_path = '/'.join(parts[_i:]) if _i < len(parts) else ''
ext = _get_archive_ext(current_archive_path)
items = []
# 计算可写性：ZIP/TAR直接可写；7z/RAR/ISO可通过提取-重打包修改
can_write = ext in ('.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2')
if not can_write:
can_write = ext in ('.7z', '.rar') and (HAS_PY7ZR or HAS_RARFILE)
# ISO в嵌套路径中，即使不能直接写入，也可以提取-修改-重打包
if ext == '.iso' and HAS_PYCDLIB:
can_write = True
# IMG 文件通过 pytsk3 处理，如果有镜像写入工具则可写
if ext == '.img' and has_image_write_tools():
can_write = True
# 关键修复：构建虚拟路径前缀时区分归档/镜像文件和普通文件夹
# 只有归档/镜像文件之间用 :: 连接，普通文件夹用 / 连接
archive_parts = []
folder_parts = []
for i, part in enumerate(parts):
# 判断每个 part 是否是归档/镜像文件
part_ext = _get_archive_ext(part)
is_part_archive = part_ext in archive_exts or part_ext in image_exts
if is_part_archive:
archive_parts.append(part)
else:
# 普通文件夹及其后续部分都视为文件夹路径
folder_parts = parts[i:]
break
# 构建前缀：归档部分用 :: 连接
# 关键修复：virtual_prefix 只包含归档链（不含文件夹路径），inner_path 保留子目录路径
# 这样归档处理器（pytsk3/pycdlib等）能正确打开到子目录位置
original_inner_path = inner_path # 保存原始 inner_path 用于 ZIP 内部查找
if virtual_prefix_override:
virtual_prefix = virtual_prefix_override
# 方案2修复：override 只包含进入本层之前已知的前缀；
# 本层循环中新提取的更深层归档段必须追加，
# 否则三层及以上的 item_path 会与父级重叠、无法继续下钻
if _vp_suffix_parts:
virtual_prefix += ''.join(_vp_suffix_parts)
if virtual_prefix and not virtual_prefix.endswith('::'):
virtual_prefix = virtual_prefix + '::'
# 关键修复：当 pre_extracted_path 由调用者提供时，virtual_prefix 可能包含
# 缓存目录路径（如 .nested_cache/xxx/），需要替换为原始归档链路径，
# 防止缓存路径泄漏到前端导致后续请求无法正确解析
if _pre_orig_archive_rel and pre_extracted_path:
_cache_prefix = os.path.relpath(pre_extracted_path, user_dir).replace('\\', '/')
if virtual_prefix.startswith(_cache_prefix + '::'):
virtual_prefix = _pre_orig_archive_rel + '::' + virtual_prefix[len(_cache_prefix) + 2:]
elif virtual_prefix == _cache_prefix + '::':
virtual_prefix = _pre_orig_archive_rel + '::'
elif archive_parts:
virtual_prefix = parts[0] + '::'
# 追加嵌套归档链的后缀（如 'BOOT/DOS71INS.IMG::'）
# _vp_suffix_parts 在归档链提取循环中构建，包含了正确的父目录+归档名
if _vp_suffix_parts:
virtual_prefix += ''.join(_vp_suffix_parts)
# 不再将 folder_parts 追加到 virtual_prefix，也不再清空 inner_path
# inner_path 已经包含子目录路径（如 'subdir'），供归档处理器定位
else:
# 没有归档部分（不应该发生，因为 parts[0] 应该是归档）
virtual_prefix = parts[0] + '::' if parts else ''
# 处理ZIP
if ext == '.zip':
try:
with zipfile.ZipFile(current_archive_path, 'r') as zf:
# 关键修复：使用 original_inner_path 进行内部查找，而不是 inner_path
prefix = (original_inner_path + '/') if original_inner_path else ''
all_infos = zf.infolist()
all_names = zf.namelist()
names = [n for n in all_names if n.startswith(prefix) and n != prefix]
seen = set()
dir_mtime_map = {}
dir_size_map = {}
for info in all_infos:
raw_fn = info.filename
decoded_fn = decode_archive_name(raw_fn)
if '/' in decoded_fn:
dir_part = '/'.join(decoded_fn.split('/')[:-1])
if dir_part.startswith(prefix) and dir_part != prefix:
rel_dir = dir_part[len(prefix):] if prefix else dir_part
# 递归累加到所有上级目录
parts_list = rel_dir.split('/')
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
if ancestor and ancestor not in dir_mtime_map:
try:
dir_mtime_map[ancestor] = datetime(*info.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[ancestor] = '-'
if not raw_fn.endswith('/'):
dir_size_map[ancestor] = dir_size_map.get(ancestor, 0) + info.file_size
for n in names:
rest = n[len(prefix):] if prefix else n
rest_parts = rest.split('/')
name = rest_parts[0]
# 始终使用统一解码策略
name = decode_archive_name(name)
if name in seen:
continue
seen.add(name)
is_dir = (len(rest_parts) > 1) or n.endswith('/')
# 关键修复：正确使用 inner_path 构建 item 路径
item_inner = '/'.join([inner_path, name]).lstrip('/') if inner_path else name
item_path = virtual_prefix + item_inner
size = '-'
mtime = '-'
can_expand_item = is_dir
if is_dir:
if name in dir_mtime_map:
mtime = dir_mtime_map[name]
if name in dir_size_map:
size = format_file_size(dir_size_map[name])
elif size == '-':
size = '0 B'
else:
matched = False
for info in all_infos:
full_name = info.filename
fn_rest = full_name[len(prefix):] if prefix else full_name
fn_display = decode_archive_name(fn_rest)
if (fn_rest == rest or fn_display == rest or fn_rest == name or fn_display == name):
size = format_file_size(info.file_size)
try:
mtime = datetime(*info.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
matched = True
break
if not matched:
for full_name in all_names:
fn_rest = full_name[len(prefix):] if prefix else full_name
fn_display = decode_archive_name(fn_rest)
if fn_rest == rest or fn_display == rest or fn_rest == name or fn_display == name:
try:
info = zf.getinfo(full_name)
size = format_file_size(info.file_size)
try:
mtime = datetime(*info.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
except Exception:
pass
break
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand_item = True
# 关键修复：is_archive 和 is_image 只对文件设置，目录始终为 False
is_archive_item = (not is_dir) and (file_ext in archive_exts)
is_image_item = (not is_dir) and (file_ext in image_exts)
items.append({
'name': name, 'path': item_path, 'is_dir': is_dir,
'size': size, 'mtime': mtime, 'is_archive': is_archive_item,
'is_image': is_image_item, 'can_expand': can_expand_item,
'can_write': can_write
})
except Exception as e:
return jsonify({'error': f'读取ZIP失败: {str(e)}'}), 500
elif ext in ('.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.tbz'):
try:
with tarfile.open(current_archive_path, 'r') as tf:
# 关键修复：使用 original_inner_path 构建 prefix
prefix = (original_inner_path + '/') if original_inner_path else ''
members = [m for m in tf.getmembers() if m.name.startswith(prefix) and m.name != prefix]
seen = set()
dir_mtime_map = {}
dir_size_map = {}
for m in members:
raw_rest = m.name[len(prefix):] if prefix else m.name
decoded_rest = decode_archive_name(raw_rest)
if '/' in decoded_rest:
dir_part = '/'.join(decoded_rest.split('/')[:-1])
# 递归累加到所有上级目录
if dir_part:
parts_list = dir_part.split('/')
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
if ancestor not in dir_mtime_map:
try:
dir_mtime_map[ancestor] = datetime.fromtimestamp(m.mtime).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[ancestor] = '-'
if m.isfile():
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
dir_size_map[ancestor] = dir_size_map.get(ancestor, 0) + m.size
for m in members:
raw_rest = m.name[len(prefix):]
raw_parts = raw_rest.split('/')
name = decode_archive_name(raw_parts[0])
decoded_parts = decode_archive_name(raw_rest).split('/')
if name in seen:
continue
seen.add(name)
is_dir = (len(decoded_parts) > 1) or m.isdir()
# 关键修复：正确使用 inner_path 构建 item 路径
item_inner = '/'.join([inner_path, name]).lstrip('/') if inner_path else name
item_path = virtual_prefix + item_inner
size = format_file_size(m.size) if m.isfile() else '-'
try:
mtime = datetime.fromtimestamp(m.mtime).strftime('%Y-%m-%d %H:%M')
except Exception:
if is_dir and name in dir_mtime_map:
mtime = dir_mtime_map[name]
else:
mtime = '-'
if is_dir:
for i in range(len(decoded_parts) - 1, 0, -1):
ancestor = '/'.join(decoded_parts[:i+1])
if ancestor in dir_size_map:
size = format_file_size(dir_size_map[ancestor])
break
if size == '-':
size = '0 B'
can_expand_item = is_dir
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand_item = True
# 关键修复：is_archive 和 is_image 只对文件设置，目录始终为 False
is_archive_item = (not is_dir) and (file_ext in archive_exts)
is_image_item = (not is_dir) and (file_ext in image_exts)
items.append({
'name': name, 'path': item_path, 'is_dir': is_dir,
'size': size, 'mtime': mtime, 'is_archive': is_archive_item,
'is_image': is_image_item, 'can_expand': can_expand_item,
'can_write': can_write
})
except Exception as e:
return jsonify({'error': f'读取TAR失败: {str(e)}'}), 500
elif ext == '.7z' and HAS_PY7ZR:
try:
with py7zr.SevenZipFile(current_archive_path, mode='r') as zf:
# 关键修复：使用 original_inner_path 构建 prefix
prefix = (original_inner_path + '/') if original_inner_path else ''
names = [n for n in zf.getnames() if n.startswith(prefix) and n != prefix]
seen = set()
file_info_map = {}
try:
archive_info = zf.archive_info
if hasattr(archive_info, 'file_info_list'):
for fi in archive_info.file_info_list:
fn = fi.get('filename', '')
file_info_map[fn] = fi
except Exception:
pass
if not file_info_map:
try:
headers = zf.headers
if hasattr(headers, 'files_info'):
for fh in headers.files_info:
if hasattr(fh, 'filename'):
fn = fh.filename
file_info_map[fn] = {'uncompressed': fh.uncompressed, 'm_time': fh.m_time}
except Exception:
pass
dir_mtime_map = {}
dir_size_map = {}
for full_fn, fi in file_info_map.items():
if full_fn.startswith(prefix) and full_fn != prefix:
rest = full_fn[len(prefix):]
if '/' in rest:
dir_part = '/'.join(rest.split('/')[:-1])
# 递归累加到所有上级目录
if dir_part:
parts_list = dir_part.split('/')
m_time = fi.get('m_time', None) if isinstance(fi, dict) else None
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
if ancestor and ancestor not in dir_mtime_map:
if m_time:
try:
dir_mtime_map[ancestor] = datetime.fromtimestamp(m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[ancestor] = '-'
else:
dir_mtime_map[ancestor] = '-'
# 累加文件大小到目录
uncompressed_val = fi.get('uncompressed', None) if isinstance(fi, dict) else None
if isinstance(uncompressed_val, (list, tuple)):
size_value = uncompressed_val[0] if uncompressed_val else 0
else:
size_value = uncompressed_val or 0
if size_value:
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
dir_size_map[ancestor] = dir_size_map.get(ancestor, 0) + size_value
for n in names:
rest = n[len(prefix):]
rest_parts = rest.split('/')
name = decode_archive_name(rest_parts[0])
if name in seen:
continue
seen.add(name)
is_dir = (len(rest_parts) > 1) or n.endswith('/')
item_inner = '/'.join([inner_path, name]).lstrip('/') if inner_path else name
item_path = virtual_prefix + item_inner
size = '-'
mtime = '-'
can_expand_item = is_dir
if is_dir:
if name in dir_mtime_map:
mtime = dir_mtime_map[name]
if name in dir_size_map:
size = format_file_size(dir_size_map[name])
elif size == '-':
size = '0 B'
else:
matched_info = None
for full_fn, fi in file_info_map.items():
fn_rest = full_fn[len(prefix):] if prefix else full_fn
if fn_rest == rest or fn_rest == name:
matched_info = fi
break
if matched_info:
uncompressed = matched_info.get('uncompressed', None) if isinstance(matched_info, dict) else None
if uncompressed and uncompressed[0]:
size = format_file_size(uncompressed[0])
m_time = matched_info.get('m_time', None) if isinstance(matched_info, dict) else None
if m_time:
try:
mtime = datetime.fromtimestamp(m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
if size == '-':
try:
headers = zf.headers
for fh in headers.files_info:
if hasattr(fh, 'filename') and (fh.filename == rest or fh.filename == name):
if hasattr(fh, 'uncompressed') and fh.uncompressed:
size = format_file_size(fh.uncompressed[0]) if fh.uncompressed[0] else '-'
if hasattr(fh, 'm_time') and fh.m_time:
try:
mtime = datetime.fromtimestamp(fh.m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
break
except Exception:
pass
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand_item = True
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({
'name': name, 'path': item_path, 'is_dir': is_dir,
'size': size, 'mtime': mtime, 'is_archive': is_archive_item,
'is_image': is_image_item, 'can_expand': can_expand_item,
'can_write': can_write
})
except Exception as e:
return jsonify({'error': f'读取7z失败: {str(e)}'}), 500
elif ext == '.rar' and HAS_RARFILE:
try:
with rarfile.RarFile(current_archive_path) as rf:
# 关键修复：使用 original_inner_path 进行内部查找
prefix = (original_inner_path + '/') if original_inner_path else ''
infolist = rf.infolist()
# Use decode_archive_name for robust decoding; avoid ad-hoc encode/decode heuristics
use_gbk = False
names = [n for n in rf.namelist() if n.startswith(prefix) and n != prefix]
seen = set()
dir_mtime_map = {}
dir_size_map = {}
for rarinfo in infolist:
rar_fn = rarinfo.filename
decoded_fn = decode_archive_name(rar_fn)
if '/' in decoded_fn:
rest = decoded_fn[len(prefix):] if prefix else decoded_fn
dir_part = '/'.join(rest.split('/')[:-1])
top_dir = dir_part.split('/')[0] if dir_part else ''
if top_dir and top_dir not in dir_mtime_map:
try:
dir_mtime_map[top_dir] = datetime(*rarinfo.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[top_dir] = '-'
if not rarinfo.is_dir():
dir_size_map[top_dir] = dir_size_map.get(top_dir, 0) + rarinfo.file_size
for n in names:
rest = n[len(prefix):]
rest_parts = rest.split('/')
name = rest_parts[0]
# Decode entry name robustly
name = decode_archive_name(name)
if name in seen:
continue
seen.add(name)
is_dir = (len(rest_parts) > 1) or n.endswith('/')
# 关键修复：正确使用 inner_path 构建 item 路径
item_inner = '/'.join([inner_path, name]).lstrip('/') if inner_path else name
item_path = virtual_prefix + item_inner
size = '-'
mtime = '-'
can_expand_item = is_dir
if is_dir:
if name in dir_mtime_map:
mtime = dir_mtime_map[name]
if name in dir_size_map:
size = format_file_size(dir_size_map[name])
else:
try:
for rarinfo in infolist:
rar_fn = rarinfo.filename
rar_rest = rar_fn[len(prefix):] if prefix else rar_fn
display_name = decode_archive_name(rar_rest) if use_gbk else rar_rest
if (rar_rest == rest or display_name == rest or rar_rest == name or display_name == name) and not rarinfo.is_dir():
size = format_file_size(rarinfo.file_size)
try:
mtime = datetime(*rarinfo.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
break
except Exception:
pass
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand_item = True
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({
'name': name, 'path': item_path, 'is_dir': is_dir,
'size': size, 'mtime': mtime, 'is_archive': is_archive_item,
'is_image': is_image_item, 'can_expand': can_expand_item,
'can_write': can_write
})
except Exception as e:
return jsonify({'error': f'读取RAR失败: {str(e)}'}), 500
elif ext in ('.iso', '.img'):
# 统一处理 ISO/IMG：pycdlib 仅处理 .iso（ISO 9660），.img 直接用 pytsk3（FAT/NTFS）
# 与 list_virtual_directory 保持一致：IMG 文件通常是原始磁盘镜像，不是 ISO 9660 格式
_iso_parsed = False
if HAS_PYCDLIB and _is_iso9660_file(current_archive_path):
try:
iso = pycdlib.PyCdlib()
iso.open(current_archive_path)
use_rr = iso.has_rock_ridge()
use_joliet = iso.has_joliet()
# 关键修复：根据 current_archive_path 是否被修改以及 folder_parts 来判断 iso_inner_path
# 如果 current_archive_path 是原始路径且 original_inner_path 非空（folder_parts 为空），使用 original_inner_path
# 如果 current_archive_path 已被修改（提取后的文件），iso_inner_path 应该为空
# 但如果 folder_parts 非空，original_inner_path 保存了文件夹路径，需要在虚拟路径前缀中体现
original_archive_path = os.path.join(user_dir, parts[0])
# 只要调用者已经通过 pre_extracted_path 提供了当前处理的归档（说明经过了从外到内的提取）
# 就应该从当前 current_archive_path 的根目录开始浏览；inner_path 只反映
# 在当前归档内部的子目录位置（由 folder_parts / original_inner_path 决定）
if pre_extracted_path and not folder_parts:
iso_inner_path = ''
elif original_inner_path:
# original_inner_path 非空，表示需要浏览 ISO 的特定子目录
iso_inner_path = original_inner_path
else:
iso_inner_path = ''
iso_start_path = '/' + iso_inner_path if iso_inner_path else '/'
# 使用 ISO9660 主视图遍历（兼容性最好，始终有目录/文件列表）
# Rock Ridge / Joliet 仅用于 stat 和文件读取的长文件名支持
walk_kwargs = {'iso_path': iso_start_path}
def iso_stat(path_str):
"""尝试用多种方式获取 ISO 文件 stat"""
try:
return iso.stat(rr_path=path_str)
except Exception:
pass
try:
return iso.stat(joliet_path=path_str)
except Exception:
pass
try:
return iso.stat(iso_path=path_str)
except Exception:
pass
# 最后尝试读取文件内容获取大小
try:
out_fp = io.BytesIO()
iso.get_file_from_iso_fp(iso_path=path_str.lstrip('/'), outfp=out_fp)
return type('StatResult', (), {'size': len(out_fp.getvalue()), 'mtime': os.path.getmtime(current_archive_path)})()
except Exception:
pass
try:
out_fp = io.BytesIO()
iso.get_file_from_iso_fp(rr_path=path_str.lstrip('/'), outfp=out_fp)
return type('StatResult', (), {'size': len(out_fp.getvalue()), 'mtime': os.path.getmtime(current_archive_path)})()
except Exception:
pass
try:
out_fp = io.BytesIO()
iso.get_file_from_iso_fp(joliet_path=path_str.lstrip('/'), outfp=out_fp)
return type('StatResult', (), {'size': len(out_fp.getvalue()), 'mtime': os.path.getmtime(current_archive_path)})()
except Exception:
pass
return None
def iso_walk(path_str):
if use_rr:
return iso.walk(rr_path=path_str)
if use_joliet:
return iso.walk(joliet_path=path_str)
return iso.walk(iso_path=path_str)
for p, dirs, files in iso.walk(**walk_kwargs):
start = '/' + iso_inner_path if iso_inner_path else '/'
if p.rstrip('/') == start.rstrip('/'):
for d in dirs:
# 去掉 ISO 版本后缀（目录一般没有 ;1，但统一处理更稳妥）
d_clean = d[:-2] if d.lower().endswith(';1') else d
item_inner = '/'.join([inner_path, d_clean]).lstrip('/') if inner_path else d_clean
item_path = virtual_prefix + item_inner
dir_mtime = '-'
dir_size = '-'
try:
dir_iso_fp = p.rstrip('/') + '/' + d
dir_stat = iso_stat(dir_iso_fp)
if dir_stat:
if hasattr(dir_stat, 'size') and dir_stat.size:
dir_size = format_file_size(dir_stat.size)
else:
total_size = 0
try:
for sub_p, _, sub_files in iso_walk(dir_iso_fp):
for sub_f in sub_files:
sub_fp = sub_p.rstrip('/') + '/' + sub_f
sub_stat = iso_stat(sub_fp)
if sub_stat and hasattr(sub_stat, 'size') and sub_stat.size:
total_size += sub_stat.size
except Exception:
total_size = 0
if total_size:
dir_size = format_file_size(total_size)
if hasattr(dir_stat, 'mtime') and dir_stat.mtime is not None:
try:
dir_mtime = datetime.fromtimestamp(dir_stat.mtime).strftime('%Y-%m-%d %H:%M')
except Exception:
pass
except Exception:
pass
if dir_mtime == '-':
dir_mtime = _get_iso_fallback_mtime(dir_mtime, current_archive_path)
file_ext = os.path.splitext(d_clean)[1].lower()
can_expand_d = file_ext in archive_exts or file_ext in image_exts
if not can_expand_d:
can_expand_d = True
is_archive_d = file_ext in archive_exts
is_image_d = file_ext in image_exts
items.append({
'name': d_clean, 'path': item_path, 'is_dir': True,
'size': dir_size if dir_size != '-' else '0 B',
'mtime': _get_iso_fallback_mtime(dir_mtime, current_archive_path),
'is_archive': is_archive_d,
'is_image': is_image_d, 'can_expand': can_expand_d,
'can_write': can_write
})
for f in files:
# 去掉 ISO9660 的版本号后缀 ;1（显示长/干净文件名）
f_clean = f[:-2] if f.lower().endswith(';1') else f
item_inner = '/'.join([inner_path, f_clean]).lstrip('/') if inner_path else f_clean
item_path = virtual_prefix + item_inner
size = '-'
f_mtime = '-'
try:
iso_fp = p.rstrip('/') + '/' + f # 内部访问仍使用 ISO 原始文件名（含 ;1）
stat = iso_stat(iso_fp)
if stat:
if hasattr(stat, 'size') and stat.size:
size = format_file_size(stat.size)
elif hasattr(stat, 'size'):
size = '0 B'
if hasattr(stat, 'mtime') and stat.mtime is not None:
try:
f_mtime = datetime.fromtimestamp(stat.mtime).strftime('%Y-%m-%d %H:%M')
except Exception:
pass
except Exception:
pass
if size == '-':
try:
iso_fp = p.rstrip('/') + '/' + f
out_fp = io.BytesIO()
for kwargs in [
{'iso_path': iso_fp.lstrip('/')},
{'rr_path': iso_fp.lstrip('/')},
{'joliet_path': iso_fp.lstrip('/')},
]:
try:
iso.get_file_from_iso_fp(outfp=out_fp, **kwargs)
fdata = out_fp.getvalue()
size = format_file_size(len(fdata)) if fdata else '0 B'
break
except Exception:
out_fp = io.BytesIO()
continue
except Exception:
pass
if size == '-':
size = '0 B'
f_mtime = _get_iso_fallback_mtime(f_mtime, current_archive_path)
file_ext = os.path.splitext(f_clean)[1].lower()
can_expand = file_ext in archive_exts or file_ext in image_exts
is_archive_f = file_ext in archive_exts
is_image_f = file_ext in image_exts
items.append({
'name': f_clean, 'path': item_path, 'is_dir': False,
'size': size, 'mtime': f_mtime, 'is_archive': is_archive_f,
'is_image': is_image_f, 'can_expand': can_expand,
'can_write': can_write
})
break
iso.close()
_iso_parsed = True
except Exception as _e:
print(f'!!! ISO BRANCH EXC: {_e!r}', flush=True)
pass
if ext in ('.iso', '.img') and not _iso_parsed and HAS_PYTSK3:
# pytsk3 回退：当 pycdlib 无法读取时（如 FAT/NTFS 文件系统的镜像、含分区表的整盘镜像）
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
_iso_parsed = False
parsed = False
try:
img = pytsk3.Img_Info(current_archive_path)
fs = None
try:
fs = pytsk3.FS_Info(img)
except Exception:
fs = None
if fs is None:
# 尝试分区表（整盘镜像）
try:
vs = pytsk3.Volume_Info(img)
for part in vs:
if part.len <= 0:
continue
try:
fs = pytsk3.FS_Info(img, offset=part.start * 512)
break
except Exception:
fs = None
except Exception:
fs = None
if fs is not None:
walk_path = '/' + original_inner_path if original_inner_path else '/'
dirobj = None
# 尝试多种路径变体打开目录
path_variants = [walk_path]
if original_inner_path:
# 添加大小写变体
path_variants.append('/' + original_inner_path.upper())
path_variants.append('/' + original_inner_path.lower())
# 如果路径包含子目录，尝试递归查找
if '/' in original_inner_path.strip('/'):
# 从根目录开始递归查找目标目录
def _find_dir_recursive(fs_obj, target_path):
"""递归搜索目录树查找目标目录，通过路径后缀匹配"""
target_parts = [p.lower() for p in target_path.strip('/').split('/') if p]
if not target_parts:
try:
return fs_obj.open_dir(path='/')
except Exception:
return None
# BFS搜索所有目录
dirs_to_check = [('/')]
while dirs_to_check:
current_path = dirs_to_check.pop(0)
try:
dir_obj = fs_obj.open_dir(path=current_path)
except Exception:
continue
for entry in dir_obj:
try:
en = entry.info.name.name
enm = en.decode('utf-8', errors='replace') if isinstance(en, bytes) else str(en)
except Exception:
continue
if not enm or enm in ('.', '..'):
continue
if enm.startswith('$') and enm.upper() in ('$FAT1', '$FAT2', '$FAT12', '$MBR', '$ORPHANFILES'):
continue
try:
if hasattr(entry.info, 'meta') and entry.info.meta:
if entry.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR:
sub_path = current_path.rstrip('/') + '/' + enm
# 构建当前目录的路径部分
sub_parts = [p.lower() for p in sub_path.strip('/').split('/') if p]
# 检查路径后缀匹配
if len(sub_parts) >= len(target_parts):
match = True
offset = len(sub_parts) - len(target_parts)
for i, tp in enumerate(target_parts):
if sub_parts[offset + i] != tp:
match = False
break
if match:
return fs_obj.open_dir(path=sub_path)
dirs_to_check.append(sub_path)
except Exception:
pass
return None
# 尝试递归查找目录
dirobj = _find_dir_recursive(fs, original_inner_path)
# 如果递归查找也失败，尝试直接打开各种路径变体
if dirobj is None:
for pv in path_variants:
try:
dirobj = fs.open_dir(path=pv)
if dirobj is not None:
break
except Exception:
dirobj = None
if dirobj is not None:
def _img_entry_name(entry):
try:
n = entry.info.name.name
name = n.decode('utf-8') if isinstance(n, bytes) else n
except Exception:
return None
if name and name in ('.', '..'):
return name
try:
if hasattr(entry.info, 'meta') and entry.info.meta:
meta = entry.info.meta
if hasattr(meta, 'name'):
current = meta.name
visited = set()
while current:
try:
n = current.name
if n in visited:
break
visited.add(n)
decoded = n.decode('utf-8', errors='replace') if isinstance(n, bytes) else str(n)
if decoded and decoded not in ('.', '..'):
parts_name = decoded.split('.', 1)
base = parts_name[0]
ext_part = parts_name[1] if len(parts_name) > 1 else ''
if len(base) > 8 or len(ext_part) > 3 or base != base.upper() or ext_part != ext_part.upper():
return decoded
except Exception:
break
if hasattr(current, 'next') and current.next:
current = current.next
else:
break
except Exception:
pass
return name
for entry in dirobj:
name = _img_entry_name(entry)
if not name or name in ('.', '..'):
continue
if name.startswith('$') and name.upper() in ('$FAT1', '$FAT2', '$FAT12', '$MBR', '$ORPHANFILES'):
continue
try:
is_dir = entry.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR
except Exception:
is_dir = False
size = '-'
mtime = '-'
try:
meta = entry.info.meta
def get_timestamp(ts):
if ts is None:
return None
if hasattr(ts, 'secs'):
return ts.secs if ts.secs > 0 else None
if isinstance(ts, (int, float)) and ts > 0:
return int(ts)
return None
ts = get_timestamp(getattr(meta, 'mtime', None))
if ts:
mtime = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
if mtime == '-':
ts = get_timestamp(getattr(meta, 'crtime', None))
if ts:
mtime = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
if mtime == '-':
ts = get_timestamp(getattr(meta, 'ctime', None))
if ts:
mtime = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
except Exception:
pass
try:
meta = entry.info.meta
meta_size = getattr(meta, 'size', None)
if meta_size is not None and not is_dir:
try:
size = format_file_size(int(meta_size))
except (TypeError, ValueError):
pass
except Exception:
pass
item_path = virtual_prefix + ('/'.join([inner_path, name]).lstrip('/') if inner_path else name)
can_expand = is_dir
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': True})
parsed = True
try:
dirobj.close()
except Exception:
pass
try:
if fs is not None:
fs.close()
except Exception:
pass
try:
img.close()
except Exception:
pass
except Exception:
parsed = False
if not parsed:
return jsonify({'items': [], 'unsupported': True})
return jsonify({'items': items, 'current_path': path})
finally:
for d in temp_dirs:
try:
if from_userdir_cache or (d.startswith(os.path.join(user_dir, '.nested_cache'))):
# user_dir 内部缓存：不立即删除，刷新活动时间，交给后台 GC 线程回收
refresh_cache_activity(d)
else:
remove_cache_entry(d)
shutil.rmtree(d, ignore_errors=True)
except Exception:
passdef api_dir_tree_with_path(path):
if 'username' not in session:
return jsonify({'error': '未登录'}), 401
# 如果调用者已传入 path，则使用它；否则从 request.args 获取
if not path:
path = request.args.get('path', '').strip('/').replace('\\', '/')
user_dir = ensure_user_upload_dir(session['username'])
logger.debug(f"api_dir_tree_with_path called: path={path}, user={session.get('username')}")
# 处理虚拟路径（archive_path::inner/path）
if '::' in path:
# 使用 split_virtual_path 统一解析
archive_rel, inner = split_virtual_path(path)
archive_rel = archive_rel.strip('/')
archive_full = os.path.join(user_dir, archive_rel)
# 安全路径检查
if not is_safe_path(user_dir, archive_full):
return jsonify({'error': '路径不安全'}), 403
# 保存原始虚拟路径前缀，用于 item_path 构建（防止链式解析后 cache 路径泄漏到前端）
_orig_archive_rel = archive_rel
# 同时保存一条虚拟链路径，用于后续构建 orig_vp。这个值在重排序/链提取中
# 始终使用用户可见的归档名（不会变成缓存目录路径）
_virtual_chain = archive_rel
# 归档链解析：当 archive_rel 存在且 inner 包含 :: 时，
# 需要迭代提取归档链中的每一层，直到找到最终的内层归档
# 关键修复：当 archive_rel 不是物理文件时（如用户在 ISO 内部浏览 IMG），
# 第一段指向虚拟路径中的文件（如 BOOT/DOS71INS.IMG），真正的物理文件在 inner 中。
# 此时需要重新排序归档链，找到最外层的物理归档文件作为起点。
nested_all_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar', '.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
_chain_extracted = []
# 只在 archive_full 不存在时，才从 inner 段中寻找物理归档文件
# 当 archive_full 已存在时，它就是正确的最外层归档（如从 ISO 内部浏览时）
# 不要因为有同名物理文件就在 inner 中触发不必要的重排序
archive_exists = archive_full is not None and os.path.exists(archive_full)
if not archive_exists and '::' in inner:
_inner_segs = inner.split('::')
_inner_segs = [s for s in _inner_segs if s]
_reorder_found = False
_original_archive_rel = archive_rel # 保存原始的 archive_rel（如 BOOT/DOS71INS.IMG）
for _si, _seg in enumerate(_inner_segs):
_seg_path = os.path.join(user_dir, _seg)
if os.path.exists(_seg_path):
_before = _inner_segs[:_si]
if _before:
_pre = '/'.join(_before)
archive_rel = _pre + '/' + _seg
archive_full = os.path.join(user_dir, archive_rel)
else:
archive_rel = _seg
archive_full = _seg_path
_after = _inner_segs[_si+1:]
# 关键修复：重排序后，把原始的 archive_rel 保留在 inner 中
# 例如：BOOT/DOS71INS.IMG::MS-DOS 7.10.iso:: -> MS-DOS 7.10.iso::BOOT/DOS71INS.IMG::
inner_parts = [_original_archive_rel] + _after
inner = '/'.join(inner_parts) if inner_parts else ''
if inner and not inner.endswith('::'):
inner += '::'
_orig_archive_rel = archive_rel
_reorder_found = True
# 提取 archive_rel 中 '/' 之前的归档段（它们也是虚拟的，需要从物理归档中逐层提取）
if _before:
_cur_arc = archive_full
_cur_rel = archive_rel
_cur_inner = '/'.join(_before)
_reorder_extracted = []
for _bi, _bseg in enumerate(_before):
_bex = _get_archive_ext(_bseg)
if _bex in nested_all_exts:
_bfn = _bseg.rsplit('/', 1)[-1] if '/' in _bseg else _bseg
_btd = _nested_cache_dir(user_dir)
_bkey = _nested_cache_key(_cur_arc, _cur_inner)
_bsub = os.path.join(_btd, _bkey)
os.makedirs(_bsub, exist_ok=True)
mark_cache_in_use(_bsub)
try:
_bexist = _find_extracted_file(_bsub, _bfn)
if not _bexist:
_extract_archive_item(_cur_arc, _cur_inner, _bsub)
_bexist = _find_extracted_file(_bsub, _bfn)
except Exception as _be:
logger.warning(f"重排序提取失败: {_bseg}, error: {_be}")
break
if _bexist:
_reorder_extracted.append(_bexist)
_cur_arc = _bexist
if _reorder_extracted:
archive_full = _reorder_extracted[-1]
archive_rel = os.path.relpath(archive_full, user_dir).replace('\\', '/')
_orig_archive_rel = archive_rel
_chain_extracted.extend(_reorder_extracted)
break
if not _reorder_found:
return jsonify({'error': '不存在'}), 404
# 保存最原始的物理归档路径（在重排序/链提取之前），
# 用于构建正确的虚拟路径前缀，防止缓存路径泄漏到前端
_virtual_chain = _orig_archive_rel
if os.path.exists(archive_full) and '::' in inner:
_current_archive = archive_full
_current_rel = archive_rel
_current_inner = inner
_extracted_files = set()
_max_iterations = 10
_iter = 0
# 跟踪从原始物理归档开始的虚拟链路径，用于构建正确的 orig_vp
while _iter < _max_iterations and '::' in _current_inner:
_iter += 1
_chk = _current_inner.split('::', 1)[0]
_chk_ext = _get_archive_ext(_chk)
if _chk_ext not in nested_all_exts:
break
# 从当前归档中提取 _chk
_chk_filename = _chk
if '/' in _chk:
_ls = _chk.rfind('/')
_chk_filename = _chk[_ls+1:]
_td = _nested_cache_dir(user_dir)
_key = _nested_cache_key(_current_archive, _chk)
_subdir = os.path.join(_td, _key)
os.makedirs(_subdir, exist_ok=True)
mark_cache_in_use(_subdir)
try:
_existing = _find_extracted_file(_subdir, _chk_filename)
if not _existing:
_existing = _find_extracted_file(_subdir, _chk)
if _existing:
refresh_cache_activity(_subdir)
else:
_extract_archive_item(_current_archive, _chk, _subdir)
_existing = _find_extracted_file(_subdir, _chk_filename)
if not _existing:
_existing = _find_extracted_file(_subdir, _chk)
if not _existing:
_existing = _find_extracted_file(_subdir, _chk_filename)
except Exception as _e:
logger.warning(f"归档链解析提取失败: {_chk}, error: {_e}")
break
if not _existing:
break
_extracted_files.add(_existing)
# 更新虚拟链：用"原始虚拟路径::当前提取段"构建正确的前缀
_virtual_chain = f"{_virtual_chain}::{_chk}"
_rem = _current_inner.split('::', 1)[1]
_current_inner = _rem
_current_archive = _existing
if _current_archive != archive_full:
archive_full = _current_archive
archive_rel = _current_rel
inner = _current_inner
_chain_extracted = list(_extracted_files)
if not os.path.exists(archive_full):
return jsonify({'error': '不存在'}), 404
# 检查 inner 是否指向嵌套归档
inner_check = inner
if '::' in inner:
inner_check = inner.split('::', 1)[0]
remaining_inner = ''
if '::' in inner:
remaining_inner = inner.split('::', 1)[1]
if inner_check:
# 关键修复：当 inner_check 包含 '/' 时（如 'subdir/xxx.img'），
# 说明归档文件在虚拟目录中，需要先提取再处理
inner_parent = ''
inner_filename = inner_check
if '/' in inner_check:
last_slash = inner_check.rfind('/')
inner_parent = inner_check[:last_slash]
inner_filename = inner_check[last_slash+1:]
nested_ext = _get_archive_ext(inner_filename)
nested_all_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar', '.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
if nested_ext in nested_all_exts or remaining_inner:
# 归档/镜像在虚拟路径里：无论是否在子目录，都统一走 nested 处理。
# inner_parent 有值：归档在外层归档的子目录里，需要先提取；
# inner_parent 为空：归档在外层归档根目录，直接提取 basename 即可。
try:
extracted, nested_temp = _get_or_extract_nested_inner(
user_dir, archive_full, archive_rel, inner_check, inner_filename)
except Exception as e:
logger.warning(f"提取虚拟目录中的归档失败: {inner_check}, error: {e}")
return jsonify({'error': f'提取嵌套归档失败: {inner_filename}'}), 500
extracted_rel = os.path.relpath(extracted, user_dir).replace('\\', '/')
nested_vp = extracted_rel
if remaining_inner:
nested_vp = nested_vp + '::' + remaining_inner
# 构建 orig_vp 的 archive 部分：使用用户可见的虚拟链路径
# _virtual_chain 从 _orig_archive_rel 开始，在链式归档提取过程中
# 累加了所有经过的归档路径（如 "MS-DOS 7.10.iso::BOOT/DOS71INS.IMG"）
# 没有链式提取时，_virtual_chain 等于 _orig_archive_rel（即物理归档路径）
_orig_for_vp = _virtual_chain
orig_vp = f"{_orig_for_vp}::{inner_check}::"
return _api_dir_tree_nested(nested_vp, user_dir,
pre_extracted_path=extracted,
pre_extracted_temp_dir=nested_temp,
virtual_prefix_override=orig_vp,
from_userdir_cache=True,
_pre_orig_archive_rel=_orig_for_vp)
ext = _get_archive_ext(archive_full)
# 方案2补充：最内层文件扩展名不可识别时，按签名纠正格式
_known_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.tbz',
'.7z', '.rar', '.iso', '.img', '.vmdk', '.vhd', '.vhdx',
'.qcow2', '.dmg', '.wim'}
if ext not in _known_exts and _is_iso9660_file(archive_full):
ext = '.iso'
items = []
# 预计算可写性（在虚拟路径各分支中都需要使用）
_archive_writable = ext in ('.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2')
_image_writable = ext in ('.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim') and has_image_write_tools()
# ZIP 支持
if ext == '.zip':
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
try:
with zipfile.ZipFile(archive_full, 'r') as zf:
prefix = (inner + '/') if inner else ''
all_infos = zf.infolist()
all_names = zf.namelist()
names = [n for n in all_names if n.startswith(prefix) and n != prefix]
seen = set()
# 构建目录 mtime 和 size 映射
dir_mtime_map = {}
dir_size_map = {}
for info in all_infos:
raw_fn = info.filename
decoded_fn = decode_archive_name(raw_fn)
if '/' in decoded_fn:
rel = decoded_fn[len(prefix):] if prefix else decoded_fn
dir_part = '/'.join(rel.split('/')[:-1])
# 递归累加到所有上级目录
if dir_part:
parts_list = dir_part.split('/')
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
if ancestor and ancestor not in dir_mtime_map:
try:
dir_mtime_map[ancestor] = datetime(*info.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[ancestor] = '-'
# 非目录条目才计入大小
if not raw_fn.endswith('/'):
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
dir_size_map[ancestor] = dir_size_map.get(ancestor, 0) + info.file_size
for n in names:
rest = n[len(prefix):]
parts = rest.split('/')
name = parts[0]
# Decode entry name robustly
name = decode_archive_name(name)
if name in seen:
continue
seen.add(name)
is_dir = (len(parts) > 1) or n.endswith('/')
item_path = f"{_orig_archive_rel}::{('/'.join([inner, name]).lstrip('/') if inner else name)}"
size = '-'
mtime = '-'
can_expand = is_dir
if is_dir:
if name in dir_mtime_map:
mtime = dir_mtime_map[name]
if name in dir_size_map:
size = format_file_size(dir_size_map[name])
elif size == '-':
size = '0 B'
logger.info(f"ZIP dir item '{name}': size={size}, mtime={mtime}")
file_ext = os.path.splitext(name.rstrip('/'))[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
else:
matched = False
for info in all_infos:
full_name = info.filename
fn_rest = full_name[len(prefix):] if prefix else full_name
fn_display = decode_archive_name(fn_rest)
if fn_rest == rest or fn_display == rest:
size = format_file_size(info.file_size)
try:
mtime = datetime(*info.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
matched = True
break
if not matched:
for full_name in all_names:
fn_rest = full_name[len(prefix):] if prefix else full_name
fn_display = decode_archive_name(fn_rest)
if fn_rest == rest or fn_display == rest:
try:
info = zf.getinfo(full_name)
size = format_file_size(info.file_size)
try:
mtime = datetime(*info.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
except Exception:
pass
break
file_ext = os.path.splitext(name)[1].lower()
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': _archive_writable or _image_writable})
except Exception as e:
return jsonify({'error': f'读取 ZIP 失败: {str(e)}'}), 500
# tar / tar.gz / tgz / bz2 支持 (在 api_dir_tree_with_path 中)
elif ext in ('.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.tbz'):
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
try:
with tarfile.open(archive_full, 'r') as tf:
prefix = (inner + '/') if inner else ''
members = [m for m in tf.getmembers() if m.name.startswith(prefix) and m.name != prefix]
seen = set()
# 构建目录mtime/size映射（使用解码后的文件名）
dir_mtime_map = {}
dir_size_map = {}
for m in members:
raw_rest = m.name[len(prefix):] if prefix else m.name
decoded_rest = decode_archive_name(raw_rest)
if '/' in decoded_rest:
dir_part = '/'.join(decoded_rest.split('/')[:-1])
# 递归累加到所有上级目录
if dir_part:
parts_list = dir_part.split('/')
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
if ancestor not in dir_mtime_map:
try:
dir_mtime_map[ancestor] = datetime.fromtimestamp(m.mtime).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[ancestor] = '-'
if m.isfile():
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
dir_size_map[ancestor] = dir_size_map.get(ancestor, 0) + m.size
for m in members:
raw_rest = m.name[len(prefix):]
raw_parts = raw_rest.split('/')
name = decode_archive_name(raw_parts[0])
decoded_parts = decode_archive_name(raw_rest).split('/')
if name in seen:
continue
seen.add(name)
is_dir = (len(decoded_parts) > 1) or m.isdir()
item_path = f"{_orig_archive_rel}::{('/'.join([inner, name]).lstrip('/') if inner else name)}"
size = format_file_size(m.size) if m.isfile() else '-'
try:
mtime = datetime.fromtimestamp(m.mtime).strftime('%Y-%m-%d %H:%M') if hasattr(m, 'mtime') else '-'
except Exception:
if is_dir and name in dir_mtime_map:
mtime = dir_mtime_map[name]
else:
mtime = '-'
if is_dir:
for i in range(len(decoded_parts) - 1, 0, -1):
ancestor = '/'.join(decoded_parts[:i+1])
if ancestor in dir_size_map:
size = format_file_size(dir_size_map[ancestor])
break
if size == '-':
size = '0 B'
can_expand = is_dir
if not is_dir:
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
# 目录如果是压缩包名也应该可展开
if is_dir:
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
file_ext = os.path.splitext(name)[1].lower()
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': _archive_writable or _image_writable})
except Exception as e:
return jsonify({'error': f'读取 TAR 失败: {str(e)}'}), 500
# 7z 支持（如果可用）(在 api_dir_tree_with_path 中)
elif ext == '.7z' and HAS_PY7ZR:
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
try:
with py7zr.SevenZipFile(archive_full, mode='r') as zf:
prefix = (inner + '/') if inner else ''
names = [n for n in zf.getnames() if n.startswith(prefix) and n != prefix]
seen = set()
# 构建文件信息查找表
file_info_map = {}
try:
archive_info = zf.archive_info
if hasattr(archive_info, 'file_info_list'):
for fi in archive_info.file_info_list:
fn = fi.get('filename', '')
file_info_map[fn] = fi
except Exception:
pass
# 构建目录 mtime 和 size 映射
dir_mtime_map = {}
dir_size_map = {}
for full_fn, fi in file_info_map.items():
if full_fn.startswith(prefix) and full_fn != prefix:
rest = full_fn[len(prefix):]
if '/' in rest:
dir_part = '/'.join(rest.split('/')[:-1])
# 递归累加到所有上级目录
if dir_part:
parts_list = dir_part.split('/')
m_time = fi.get('m_time', None) if isinstance(fi, dict) else None
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
if ancestor and ancestor not in dir_mtime_map:
if m_time:
try:
dir_mtime_map[ancestor] = datetime.fromtimestamp(m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[ancestor] = '-'
else:
dir_mtime_map[ancestor] = '-'
# 累加文件大小到目录
uncompressed_val = fi.get('uncompressed', None) if isinstance(fi, dict) else None
if isinstance(uncompressed_val, (list, tuple)):
size_value = uncompressed_val[0] if uncompressed_val else 0
else:
size_value = uncompressed_val or 0
if size_value:
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
dir_size_map[ancestor] = dir_size_map.get(ancestor, 0) + size_value
for n in names:
rest = n[len(prefix):]
parts = rest.split('/')
name = decode_archive_name(parts[0])
if name in seen:
continue
seen.add(name)
is_dir = (len(parts) > 1) or n.endswith('/')
item_path = f"{_orig_archive_rel}::{('/'.join([inner, name]).lstrip('/') if inner else name)}"
size = '-'
mtime = '-'
can_expand = is_dir
if is_dir:
if name in dir_mtime_map:
mtime = dir_mtime_map[name]
if name in dir_size_map:
size = format_file_size(dir_size_map[name])
elif size == '-':
size = '0 B'
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
else:
matched_rest = rest
if rest in file_info_map:
fi = file_info_map[rest]
uncompressed = fi.get('uncompressed', None) if isinstance(fi, dict) else None
if uncompressed and uncompressed[0]:
size = format_file_size(uncompressed[0])
m_time = fi.get('m_time', None) if isinstance(fi, dict) else None
if m_time:
try:
mtime = datetime.fromtimestamp(m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
if size == '-' and name in file_info_map:
fi = file_info_map[name]
uncompressed = fi.get('uncompressed', None) if isinstance(fi, dict) else None
if uncompressed and uncompressed[0]:
size = format_file_size(uncompressed[0])
m_time = fi.get('m_time', None) if isinstance(fi, dict) else None
if m_time:
try:
mtime = datetime.fromtimestamp(m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
if size == '-':
try:
headers = zf.headers
for fh in headers.files_info:
if hasattr(fh, 'filename') and (fh.filename == rest or fh.filename == name):
if hasattr(fh, 'uncompressed') and fh.uncompressed:
size = format_file_size(fh.uncompressed[0]) if fh.uncompressed[0] else '-'
if hasattr(fh, 'm_time') and fh.m_time:
try:
mtime = datetime.fromtimestamp(fh.m_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
break
except Exception:
pass
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
file_ext = os.path.splitext(name)[1].lower()
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': _archive_writable or _image_writable})
except Exception as e:
return jsonify({'error': f'读取 7z 失败: {str(e)}'}), 500
# RAR 支持（如果可用）- api_dir_tree_with_path
elif ext == '.rar' and HAS_RARFILE:
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
try:
with rarfile.RarFile(archive_full) as rf:
prefix = (inner + '/') if inner else ''
infolist = rf.infolist()
# 增强编码检测
use_gbk = False
try:
for info in infolist:
fn = info.filename
if isinstance(fn, str):
try:
fn.encode('utf-8')
except UnicodeEncodeError:
use_gbk = True
break
if any(ord(c) > 127 and ord(c) < 256 for c in fn):
try:
fn.encode('latin-1').decode('gbk')
use_gbk = True
break
except Exception:
pass
except Exception:
pass
names = [n for n in rf.namelist() if n.startswith(prefix) and n != prefix]
seen = set()
# 构建目录 mtime 和 size 映射
dir_mtime_map = {}
dir_size_map = {}
for rarinfo in infolist:
rar_fn = rarinfo.filename
decoded_fn = decode_archive_name(rar_fn)
if '/' in decoded_fn:
rest = decoded_fn[len(prefix):] if prefix else decoded_fn
dir_part = '/'.join(rest.split('/')[:-1])
# 递归累加到所有上级目录
if dir_part:
parts_list = dir_part.split('/')
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
if ancestor and ancestor not in dir_mtime_map:
try:
dir_mtime_map[ancestor] = datetime(*rarinfo.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
dir_mtime_map[ancestor] = '-'
if not rarinfo.is_dir():
for i in range(len(parts_list)):
ancestor = '/'.join(parts_list[:i+1])
dir_size_map[ancestor] = dir_size_map.get(ancestor, 0) + rarinfo.file_size
for n in names:
rest = n[len(prefix):]
parts = rest.split('/')
name = parts[0]
# Decode entry name robustly
name = decode_archive_name(name)
if name in seen:
continue
seen.add(name)
is_dir = (len(parts) > 1) or n.endswith('/')
item_path = f"{_orig_archive_rel}::{('/'.join([inner, name]).lstrip('/') if inner else name)}"
size = '-'
mtime = '-'
can_expand = is_dir
if is_dir:
if name in dir_mtime_map:
mtime = dir_mtime_map[name]
if name in dir_size_map:
size = format_file_size(dir_size_map[name])
elif size == '-':
size = '0 B'
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
else:
try:
for rarinfo in infolist:
rar_fn = rarinfo.filename
rar_rest = rar_fn[len(prefix):] if prefix else rar_fn
display_name = decode_archive_name(rar_rest)
if (rar_rest == rest or display_name == rest or rar_rest == name or display_name == name) and not rarinfo.is_dir():
size = format_file_size(rarinfo.file_size)
try:
mtime = datetime(*rarinfo.date_time).strftime('%Y-%m-%d %H:%M')
except Exception:
mtime = '-'
break
except Exception:
pass
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
file_ext = os.path.splitext(name)[1].lower()
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': _archive_writable or _image_writable})
except Exception as e:
return jsonify({'error': f'读取 RAR 失败: {str(e)}'}), 500
else:
# 镜像或未支持的格式
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
if ext in image_exts:
# 按二进制签名判断真实格式，而不是按扩展名。
_iso_ok = False
if HAS_PYCDLIB and _is_iso9660_file(archive_full):
try:
iso = pycdlib.PyCdlib()
iso.open(archive_full)
use_rr = iso.has_rock_ridge()
use_joliet = iso.has_joliet()
if use_rr:
walk_kwargs = {'rr_path': '/' + inner if inner else '/'}
elif use_joliet:
walk_kwargs = {'joliet_path': '/' + inner if inner else '/'}
else:
walk_kwargs = {'iso_path': '/' + inner if inner else '/'}
def iso_stat(fp):
for kwargs in [{'rr_path': fp}, {'joliet_path': fp}, {'iso_path': fp}]:
try:
stat_res = iso.stat(**kwargs)
s = getattr(stat_res, 'size', None) or getattr(stat_res, 'st_size', None)
t = getattr(stat_res, 'mtime', None) or getattr(stat_res, 'st_mtime', None)
if s is not None or t is not None:
return (s, t)
except Exception:
pass
return None
def iso_walk_dir(path_str):
if use_rr:
return iso.walk(rr_path=path_str)
if use_joliet:
return iso.walk(joliet_path=path_str)
return iso.walk(iso_path=path_str)
for p, dirs, files in iso.walk(**walk_kwargs):
start = '/' + inner if inner else '/'
if p.rstrip('/') == start.rstrip('/'):
for d in dirs:
# 去掉 ISO ;1 版本号
d_clean = d[:-2] if d.lower().endswith(';1') else d
item_path = f"{_orig_archive_rel}::{('/'.join([inner, d_clean]).lstrip('/') if inner else d_clean)}"
dir_mtime = '-'
dir_size = '-'
try:
dir_iso_fp = p.rstrip('/') + '/' + d
stat_result = iso_stat(dir_iso_fp)
if stat_result:
dir_s, dir_t = stat_result
if dir_t:
dir_mtime = datetime.fromtimestamp(dir_t).strftime('%Y-%m-%d %H:%M')
if dir_s:
dir_size = format_file_size(dir_s)
else:
total_size = 0
for sub_p, _, sub_files in iso_walk_dir(dir_iso_fp):
for sub_f in sub_files:
sub_fp = sub_p.rstrip('/') + '/' + sub_f
sub_stat = iso_stat(sub_fp)
if sub_stat and sub_stat[0]:
total_size += sub_stat[0]
if total_size:
dir_size = format_file_size(total_size)
except Exception:
pass
if dir_mtime == '-':
try:
dir_mtime = datetime.fromtimestamp(os.path.getmtime(archive_full)).strftime('%Y-%m-%d %H:%M')
except Exception:
pass
file_ext = os.path.splitext(d_clean)[1].lower()
can_expand_d = file_ext in archive_exts or file_ext in image_exts
if not can_expand_d:
can_expand_d = True
is_archive_d = file_ext in archive_exts
is_image_d = file_ext in image_exts
items.append({'name': d_clean, 'path': item_path, 'is_dir': True, 'size': dir_size if dir_size != '-' else '0 B', 'mtime': dir_mtime, 'is_archive': is_archive_d, 'is_image': is_image_d, 'can_expand': can_expand_d, 'can_write': _archive_writable or _image_writable})
for f in files:
f_clean = f[:-2] if f.lower().endswith(';1') else f
item_path = f"{_orig_archive_rel}::{('/'.join([inner, f_clean]).lstrip('/') if inner else f_clean)}"
size = '-'
f_mtime = '-'
try:
iso_fp = p.rstrip('/') + '/' + f
stat_result = iso_stat(iso_fp)
if stat_result:
f_s, f_t = stat_result
if f_s:
size = format_file_size(f_s)
if f_t:
f_mtime = datetime.fromtimestamp(f_t).strftime('%Y-%m-%d %H:%M')
elif size == '-':
try:
bio = iso.get_file_from_iso_fp(iso_fp)
size_bytes = bio.getbuffer().nbytes
if size_bytes > 0:
size = format_file_size(size_bytes)
except Exception:
pass
except Exception:
pass
if f_mtime == '-':
try:
f_mtime = datetime.fromtimestamp(os.path.getmtime(archive_full)).strftime('%Y-%m-%d %H:%M')
except Exception:
pass
file_ext = os.path.splitext(f_clean)[1].lower()
can_expand = file_ext in archive_exts or file_ext in image_exts
is_archive_f = file_ext in archive_exts
is_image_f = file_ext in image_exts
items.append({'name': f_clean, 'path': item_path, 'is_dir': False, 'size': size, 'mtime': f_mtime, 'is_archive': is_archive_f, 'is_image': is_image_f, 'can_expand': can_expand, 'can_write': _archive_writable or _image_writable})
break
iso.close()
_iso_ok = True
except Exception:
_iso_ok = False
# 不直接 500，回退到 pytsk3
if _iso_ok:
pass
elif HAS_PYTSK3:
parsed = False
try:
img = pytsk3.Img_Info(archive_full)
fs = None
try:
fs = pytsk3.FS_Info(img)
except Exception:
fs = None
# 整盘镜像（含 MBR/分区表）：尝试第一个分区
if fs is None:
try:
vs = pytsk3.Volume_Info(img)
for part in vs:
if part.len <= 0:
continue
try:
fs = pytsk3.FS_Info(img, offset=part.start * 512)
break
except Exception:
fs = None
except Exception:
fs = None
if fs is None:
parsed = False
else:
walk_path = '/' + inner if inner else '/'
try:
dirobj = fs.open_dir(path=walk_path)
except Exception:
dirobj = None
if dirobj is not None:
# Helper function to get entry name and check LFN
def _img_entry_name(entry):
"""获取镜像条目的文件名，优先长文件名"""
try:
n = entry.info.name.name
name = n.decode('utf-8') if isinstance(n, bytes) else n
except Exception:
return None
if name and name in ('.', '..'):
return name
try:
# 尝试遍历 name 链获取长文件名
if hasattr(entry.info, 'meta') and entry.info.meta:
meta = entry.info.meta
if hasattr(meta, 'name'):
current = meta.name
visited = set()
while current:
try:
n = current.name
if n in visited:
break
visited.add(n)
decoded = n.decode('utf-8', errors='replace') if isinstance(n, bytes) else str(n)
if decoded and decoded not in ('.', '..'):
parts = decoded.split('.', 1)
base = parts[0]
ext = parts[1] if len(parts) > 1 else ''
# 长文件名特征：名称>8字符 或 扩展名>3字符 或 包含小写
if len(base) > 8 or len(ext) > 3 or base != base.upper() or ext != ext.upper():
return decoded
except Exception:
break
if hasattr(current, 'next') and current.next:
current = current.next
else:
break
except Exception:
pass
return name
for entry in dirobj:
name = _img_entry_name(entry)
if not name or name in ('.', '..'):
continue
# 过滤 pytsk3 的系统/元数据条目（$MBR/$FAT1/$FAT2/$OrphanFiles）
if name.startswith('$') and name.upper() in ('$FAT1', '$FAT2', '$FAT12', '$MBR', '$ORPHANFILES'):
continue
try:
is_dir = entry.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR
except Exception:
is_dir = False
size = '-'
mtime = '-'
try:
meta = entry.info.meta
def get_timestamp(ts):
if ts is None:
return None
if hasattr(ts, 'secs'):
return ts.secs if ts.secs > 0 else None
if isinstance(ts, (int, float)) and ts > 0:
return int(ts)
return None
ts = get_timestamp(getattr(meta, 'mtime', None))
if ts:
mtime = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
if mtime == '-':
ts = get_timestamp(getattr(meta, 'crtime', None))
if ts:
mtime = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
if mtime == '-':
ts = get_timestamp(getattr(meta, 'ctime', None))
if ts:
mtime = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M')
except Exception:
pass
try:
meta = entry.info.meta
meta_size = getattr(meta, 'size', None)
if meta_size is not None and not is_dir:
try:
size = format_file_size(int(meta_size))
except (TypeError, ValueError):
pass
except Exception:
pass
item_path = f"{_orig_archive_rel}::{('/'.join([inner, name]).lstrip('/') if inner else name)}"
can_expand = is_dir
file_ext = os.path.splitext(name)[1].lower()
if file_ext in archive_exts or file_ext in image_exts:
can_expand = True
is_archive_item = file_ext in archive_exts
is_image_item = file_ext in image_exts
img_can_write = True
items.append({'name': name, 'path': item_path, 'is_dir': is_dir, 'size': size, 'mtime': mtime, 'is_archive': is_archive_item, 'is_image': is_image_item, 'can_expand': can_expand, 'can_write': img_can_write})
parsed = True
except Exception:
parsed = parsed or False
finally:
try:
if fs is not None:
fs.close()
except Exception:
pass
try:
img.close()
except Exception:
pass
if not parsed:
return jsonify({'items': [], 'unsupported': True})
else:
return jsonify({'items': [], 'unsupported': True})
for it in items:
it['can_write'] = _archive_writable or _image_writable
return jsonify({'items': items, 'current_path': path})
# 普通目录（物理）
target_dir = os.path.join(user_dir, path) if path else user_dir
if not is_safe_path(user_dir, target_dir):
return jsonify({'error': '路径不安全'}), 403
if not os.path.exists(target_dir):
return jsonify({'error': '不存在'}), 404
# 如果 path 指向一个文件（归档/镜像），尝试展开其内部
# 注意：如果 path 已经包含 :: 则说明已进入虚拟路径，不应再次处理
if os.path.isfile(target_dir) and '::' not in path and path.count('::') < 1:
ext = os.path.splitext(target_dir)[1].lower()
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
if ext in archive_exts or ext in image_exts:
# 将 path 转为虚拟路径格式并重新调用
return api_dir_tree_with_path(path + '::')
return jsonify({'items': [], 'current_path': path})
items = []
for item in os.listdir(target_dir):
if item.startswith('.'):
continue
item_path = os.path.join(target_dir, item)
rel_path = os.path.join(path, item).replace('\\', '/') if path else item
if os.path.isfile(item_path):
size = os.path.getsize(item_path)
mtime = datetime.fromtimestamp(os.path.getmtime(item_path)).strftime('%Y-%m-%d %H:%M')
ext = os.path.splitext(item)[1].lower()
archive_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.7z', '.rar'}
image_exts = {'.iso', '.img', '.vmdk', '.vhd', '.vhdx', '.qcow2', '.dmg', '.wim'}
is_archive = ext in archive_exts
is_image = ext in image_exts
# 仅当服务器具备对应解析库时才标记可展开
can_expand = False
if is_archive:
if ext in ('.zip',):
can_expand = True
elif ext in ('.tar', '.tgz', '.tar.gz', '.gz', '.bz2', '.tbz'):
can_expand = True
elif ext == '.7z' and HAS_PY7ZR:
can_expand = True
elif ext == '.rar' and HAS_RARFILE:
can_expand = True
elif is_image:
if ext == '.iso' and HAS_PYCDLIB:
can_expand = True
elif HAS_PYTSK3:
can_expand = True
# 路径选择框只显示可展开的归档/镜像文件，普通文件不显示
# /api/dir_tree 应该显示所有文件，不跳过
# 是否可写（默认仅对常见归档开启写入，对镜像需额外工具支持）
can_write = False
writable_exts = {'.zip', '.tar', '.tgz', '.tar.gz', '.gz', '.bz2'}
if is_archive and ext in writable_exts:
can_write = True
elif is_image and has_image_write_tools():
can_write = True
items.append({
'name': item,
'path': rel_path,
'is_dir': False,
'size': format_file_size(size),
'mtime': mtime,
'is_archive': is_archive,
'is_image': is_image,
'can_expand': can_expand,
'can_write': can_write
})
else:
# 目录显示大小和修改时间
dir_size = 0
dir_mtime = 0
try:
dir_mtime = os.path.getmtime(item_path)
# 计算目录大小（可选，只计算直接子项）
for root, dirs, files in os.walk(item_path):
for f in files:
try:
fp = os.path.join(root, f)
dir_size += os.path.getsize(fp)
except (OSError, FileNotFoundError):
pass
except (OSError, FileNotFoundError):
dir_mtime = 0
items.append({
'name': item,
'path': rel_path,
'is_dir': True,
'size': format_file_size(dir_size) if dir_size > 0 else '-',
'mtime': datetime.fromtimestamp(dir_mtime).strftime('%Y-%m-%d %H:%M') if dir_mtime else '-',
'is_archive': False,
'is_image': False,
'can_expand': True,
'can_write': True
})
items.sort(key=lambda x: (not x['is_dir'], x['name'].lower()))
return jsonify({'items': items, 'current_path': path})ISO/IMG 专用提取函数
def _extract_iso_item(iso_path, inner_path, temp_dir):
"""从ISO/镜像文件中提取指定路径的项到临时目录，支持pycdlib和pytsk3回退"""
# 先尝试 pycdlib
if HAS_PYCDLIB and _is_iso9660_file(iso_path):
try:
iso = pycdlib.PyCdlib()
iso.open(iso_path)
try:
inner_lower = inner_path.lower()
use_rr = iso.has_rock_ridge()
use_joliet = iso.has_joliet()
# 按优先级构建路径系统列表：Rock Ridge > Joliet > ISO标准路径
path_systems = []
if use_rr:
path_systems.append('rr_path')
if use_joliet:
path_systems.append('joliet_path')
path_systems.append('iso_path')
def _strip_version_number(filename):
"""移除ISO 9660版本号后缀（如 ;1）"""
semicolon_idx = filename.rfind(';')
if semicolon_idx >= 0:
return filename[:semicolon_idx]
return filename
def _path_matches(walked_path, filename, inner_lower):
"""检查路径是否匹配目标inner_path"""
full_path = (walked_path.rstrip('/') + '/' + filename).lstrip('/')
full_path_clean = _strip_version_number(full_path)
return full_path_clean.lower() == inner_lower or full_path_clean.lower().startswith(inner_lower + '/')
def _build_dst_name(walked_path, filename, inner_lower, inner_path_orig):
"""构建目标文件名"""
full_path = (walked_path.rstrip('/') + '/' + filename).lstrip('/')
full_path_clean = _strip_version_number(full_path)
if full_path_clean.lower() == inner_lower:
return inner_path_orig
else:
sub_after_prefix = full_path_clean[len(inner_path_orig):]
return inner_path_orig + sub_after_prefix.replace(os.sep, '/')
# 依次尝试每种路径系统，找到匹配的文件即停止
for path_type in path_systems:
iso_to_dst = {}
try:
for dir_path, dirs, files in iso.walk(**{path_type: '/'}):
for f in files:
if _path_matches(dir_path, f, inner_lower):
dst_name = _build_dst_name(dir_path, f, inner_lower, inner_path)
full_iso_path = (dir_path.rstrip('/') + '/' + f).lstrip('/')
iso_to_dst['/' + full_iso_path] = (dst_name, path_type)
dst_path = os.path.join(temp_dir, dst_name)
os.makedirs(os.path.dirname(dst_path), exist_ok=True)
except Exception:
continue
if iso_to_dst:
# 找到匹配文件，执行提取
for iso_fp, (dst_name, pt) in iso_to_dst.items():
dst_path = os.path.join(temp_dir, dst_name)
os.makedirs(os.path.dirname(dst_path), exist_ok=True)
with open(dst_path, 'wb') as fp:
iso.get_file_from_iso_fp(fp, **{pt: iso_fp})
return
# 所有路径系统都未找到匹配文件
raise FileNotFoundError(f"ISO中未找到: {inner_path}")
finally:
iso.close()
except Exception as _e:
logger.warning(f"pycdlib extract failed, falling back to pytsk3: {_e}")
# pytsk3 回退
if HAS_PYTSK3:
try:
img = pytsk3.Img_Info(iso_path)
fs = pytsk3.FS_Info(img)
try:
# 尝试打开目标路径
walk_path = '/' + inner_path
dirobj = fs.open_dir(path=walk_path)
# 是目录，提取所有文件
for entry in dirobj:
try:
name = entry.info.name.name.decode('utf-8') if isinstance(entry.info.name.name, bytes) else entry.info.name.name
except Exception:
continue
if name in ['.', '..']:
continue
dst_name = os.path.join(temp_dir, inner_path, name) if inner_path else os.path.join(temp_dir, name)
os.makedirs(os.path.dirname(dst_name), exist_ok=True)
if entry.info.meta.type != pytsk3.TSK_FS_META_TYPE_DIR:
try:
meta_size = getattr(entry.info.meta, 'size', None)
if hasattr(meta_size, '__int__'):
meta_size = int(meta_size)
if meta_size and meta_size > 0:
with open(dst_name, 'wb') as fp:
entry.get_file().write_to_fd(fp.fileno())
except Exception:
pass
return
except Exception as _e:
logger.warning(f"pytsk3 open_dir failed for {walk_path}, trying as file: {_e}")
# 尝试作为文件提取
try:
walk_path = '/' + inner_path
obj = fs.open_file(path=walk_path)
dst_name = os.path.join(temp_dir, inner_path)
os.makedirs(os.path.dirname(dst_name), exist_ok=True)
with open(dst_name, 'wb') as fp:
obj.get_file().write_to_fd(fp.fileno())
return
except Exception:
pass
finally:
try:
fs.close()
except Exception:
pass
try:
img.close()
except Exception:
pass
except Exception as _e:
logger.warning(f"pytsk3 fallback also failed: {_e}")
raise Exception(f"无法从镜像文件提取 {inner_path}: pycdlib 和 pytsk3 均失败")def _extract_iso_to_dir(iso_path, temp_dir):
"""使用pycdlib或pytsk3将ISO/镜像文件内容提取到临时目录"""
# 先尝试 pycdlib
if HAS_PYCDLIB and _is_iso9660_file(iso_path):
try:
iso = pycdlib.PyCdlib()
iso.open(iso_path)
try:
for dir_path, dirs, files in iso.walk(iso_path='/'):
rel_dir = dir_path.lstrip('/').rstrip('/')
if rel_dir:
target_dir = os.path.join(temp_dir, rel_dir.replace('/', os.sep))
os.makedirs(target_dir, exist_ok=True)
for f in files:
src_iso_path = (dir_path.rstrip('/') + '/' + f).lstrip('/')
if rel_dir:
dst_path = os.path.join(temp_dir, rel_dir.replace('/', os.sep), f)
else:
dst_path = os.path.join(temp_dir, f)
os.makedirs(os.path.dirname(dst_path), exist_ok=True)
with open(dst_path, 'wb') as fp:
iso.get_file_from_iso_fp(fp, iso_path='/' + src_iso_path)
finally:
iso.close()
return
except Exception as _e:
logger.warning(f"pycdlib extract to dir failed, falling back to pytsk3: {_e}")
# pytsk3 回退
if HAS_PYTSK3:
try:
img = pytsk3.Img_Info(iso_path)
fs = pytsk3.FS_Info(img)
try:
def _walk_tsk_dir(path_str):
try:
dirobj = fs.open_dir(path=path_str)
for entry in dirobj:
try:
name = entry.info.name.name.decode('utf-8') if isinstance(entry.info.name.name, bytes) else entry.info.name.name
except Exception:
continue
if name in ['.', '..']:
continue
child_path = path_str.rstrip('/') + '/' + name
if entry.info.meta.type == pytsk3.TSK_FS_META_TYPE_DIR:
rel = child_path.lstrip('/')
if rel:
os.makedirs(os.path.join(temp_dir, rel), exist_ok=True)
_walk_tsk_dir(child_path)
else:
rel = child_path.lstrip('/')
dst_path = os.path.join(temp_dir, rel)
os.makedirs(os.path.dirname(dst_path), exist_ok=True)
try:
meta_size = getattr(entry.info.meta, 'size', None)
if hasattr(meta_size, '__int__'):
meta_size = int(meta_size)
if meta_size and meta_size > 0:
with open(dst_path, 'wb') as fp:
entry.get_file().write_to_fd(fp.fileno())
except Exception:
pass
except Exception:
pass
_walk_tsk_dir('/')
return
finally:
try:
fs.close()
except Exception:
pass
try:
img.close()
except Exception:
pass
except Exception as _e:
logger.warning(f"pytsk3 fallback extract to dir also failed: {_e}")
raise Exception(f"无法从镜像文件提取内容: pycdlib 和 pytsk3 均失败")
关于 LearnKu
粤公网安备 44030502004330号
推荐文章：