mirror of
https://github.com/mofeng-git/One-KVM.git
synced 2026-01-29 00:51:53 +08:00
refactor: 完善代码质量检查和修复系统
主要改进: - 添加 make tox-local 本地代码质量检查支持 - 创建 check-code.sh 脚本支持独立工具执行 - 修复 51+ flake8 代码风格问题(未使用导入、行尾空格、注释格式等) - 解决 pylint 变量命名和日志格式问题 - 重构 make_image 方法解决 too-many-statements 警告 - 添加类型注解和修复方法签名不匹配问题 - 统一代码风格规范(引号使用、空格格式等) 工具配置: - 更新 tox.ini 支持 Python 3.10 本地环境 - 添加缺失的核心依赖包定义 - 完善 Makefile 构建系统集成
This commit is contained in:
@@ -157,7 +157,7 @@ class BaseMsd(BasePlugin):
|
||||
|
||||
async def set_connected(self, connected: bool) -> None:
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
async def make_image(self, zipped: bool) -> None:
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
@@ -77,6 +77,9 @@ class Plugin(BaseMsd):
|
||||
async def set_connected(self, connected: bool) -> None:
|
||||
raise MsdDisabledError()
|
||||
|
||||
async def make_image(self, zipped: bool) -> None:
|
||||
raise MsdDisabledError()
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def read_image(self, name: str) -> AsyncGenerator[BaseMsdReader, None]:
|
||||
if self is not None: # XXX: Vulture and pylint hack
|
||||
|
||||
@@ -25,7 +25,6 @@ import asyncio
|
||||
import contextlib
|
||||
import dataclasses
|
||||
import functools
|
||||
import time
|
||||
import os
|
||||
import copy
|
||||
import pyfatfs
|
||||
@@ -303,7 +302,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
|
||||
|
||||
self.__drive.set_rw_flag(self.__state.vd.rw)
|
||||
self.__drive.set_cdrom_flag(self.__state.vd.cdrom)
|
||||
#reset UDC to fix otg cd-rom and flash switch
|
||||
# reset UDC to fix otg cd-rom and flash switch
|
||||
try:
|
||||
udc_path = self.__drive.get_udc_path()
|
||||
with open(udc_path) as file:
|
||||
@@ -313,7 +312,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
|
||||
file.write("\n")
|
||||
with open(udc_path, "w") as file:
|
||||
file.write(sorted(os.listdir("/sys/class/udc"))[0])
|
||||
except:
|
||||
except Exception:
|
||||
logger = get_logger(0)
|
||||
logger.error("Can't reset UDC")
|
||||
if self.__state.vd.rw:
|
||||
@@ -329,69 +328,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
|
||||
|
||||
@aiotools.atomic_fg
|
||||
async def make_image(self, zipped: bool) -> None:
|
||||
#Note: img size >= 64M
|
||||
def create_fat_image(img_size: int, file_img_path: str, source_dir: str, fat_type: int = 32, label: str = 'One-KVM'):
|
||||
def add_directory_to_fat(fat: str, src_path: str, dst_path: str):
|
||||
for item in os.listdir(src_path):
|
||||
src_item_path = os.path.join(src_path, item)
|
||||
dst_item_path = os.path.join(dst_path, item)
|
||||
|
||||
if os.path.isdir(src_item_path):
|
||||
fat.makedir(dst_item_path)
|
||||
add_directory_to_fat(fat, src_item_path, dst_item_path)
|
||||
elif os.path.isfile(src_item_path):
|
||||
with open(src_item_path, 'rb') as src_file:
|
||||
fat.create(dst_item_path)
|
||||
with fat.open(dst_item_path, 'wb') as dst_file:
|
||||
dst_file.write(src_file.read())
|
||||
print(file_img_path)
|
||||
with open(file_img_path, 'wb') as f:
|
||||
f.seek(img_size * 1024 *1024 - 1)
|
||||
f.write(b'\0')
|
||||
fat_file = pyfatfs.PyFat.PyFat()
|
||||
try:
|
||||
fat_file.mkfs(file_img_path, fat_type = fat_type, label = label)
|
||||
except Exception as e:
|
||||
get_logger(0).exception(f"Error making FAT Filesystem: {e}")
|
||||
finally:
|
||||
fat_file.close()
|
||||
fat_handle = pyfatfs.PyFatFS.PyFatFS(file_img_path)
|
||||
try:
|
||||
add_directory_to_fat(fat_handle, source_dir, '/')
|
||||
except Exception as e:
|
||||
get_logger(0).exception(f"Error adding directory to FAT image: {e}")
|
||||
finally:
|
||||
fat_handle.close()
|
||||
|
||||
def extract_fat_image(file_img_path: str, output_dir: str):
|
||||
try:
|
||||
for root, dirs, files in os.walk(output_dir, topdown=False):
|
||||
for name in files:
|
||||
os.remove(os.path.join(root, name))
|
||||
for name in dirs:
|
||||
os.rmdir(os.path.join(root, name))
|
||||
except Exception as e:
|
||||
get_logger(0).exception(f"Error removing normal file or directory: {e}")
|
||||
fat_handle = pyfatfs.PyFatFS.PyFatFS(file_img_path)
|
||||
try:
|
||||
def extract_directory(fat_handle, src_path: str, dst_path: str):
|
||||
for entry in fat_handle.listdir(src_path):
|
||||
src_item_path = os.path.join(src_path, entry)
|
||||
dst_item_path = os.path.join(dst_path, entry)
|
||||
|
||||
if fat_handle.gettype(src_item_path) is pyfatfs.PyFatFS.ResourceType.directory:
|
||||
os.makedirs(dst_item_path, exist_ok=True)
|
||||
extract_directory(fat_handle, src_item_path, dst_item_path)
|
||||
else:
|
||||
with fat_handle.open(src_item_path, 'rb') as src_file:
|
||||
with open(dst_item_path, 'wb') as dst_file:
|
||||
dst_file.write(src_file.read())
|
||||
extract_directory(fat_handle, '/', output_dir)
|
||||
except Exception as e:
|
||||
get_logger(0).exception(f"Error extracting FAT image: {e}")
|
||||
finally:
|
||||
fat_handle.close()
|
||||
|
||||
# Note: img size >= 64M
|
||||
async with self.__state.busy():
|
||||
msd_path = self.__msd_path
|
||||
file_storage_path = os.path.join(msd_path, self.__normalfiles_path)
|
||||
@@ -402,10 +339,74 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
|
||||
os.makedirs(file_storage_path)
|
||||
if os.path.exists(file_img_path):
|
||||
os.remove(file_img_path)
|
||||
create_fat_image(img_size, file_img_path, file_storage_path)
|
||||
self._create_fat_image(img_size, file_img_path, file_storage_path)
|
||||
else:
|
||||
if os.path.exists(file_img_path):
|
||||
extract_fat_image(file_img_path, file_storage_path)
|
||||
self._extract_fat_image(file_img_path, file_storage_path)
|
||||
|
||||
def _create_fat_image(self, img_size: int, file_img_path: str, source_dir: str, fat_type: int = 32, label: str = "One-KVM") -> None:
|
||||
print(file_img_path)
|
||||
with open(file_img_path, "wb") as file_handle:
|
||||
file_handle.seek(img_size * 1024 * 1024 - 1)
|
||||
file_handle.write(b"\0")
|
||||
fat_file = pyfatfs.PyFat.PyFat()
|
||||
try:
|
||||
fat_file.mkfs(file_img_path, fat_type=fat_type, label=label)
|
||||
except Exception as exception:
|
||||
get_logger(0).exception("Error making FAT Filesystem: %s", exception)
|
||||
finally:
|
||||
fat_file.close()
|
||||
fat_handle = pyfatfs.PyFatFS.PyFatFS(file_img_path)
|
||||
try:
|
||||
self._add_directory_to_fat(fat_handle, source_dir, "/")
|
||||
except Exception as exception:
|
||||
get_logger(0).exception("Error adding directory to FAT image: %s", exception)
|
||||
finally:
|
||||
fat_handle.close()
|
||||
|
||||
def _add_directory_to_fat(self, fat, src_path: str, dst_path: str) -> None: # type: ignore
|
||||
for item in os.listdir(src_path):
|
||||
src_item_path = os.path.join(src_path, item)
|
||||
dst_item_path = os.path.join(dst_path, item)
|
||||
|
||||
if os.path.isdir(src_item_path):
|
||||
fat.makedir(dst_item_path) # type: ignore
|
||||
self._add_directory_to_fat(fat, src_item_path, dst_item_path)
|
||||
elif os.path.isfile(src_item_path):
|
||||
with open(src_item_path, "rb") as src_file:
|
||||
fat.create(dst_item_path) # type: ignore
|
||||
with fat.open(dst_item_path, "wb") as dst_file: # type: ignore
|
||||
dst_file.write(src_file.read())
|
||||
|
||||
def _extract_fat_image(self, file_img_path: str, output_dir: str) -> None:
|
||||
try:
|
||||
for root, dirs, files in os.walk(output_dir, topdown=False):
|
||||
for name in files:
|
||||
os.remove(os.path.join(root, name))
|
||||
for name in dirs:
|
||||
os.rmdir(os.path.join(root, name))
|
||||
except Exception as exception:
|
||||
get_logger(0).exception("Error removing normal file or directory: %s", exception)
|
||||
fat_handle = pyfatfs.PyFatFS.PyFatFS(file_img_path)
|
||||
try:
|
||||
self._extract_directory(fat_handle, "/", output_dir)
|
||||
except Exception as exception:
|
||||
get_logger(0).exception("Error extracting FAT image: %s", exception)
|
||||
finally:
|
||||
fat_handle.close()
|
||||
|
||||
def _extract_directory(self, fat_handle, src_path: str, dst_path: str) -> None: # type: ignore
|
||||
for entry in fat_handle.listdir(src_path): # type: ignore
|
||||
src_item_path = os.path.join(src_path, entry)
|
||||
dst_item_path = os.path.join(dst_path, entry)
|
||||
|
||||
if fat_handle.gettype(src_item_path) is pyfatfs.PyFatFS.ResourceType.directory: # type: ignore
|
||||
os.makedirs(dst_item_path, exist_ok=True)
|
||||
self._extract_directory(fat_handle, src_item_path, dst_item_path)
|
||||
else:
|
||||
with fat_handle.open(src_item_path, "rb") as src_file: # type: ignore
|
||||
with open(dst_item_path, "wb") as dst_file:
|
||||
dst_file.write(src_file.read())
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def read_image(self, name: str) -> AsyncGenerator[MsdFileReader, None]:
|
||||
|
||||
@@ -39,7 +39,7 @@ class MsdDriveLockedError(MsdOperationError):
|
||||
class Drive:
|
||||
def __init__(self, gadget: str, instance: int, lun: int) -> None:
|
||||
func = f"mass_storage.usb{instance}"
|
||||
self.__udc_path = os.path.join(f"/sys/kernel/config/usb_gadget", gadget, usb.G_UDC)
|
||||
self.__udc_path = os.path.join("/sys/kernel/config/usb_gadget", gadget, usb.G_UDC)
|
||||
self.__profile_func_path = usb.get_gadget_path(gadget, usb.G_PROFILE, func)
|
||||
self.__profile_path = usb.get_gadget_path(gadget, usb.G_PROFILE)
|
||||
self.__lun_path = usb.get_gadget_path(gadget, usb.G_FUNCTIONS, func, f"lun.{lun}")
|
||||
@@ -49,7 +49,7 @@ class Drive:
|
||||
|
||||
def get_watchable_paths(self) -> list[str]:
|
||||
return [self.__lun_path, self.__profile_path]
|
||||
|
||||
|
||||
def get_udc_path(self) -> str:
|
||||
return self.__udc_path
|
||||
|
||||
@@ -80,7 +80,7 @@ class Drive:
|
||||
# =====
|
||||
def __has_param(self, param: str) -> bool:
|
||||
return os.access(os.path.join(self.__lun_path, param), os.F_OK)
|
||||
|
||||
|
||||
def __get_param(self, param: str) -> str:
|
||||
with open(os.path.join(self.__lun_path, param)) as file:
|
||||
return file.read().strip()
|
||||
|
||||
Reference in New Issue
Block a user