From dd5d9a11ad8459814d4f0a57e925e3f317014de6 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 7 Nov 2025 20:52:44 +0800 Subject: [PATCH] =?UTF-8?q?create:=20=E5=88=9B=E5=BB=BA=20svfs=20=E4=BB=93?= =?UTF-8?q?=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 21 + README.md | 0 Scripts/clean_pycache.py | 12 + Scripts/code_content.py | 33 ++ pyproject.toml | 27 ++ src/svfs/__init__.py | 0 src/svfs/cli/__init__.py | 0 src/svfs/cli/packer.py | 40 ++ src/svfs/cli/unpacker.py | 77 ++++ src/svfs/core.py | 744 ++++++++++++++++++++++++++++++++++++ src/svfs/folder_packer.py | 103 +++++ src/svfs/folder_unpacker.py | 87 +++++ uv.lock | 8 + 13 files changed, 1152 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 Scripts/clean_pycache.py create mode 100644 Scripts/code_content.py create mode 100644 pyproject.toml create mode 100644 src/svfs/__init__.py create mode 100644 src/svfs/cli/__init__.py create mode 100644 src/svfs/cli/packer.py create mode 100644 src/svfs/cli/unpacker.py create mode 100644 src/svfs/core.py create mode 100644 src/svfs/folder_packer.py create mode 100644 src/svfs/folder_unpacker.py create mode 100644 uv.lock diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c8af3d0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,21 @@ +# Python-generated files +__pycache__/ +*.py[oc] +build/ +dist/ +wheels/ +*.egg-info + +# Virtual environments +.venv +resources + +# output files +output.md + +# vscode +.vscode/ +.python-version + +# svfs +*.svfs \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/Scripts/clean_pycache.py b/Scripts/clean_pycache.py new file mode 100644 index 0000000..070343f --- /dev/null +++ b/Scripts/clean_pycache.py @@ -0,0 +1,12 @@ +import os +import shutil +# 需要遍历的目录 +root_dir = "./" +# 遍历目录 +for dirpath, dirnames, filenames in os.walk(root_dir): + if "__pycache__" in dirnames: + # 获取 __pycache__ 目录的全路径 + pycache_dir = os.path.join(dirpath, "__pycache__") + # 删除目录 + shutil.rmtree(pycache_dir) + print(f"Removed: {pycache_dir}") diff --git a/Scripts/code_content.py b/Scripts/code_content.py new file mode 100644 index 0000000..eea10a7 --- /dev/null +++ b/Scripts/code_content.py @@ -0,0 +1,33 @@ +import os +import sys + + +def generate_markdown_from_py_files(directory, output_file): + with open(output_file, "w", encoding="utf-8") as md_file: + for root, dirs, files in os.walk(directory): + # 排除 venv 目录 + dirs[:] = [d for d in dirs if d != ".venv"] + dirs[:] = [d for d in dirs if d != ".vscode"] + dirs[:] = [d for d in dirs if d != "Scripts"] + dirs[:] = [d for d in dirs if d != "build"] + for file in files: + if ( + file.endswith(".py") + or file.endswith(".rs") + or file.endswith(".toml") + ): + file_path = os.path.join(root, file) + md_file.write(f"`{file_path}`\n") + md_file.write("```python\n") + with open(file_path, "r", encoding="utf-8") as py_file: + md_file.write(py_file.read()) + md_file.write("\n```\n\n") + + +if __name__ == "__main__": + # 指定目录和输出文件名 + target_directory = sys.argv[1] # 替换为你的目标目录 + output_markdown_file = "output.md" # 输出的 Markdown 文件名 + + generate_markdown_from_py_files(target_directory, output_markdown_file) + print(f"Markdown 文件已生成：{output_markdown_file}") diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..7fc9a80 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,27 @@ +[project] +dependencies = [] +description = "Add your description here" +name = "svfs" +readme = "README.md" +requires-python = ">=3.7" +version = "0.1.0" + +[[tool.uv.index]] +default = true +url = "https://pypi.tuna.tsinghua.edu.cn/simple" + +[tool.uv.workspace] +members = [ + "devmo", +] + +[build-system] +build-backend = "hatchling.build" +requires = ["hatchling"] + +[project.scripts] +vfspack = "svfs.cli.packer:pack" +vfsunpack = "svfs.cli.unpacker:unpack" + +[tool.hatch.build.targets.wheel] +packages = ["src/svfs"] diff --git a/src/svfs/__init__.py b/src/svfs/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/svfs/cli/__init__.py b/src/svfs/cli/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/svfs/cli/packer.py b/src/svfs/cli/packer.py new file mode 100644 index 0000000..182c9b6 --- /dev/null +++ b/src/svfs/cli/packer.py @@ -0,0 +1,40 @@ +#!/usr/bin/env python3 +""" +文件夹打包工具 - 将文件夹打包为SQLite虚拟文件系统 +""" + +import os +import argparse +from ..folder_packer import FolderPacker +from ..core import SQLiteVFS + + +def pack(): + parser = argparse.ArgumentParser(description="将文件夹打包为SQLite虚拟文件系统") + parser.add_argument("source_folder", help="要打包的源文件夹路径") + parser.add_argument("output_db", help="输出的SQLite数据库文件路径") + parser.add_argument("--name", default="packed_filesystem", help="文件系统名称") + parser.add_argument("--compress", action="store_true", help="启用压缩") + parser.add_argument("--exclude", nargs="*", help="要排除的文件模式") + + args = parser.parse_args() + + if not os.path.exists(args.source_folder): + print(f"错误: 源文件夹不存在: {args.source_folder}") + return + + # 创建打包器 + packer = FolderPacker( + sqlite_vfs=SQLiteVFS(args.output_db, args.compress), + exclude_patterns=args.exclude or [], + fs_name=args.name, + ) + + try: + packer.pack_folder(args.source_folder) + finally: + packer.sqlite_vfs.close() + + +if __name__ == "__main__": + pack() diff --git a/src/svfs/cli/unpacker.py b/src/svfs/cli/unpacker.py new file mode 100644 index 0000000..0b0ed38 --- /dev/null +++ b/src/svfs/cli/unpacker.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python3 +""" +文件夹解包工具 - 从SQLite虚拟文件系统导出文件 +""" + +import os +import argparse +from ..core import SQLiteVFS +from ..folder_unpacker import FolderUnpacker + + +def unpack(): + parser = argparse.ArgumentParser(description="从SQLite虚拟文件系统导出文件") + parser.add_argument("db_file", help="SQLite数据库文件路径") + parser.add_argument("export_folder", help="导出目标文件夹路径") + parser.add_argument("--path", help="要导出的特定文件或目录路径（默认导出全部）") + parser.add_argument( + "--list-root", action="store_true", help="列出根目录内容而不导出" + ) + parser.add_argument("--debug", action="store_true", help="显示调试信息") + parser.add_argument( + "--ignore-timestamps", + action="store_true", + help="忽略时间戳错误，只导出文件内容", + ) + + args = parser.parse_args() + + if not os.path.exists(args.db_file): + print(f"错误: 数据库文件不存在: {args.db_file}") + return + + # 创建解包器 + unpacker = FolderUnpacker(sqlite_vfs=SQLiteVFS(args.db_file)) + + try: + if args.list_root: + # 只列出根目录内容 + print("根目录内容:") + items = unpacker.list_root() + for item in items: + file_type = "DIR" if item["is_directory"] else "FILE" + print(f" {file_type}: {item['name']} (路径: {item['path']})") + return + + if args.path: + # 导出特定路径 + if unpacker.sqlite_vfs.is_dir(args.path): + local_dir = os.path.join( + args.export_folder, os.path.basename(args.path) + ) + unpacker.unpack_directory(args.path, local_dir) + else: + local_file = os.path.join( + args.export_folder, os.path.basename(args.path) + ) + if args.ignore_timestamps: + unpacker.sqlite_vfs.export_file_safe(args.path, local_file) + else: + unpacker.unpack_file(args.path, local_file) + else: + # 导出整个文件系统 + if args.ignore_timestamps: + # 使用安全导出方法 + unpacker.sqlite_vfs.export_filesystem = ( + lambda export_root: unpacker.sqlite_vfs.export_filesystem( + export_root + ) + ) + unpacker.unpack_filesystem(args.export_folder) + + finally: + unpacker.sqlite_vfs.close() + + +if __name__ == "__main__": + unpack() diff --git a/src/svfs/core.py b/src/svfs/core.py new file mode 100644 index 0000000..6c3508d --- /dev/null +++ b/src/svfs/core.py @@ -0,0 +1,744 @@ +#!/usr/bin/env python3 +""" +SQLite虚拟文件系统实现 +""" + +import sqlite3 +import os +import zlib +from datetime import datetime + + +class SQLiteVFS: + def __init__(self, db_path, compress=False): + self.db_path = db_path + self.compress = compress + self.conn = None + self.total_files = 0 + self.total_size = 0 + self.open_files = {} + self.connect() + + def connect(self): + """连接到文件系统数据库""" + self.conn = sqlite3.connect(self.db_path) + self.conn.row_factory = sqlite3.Row + self._create_tables() + + def clean(self): + if self.conn is None: + raise ValueError("数据库连接未建立") + cursor = self.conn.cursor() + cursor.execute("DELETE FROM files") + cursor.execute("DELETE FROM filesystem_metadata") + + def add_vfs_metadata(self, name): + """添加文件系统元数据""" + if self.conn is None: + raise ValueError("数据库连接未建立") + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO filesystem_metadata (name) + VALUES (?) + """, + (name,), + ) + self.conn.commit() + + def update_vfs_metadata(self, name, total_files, total_size): + """更新文件系统元数据""" + if self.conn is None: + raise ValueError("数据库连接未建立") + cursor = self.conn.cursor() + cursor.execute( + """ + UPDATE filesystem_metadata + SET total_files = ?, total_size = ? + WHERE name = ? + """, + (total_files, total_size, name), + ) + self.conn.commit() + + def _create_tables(self): + """创建必要的表""" + if self.conn is None: + raise ValueError("数据库连接未建立") + cursor = self.conn.cursor() + + # 创建文件系统元数据表 + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS filesystem_metadata ( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL, + created_time DATETIME DEFAULT CURRENT_TIMESTAMP, + total_files INTEGER DEFAULT 0, + total_size INTEGER DEFAULT 0 + ) + """ + ) + + # 创建文件表 + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS files ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + path TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + parent_path TEXT, + is_directory BOOLEAN NOT NULL DEFAULT 0, + file_size INTEGER DEFAULT 0, + created_time DATETIME, + modified_time DATETIME, + permissions INTEGER DEFAULT 0644, + content BLOB, + compressed BOOLEAN DEFAULT 0 + ) + """ + ) + + # 创建索引 + cursor.execute("CREATE INDEX IF NOT EXISTS idx_files_path ON files(path)") + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_files_parent ON files(parent_path)" + ) + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_files_directory ON files(is_directory)" + ) + + self.conn.commit() + + def add_directory(self, rel_path, source_folder): + """添加目录到数据库""" + try: + abs_path = os.path.join(source_folder, rel_path) + stat = os.stat(abs_path) + + if self.conn is None: + raise ValueError("数据库连接未建立") + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO files + (path, name, parent_path, is_directory, created_time, modified_time, permissions) + VALUES (?, ?, ?, 1, ?, ?, ?) + """, + ( + rel_path.replace("\\", "/"), + os.path.basename(rel_path), + # 将 \ 替换为 / + os.path.dirname(rel_path).replace("\\", "/"), + datetime.fromtimestamp(stat.st_ctime), + datetime.fromtimestamp(stat.st_mtime), + stat.st_mode & 0o777, + ), + ) + + except Exception as e: + print(f"添加目录失败 {rel_path}: {e}") + + else: + self.conn.commit() + + def add_file(self, abs_path, rel_path): + """添加文件到数据库""" + try: + stat = os.stat(abs_path) + file_size = stat.st_size + + # 读取文件内容 + with open(abs_path, "rb") as f: + content = f.read() + + # 压缩处理 + compressed = False + if self.compress and len(content) > 1024: # 大于1KB的文件才压缩 + compressed_content = zlib.compress(content, level=6) + if len(compressed_content) < len(content): + content = compressed_content + compressed = True + + if self.conn is None: + raise ValueError("数据库连接未建立") + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO files + (path, name, parent_path, is_directory, file_size, + created_time, modified_time, permissions, content, compressed) + VALUES (?, ?, ?, 0, ?, ?, ?, ?, ?, ?) + """, + ( + rel_path.replace("\\", "/"), + os.path.basename(rel_path), + os.path.dirname(rel_path).replace("\\", "/"), + file_size, + datetime.fromtimestamp(stat.st_ctime), + datetime.fromtimestamp(stat.st_mtime), + stat.st_mode & 0o777, + sqlite3.Binary(content), + compressed, + ), + ) + + self.total_files += 1 + self.total_size += file_size + + print(f"添加文件: {rel_path} ({file_size} bytes)") + + except Exception as e: + print(f"添加文件失败 {rel_path}: {e}") + + def get_file_info(self, path): + """获取文件信息""" + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + cursor = self.conn.cursor() + cursor.execute("SELECT * FROM files WHERE path = ?", (path.lstrip("/"),)) + row = cursor.fetchone() + if row: + # 将时间字符串转换为datetime对象 + return self._parse_row_datetime(row) + return None + + def list_directory(self, path): + """列出目录内容""" + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + if path == "/": + sql = """ + SELECT * FROM files + ORDER BY is_directory DESC, name ASC + """ + else: + sql = f""" + SELECT * FROM files + WHERE path = '{path.lstrip("/")}' + ORDER BY is_directory DESC, name ASC + """ + cursor = self.conn.cursor() + cursor.execute(sql) + rows = cursor.fetchall() + # 将时间字符串转换为datetime对象 + return [self._parse_row_datetime(row) for row in rows] + + def _parse_row_datetime(self, row): + """将行中的时间字符串转换为datetime对象""" + from datetime import datetime + + parsed_row = dict(row) + + # 处理created_time + if parsed_row["created_time"] and isinstance(parsed_row["created_time"], str): + try: + parsed_row["created_time"] = datetime.fromisoformat( + parsed_row["created_time"].replace("Z", "+00:00") + ) + except ValueError: + # 如果格式不匹配，尝试其他格式 + try: + parsed_row["created_time"] = datetime.strptime( + parsed_row["created_time"], "%Y-%m-%d %H:%M:%S" + ) + except ValueError: + # 如果无法解析，保持原样 + pass + + # 处理modified_time + if parsed_row["modified_time"] and isinstance(parsed_row["modified_time"], str): + try: + parsed_row["modified_time"] = datetime.fromisoformat( + parsed_row["modified_time"].replace("Z", "+00:00") + ) + except ValueError: + try: + parsed_row["modified_time"] = datetime.strptime( + parsed_row["modified_time"], "%Y-%m-%d %H:%M:%S" + ) + except ValueError: + pass + + return parsed_row + + def read_file(self, path, offset=0, size=None): + """读取文件内容""" + file_info = self.get_file_info(path) + if not file_info: + raise FileNotFoundError(f"文件不存在: {path}") + + if file_info["is_directory"]: + raise IsADirectoryError(f"这是一个目录: {path}") + + content = file_info["content"] + + # 解压缩 + if file_info["compressed"]: + content = zlib.decompress(content) + + if size is None: + size = len(content) - offset + + return content[offset : offset + size] + + def mv_file(self, src, dest): + """移动文件""" + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + cursor = self.conn.cursor() + cursor.execute( + """ + UPDATE files + SET path = ?, name = ?, parent_path = ? + WHERE path = ? + """, + ( + dest.lstrip("/"), + os.path.basename(dest), + os.path.dirname(dest).lstrip("/"), + src.lstrip("/"), + ), + ) + self.conn.commit() + + def mk_dir(self, path): + """创建目录""" + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO files + (path, name, parent_path, is_directory, created_time, modified_time, permissions) + VALUES (?, ?, ?, 1, ?, ?, ?) + """, + ( + path.lstrip("/"), + os.path.basename(path), + os.path.dirname(path).lstrip("/"), + datetime.now(), + datetime.now(), + 0o755, + ), + ) + self.conn.commit() + + def mv_dir(self, src, dest): + """移动目录""" + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + + cursor = self.conn.cursor() + + if src == "/": + # 移动根目录下所有内容到指定目录 + # 先处理目录，避免路径冲突 + cursor.execute( + """ + UPDATE files + SET path = ? || '/' || path, + parent_path = ? || '/' || parent_path + WHERE path != '' AND is_directory = 1 + """, + (dest.lstrip("/"), dest.lstrip("/")), + ) + + # 再处理文件 + cursor.execute( + """ + UPDATE files + SET path = ? || '/' || path, + parent_path = CASE + WHEN parent_path = '' THEN ? + ELSE ? || '/' || parent_path + END + WHERE is_directory = 0 + """, + (dest.lstrip("/"), dest.lstrip("/"), dest.lstrip("/")), + ) + else: + # 移动指定目录到目标位置 + # 更新目录本身 + cursor.execute( + """ + UPDATE files + SET path = ?, + name = ?, + parent_path = ? + WHERE path = ? + """, + ( + dest.lstrip("/"), + os.path.basename(dest), + os.path.dirname(dest).lstrip("/"), + src.lstrip("/"), + ), + ) + + # 更新目录下的所有内容 + cursor.execute( + """ + UPDATE files + SET path = ? || SUBSTR(path, ?), + parent_path = ? || SUBSTR(parent_path, ?) + WHERE path LIKE ? || '%' + """, + ( + dest.lstrip("/"), + len(src.lstrip("/")) + 1, + dest.lstrip("/"), + len(src.lstrip("/")) + 1, + src.lstrip("/"), + ), + ) + + self.conn.commit() + + def get_stats(self): + """获取文件系统统计信息""" + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + cursor = self.conn.cursor() + cursor.execute("SELECT * FROM filesystem_metadata WHERE id = 1") + return cursor.fetchone() + + def remove(self, path): + """删除文件或目录""" + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + if self.is_dir(path): # 检查是否为目录 + self._remove_directory(path) + else: # 检查是否为文件 + self._remove_file(path) + + def _remove_directory(self, path): + """删除目录""" + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + cursor = self.conn.cursor() + cursor.execute( + """ + DELETE FROM files + WHERE path LIKE ? || '%' + """, + (path.lstrip("/"),), + ) + self.total_files -= cursor.rowcount + self.total_size -= cursor.rowcount + self.conn.commit() + + def _remove_file(self, path): + """删除文件""" + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + cursor = self.conn.cursor() + cursor.execute("DELETE FROM files WHERE path = ?", (path.lstrip("/"),)) + self.total_files -= cursor.rowcount + self.total_size -= cursor.rowcount + self.conn.commit() + + def exists(self, path): + """检查文件是否存在""" + return self.get_file_info(path) is not None + + def is_dir(self, path): + """检查是否为目录""" + file_info = self.get_file_info(path) + return file_info and file_info["is_directory"] + + def is_file(self, path): + """检查是否为文件""" + file_info = self.get_file_info(path) + return file_info and not file_info["is_directory"] + + def close(self): + """关闭连接""" + if self.conn: + self.conn.close() + + def export_file(self, vfs_path, local_path): + """从虚拟文件系统导出文件到本地文件系统""" + try: + file_info = self.get_file_info(vfs_path) + if not file_info: + raise FileNotFoundError(f"虚拟文件系统中找不到文件: {vfs_path}") + + if file_info["is_directory"]: + raise IsADirectoryError(f"无法导出目录: {vfs_path}") + + # 确保目标目录存在 + os.makedirs(os.path.dirname(local_path), exist_ok=True) + + # 读取文件内容 + content = self.read_file(vfs_path) + + # 写入本地文件 + with open(local_path, "wb") as f: + f.write(content) + + # 恢复文件属性（时间戳和权限） + if file_info["created_time"]: + if isinstance(file_info["created_time"], str): + # 如果是字符串，转换为datetime对象 + from datetime import datetime + + created_time = datetime.fromisoformat( + file_info["created_time"].replace("Z", "+00:00") + ) + else: + created_time = file_info["created_time"] + created_timestamp = created_time.timestamp() + os.utime(local_path, (created_timestamp, created_timestamp)) + if file_info["modified_time"]: + if isinstance(file_info["modified_time"], str): + # 如果是字符串，转换为datetime对象 + from datetime import datetime + + modified_time = datetime.fromisoformat( + file_info["modified_time"].replace("Z", "+00:00") + ) + else: + modified_time = file_info["modified_time"] + modified_timestamp = modified_time.timestamp() + # 保持访问时间不变，只修改修改时间 + current_atime = os.path.getatime(local_path) + os.utime(local_path, (current_atime, modified_timestamp)) + + if file_info["permissions"]: + os.chmod(local_path, file_info["permissions"]) + + print(f"导出成功: {vfs_path} -> {local_path}") + return True + + except Exception as e: + print(f"导出文件失败: {e}") + import traceback + + traceback.print_exc() # 打印详细错误信息 + return False + + def export_directory(self, vfs_path, local_path): + """从虚拟文件系统导出整个目录到本地文件系统""" + try: + # 处理根目录的特殊情况 + if vfs_path == "/": + # 检查根目录是否存在 + if not self._has_root_directory(): + # 如果根目录不存在，直接导出根目录下的所有内容 + return self._export_root_contents(local_path) + + if not self.is_dir(vfs_path): + # 检查是否是有效的目录路径 + if not self._is_valid_directory_path(vfs_path): + raise NotADirectoryError(f"路径不是目录: {vfs_path}") + else: + # 可能是根目录的另一种表示 + return self._export_root_contents(local_path) + + # 创建本地目录 + os.makedirs(local_path, exist_ok=True) + + # 遍历目录内容 + items = self.list_directory(vfs_path) + exported_count = 0 + + for item in items: + item_vfs_path = item["path"] + item_local_path = os.path.join(local_path, item["name"]) + + if item["is_directory"]: + # 递归导出子目录 + count = self.export_directory(item_vfs_path, item_local_path) + exported_count += count + else: + # 导出文件 + if self.export_file_safe(item_vfs_path, item_local_path): + exported_count += 1 + + print(f"目录导出完成: {vfs_path} -> {local_path} ({exported_count} 个项目)") + return exported_count + + except Exception as e: + print(f"导出目录失败: {e}") + return 0 + + def _has_root_directory(self): + """检查是否存在根目录记录""" + if self.conn is None: + return False + cursor = self.conn.cursor() + # 检查是否有路径为 "" 或 "/" 的目录 + cursor.execute( + "SELECT COUNT(*) as count FROM files WHERE path = '' OR path = '/'" + ) + result = cursor.fetchone() + return result["count"] > 0 + + def _export_root_contents(self, local_path): + """导出根目录下的所有内容""" + try: + # 创建本地目录 + os.makedirs(local_path, exist_ok=True) + + # 获取所有父路径为空的文件/目录（这些就是根目录下的内容） + if self.conn is None: + return 0 + + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT * FROM files + WHERE parent_path IS NULL OR parent_path = '' OR parent_path = '/' + ORDER BY is_directory DESC, name ASC + """ + ) + + items = cursor.fetchall() + exported_count = 0 + + for item in items: + item_vfs_path = item["path"] + item_local_path = os.path.join(local_path, item["name"]) + + if item["is_directory"]: + # 递归导出子目录 + count = self.export_directory(item_vfs_path, item_local_path) + exported_count += count + else: + # 导出文件 + if self.export_file(item_vfs_path, item_local_path): + exported_count += 1 + + print(f"根目录导出完成: -> {local_path} ({exported_count} 个项目)") + return exported_count + + except Exception as e: + print(f"导出根目录内容失败: {e}") + return 0 + + def _is_valid_directory_path(self, path): + """检查是否是有效的目录路径""" + if self.conn is None: + return False + cursor = self.conn.cursor() + # 检查是否有以此路径为父路径的文件/目录 + cursor.execute( + "SELECT COUNT(*) as count FROM files WHERE parent_path = ?", + (path.lstrip("/"),), + ) + result = cursor.fetchone() + return result["count"] > 0 + + def export_filesystem(self, export_root): + """导出整个文件系统到指定目录""" + try: + print(f"开始导出文件系统到: {export_root}") + + # 确保导出目录存在 + os.makedirs(export_root, exist_ok=True) + + # 从根目录开始导出 + total_exported = self.export_directory("/", export_root) + + print(f"文件系统导出完成! 共导出 {total_exported} 个文件/目录") + return total_exported + + except Exception as e: + print(f"导出文件系统失败: {e}") + return 0 + + def export_file_safe(self, vfs_path, local_path): + """安全导出文件 - 即使时间戳处理失败也能导出文件内容""" + try: + file_info = self.get_file_info(vfs_path) + if not file_info: + raise FileNotFoundError(f"虚拟文件系统中找不到文件: {vfs_path}") + + if file_info["is_directory"]: + raise IsADirectoryError(f"无法导出目录: {vfs_path}") + + # 确保目标目录存在 + os.makedirs(os.path.dirname(local_path), exist_ok=True) + + # 读取文件内容 + content = self.read_file(vfs_path) + + # 写入本地文件 + with open(local_path, "wb") as f: + f.write(content) + + # 尝试设置时间戳，但即使失败也不影响文件导出 + try: + if file_info["created_time"]: + if isinstance(file_info["created_time"], str): + from datetime import datetime + + created_time = datetime.fromisoformat( + file_info["created_time"].replace("Z", "+00:00") + ) + else: + created_time = file_info["created_time"] + created_timestamp = created_time.timestamp() + os.utime(local_path, (created_timestamp, created_timestamp)) + except Exception as e: + print(f"警告: 无法设置创建时间 {vfs_path}: {e}") + + try: + if file_info["modified_time"]: + if isinstance(file_info["modified_time"], str): + from datetime import datetime + + modified_time = datetime.fromisoformat( + file_info["modified_time"].replace("Z", "+00:00") + ) + else: + modified_time = file_info["modified_time"] + modified_timestamp = modified_time.timestamp() + current_atime = os.path.getatime(local_path) + os.utime(local_path, (current_atime, modified_timestamp)) + except Exception as e: + print(f"警告: 无法设置修改时间 {vfs_path}: {e}") + + try: + if file_info["permissions"]: + os.chmod(local_path, file_info["permissions"]) + except Exception as e: + print(f"警告: 无法设置权限 {vfs_path}: {e}") + + print(f"导出成功: {vfs_path} -> {local_path}") + return True + + except Exception as e: + print(f"导出文件失败 {vfs_path}: {e}") + return False + + +# 使用示例 +def demo_usage(): + # 创建文件系统实例 + vfs = SQLiteVFS("project.bin") + vfs.connect() + + # 获取统计信息 + stats = vfs.get_stats() + if not stats: + return + print(f"文件系统: {stats['name']}") + print(f"文件数量: {stats['total_files']}") + print(f"总大小: {stats['total_size'] / 1024 / 1024:.2f} MB") + + # 列出根目录 + print("\n根目录内容:") + for item in vfs.list_directory("/"): + file_type = "DIR" if item["is_directory"] else "FILE" + print(f" {file_type}: {item['name']} ({item['file_size']} bytes)") + + # 读取文件示例 + try: + content = vfs.read_file(r"Lib\site-packages\_virtualenv.py") + print(f"\n文件内容: {content.decode('utf-8')}") + except Exception as e: + print(f"读取文件失败: {e}") + + vfs.close() diff --git a/src/svfs/folder_packer.py b/src/svfs/folder_packer.py new file mode 100644 index 0000000..1383a11 --- /dev/null +++ b/src/svfs/folder_packer.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +""" +文件夹打包工具 - 将文件夹打包为SQLite虚拟文件系统 +""" + +import os +from datetime import datetime +from .core import SQLiteVFS + + +class FolderPacker: + def __init__( + self, sqlite_vfs: SQLiteVFS, exclude_patterns=None, fs_name="packed_filesystem" + ): + self.sqlite_vfs = sqlite_vfs + self.exclude_patterns = exclude_patterns or [] + self.total_files = 0 + self.total_size = 0 + self.fs_name = fs_name + + def should_exclude(self, filepath): + """检查文件是否应该被排除""" + filename = os.path.basename(filepath) + for pattern in self.exclude_patterns: + if pattern in filename: + return True + return False + + def pack_folder(self, source_folder): + """打包整个文件夹""" + print(f"开始打包文件夹: {source_folder}") + + # 记录开始时间 + start_time = datetime.now() + + self.sqlite_vfs.clean() + + # 插入文件系统元数据 + self.sqlite_vfs.add_vfs_metadata(self.fs_name) + + self.total_files = 0 + self.total_size = 0 + + # 遍历文件夹 + for root, dirs, files in os.walk(source_folder): + # 处理目录 + for dir_name in dirs: + dir_path = os.path.join(root, dir_name) + rel_path = os.path.relpath(dir_path, source_folder) + + if self.should_exclude(rel_path): + continue + + self.sqlite_vfs.add_directory(rel_path, source_folder) + + # 处理文件 + for file_name in files: + file_path = os.path.join(root, file_name) + rel_path = os.path.relpath(file_path, source_folder) + + if self.should_exclude(rel_path): + continue + + file_size = os.path.getsize(file_path) + self.sqlite_vfs.add_file(file_path, rel_path) + self.total_size += file_size + self.total_files += 1 + + # 更新统计信息 + self.sqlite_vfs.update_vfs_metadata( + self.fs_name, self.total_files, self.total_size + ) + + # 计算耗时 + end_time = datetime.now() + duration = (end_time - start_time).total_seconds() + + print(f"打包完成!") + print(f"文件数量: {self.total_files}") + print(f"总大小: {self.total_size / 1024 / 1024:.2f} MB") + print(f"耗时: {duration:.2f} 秒") + print(f"数据库文件: {self.sqlite_vfs.db_path}") + + def add_file(self, abs_path, rel_path): + """添加单个文件""" + self.sqlite_vfs.add_file(abs_path, rel_path) + self.total_size += os.path.getsize(abs_path) + self.total_files += 1 + self.sqlite_vfs.update_vfs_metadata( + self.fs_name, self.total_files, self.total_size + ) + print(f"添加文件: {rel_path}") + print(f"文件大小: {os.path.getsize(abs_path) / 1024 / 1024:.2f} MB") + print(f"数据库文件: {self.sqlite_vfs.db_path}") + self.sqlite_vfs.export_file_safe(rel_path, abs_path) + print(f"已保存到: {abs_path}") + print(f"已保存到: {self.sqlite_vfs.db_path}") + + def add_directory(self, abs_path, rel_path): + """添加目录""" + self.sqlite_vfs.add_directory(rel_path, abs_path) + print(f"添加目录: {rel_path}") + print(f"数据库文件: {self.sqlite_vfs.db_path}") diff --git a/src/svfs/folder_unpacker.py b/src/svfs/folder_unpacker.py new file mode 100644 index 0000000..295b1da --- /dev/null +++ b/src/svfs/folder_unpacker.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +""" +文件夹解包工具 - 从SQLite虚拟文件系统导出文件 +""" + +from .core import SQLiteVFS + + +class FolderUnpacker: + def __init__(self, sqlite_vfs: SQLiteVFS): + self.sqlite_vfs = sqlite_vfs + + def check_database(self): + """检查数据库结构和内容""" + try: + if self.sqlite_vfs.conn is None: + return False, "数据库未连接" + + cursor = self.sqlite_vfs.conn.cursor() + + # 检查表是否存在 + cursor.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='files'" + ) + if not cursor.fetchone(): + return False, "files表不存在" + + # 检查是否有数据 + cursor.execute("SELECT COUNT(*) as count FROM files") + result = cursor.fetchone() + file_count = result["count"] + + # 检查根目录内容 + cursor.execute( + """ + SELECT COUNT(*) as count FROM files + WHERE parent_path IS NULL OR parent_path = '' OR parent_path = '/' + """ + ) + root_count = cursor.fetchone()["count"] + + return ( + True, + f"数据库正常，共有 {file_count} 个文件/目录，根目录下有 {root_count} 个项目", + ) + + except Exception as e: + return False, f"数据库检查失败: {e}" + + def unpack_filesystem(self, export_folder): + """解包整个文件系统""" + # 先检查数据库 + status, message = self.check_database() + print(f"数据库状态: {message}") + + if not status: + print("数据库存在问题，无法导出") + return 0 + + return self.sqlite_vfs.export_filesystem(export_folder) + + def unpack_file(self, vfs_path, local_path): + """解包单个文件""" + return self.sqlite_vfs.export_file(vfs_path, local_path) + + def unpack_directory(self, vfs_path, local_path): + """解包目录""" + return self.sqlite_vfs.export_directory(vfs_path, local_path) + + def list_root(self): + """列出根目录内容""" + try: + if self.sqlite_vfs.conn is None: + return [] + + cursor = self.sqlite_vfs.conn.cursor() + cursor.execute( + """ + SELECT * FROM files + WHERE parent_path IS NULL OR parent_path = '' OR parent_path = '/' + ORDER BY is_directory DESC, name ASC + """ + ) + return cursor.fetchall() + except Exception as e: + print(f"列出根目录失败: {e}") + return [] diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..36bf3d0 --- /dev/null +++ b/uv.lock @@ -0,0 +1,8 @@ +version = 1 +revision = 3 +requires-python = ">=3.7" + +[[package]] +name = "svfs" +version = "0.1.0" +source = { editable = "." }