import os import ast import traceback from lib.Projectrans.Utils import compute_sha1, get_file_dependencies, find_reachable_files_from_entries from lib.Projectrans.DeclarationGenerator import DeclarationGenerator from StubGen import PythonToStubConverter class Phase1Generator: """阶段一：从源文件生成声明接口""" def __init__(self, src_root: str, temp_dir: str, include_dirs: list = None, entry_files: list = None, target_triple: str = None, target_datalayout: str = None): self.src_root = os.path.abspath(src_root) self.temp_dir = os.path.abspath(temp_dir) self.include_dirs = include_dirs or [] self.sha1_map: dict[str, str] = {} self.include_py_map: dict[str, str] = {} self.entry_files = entry_files self.target_triple = target_triple self.target_datalayout = target_datalayout os.makedirs(self.temp_dir, exist_ok=True) def _get_needed_include_files(self, reachable_source_files: set) -> list: """从可达源文件收集所有被引用（含传递依赖）的 include 文件""" include_file_map = {} for includes_dir in self.include_dirs: if not os.path.isdir(includes_dir): continue for root, dirs, files in os.walk(includes_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if fname.endswith('.py') or fname.endswith('.pyi'): src_path = os.path.join(root, fname) rel_from_inc = os.path.relpath(src_path, includes_dir) ModulePath = rel_from_inc.replace(os.sep, '.').replace('/', '.') module_name = os.path.splitext(ModulePath)[0] info = (src_path, rel_from_inc, includes_dir) include_file_map[module_name] = info top_pkg = module_name.split('.')[0] if top_pkg not in include_file_map: include_file_map[top_pkg] = info imported_from_src = set() for src_path in reachable_source_files: deps = get_file_dependencies(src_path, self.src_root) imported_from_src.update(deps) needed = set() queue = list(imported_from_src) while queue: mod_name = queue.pop(0) if mod_name in needed: continue if mod_name in include_file_map: needed.add(mod_name) src_path = include_file_map[mod_name][0] includes_dir = include_file_map[mod_name][2] deps = get_file_dependencies(src_path, includes_dir) for dep in deps: if dep in include_file_map and dep not in needed: queue.append(dep) needed_infos = [] for mod_name in sorted(needed): if mod_name in include_file_map: info = include_file_map[mod_name] if info not in needed_infos: needed_infos.append(info) return needed_infos def run(self): """扫描源目录，只处理入口文件可达的 .py 文件（导入遍历）""" if self.entry_files: py_files = list(self.entry_files) else: main_py = os.path.join(self.src_root, 'main.py') if os.path.exists(main_py): py_files = [main_py] else: py_files = [] for root, dirs, files in os.walk(self.src_root): dirs[:] = [d for d in dirs if d not in ('__pycache__',)] for file in files: if file.endswith('.py'): py_files.append(os.path.join(root, file)) if not py_files: print("[阶段一] 未找到入口文件或源文件") return if self.entry_files: reachable = find_reachable_files_from_entries(self.src_root, py_files) else: reachable = find_reachable_files_from_entries(self.src_root, py_files) print(f"[阶段一] 找到 {len(reachable)} 个可达源文件（从入口遍历）") for i, src_path in enumerate(sorted(reachable), 1): rel = os.path.relpath(src_path, self.src_root) print(f"[{i}/{len(reachable)}] 生成签名: {rel}") try: self._process_file_pyi(src_path, rel) except Exception as e: print(f" [错误] {e}") traceback.print_exc() needed_includes = self._get_needed_include_files(reachable) self._process_include_py_files_pyi(needed_includes) struct_names, enum_names, struct_sha1_map, exception_names = self._build_struct_registry() print(f"\n[阶段一] 结构体注册表: {len(struct_names)} 个类型, {len(enum_names)} 个枚举") for name in sorted(struct_names): print(f" {name}") for i, src_path in enumerate(sorted(reachable), 1): rel = os.path.relpath(src_path, self.src_root) try: self._process_file_stub(src_path, rel, struct_names, enum_names=enum_names, struct_sha1_map=struct_sha1_map, exception_names=exception_names) except Exception as e: print(f" [错误] {e}") traceback.print_exc() self._process_include_py_files_stub(struct_names, needed_includes, enum_names=enum_names, struct_sha1_map=struct_sha1_map, exception_names=exception_names) print(f"\n[阶段一完成] 声明接口生成到: {self.temp_dir}") print(f"SHA1 映射表 ({len(self.sha1_map)} 个文件):") for sha1, rel in sorted(self.sha1_map.items()): print(f" {sha1} -> {rel}") def _collect_include_py_files(self): include_py_files = [] for includes_dir in self.include_dirs: if not os.path.isdir(includes_dir): continue for root, dirs, files in os.walk(includes_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if fname.endswith('.py') or fname.endswith('.pyi'): src_path = os.path.join(root, fname) rel_from_inc = os.path.relpath(src_path, includes_dir) include_py_files.append((src_path, rel_from_inc, includes_dir)) return include_py_files def _process_include_py_files_pyi(self, needed_includes: list = None): if needed_includes is not None: include_py_files = needed_includes else: include_py_files = self._collect_include_py_files() if not include_py_files: return print(f"\n[阶段一-includes] 处理 {len(include_py_files)} 个被引用的 Python 库文件") for src_path, rel_from_inc, includes_dir in include_py_files: ModulePath = rel_from_inc.replace(os.sep, '.').replace('/', '.') module_name = os.path.splitext(ModulePath)[0] try: with open(src_path, 'r', encoding='utf-8') as f: content = f.read() sha1 = compute_sha1(content) self.sha1_map[sha1] = f"includes/{rel_from_inc}" top_module = rel_from_inc.split(os.sep)[0].split('/')[0] if top_module not in self.include_py_map: self.include_py_map[top_module] = sha1 self.include_py_map[module_name] = sha1 sig_path = os.path.join(self.temp_dir, f"{sha1}.pyi") if os.path.isfile(sig_path): print(f" 缓存命中: {rel_from_inc} -> {sha1}.pyi") continue sig_content = PythonToStubConverter.convert(content, module_name) with open(sig_path, 'w', encoding='utf-8', newline='\n') as f: f.write(sig_content) print(f" 生成签名: {rel_from_inc} -> {sha1}.pyi") except Exception as e: print(f" [错误] {rel_from_inc}: {e}") def _collect_typedef_map(self): import ast as _ast typedef_map = {} if not os.path.isdir(self.temp_dir): return typedef_map for fname in os.listdir(self.temp_dir): if not fname.endswith('.pyi'): continue pyi_path = os.path.join(self.temp_dir, fname) try: with open(pyi_path, 'r', encoding='utf-8') as f: content = f.read() tree = _ast.parse(content) for node in _ast.iter_child_nodes(tree): if isinstance(node, _ast.AnnAssign) and isinstance(node.target, _ast.Name): var_name = node.target.id is_typedef = False if isinstance(node.annotation, _ast.Attribute) and hasattr(node.annotation, 'attr') and node.annotation.attr == 'CTypedef': is_typedef = True elif isinstance(node.annotation, _ast.Name) and node.annotation.id == 'CTypedef': is_typedef = True elif isinstance(node.annotation, _ast.BinOp) and isinstance(node.annotation.left, _ast.Attribute) and hasattr(node.annotation.left, 'attr') and node.annotation.left.attr == 'CTypedef': is_typedef = True if is_typedef and node.value and var_name not in typedef_map: typedef_map[var_name] = node.value except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _ConfigMode if _ConfigMode == "strict": raise _vlog().warning(f"收集 typedef 映射失败: {_e}", "Exception") return typedef_map def _process_include_py_files_stub(self, struct_names: set, needed_includes: list = None, enum_names: set = None, struct_sha1_map: dict = None, exception_names: set = None): if needed_includes is not None: include_py_files = needed_includes else: include_py_files = self._collect_include_py_files() if not include_py_files: return typedef_map = self._collect_typedef_map() for src_path, rel_from_inc, includes_dir in include_py_files: ModulePath = rel_from_inc.replace(os.sep, '.').replace('/', '.') module_name = os.path.splitext(ModulePath)[0] try: with open(src_path, 'r', encoding='utf-8') as f: content = f.read() sha1 = compute_sha1(content) stub_path = os.path.join(self.temp_dir, f"{sha1}.stub.ll") if os.path.isfile(stub_path): print(f" 缓存命中: {rel_from_inc} -> {sha1}.stub.ll") continue sig_path = os.path.join(self.temp_dir, f"{sha1}.pyi") with open(sig_path, 'r', encoding='utf-8') as f: sig_content = f.read() self._generate_decl_ll(sig_content, stub_path, src_path, struct_names, enum_names=enum_names, module_sha1=sha1, struct_sha1_map=struct_sha1_map, exception_names=exception_names, typedef_map=typedef_map) print(f" 生成声明: {rel_from_inc} -> {sha1}.stub.ll") except Exception as e: print(f" [错误] {rel_from_inc}: {e}") def _process_file_pyi(self, src_path: str, rel_path: str): with open(src_path, 'r', encoding='utf-8') as f: content = f.read() sha1 = compute_sha1(content) self.sha1_map[sha1] = rel_path sig_path = os.path.join(self.temp_dir, f"{sha1}.pyi") if os.path.isfile(sig_path): print(f" -> {sha1}.pyi (缓存)") return module_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') sig_content = PythonToStubConverter.convert(content, module_name) with open(sig_path, 'w', encoding='utf-8', newline='\n') as f: f.write(sig_content) print(f" -> {sha1}.pyi (签名)") def _process_file_stub(self, src_path: str, rel_path: str, struct_names: set, enum_names: set = None, struct_sha1_map: dict = None, exception_names: set = None): with open(src_path, 'r', encoding='utf-8') as f: content = f.read() sha1 = compute_sha1(content) stub_path = os.path.join(self.temp_dir, f"{sha1}.stub.ll") if os.path.isfile(stub_path): print(f" -> {sha1}.stub.ll (缓存)") return sig_path = os.path.join(self.temp_dir, f"{sha1}.pyi") with open(sig_path, 'r', encoding='utf-8') as f: sig_content = f.read() typedef_map = self._collect_typedef_map() self._generate_decl_ll(sig_content, stub_path, src_path, struct_names, enum_names=enum_names, module_sha1=sha1, struct_sha1_map=struct_sha1_map, exception_names=exception_names, typedef_map=typedef_map) print(f" -> {sha1}.stub.ll (声明)") def _build_struct_registry(self): struct_names = set() enum_names = set() exception_names = set() struct_sha1_map = {} valid_sha1_keys = set(self.sha1_map.keys()) for fname in os.listdir(self.temp_dir): if not fname.endswith('.pyi'): continue sha1_key = fname.replace('.pyi', '') if sha1_key not in valid_sha1_keys: continue pyi_path = os.path.join(self.temp_dir, fname) try: with open(pyi_path, 'r', encoding='utf-8') as f: content = f.read() tree = ast.parse(content) for node in ast.iter_child_nodes(tree): if isinstance(node, ast.ClassDef): is_enum = False is_exception = False if node.bases: for base in node.bases: if isinstance(base, ast.Attribute) and hasattr(base, 'attr'): if base.attr in ('CEnum', 'Enum'): is_enum = True break elif base.attr == 'Exception' or base.attr in exception_names: is_exception = True break elif isinstance(base, ast.Name) and hasattr(base, 'id'): if base.id in ('CEnum', 'Enum'): is_enum = True break elif base.id == 'Exception' or base.id in exception_names: is_exception = True break if is_enum: enum_names.add(node.name) elif is_exception: exception_names.add(node.name) else: struct_names.add(node.name) struct_sha1_map[node.name] = sha1_key except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _ConfigMode if _ConfigMode == "strict": raise _vlog().warning(f"构建结构体/枚举注册表失败: {_e}", "Exception") return struct_names, enum_names, struct_sha1_map, exception_names def _generate_decl_ll(self, pyi_content: str, ll_path: str, src_path: str, struct_names: set = None, enum_names: set = None, module_sha1: str = None, struct_sha1_map: dict = None, exception_names: set = None, typedef_map: dict = None): try: decl_gen = DeclarationGenerator(struct_names=struct_names, enum_names=enum_names, module_sha1=module_sha1, target_triple=self.target_triple, target_datalayout=self.target_datalayout, struct_sha1_map=struct_sha1_map, exception_names=exception_names, typedef_map=typedef_map) decl_ll = decl_gen.generate(pyi_content, src_path) with open(ll_path, 'w', encoding='utf-8', newline='\n') as f: f.write(decl_ll) except Exception as e: print(f" [警告] .ll 声明生成失败: {e}") with open(ll_path, 'w', encoding='utf-8') as f: f.write(f"; declaration for {os.path.basename(src_path)}\n; error: {e}\n")