1337 lines
49 KiB
Python
1337 lines
49 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
项目数据管理器模块
|
||
|
||
这个模块提供了完整的项目管理功能,包括:
|
||
- 项目数据的创建、读取、更新、删除 (CRUD)
|
||
- 项目文件结构的自动生成
|
||
- 项目验证和重名检查
|
||
- 项目数据的持久化存储
|
||
- 基于Qt信号的事件通知系统
|
||
|
||
主要类:
|
||
- Project: 项目数据模型类
|
||
- ProjectManager: 项目管理器类,提供所有项目管理功能
|
||
|
||
作者: MetaCore Team
|
||
版本: 1.0.0
|
||
创建时间: 2024
|
||
"""
|
||
|
||
# 标准库导入
|
||
import json # JSON数据序列化和反序列化
|
||
import os # 操作系统接口,用于文件和目录操作
|
||
import platform
|
||
import stat
|
||
import threading
|
||
import time
|
||
from datetime import datetime # 日期时间处理
|
||
from pathlib import Path # 现代路径处理
|
||
from typing import List, Dict, Optional # 类型提示
|
||
import glob
|
||
import subprocess
|
||
from PyQt5.QtWidgets import QDialog
|
||
from ui.widget import UniversalMessageDialog
|
||
|
||
# 第三方库导入
|
||
from PyQt5.QtCore import QObject, pyqtSignal, QFileSystemWatcher, QTimer # Qt核心对象和信号系统
|
||
|
||
|
||
class PathUtils:
|
||
"""跨平台路径处理工具类"""
|
||
|
||
@staticmethod
|
||
def normalize_path(path_str: str) -> Path:
|
||
"""规范化路径为Path对象"""
|
||
if not path_str:
|
||
return Path()
|
||
return Path(path_str).resolve()
|
||
|
||
@staticmethod
|
||
def ensure_path_exists(path: Path, is_file: bool = False) -> bool:
|
||
"""确保路径存在,如果是文件则确保其父目录存在"""
|
||
try:
|
||
if is_file:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
else:
|
||
path.mkdir(parents=True, exist_ok=True)
|
||
return True
|
||
except Exception as e:
|
||
print(f"创建路径失败 {path}: {e}")
|
||
return False
|
||
|
||
@staticmethod
|
||
def safe_path_join(*parts) -> Path:
|
||
"""安全地连接路径部分"""
|
||
if not parts:
|
||
return Path()
|
||
return Path(parts[0]).joinpath(*parts[1:])
|
||
|
||
@staticmethod
|
||
def get_relative_path(path: Path, base: Path) -> Path:
|
||
"""获取相对路径"""
|
||
try:
|
||
return path.relative_to(base)
|
||
except ValueError:
|
||
return path
|
||
|
||
@staticmethod
|
||
def is_valid_path(path_str: str) -> bool:
|
||
"""检查路径字符串是否有效"""
|
||
if not path_str:
|
||
return False
|
||
|
||
# 检查非法字符(跨平台)
|
||
invalid_chars = ['<', '>', ':', '"', '|', '?', '*']
|
||
if platform.system().lower() == 'windows':
|
||
# Windows额外的限制
|
||
invalid_chars.extend(['\\', '/'])
|
||
|
||
return not any(char in path_str for char in invalid_chars)
|
||
|
||
|
||
class PlatformUtils:
|
||
"""跨平台工具类"""
|
||
|
||
@staticmethod
|
||
def get_system_type() -> str:
|
||
"""获取系统类型"""
|
||
return platform.system().lower()
|
||
|
||
@staticmethod
|
||
def get_executable_extension() -> str:
|
||
"""获取可执行文件扩展名"""
|
||
return '.exe' if PlatformUtils.get_system_type() == 'windows' else ''
|
||
|
||
@staticmethod
|
||
def get_script_extension() -> str:
|
||
"""获取脚本文件扩展名"""
|
||
system = PlatformUtils.get_system_type()
|
||
if system == 'windows':
|
||
return '.bat'
|
||
else:
|
||
return '.sh'
|
||
|
||
@staticmethod
|
||
def get_python_executable(venv_path: Path) -> Path:
|
||
"""获取虚拟环境中的Python可执行文件路径"""
|
||
system = PlatformUtils.get_system_type()
|
||
if system == 'windows':
|
||
return venv_path / 'Scripts' / 'python.exe'
|
||
else:
|
||
return venv_path / 'bin' / 'python'
|
||
|
||
@staticmethod
|
||
def find_venv_path(project_path: Path) -> Optional[Path]:
|
||
"""
|
||
查找项目中的虚拟环境路径
|
||
根据平台规定虚拟环境目录名称:
|
||
- Windows: .venv
|
||
- Ubuntu/Linux: venv
|
||
- macOS: venv
|
||
"""
|
||
system = PlatformUtils.get_system_type()
|
||
|
||
# 根据平台确定虚拟环境目录名
|
||
if system == 'windows':
|
||
venv_name = '.venv'
|
||
else: # Linux, macOS等
|
||
venv_name = 'venv'
|
||
|
||
venv_path = project_path / venv_name
|
||
|
||
# 检查虚拟环境是否存在
|
||
if venv_path.exists() and venv_path.is_dir():
|
||
# 验证是否是有效的虚拟环境目录
|
||
python_exec = PlatformUtils.get_python_executable(venv_path)
|
||
if python_exec.exists():
|
||
print(f"找到虚拟环境: {venv_path}")
|
||
return venv_path
|
||
else:
|
||
print(f"虚拟环境目录存在但无效(缺少Python解释器): {venv_path}")
|
||
return None
|
||
else:
|
||
print(f"虚拟环境不存在: {venv_path} (平台: {system})")
|
||
return None
|
||
|
||
@staticmethod
|
||
def get_pycharm_paths() -> List[Path]:
|
||
"""获取各平台PyCharm可能的安装路径"""
|
||
system = PlatformUtils.get_system_type()
|
||
home = Path.home()
|
||
|
||
if system == "windows":
|
||
return [
|
||
Path("pycharm.exe"),
|
||
Path("pycharm64.exe"),
|
||
*Path("C:/Program Files/JetBrains").glob("PyCharm*/bin/pycharm64.exe"),
|
||
*Path("C:/Program Files (x86)/JetBrains").glob("PyCharm*/bin/pycharm.exe"),
|
||
*home.glob("AppData/Local/JetBrains/Toolbox/apps/PyCharm-P/ch-*/bin/pycharm64.exe"),
|
||
*home.glob("AppData/Local/JetBrains/Toolbox/apps/PyCharm-P/ch-*/bin/pycharm.exe")
|
||
]
|
||
elif system == "darwin": # macOS
|
||
return [
|
||
Path("pycharm"),
|
||
*Path("/Applications").glob("PyCharm*.app/Contents/MacOS/pycharm"),
|
||
Path("/Applications/PyCharm.app/Contents/MacOS/pycharm"),
|
||
*home.glob("Applications/PyCharm*.app/Contents/MacOS/pycharm")
|
||
]
|
||
else: # Linux
|
||
return [
|
||
Path("pycharm"),
|
||
Path("pycharm.sh"),
|
||
*Path("/opt").glob("pycharm*/bin/pycharm.sh"),
|
||
Path("/usr/local/bin/pycharm"),
|
||
Path("/usr/bin/pycharm"),
|
||
*home.glob(".local/share/JetBrains/Toolbox/apps/PyCharm-P/ch-*/bin/pycharm.sh")
|
||
]
|
||
|
||
@staticmethod
|
||
def find_executable(paths: List[Path]) -> Optional[Path]:
|
||
"""在指定路径列表中查找可执行文件"""
|
||
system = PlatformUtils.get_system_type()
|
||
|
||
# 首先尝试直接调用(如果在PATH中)
|
||
try:
|
||
cmd = "pycharm" + PlatformUtils.get_executable_extension()
|
||
subprocess.run([cmd, "--version"], check=True,
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
return Path(cmd)
|
||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||
pass
|
||
|
||
# 在指定路径中查找
|
||
for path in paths:
|
||
if path.exists():
|
||
if system == "windows" or os.access(path, os.X_OK):
|
||
return path
|
||
|
||
return None
|
||
|
||
|
||
class Project:
|
||
"""
|
||
项目数据模型类
|
||
|
||
这个类表示一个项目的所有基本信息,包括项目的元数据、路径信息和描述。
|
||
它提供了项目数据的序列化和反序列化功能,用于数据持久化。
|
||
|
||
属性:
|
||
id (int): 项目的唯一标识符
|
||
title (str): 项目名称
|
||
date (str): 项目创建日期,格式为 "YYYY-MM-DD HH:MM:SS"
|
||
type (str): 项目类型,如 "empty", "web", "mobile" 等
|
||
image (str): 项目图标路径
|
||
path (Path): 项目选择的基础路径(用户选择的父目录)
|
||
project_dir (Path): 实际项目目录的完整路径
|
||
description (str): 项目描述信息
|
||
status (str): 项目状态
|
||
"""
|
||
|
||
def __init__(self, id: int, title: str, date: str, type: str, image: str,
|
||
path: str = "", project_dir: str = "", description: str = "",
|
||
status: str = "normal"):
|
||
"""
|
||
初始化项目对象
|
||
|
||
Args:
|
||
id (int): 项目唯一标识符
|
||
title (str): 项目名称
|
||
date (str): 创建日期字符串
|
||
type (str): 项目类型
|
||
image (str): 项目图标
|
||
path (str, optional): 基础路径. Defaults to "".
|
||
project_dir (str, optional): 项目目录路径. Defaults to "".
|
||
description (str, optional): 项目描述. Defaults to "".
|
||
status (str, optional): 项目状态. Defaults to "normal".
|
||
"""
|
||
self.id = id
|
||
self.title = title
|
||
self.date = date
|
||
self.type = type
|
||
self.image = image
|
||
# 使用PathUtils处理路径
|
||
self.path = PathUtils.normalize_path(path) if path else Path()
|
||
self.project_dir = PathUtils.normalize_path(project_dir) if project_dir else Path()
|
||
self.description = description
|
||
self.status = status
|
||
|
||
self.pycharm_process = None
|
||
self.pycharm_ready = False
|
||
|
||
def to_dict(self) -> Dict:
|
||
"""
|
||
将项目对象转换为字典格式
|
||
|
||
Returns:
|
||
Dict: 包含项目所有属性的字典
|
||
"""
|
||
return {
|
||
'id': self.id,
|
||
'title': self.title,
|
||
'date': self.date,
|
||
'type': self.type,
|
||
'image': self.image,
|
||
'path': str(self.path),
|
||
'project_dir': str(self.project_dir),
|
||
'description': self.description,
|
||
'status': self.status
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: Dict) -> 'Project':
|
||
"""
|
||
从字典创建项目对象
|
||
|
||
Args:
|
||
data (Dict): 包含项目数据的字典
|
||
|
||
Returns:
|
||
Project: 新创建的项目对象
|
||
"""
|
||
return cls(
|
||
id=data.get('id', 0),
|
||
title=data.get('title', ''),
|
||
date=data.get('date', ''),
|
||
type=data.get('type', ''),
|
||
image=data.get('image', ''),
|
||
path=data.get('path', ''),
|
||
project_dir=data.get('project_dir', ''),
|
||
description=data.get('description', ''),
|
||
status=data.get('status', 'normal')
|
||
)
|
||
|
||
|
||
class ProjectManager(QObject):
|
||
"""
|
||
项目管理器类
|
||
|
||
这是整个项目管理系统的核心类,继承自QObject以支持Qt信号系统。
|
||
它负责管理所有项目的生命周期,包括创建、读取、更新、删除操作,
|
||
以及项目数据的持久化存储和验证功能。
|
||
"""
|
||
|
||
# ==================== 信号定义 ====================
|
||
projects_changed = pyqtSignal() # 项目列表发生变化时发出
|
||
project_added = pyqtSignal(Project) # 添加新项目时发出
|
||
project_removed = pyqtSignal(int) # 删除项目时发出
|
||
project_updated = pyqtSignal(Project) # 更新项目时发出
|
||
pycharm_started = pyqtSignal() # PyCharm启动完成信号
|
||
project_method_called = pyqtSignal(str) # 项目方法调用完成信号
|
||
check_pycharm_ready_signal = pyqtSignal(str) # 检查PyCharm准备信号
|
||
pycharm_check_complete = pyqtSignal(bool, str) # PyCharm检查完成信号
|
||
|
||
def __init__(self):
|
||
"""初始化项目管理器"""
|
||
super().__init__()
|
||
self.projects: List[Project] = []
|
||
self.data_file = PathUtils.safe_path_join("data", "projects.json")
|
||
|
||
# 初始化文件系统监控器
|
||
self.file_watcher = QFileSystemWatcher()
|
||
self.file_watcher.directoryChanged.connect(self.on_directory_changed)
|
||
self.file_watcher.fileChanged.connect(self.on_file_changed)
|
||
|
||
# 监控的项目路径映射 {路径字符串: 项目ID}
|
||
self.watched_paths = {}
|
||
|
||
self.load_projects()
|
||
self.setup_project_monitoring()
|
||
|
||
# 连接信号到槽
|
||
self.check_pycharm_ready_signal.connect(self._on_check_pycharm_ready)
|
||
|
||
# 添加定时器用于轮询检查
|
||
self.pycharm_check_timer = QTimer()
|
||
self.pycharm_check_timer.timeout.connect(self._check_pycharm_status)
|
||
self.target_project_path = None
|
||
|
||
# 添加缺失的属性
|
||
self.max_wait_time = 180
|
||
self.check_start_time = 0
|
||
self.current_opened_project = None
|
||
self.project_run_started = False
|
||
self.project_run_completed = False
|
||
|
||
# ==================== 数据持久化方法 ====================
|
||
|
||
def load_projects(self):
|
||
"""从JSON文件加载项目数据"""
|
||
try:
|
||
if self.data_file.exists():
|
||
with open(self.data_file, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
self.projects = [Project.from_dict(item) for item in data]
|
||
else:
|
||
self.create_default_projects()
|
||
except Exception as e:
|
||
print(f"加载项目数据失败: {e}")
|
||
self.create_default_projects()
|
||
|
||
def save_projects(self):
|
||
"""将项目数据保存到JSON文件"""
|
||
try:
|
||
# 确保数据文件的目录存在
|
||
PathUtils.ensure_path_exists(self.data_file, is_file=True)
|
||
|
||
with open(self.data_file, 'w', encoding='utf-8') as f:
|
||
data = [project.to_dict() for project in self.projects]
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
print(f"保存项目数据失败: {e}")
|
||
|
||
def create_default_projects(self):
|
||
"""创建默认项目数据"""
|
||
default_projects = [
|
||
Project(1, "示例项目1", "2024-06-08 15:56:35", "empty", "📁",
|
||
"", "", "这是一个示例空白项目"),
|
||
Project(2, "示例项目2", "2023-01-10 12:09:04", "empty", "📁",
|
||
"", "", "这是另一个示例空白项目"),
|
||
Project(3, "我的第一个项目", "2024-06-07 06:57:46", "empty", "📁",
|
||
"", "", "从这里开始您的第一个项目"),
|
||
]
|
||
self.projects = default_projects
|
||
|
||
# ==================== 项目查询方法 ====================
|
||
|
||
def get_all_projects(self) -> List[Project]:
|
||
"""获取所有项目列表"""
|
||
return self.projects.copy()
|
||
|
||
def get_project_by_id(self, project_id: int) -> Optional[Project]:
|
||
"""根据ID获取项目"""
|
||
for project in self.projects:
|
||
if project.id == project_id:
|
||
return project
|
||
return None
|
||
|
||
def get_project_by_name(self, name: str) -> Optional[Project]:
|
||
"""根据项目名称获取特定项目"""
|
||
for project in self.projects:
|
||
if project.title == name:
|
||
return project
|
||
return None
|
||
|
||
# ==================== 项目验证方法 ====================
|
||
|
||
def is_project_name_exists(self, name: str) -> bool:
|
||
"""检查项目名称是否已存在"""
|
||
return self.get_project_by_name(name) is not None
|
||
|
||
def is_project_directory_exists(self, path: Path, name: str) -> bool:
|
||
"""检查项目目录是否已存在于文件系统中"""
|
||
project_dir = path / name
|
||
return project_dir.exists()
|
||
|
||
def is_project_path_exists(self, path: Path) -> bool:
|
||
"""检查项目路径是否已存在于项目列表中"""
|
||
normalized_path = PathUtils.normalize_path(str(path))
|
||
|
||
for project in self.projects:
|
||
if project.path and PathUtils.normalize_path(str(project.path)) == normalized_path:
|
||
return True
|
||
if project.project_dir and PathUtils.normalize_path(str(project.project_dir)) == normalized_path:
|
||
return True
|
||
|
||
return False
|
||
|
||
def validate_project_creation(self, name: str, path: str) -> tuple[bool, str]:
|
||
"""验证项目创建的有效性"""
|
||
if not name or not name.strip():
|
||
return False, "请输入项目名称"
|
||
|
||
if not PathUtils.is_valid_path(name):
|
||
return False, f"项目名称包含非法字符"
|
||
|
||
path_obj = PathUtils.normalize_path(path)
|
||
if self.is_project_directory_exists(path_obj, name):
|
||
return False, f"目录已存在"
|
||
|
||
return True, ""
|
||
|
||
def validate_project_import(self, name: str, path: str) -> tuple[bool, str]:
|
||
"""验证项目导入的有效性"""
|
||
if not name or not name.strip():
|
||
return False, "项目名称不能为空"
|
||
|
||
if not PathUtils.is_valid_path(name):
|
||
return False, f"项目名称包含非法字符"
|
||
|
||
path_obj = PathUtils.normalize_path(path)
|
||
if not path_obj.exists():
|
||
return False, f"文件夹不存在"
|
||
|
||
if not path_obj.is_dir():
|
||
return False, f"选择的路径不是文件夹"
|
||
|
||
if self.is_project_path_exists(path_obj):
|
||
return False, f"该文件夹已被导入"
|
||
|
||
return True, ""
|
||
|
||
def validate_project_open(self, project_path: str) -> tuple[bool, str]:
|
||
"""验证项目是否可以打开"""
|
||
if not project_path:
|
||
return False, "项目路径不存在或无效"
|
||
|
||
path_obj = PathUtils.normalize_path(project_path)
|
||
|
||
if not path_obj.exists():
|
||
return False, "项目路径不存在或无效"
|
||
|
||
if not path_obj.is_dir():
|
||
return False, "项目路径不是有效的文件夹"
|
||
|
||
if not os.access(path_obj, os.R_OK):
|
||
return False, "没有权限访问项目文件夹"
|
||
|
||
try:
|
||
if not any(path_obj.iterdir()):
|
||
return False, "项目文件夹为空"
|
||
except PermissionError:
|
||
return False, "没有权限读取项目文件夹内容"
|
||
|
||
return True, ""
|
||
|
||
# ==================== 项目CRUD操作方法 ====================
|
||
|
||
def add_project(self, title: str, description: str, project_type: str, path: str) -> Project:
|
||
"""添加新项目"""
|
||
new_id = max([p.id for p in self.projects], default=0) + 1
|
||
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
path_obj = PathUtils.normalize_path(path)
|
||
project_dir = path_obj / title
|
||
|
||
try:
|
||
PathUtils.ensure_path_exists(project_dir)
|
||
self._create_project_structure(project_dir, project_type, title, description)
|
||
except Exception as e:
|
||
raise Exception(f"创建项目目录失败: {str(e)}")
|
||
|
||
# 创建项目图标路径
|
||
filename = f'{title}.png'
|
||
full_path = project_dir / filename
|
||
|
||
new_project = Project(
|
||
id=new_id,
|
||
title=title,
|
||
date=current_time,
|
||
type=project_type,
|
||
image=str(full_path),
|
||
path=str(path_obj),
|
||
project_dir=str(project_dir),
|
||
description=description
|
||
)
|
||
|
||
self.projects.insert(0, new_project)
|
||
self.add_project_to_watcher(new_project)
|
||
self.project_added.emit(new_project)
|
||
self.projects_changed.emit()
|
||
self.save_projects()
|
||
|
||
return new_project
|
||
|
||
def import_project(self, title: str, description: str, project_type: str, project_dir: str) -> Project:
|
||
"""导入现有项目文件夹"""
|
||
project_dir_path = PathUtils.normalize_path(project_dir)
|
||
|
||
if not project_dir_path.exists():
|
||
raise Exception(f"项目目录不存在: {project_dir}")
|
||
|
||
if not project_dir_path.is_dir():
|
||
raise Exception(f"路径不是文件夹: {project_dir}")
|
||
|
||
new_id = max([p.id for p in self.projects], default=0) + 1
|
||
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
parent_path = project_dir_path.parent
|
||
filename = f'{title}.png'
|
||
full_path = project_dir_path / filename
|
||
|
||
new_project = Project(
|
||
id=new_id,
|
||
title=title,
|
||
date=current_time,
|
||
type=project_type,
|
||
image=str(full_path),
|
||
path=str(parent_path),
|
||
project_dir=str(project_dir_path),
|
||
description=description
|
||
)
|
||
|
||
self.projects.insert(0, new_project)
|
||
self.add_project_to_watcher(new_project)
|
||
self.project_added.emit(new_project)
|
||
self.projects_changed.emit()
|
||
self.save_projects()
|
||
|
||
return new_project
|
||
|
||
def _create_project_structure(self, project_dir: Path, project_type: str, title: str, description: str):
|
||
"""创建项目文件结构"""
|
||
try:
|
||
# 创建基本目录结构
|
||
directories = ['models', 'textures', 'scenes']
|
||
|
||
for directory in directories:
|
||
dir_path = project_dir / directory
|
||
PathUtils.ensure_path_exists(dir_path)
|
||
|
||
# 创建项目配置文件
|
||
project_config = {
|
||
"name": title,
|
||
"path": str(project_dir),
|
||
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"last_modified": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"version": "1.0.0",
|
||
"engine_version": "1.0.0"
|
||
}
|
||
|
||
config_file = project_dir / 'project.json'
|
||
with open(config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(project_config, f, ensure_ascii=False, indent=2)
|
||
|
||
self._create_type_specific_files(project_dir, project_type)
|
||
|
||
except Exception as e:
|
||
print(f"创建项目文件结构失败: {e}")
|
||
|
||
def _create_type_specific_files(self, project_dir: Path, project_type: str):
|
||
"""根据项目类型创建特定文件"""
|
||
try:
|
||
if project_type == 'empty':
|
||
# 创建配置目录和文件
|
||
config_dir = project_dir / 'config'
|
||
PathUtils.ensure_path_exists(config_dir)
|
||
|
||
config_content = {
|
||
'project_settings': {
|
||
'version': '1.0.0',
|
||
'build_target': 'development',
|
||
'dependencies': []
|
||
},
|
||
'custom_settings': {}
|
||
}
|
||
|
||
config_file = config_dir / 'settings.json'
|
||
with open(config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(config_content, f, ensure_ascii=False, indent=2)
|
||
|
||
# 创建源代码目录和示例文件
|
||
src_dir = project_dir / 'src'
|
||
PathUtils.ensure_path_exists(src_dir)
|
||
|
||
src_file = src_dir / 'main.py'
|
||
with open(src_file, 'w', encoding='utf-8') as f:
|
||
f.write('''#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
主程序入口文件
|
||
"""
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("Hello, MetaCore!")
|
||
print("这是一个空白项目模板,您可以在此基础上开始开发。")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
''')
|
||
|
||
except Exception as e:
|
||
print(f"创建项目类型特定文件失败: {e}")
|
||
|
||
def remove_project(self, project_id: int) -> bool:
|
||
"""删除项目"""
|
||
for i, project in enumerate(self.projects):
|
||
if project.id == project_id:
|
||
self.remove_project_from_watcher(project)
|
||
del self.projects[i]
|
||
self.project_removed.emit(project_id)
|
||
self.projects_changed.emit()
|
||
self.save_projects()
|
||
return True
|
||
return False
|
||
|
||
def update_project(self, project: Project):
|
||
"""更新项目"""
|
||
for i, p in enumerate(self.projects):
|
||
if p.id == project.id:
|
||
self.projects[i] = project
|
||
self.project_updated.emit(project)
|
||
self.projects_changed.emit()
|
||
self.save_projects()
|
||
break
|
||
|
||
def cleanup_old_preview_images(self, project_id: int):
|
||
"""清理指定项目的旧预览图"""
|
||
try:
|
||
preview_dir = Path.cwd() / 'MetaCore' / 'Resources' / 'ProjectPreviews'
|
||
if not preview_dir.exists():
|
||
return
|
||
|
||
# 查找该项目的所有预览图
|
||
pattern = f"*preview_{project_id}_*.png"
|
||
old_previews = list(preview_dir.glob(pattern))
|
||
|
||
# 按修改时间排序,保留最新的一个,删除其他的
|
||
if len(old_previews) > 1:
|
||
old_previews.sort(key=lambda x: x.stat().st_mtime, reverse=True)
|
||
# 保留最新的,删除其他的
|
||
for old_preview in old_previews[1:]:
|
||
try:
|
||
old_preview.unlink()
|
||
print(f"已删除旧预览图: {old_preview}")
|
||
except Exception as e:
|
||
print(f"删除旧预览图失败 {old_preview}: {e}")
|
||
|
||
except Exception as e:
|
||
print(f"清理旧预览图时发生错误: {e}")
|
||
|
||
def get_projects_by_type(self, project_type: str) -> List[Project]:
|
||
"""根据类型获取项目"""
|
||
if project_type in ["overview", "all"]:
|
||
return self.get_all_projects()
|
||
elif project_type in ["management", "resource_category", "resource_management", "system_settings"]:
|
||
return []
|
||
else:
|
||
return [p for p in self.projects if p.type == project_type]
|
||
|
||
def search_projects(self, keyword: str) -> List[Project]:
|
||
"""搜索项目"""
|
||
if not keyword:
|
||
return self.get_all_projects()
|
||
|
||
keyword = keyword.lower()
|
||
return [p for p in self.projects if keyword in p.title.lower()]
|
||
|
||
def get_project_types(self) -> List[str]:
|
||
"""获取所有项目类型"""
|
||
types = set(p.type for p in self.projects)
|
||
return sorted(list(types))
|
||
|
||
# ==================== 项目监控方法 ====================
|
||
|
||
def setup_project_monitoring(self):
|
||
"""设置项目监控"""
|
||
for project in self.projects:
|
||
self.add_project_to_watcher(project)
|
||
|
||
# 添加定时检查机制
|
||
self.check_timer = QTimer()
|
||
self.check_timer.timeout.connect(self.check_all_projects_existence)
|
||
self.check_timer.start(3000) # 每3秒检查一次
|
||
|
||
def check_all_projects_existence(self):
|
||
"""检查所有项目是否仍然存在"""
|
||
deleted_projects = []
|
||
restored_projects = []
|
||
|
||
for project in self.projects:
|
||
if project.project_dir:
|
||
project_path = PathUtils.normalize_path(str(project.project_dir))
|
||
if project.status == 'normal' and not project_path.exists():
|
||
deleted_projects.append(project)
|
||
elif project.status == 'pending_delete' and project_path.exists():
|
||
restored_projects.append(project)
|
||
|
||
if deleted_projects:
|
||
self.handle_multiple_projects_deleted(deleted_projects)
|
||
|
||
if restored_projects:
|
||
self.handle_multiple_projects_restored(restored_projects)
|
||
|
||
def handle_multiple_projects_deleted(self, deleted_projects):
|
||
"""处理多个项目目录被删除的情况"""
|
||
from PyQt5.QtWidgets import QMessageBox
|
||
|
||
if len(deleted_projects) == 1:
|
||
self.handle_project_directory_deleted(deleted_projects[0])
|
||
else:
|
||
project_names = [p.title for p in deleted_projects]
|
||
project_list = "\n".join([f"• {name}" for name in project_names])
|
||
|
||
# reply = QMessageBox.question(
|
||
# None,
|
||
# "多个项目目录已删除",
|
||
# f"检测到以下 {len(deleted_projects)} 个项目的目录已被删除:\n\n"
|
||
# f"{project_list}\n\n"
|
||
# f"是否要从项目列表中移除这些项目?",
|
||
# QMessageBox.Yes | QMessageBox.No,
|
||
# QMessageBox.Yes
|
||
# )
|
||
reply = UniversalMessageDialog.show_info("多个项目目录已删除", f"检测到以下 {len(deleted_projects)} 个项目的目录已被删除:\n\n"
|
||
f"{project_list}\n\n"
|
||
f"是否要从项目列表中移除这些项目?",
|
||
True, "确定", "取消")
|
||
|
||
if reply == QDialog.Accepted:
|
||
for project in deleted_projects:
|
||
self.remove_project_from_watcher(project)
|
||
self.remove_project(project.id)
|
||
print(f"已自动移除 {len(deleted_projects)} 个被删除的项目")
|
||
else:
|
||
for project in deleted_projects:
|
||
self.remove_project_from_watcher(project)
|
||
project.status = 'pending_delete'
|
||
self.project_updated.emit(project)
|
||
self.projects_changed.emit()
|
||
self.save_projects()
|
||
print(f"已将 {len(deleted_projects)} 个项目标记为待删除状态")
|
||
|
||
def handle_multiple_projects_restored(self, restored_projects):
|
||
"""处理多个项目目录恢复的情况"""
|
||
for project in restored_projects:
|
||
self.auto_restore_project(project)
|
||
|
||
def auto_restore_project(self, project: Project):
|
||
"""自动恢复项目到正常状态"""
|
||
if project.status == 'pending_delete' and project.project_dir:
|
||
project_path = PathUtils.normalize_path(str(project.project_dir))
|
||
if project_path.exists():
|
||
project.status = 'normal'
|
||
self.add_project_to_watcher(project)
|
||
self.project_updated.emit(project)
|
||
self.projects_changed.emit()
|
||
self.save_projects()
|
||
print(f"项目已自动恢复: {project.title}")
|
||
return True
|
||
return False
|
||
|
||
def add_project_to_watcher(self, project: Project):
|
||
"""将项目添加到文件监控器"""
|
||
if not project.project_dir:
|
||
return
|
||
|
||
project_path = PathUtils.normalize_path(str(project.project_dir))
|
||
if not project_path.exists():
|
||
return
|
||
|
||
path_str = str(project_path)
|
||
if path_str not in self.watched_paths:
|
||
self.file_watcher.addPath(path_str)
|
||
self.watched_paths[path_str] = project.id
|
||
print(f"开始监控项目目录: {path_str}")
|
||
|
||
def remove_project_from_watcher(self, project: Project):
|
||
"""从文件监控器中移除项目"""
|
||
if not project.project_dir:
|
||
return
|
||
|
||
path_str = str(PathUtils.normalize_path(str(project.project_dir)))
|
||
if path_str in self.watched_paths:
|
||
self.file_watcher.removePath(path_str)
|
||
del self.watched_paths[path_str]
|
||
print(f"停止监控项目目录: {path_str}")
|
||
|
||
def on_directory_changed(self, path: str):
|
||
"""目录变化处理"""
|
||
print(f"检测到目录变化: {path}")
|
||
|
||
if path in self.watched_paths:
|
||
path_id = self.watched_paths[path]
|
||
if isinstance(path_id, int):
|
||
self.check_project_directory(path_id, path)
|
||
|
||
def on_file_changed(self, path: str):
|
||
"""文件变化处理"""
|
||
print(f"检测到文件变化: {path}")
|
||
|
||
def check_project_directory(self, project_id: int, directory_path: str):
|
||
"""检查项目目录状态"""
|
||
project = self.get_project_by_id(project_id)
|
||
if not project:
|
||
return
|
||
|
||
project_path = PathUtils.normalize_path(str(project.project_dir))
|
||
if project.status == 'normal' and not project_path.exists():
|
||
print(f"项目目录已被删除: {project_path}")
|
||
self.handle_project_directory_deleted(project)
|
||
|
||
def handle_project_directory_deleted(self, project: Project):
|
||
"""处理项目目录被删除的情况"""
|
||
from PyQt5.QtWidgets import QMessageBox
|
||
|
||
if project.status != 'normal':
|
||
return
|
||
|
||
# reply = QMessageBox.question(
|
||
# None,
|
||
# "项目目录已删除",
|
||
# f"检测到项目 \"{project.title}\" 的目录已被删除:\n{project.project_dir}\n\n"
|
||
# f"是否要从项目列表中移除此项目?",
|
||
# QMessageBox.Yes | QMessageBox.No,
|
||
# QMessageBox.Yes
|
||
# )
|
||
reply = UniversalMessageDialog.show_info("项目目录已删除", f"检测到项目 \"{project.title}\" 的目录已被删除:\n{project.project_dir}\n\n"
|
||
f"是否要从项目列表中移除此项目?",
|
||
True, "确定", "取消")
|
||
|
||
if reply == QDialog.Accepted:
|
||
self.remove_project_from_watcher(project)
|
||
self.remove_project(project.id)
|
||
print(f"已自动移除被删除的项目: {project.title}")
|
||
else:
|
||
self.remove_project_from_watcher(project)
|
||
project.status = 'pending_delete'
|
||
self.project_updated.emit(project)
|
||
self.projects_changed.emit()
|
||
self.save_projects()
|
||
print(f"项目标记为待删除状态: {project.title}")
|
||
|
||
def restore_project(self, project_id: int):
|
||
"""恢复项目到正常状态(手动恢复)"""
|
||
project = self.get_project_by_id(project_id)
|
||
if project and project.status == 'pending_delete':
|
||
project_path = PathUtils.normalize_path(str(project.project_dir))
|
||
if project_path.exists():
|
||
return self.auto_restore_project(project)
|
||
else:
|
||
print(f"项目目录仍不存在,无法恢复: {project.title}")
|
||
return False
|
||
return False
|
||
|
||
def confirm_delete_project(self, project_id: int):
|
||
"""确认删除项目(用于待删除状态的项目)"""
|
||
project = self.get_project_by_id(project_id)
|
||
if project and project.status == 'pending_delete':
|
||
self.remove_project(project_id)
|
||
print(f"已确认删除项目: {project.title}")
|
||
return True
|
||
return False
|
||
|
||
# ==================== PyCharm集成方法 ====================
|
||
|
||
def run_python_in_venv(self, venv_path: str, script_path: str, args=None, cwd=None, timeout=30):
|
||
"""在指定虚拟环境中运行Python脚本"""
|
||
try:
|
||
venv_path_obj = PathUtils.normalize_path(venv_path)
|
||
script_path_obj = PathUtils.normalize_path(script_path)
|
||
|
||
# 获取虚拟环境中的Python解释器
|
||
python_executable = PlatformUtils.get_python_executable(venv_path_obj)
|
||
|
||
# 构建命令
|
||
cmd = [str(python_executable), str(script_path_obj)]
|
||
if args:
|
||
# 将所有参数转换为字符串,处理 Path 对象
|
||
str_args = [str(arg) for arg in args]
|
||
print("参数:", str_args)
|
||
cmd.extend(str_args)
|
||
|
||
# 设置环境变量
|
||
env = os.environ.copy()
|
||
env['VIRTUAL_ENV'] = str(venv_path_obj)
|
||
|
||
# 强制子进程使用 UTF-8 编码,避免 emoji 和特殊字符编码错误
|
||
env['PYTHONIOENCODING'] = 'utf-8'
|
||
|
||
system = PlatformUtils.get_system_type()
|
||
if system == 'windows':
|
||
env['PATH'] = str(venv_path_obj / 'Scripts') + os.pathsep + env['PATH']
|
||
else:
|
||
env['PATH'] = str(venv_path_obj / 'bin') + os.pathsep + env['PATH']
|
||
|
||
# 根据平台确定编码
|
||
# 注意:子进程会使用 UTF-8(通过 PYTHONIOENCODING),这里用于读取子进程输出
|
||
encoding = 'utf-8' # 统一使用 UTF-8
|
||
|
||
# 执行命令
|
||
process = subprocess.Popen(
|
||
cmd,
|
||
cwd=cwd,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
encoding=encoding,
|
||
errors='replace', # 遇到无法解码的字符时用替换字符代替,避免崩溃
|
||
env=env
|
||
)
|
||
|
||
return process, "", ""
|
||
|
||
except Exception as e:
|
||
return None, "", f"在虚拟环境中运行Python脚本时出错: {str(e)}"
|
||
|
||
def run_project_command(self, project_path: str, target_project_path: str):
|
||
"""通过PyCharm命令行接口运行项目"""
|
||
try:
|
||
project_path_obj = PathUtils.normalize_path(project_path)
|
||
print(f"Finding entry point for project: {project_path_obj}")
|
||
|
||
entry_point = self.find_project_entry_point(project_path_obj)
|
||
if not entry_point:
|
||
print(f"No entry point found for project: {project_path_obj}")
|
||
return False
|
||
|
||
# 使用跨平台的虚拟环境查找方法
|
||
venv_path = PlatformUtils.find_venv_path(project_path_obj)
|
||
if not venv_path:
|
||
print(f"错误: 未找到虚拟环境")
|
||
return False
|
||
|
||
print(f"虚拟环境路径: {venv_path}")
|
||
print(f"脚本路径: {entry_point}")
|
||
|
||
if not entry_point.exists():
|
||
print(f"错误: 脚本文件不存在: {entry_point}")
|
||
return False
|
||
|
||
print(f'target_project_path:{target_project_path}')
|
||
process, stdout, stderr = self.run_python_in_venv(str(venv_path), str(entry_point), [target_project_path])
|
||
|
||
try:
|
||
stdout, stderr = process.communicate(timeout=2)
|
||
if process.returncode == 0:
|
||
print("运行成功,输出:", stdout)
|
||
return True
|
||
else:
|
||
print("运行失败,错误:", stderr)
|
||
return False
|
||
except subprocess.TimeoutExpired:
|
||
print("进程仍在运行(认为启动成功)")
|
||
return True
|
||
|
||
except subprocess.TimeoutExpired:
|
||
print("PyCharm run command timeout - project may be running")
|
||
return True
|
||
except Exception as e:
|
||
print(f"PyCharm run command failed: {e}")
|
||
|
||
return False
|
||
|
||
def launch_pycharm_and_open_project(self, project_path: str, target_project_path: str):
|
||
"""启动PyCharm,等待加载完成后调用项目方法打开指定项目"""
|
||
try:
|
||
if not self.start_pycharm(project_path):
|
||
return False
|
||
|
||
self.wait_for_pycharm_ready_and_run_project(project_path, target_project_path)
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"启动PyCharm并打开项目失败: {e}")
|
||
return False
|
||
|
||
def start_pycharm(self, project_path: str):
|
||
"""启动PyCharm并打开指定项目"""
|
||
try:
|
||
project_path_obj = PathUtils.normalize_path(project_path)
|
||
|
||
if not project_path_obj.exists():
|
||
print(f"Project path does not exist: {project_path_obj}")
|
||
return False
|
||
|
||
self.current_opened_project = str(project_path_obj)
|
||
|
||
# 获取PyCharm路径
|
||
pycharm_paths = PlatformUtils.get_pycharm_paths()
|
||
pycharm_cmd = PlatformUtils.find_executable(pycharm_paths)
|
||
|
||
if not pycharm_cmd:
|
||
print("PyCharm not found. Please install PyCharm or add it to PATH.")
|
||
return False
|
||
|
||
# 构建启动命令
|
||
cmd = [str(pycharm_cmd), str(project_path_obj)]
|
||
|
||
# 启动PyCharm进程
|
||
system = PlatformUtils.get_system_type()
|
||
if system == "windows":
|
||
self.pycharm_process = subprocess.Popen(
|
||
cmd,
|
||
shell=True,
|
||
creationflags=subprocess.CREATE_NEW_CONSOLE
|
||
)
|
||
else:
|
||
self.pycharm_process = subprocess.Popen(
|
||
cmd,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL
|
||
)
|
||
|
||
print(f"Starting PyCharm with project: {project_path_obj}")
|
||
self.pycharm_ready = False
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"Failed to start PyCharm: {e}")
|
||
return False
|
||
|
||
def wait_for_pycharm_ready_and_run_project(self, project_path: str, target_project_path: str):
|
||
"""等待PyCharm启动完成并运行项目"""
|
||
print(f"Setting up PyCharm ready check and project run for: {project_path}")
|
||
self.project_path = project_path
|
||
self.target_project_path = target_project_path
|
||
self.check_start_time = time.time()
|
||
self.project_run_started = False
|
||
self.project_run_completed = False
|
||
|
||
self.pycharm_check_timer.start(3000) # 每3秒检查一次
|
||
print("PyCharm check timer started")
|
||
|
||
def _check_pycharm_status(self):
|
||
"""定时检查PyCharm状态的槽函数"""
|
||
try:
|
||
elapsed_time = time.time() - self.check_start_time
|
||
print(f"Checking PyCharm status... ({elapsed_time:.1f}s elapsed)")
|
||
|
||
if elapsed_time > self.max_wait_time:
|
||
print("Timeout waiting for PyCharm to be ready")
|
||
self.pycharm_check_timer.stop()
|
||
return
|
||
|
||
if self.is_pycharm_ready():
|
||
print("PyCharm is ready")
|
||
self.pycharm_ready = True
|
||
self.pycharm_started.emit()
|
||
|
||
if not self.project_run_started:
|
||
print(f"Running project main function: {self.project_path}")
|
||
if self.run_project_main_function(self.project_path):
|
||
self.project_run_started = True
|
||
self.pycharm_check_timer.stop()
|
||
print("Project main function started, waiting for completion...")
|
||
else:
|
||
print("Failed to run project main function")
|
||
self.pycharm_check_timer.stop()
|
||
return
|
||
else:
|
||
print("PyCharm not ready yet, continuing to wait...")
|
||
|
||
except Exception as e:
|
||
print(f"Error in _check_pycharm_status: {e}")
|
||
self.pycharm_check_timer.stop()
|
||
|
||
def run_project_main_function(self, project_path: str):
|
||
"""运行项目主函数"""
|
||
try:
|
||
if self.run_project_via_pycharm_command(project_path):
|
||
return True
|
||
|
||
print("All methods to run project main function failed")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"Error running project main function: {e}")
|
||
return False
|
||
|
||
def run_project_via_pycharm_command(self, project_path: str):
|
||
"""通过PyCharm命令行接口运行项目"""
|
||
try:
|
||
project_path_obj = PathUtils.normalize_path(project_path)
|
||
entry_point = self.find_project_entry_point(project_path_obj)
|
||
|
||
if not entry_point:
|
||
print(f"No entry point found for project: {project_path_obj}")
|
||
return False
|
||
|
||
# 构建运行命令
|
||
run_cmd = [
|
||
"python3.10",
|
||
str(entry_point),
|
||
self.target_project_path
|
||
]
|
||
|
||
result = subprocess.Popen(run_cmd)
|
||
print(f'result:{result.returncode}')
|
||
|
||
if result.returncode == 0:
|
||
print(f"Successfully started project run: {project_path_obj}")
|
||
return True
|
||
|
||
except subprocess.TimeoutExpired:
|
||
print("PyCharm run command timeout - project may be running")
|
||
return True
|
||
except Exception as e:
|
||
print(f"PyCharm run command failed: {e}")
|
||
|
||
return False
|
||
|
||
def find_project_entry_point(self, project_path: Path) -> Optional[Path]:
|
||
"""查找项目的入口点文件"""
|
||
entry_point_names = [
|
||
'Start_Run.py',
|
||
'main.py',
|
||
'app.py',
|
||
'run.py',
|
||
'start.py',
|
||
'application.py',
|
||
]
|
||
|
||
# 在项目根目录中查找
|
||
for entry_name in entry_point_names:
|
||
entry_path = project_path / entry_name
|
||
if entry_path.exists():
|
||
print(f"Found entry point: {entry_path}")
|
||
return entry_path
|
||
|
||
# 如果没有找到标准入口点,返回src/main.py
|
||
default_entry = project_path / 'src' / 'main.py'
|
||
print(f"Using default entry point: {default_entry}")
|
||
return default_entry if default_entry.parent.exists() else None
|
||
|
||
def find_project_point(self, project_path: str, entry_point_name: str) -> Optional[str]:
|
||
"""使用pathlib安全地查找项目的入口点文件"""
|
||
try:
|
||
root = PathUtils.normalize_path(project_path)
|
||
except Exception:
|
||
print(f"❌ 错误: 项目路径不存在或不是一个目录 -> '{project_path}'")
|
||
return None
|
||
|
||
print(f"🔍 正在搜索项目: {root}")
|
||
|
||
# 定义要搜索的目录列表
|
||
search_relative_dirs = ['src', '.']
|
||
|
||
# 遍历搜索
|
||
for rel_dir in search_relative_dirs:
|
||
search_dir = root / rel_dir
|
||
|
||
if not search_dir.is_dir():
|
||
continue
|
||
|
||
print(f" -> 正在检查目录: {search_dir}")
|
||
entry_path = search_dir / entry_point_name
|
||
|
||
if entry_path.is_file():
|
||
print(f"✅ 找到入口文件: {entry_path}")
|
||
return str(entry_path)
|
||
|
||
print(f"❌ 未能在项目中找到常见的入口文件。")
|
||
return None
|
||
|
||
def _on_check_pycharm_ready(self, target_project_path: str):
|
||
"""处理检查PyCharm准备信号的槽函数"""
|
||
pass
|
||
|
||
def is_pycharm_ready(self):
|
||
"""检查PyCharm是否准备就绪"""
|
||
try:
|
||
if self.pycharm_process and self.pycharm_process.poll() is not None:
|
||
print("PyCharm process has terminated")
|
||
return False
|
||
|
||
if self.check_pycharm_responsive():
|
||
return True
|
||
|
||
if self.is_pycharm_window_visible():
|
||
time.sleep(2)
|
||
return True
|
||
|
||
return False
|
||
except Exception as e:
|
||
print(f"Error checking PyCharm status: {e}")
|
||
return False
|
||
|
||
def check_pycharm_responsive(self):
|
||
"""检查PyCharm是否响应命令"""
|
||
try:
|
||
system = PlatformUtils.get_system_type()
|
||
encoding = 'gbk' if system == 'windows' else 'utf-8'
|
||
|
||
result = subprocess.run(
|
||
["pycharm", "--version"],
|
||
capture_output=True,
|
||
text=True,
|
||
encoding=encoding,
|
||
errors='replace',
|
||
timeout=5
|
||
)
|
||
if result.returncode == 0:
|
||
print("PyCharm is responsive")
|
||
return True
|
||
except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError):
|
||
pass
|
||
return False
|
||
|
||
def is_pycharm_window_visible(self):
|
||
"""检查PyCharm窗口是否可见(跨平台实现)"""
|
||
try:
|
||
system = PlatformUtils.get_system_type()
|
||
|
||
if system == "windows":
|
||
return self._check_pycharm_window_windows()
|
||
elif system == "darwin": # macOS
|
||
return self._check_pycharm_window_macos()
|
||
else: # Linux
|
||
return self._check_pycharm_window_linux()
|
||
|
||
except Exception as e:
|
||
print(f"Error checking PyCharm window: {e}")
|
||
return True
|
||
|
||
def _check_pycharm_window_windows(self):
|
||
"""检查Windows上的PyCharm窗口"""
|
||
try:
|
||
import win32gui
|
||
import win32process
|
||
|
||
def enum_windows_callback(hwnd, pid_list):
|
||
_, window_pid = win32process.GetWindowThreadProcessId(hwnd)
|
||
if window_pid in pid_list:
|
||
window_title = win32gui.GetWindowText(hwnd)
|
||
if "pycharm" in window_title.lower():
|
||
pid_list['found'] = True
|
||
return True
|
||
|
||
if self.pycharm_process:
|
||
pycharm_pid = self.pycharm_process.pid
|
||
pid_list = {'found': False, pycharm_pid: True}
|
||
|
||
win32gui.EnumWindows(enum_windows_callback, pid_list)
|
||
return pid_list['found']
|
||
|
||
return False
|
||
except ImportError:
|
||
return True
|
||
except Exception:
|
||
return True
|
||
|
||
def _check_pycharm_window_macos(self):
|
||
"""检查macOS上的PyCharm窗口"""
|
||
try:
|
||
result = subprocess.run(
|
||
['osascript', '-e', 'tell application "System Events" to count windows of process "PyCharm"'],
|
||
capture_output=True,
|
||
text=True,
|
||
encoding='utf-8',
|
||
errors='replace'
|
||
)
|
||
if result.returncode == 0 and result.stdout.strip().isdigit():
|
||
window_count = int(result.stdout.strip())
|
||
return window_count > 0
|
||
return True
|
||
except Exception:
|
||
return True
|
||
|
||
def _check_pycharm_window_linux(self):
|
||
"""检查Linux上的PyCharm窗口"""
|
||
try:
|
||
result = subprocess.run(
|
||
['xdotool', 'search', '--name', 'PyCharm'],
|
||
capture_output=True,
|
||
text=True,
|
||
encoding='utf-8',
|
||
errors='replace'
|
||
)
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
return True
|
||
return True
|
||
except FileNotFoundError:
|
||
return True
|
||
except Exception:
|
||
return True
|
||
|
||
def stop_pycharm(self):
|
||
"""停止PyCharm进程"""
|
||
try:
|
||
if self.pycharm_process and self.pycharm_process.poll() is None:
|
||
self.pycharm_process.terminate()
|
||
self.pycharm_process.wait(timeout=10)
|
||
print("PyCharm process terminated")
|
||
except Exception as e:
|
||
print(f"Error stopping PyCharm: {e}")
|