burgess/fastcopy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

finish fragment files copy

Browse Source
This commit is contained in:
Burgess Leo
2025-05-22 13:03:09 +08:00
parent 0c98dfecda
commit 3347abe02f
8 changed files with 252 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
fake_main.py
View File

@@ -1,14 +1,7 @@
from files_utils.files_list import GetFilesDBNodeInfo, GetFilesDBPathInfo, SortFragmentsByStartByte
from files_utils.files_save import CopyFileFromBytes
def GetSortFragments(files_list: list) -> list:
path_info = GetFilesDBPathInfo(db_path="./src/db_ntfs_info.db", table_name="db_path", files_path=files_list)
node_info = GetFilesDBNodeInfo(db_path="./src/db_ntfs_info.db", table_name="db_node", path_records=path_info)
result = SortFragmentsByStartByte(node_info)
return result
from files_utils.files_save import CopySingleFragmentFiles, CopyMultiFragmentFiles
from files_utils.files_sort import GetSortFragments
fragment_lists = {}
test_files = [
r"Y:\CloudMusic\AGA - MIZU.mp3",
r"Y:\CloudMusic\AGA - 一.mp3",
@@ -20,7 +13,10 @@ test_files = [
r"Y:\CloudMusic\Color Music Choir - Something Just Like This (Live).mp3"
]
sort_fragments = GetSortFragments(test_files)
sort_fragments = GetSortFragments(db_path="./src/db_ntfs_info.db", files_list=test_files)
for item in sort_fragments:
if item["extent_count"] == 1:
CopyFileFromBytes(item, target_path=r"Z:\test_files")
extent_count = item['extent_count']
if extent_count == 1:
CopySingleFragmentFiles(item, target_path=r"Z:\test_files")
elif extent_count > 1:
CopyMultiFragmentFiles(item, fragment_lists=fragment_lists, target_path=r"Z:\test_files")

87
files_utils/files_save.py
View File

@@ -1,7 +1,7 @@
import os
def extract_drive_letter(path: str) -> str:
def ExtractVolumeLetter(path: str) -> str:
"""从绝对路径中提取盘符"""
drive = os.path.splitdrive(path)[0]
if not drive:
@@ -9,7 +9,7 @@ def extract_drive_letter(path: str) -> str:
return drive[0].upper() # 返回 'Y'
def CopyFileFromBytes(source_data_dict, target_path):
def CopySingleFragmentFiles(source_data_dict, target_path):
"""
根据起始字节和长度，从磁盘中读取数据并保存为目标文件
@@ -29,7 +29,7 @@ def CopyFileFromBytes(source_data_dict, target_path):
print("错误：缺少必要的文件信息")
return
source_disk_path = extract_drive_letter(absolute_path)
source_disk_path = ExtractVolumeLetter(absolute_path)
target_file_path = os.path.join(target_path, file_name)
try:
@@ -60,14 +60,75 @@ def CopyFileFromBytes(source_data_dict, target_path):
print(f"发生错误: {str(e)}")
if __name__ == "__main__":
test_dict = {
'absolute_path': 'Y:\\CloudMusic\\Aaron Zigman - Main Title.mp3',
'filename': 'Aaron Zigman - Main Title.mp3',
'extent_count': 1,
'start_byte': 687685632,
'length': 7163904,
'fragment_index': 1
}
def ReadDiskBytes(volume_letter: str, start_byte: int, length: int) -> bytes:
"""
从指定磁盘的指定起始位置读取指定长度的字节。
CopyFileFromBytes(test_dict, target_path=r"Z:\RecoveredFiles")
:param volume_letter: 盘符（如 "Y"）
:param start_byte: 起始字节位置（整数）
:param length: 要读取的字节数（整数）
:return: 读取到的原始字节数据（bytes）
"""
if not isinstance(volume_letter, str) or len(volume_letter.strip()) != 1:
raise ValueError("drive_letter 必须是单个字母，如 'Y'")
# 构建 Windows 设备路径格式：\\.\Y:
disk_path = f"\\\\.\\{volume_letter.strip().upper()}:"
try:
with open(disk_path, "rb") as disk:
disk.seek(start_byte)
data = disk.read(length)
return data
except PermissionError:
raise PermissionError("权限不足，请以管理员身份运行程序")
except Exception as e:
raise RuntimeError(f"读取磁盘失败：{e}")
def CopyMultiFragmentFiles(
item: dict,
fragment_lists: dict,
target_path: str
):
"""
处理具有多个分片的文件，读取并按顺序拼接内容，最终写入磁盘。
:param item: 包含文件分片信息的字典
:param fragment_lists: 存储各文件分片内容的字典
:param target_path: 恢复文件的目标保存路径
:return: None
"""
file_name = item['filename']
extent_count = item['extent_count']
fragment_index = item['fragment_index']
start_byte = item['start_byte']
length_byte = item['length']
volume_letter = ExtractVolumeLetter(item['absolute_path'])
# 读取分片内容
fragment_content = ReadDiskBytes(volume_letter, start_byte, length_byte)
# 如果还没有为这个文件创建列表，则初始化
if file_name not in fragment_lists:
fragment_lists[file_name] = [None] * extent_count
# 将内容插入到指定位置
if fragment_index <= extent_count:
fragment_lists[file_name][fragment_index - 1] = fragment_content
print(f"已写入 {file_name} 的第 {fragment_index} 个片段。")
else:
print(f"警告：{file_name} 的 fragment_index 超出范围：{fragment_index} / {extent_count}")
# 检查是否所有分片都已加载
fragments = fragment_lists[file_name]
if None not in fragments:
full_content = b''.join(fragments)
target_file_path = os.path.join(target_path, file_name)
try:
with open(target_file_path, 'wb') as f:
f.write(full_content)
print(f"已成功恢复文件：{file_name}")
except Exception as e:
print(f"写入文件失败：{file_name}，错误：{e}")

20
files_utils/files_list.py → files_utils/files_sort.py
View File

@@ -141,20 +141,8 @@ def SortFragmentsByStartByte(file_extents_list: list) -> list:
return all_fragments
if __name__ == "__main__":
test_files = [
r"Y:\CloudMusic\AGA - MIZU.mp3",
r"Y:\CloudMusic\AGA - 一.mp3",
r"Y:\CloudMusic\Aaron Zigman - Main Title.mp3",
r"Y:\CloudMusic\Anson Seabra - Keep Your Head Up Princess.mp3",
r"Y:\CloudMusic\Anthony Keyrouz,Romy Wave - Something Just Like This (feat. Romy Wave).mp3",
r"Y:\CloudMusic\Ava Max - Sweet but Psycho.mp3",
r"Y:\CloudMusic\Cecilia Cheung - Turn Into Fireworks and Fall for You.mp3",
r"Y:\CloudMusic\Color Music Choir - Something Just Like This (Live).mp3"
]
path_info = GetFilesDBPathInfo(files_path=test_files)
node_info = GetFilesDBNodeInfo(path_records=path_info)
def GetSortFragments(db_path: str = "../src/db_ntfs_info.db", files_list: list = None) -> list:
path_info = GetFilesDBPathInfo(db_path=db_path, table_name="db_path", files_path=files_list)
node_info = GetFilesDBNodeInfo(db_path=db_path, table_name="db_node", path_records=path_info)
result = SortFragmentsByStartByte(node_info)
for item in result:
print(item)
return result

37
files_utils/public.py
View File

@@ -1,37 +0,0 @@
def ReadDiskBytes(volume_letter: str, start_byte: int, length: int) -> bytes:
"""
从指定磁盘的指定起始位置读取指定长度的字节。
:param volume_letter: 盘符（如 "Y"）
:param start_byte: 起始字节位置（整数）
:param length: 要读取的字节数（整数）
:return: 读取到的原始字节数据（bytes）
"""
if not isinstance(volume_letter, str) or len(volume_letter.strip()) != 1:
raise ValueError("drive_letter 必须是单个字母，如 'Y'")
# 构建 Windows 设备路径格式：\\.\Y:
disk_path = f"\\\\.\\{volume_letter.strip().upper()}:"
try:
with open(disk_path, "rb") as disk:
disk.seek(start_byte)
data = disk.read(length)
return data
except PermissionError:
raise PermissionError("权限不足，请以管理员身份运行程序")
except Exception as e:
raise RuntimeError(f"读取磁盘失败：{e}")
if __name__ == "__main__":
drive = "Y"
start = 687685632
size = 7163904
try:
content = ReadDiskBytes(drive, start, size)
print(f"成功读取 {len(content)} 字节内容。前100字节为：")
print(content[:100])
except Exception as e:
print("错误：", e)

36
test/fake_main.py Normal file
View File

@@ -0,0 +1,36 @@
from files_save import CopyMultiFragmentFiles, CopySingleFragmentFiles
target_path = r"Z:\Recovered"
# 存储各个文件的分片内容列表
fragment_lists = {}
test_file_sort = [{'absolute_path': 'Y:\\CloudMusic\\AGA - 一.mp3', 'filename': 'AGA - 一.mp3', 'extent_count': 2,
'start_byte': 23162880, 'length': 69632, 'fragment_index': 1},
{'absolute_path': 'Y:\\CloudMusic\\Aaron Zigman - Main Title.mp3',
'filename': 'Aaron Zigman - Main Title.mp3', 'extent_count': 1, 'start_byte': 687685632,
'length': 7163904, 'fragment_index': 1},
{'absolute_path': 'Y:\\CloudMusic\\AGA - MIZU.mp3', 'filename': 'AGA - MIZU.mp3', 'extent_count': 1,
'start_byte': 694849536, 'length': 8126464, 'fragment_index': 1},
{'absolute_path': 'Y:\\CloudMusic\\AGA - 一.mp3', 'filename': 'AGA - 一.mp3', 'extent_count': 2,
'start_byte': 702976000, 'length': 10870784, 'fragment_index': 2},
{'absolute_path': 'Y:\\CloudMusic\\Anson Seabra - Keep Your Head Up Princess.mp3',
'filename': 'Anson Seabra - Keep Your Head Up Princess.mp3', 'extent_count': 1,
'start_byte': 713846784, 'length': 7970816, 'fragment_index': 1}, {
'absolute_path': 'Y:\\CloudMusic\\Anthony Keyrouz,Romy Wave - Something Just Like This (feat. Romy Wave).mp3',
'filename': 'Anthony Keyrouz,Romy Wave - Something Just Like This (feat. Romy Wave).mp3',
'extent_count': 1, 'start_byte': 721817600, 'length': 9179136, 'fragment_index': 1},
{'absolute_path': 'Y:\\CloudMusic\\Ava Max - Sweet but Psycho.mp3',
'filename': 'Ava Max - Sweet but Psycho.mp3', 'extent_count': 1, 'start_byte': 731000832,
'length': 7938048, 'fragment_index': 1},
{'absolute_path': 'Y:\\CloudMusic\\Cecilia Cheung - Turn Into Fireworks and Fall for You.mp3',
'filename': 'Cecilia Cheung - Turn Into Fireworks and Fall for You.mp3', 'extent_count': 1,
'start_byte': 738938880, 'length': 6791168, 'fragment_index': 1},
{'absolute_path': 'Y:\\CloudMusic\\Color Music Choir - Something Just Like This (Live).mp3',
'filename': 'Color Music Choir - Something Just Like This (Live).mp3', 'extent_count': 1,
'start_byte': 745730048, 'length': 6193152, 'fragment_index': 1}]
for item in test_file_sort:
extent_count = item['extent_count']
if extent_count == 1:
CopySingleFragmentFiles(item, target_path)
elif extent_count > 1:
CopyMultiFragmentFiles(item, fragment_lists, target_path)

112
test/files_save.py
View File

@@ -1,7 +1,7 @@
import os
def extract_drive_letter(path: str) -> str:
def ExtractVolumeLetter(path: str) -> str:
"""从绝对路径中提取盘符"""
drive = os.path.splitdrive(path)[0]
if not drive:
@@ -9,7 +9,7 @@ def extract_drive_letter(path: str) -> str:
return drive[0].upper() # 返回 'Y'
def CopyFileFromBytes(source_data_dict, target_path):
def CopySingleFragmentFiles(source_data_dict, target_path):
"""
根据起始字节和长度，从磁盘中读取数据并保存为目标文件
@@ -29,7 +29,7 @@ def CopyFileFromBytes(source_data_dict, target_path):
print("错误：缺少必要的文件信息")
return
source_disk_path = extract_drive_letter(absolute_path)
source_disk_path = ExtractVolumeLetter(absolute_path)
target_file_path = os.path.join(target_path, file_name)
try:
@@ -60,13 +60,101 @@ def CopyFileFromBytes(source_data_dict, target_path):
print(f"发生错误: {str(e)}")
test_dict = {
'absolute_path': 'Y:\\CloudMusic\\Aaron Zigman - Main Title.mp3',
'filename': 'Aaron Zigman - Main Title.mp3',
'extent_count': 1,
'start_byte': 687685632,
'length': 7163904,
'fragment_index': 1
}
def ReadDiskBytes(volume_letter: str, start_byte: int, length: int) -> bytes:
"""
从指定磁盘的指定起始位置读取指定长度的字节。
CopyFileFromBytes(test_dict, target_path=r"Z:\RecoveredFiles")
:param volume_letter: 盘符（如 "Y"）
:param start_byte: 起始字节位置（整数）
:param length: 要读取的字节数（整数）
:return: 读取到的原始字节数据（bytes）
"""
if not isinstance(volume_letter, str) or len(volume_letter.strip()) != 1:
raise ValueError("drive_letter 必须是单个字母，如 'Y'")
# 构建 Windows 设备路径格式：\\.\Y:
disk_path = f"\\\\.\\{volume_letter.strip().upper()}:"
try:
with open(disk_path, "rb") as disk:
disk.seek(start_byte)
data = disk.read(length)
return data
except PermissionError:
raise PermissionError("权限不足，请以管理员身份运行程序")
except Exception as e:
raise RuntimeError(f"读取磁盘失败：{e}")
# if __name__ == "__main__":
# drive = "Y"
# start = 687685632
# size = 7163904
#
# try:
# content = ReadDiskBytes(drive, start, size)
# print(f"成功读取 {len(content)} 字节内容。前100字节为：")
# print(content[:100])
# except Exception as e:
# print("错误：", e)
def CopyMultiFragmentFiles(
item: dict,
fragment_lists: dict,
target_path: str
):
"""
处理具有多个分片的文件，读取并按顺序拼接内容，最终写入磁盘。
:param item: 包含文件分片信息的字典
:param fragment_lists: 存储各文件分片内容的字典
:param target_path: 恢复文件的目标保存路径
:return: None
"""
file_name = item['filename']
extent_count = item['extent_count']
fragment_index = item['fragment_index']
start_byte = item['start_byte']
length_byte = item['length']
volume_letter = ExtractVolumeLetter(item['absolute_path'])
# 读取分片内容
fragment_content = ReadDiskBytes(volume_letter, start_byte, length_byte)
# 如果还没有为这个文件创建列表，则初始化
if file_name not in fragment_lists:
fragment_lists[file_name] = [None] * extent_count
# 将内容插入到指定位置
if fragment_index <= extent_count:
fragment_lists[file_name][fragment_index - 1] = fragment_content
print(f"已写入 {file_name} 的第 {fragment_index} 个片段。")
else:
print(f"警告：{file_name} 的 fragment_index 超出范围：{fragment_index} / {extent_count}")
# 检查是否所有分片都已加载
fragments = fragment_lists[file_name]
if None not in fragments:
full_content = b''.join(fragments)
target_file_path = os.path.join(target_path, file_name)
try:
with open(target_file_path, 'wb') as f:
f.write(full_content)
print(f"已成功恢复文件：{file_name}")
except Exception as e:
print(f"写入文件失败：{file_name}，错误：{e}")
if __name__ == "__main__":
test_dict = {
'absolute_path': 'Y:\\CloudMusic\\Aaron Zigman - Main Title.mp3',
'filename': 'Aaron Zigman - Main Title.mp3',
'extent_count': 1,
'start_byte': 687685632,
'length': 7163904,
'fragment_index': 1
}
CopySingleFragmentFiles(test_dict, target_path=r"Z:\RecoveredFiles")

34
test/files_list.py → test/files_sort.py
View File

@@ -158,6 +158,7 @@ def sort_fragments_by_start_byte(file_extents_list: list) -> list:
for file_info in file_extents_list:
absolute_path = file_info['absolute_path']
filename = file_info['name']
extent_count = file_info['extent_count']
fragments = file_info['fragments']
# 对当前文件的片段排序（虽然通常已经是有序的）
@@ -168,6 +169,7 @@ def sort_fragments_by_start_byte(file_extents_list: list) -> list:
all_fragments.append({
'absolute_path': absolute_path,
'filename': filename,
'extent_count': extent_count,
'start_byte': fragment['start_byte'],
'length': fragment['length'],
'fragment_index': idx
@@ -190,9 +192,31 @@ if __name__ == "__main__":
r"Y:\CloudMusic\Cecilia Cheung - Turn Into Fireworks and Fall for You.mp3",
r"Y:\CloudMusic\Color Music Choir - Something Just Like This (Live).mp3"
]
path_info = GetFilesDBPathInfo(files_path=test_files)
file_extents_data = GetFilesDBNodeInfo(path_records=path_info)
result = sort_fragments_by_start_byte(file_extents_data)
test_files_sort = [
{'absolute_path': 'Y:\\CloudMusic\\AGA - MIZU.mp3', 'name': 'AGA - MIZU.mp3', 'path_id': 6, 'extent_count': 1,
'fragments': [{'start_byte': 694849536, 'length': 8126464}]},
{'absolute_path': 'Y:\\CloudMusic\\AGA - 一.mp3', 'name': 'AGA - 一.mp3', 'path_id': 7, 'extent_count': 2,
'fragments': [{'start_byte': 702976000, 'length': 10870784}, {'start_byte': 23162880, 'length': 69632}]},
{'absolute_path': 'Y:\\CloudMusic\\Aaron Zigman - Main Title.mp3', 'name': 'Aaron Zigman - Main Title.mp3',
'path_id': 5, 'extent_count': 1, 'fragments': [{'start_byte': 687685632, 'length': 7163904}]},
{'absolute_path': 'Y:\\CloudMusic\\Anson Seabra - Keep Your Head Up Princess.mp3',
'name': 'Anson Seabra - Keep Your Head Up Princess.mp3', 'path_id': 8, 'extent_count': 1,
'fragments': [{'start_byte': 713846784, 'length': 7970816}]},
{'absolute_path': 'Y:\\CloudMusic\\Anthony Keyrouz,Romy Wave - Something Just Like This (feat. Romy Wave).mp3',
'name': 'Anthony Keyrouz,Romy Wave - Something Just Like This (feat. Romy Wave).mp3', 'path_id': 9,
'extent_count': 1, 'fragments': [{'start_byte': 721817600, 'length': 9179136}]},
{'absolute_path': 'Y:\\CloudMusic\\Ava Max - Sweet but Psycho.mp3', 'name': 'Ava Max - Sweet but Psycho.mp3',
'path_id': 10, 'extent_count': 1, 'fragments': [{'start_byte': 731000832, 'length': 7938048}]},
{'absolute_path': 'Y:\\CloudMusic\\Cecilia Cheung - Turn Into Fireworks and Fall for You.mp3',
'name': 'Cecilia Cheung - Turn Into Fireworks and Fall for You.mp3', 'path_id': 11, 'extent_count': 1,
'fragments': [{'start_byte': 738938880, 'length': 6791168}]},
{'absolute_path': 'Y:\\CloudMusic\\Color Music Choir - Something Just Like This (Live).mp3',
'name': 'Color Music Choir - Something Just Like This (Live).mp3', 'path_id': 12, 'extent_count': 1,
'fragments': [{'start_byte': 745730048, 'length': 6193152}]}]
for item in result:
print(item)
# path_info = GetFilesDBPathInfo(files_path=test_files)
# file_extents_data = GetFilesDBNodeInfo(path_records=path_info)
result = sort_fragments_by_start_byte(test_files_sort)
print(result)
# for item in result:
# print(item)

0
main.py → test/folders_save.py
View File