from __future__ import annotations import os import ast import json import shutil import traceback from lib.Projectrans.Utils import compute_sha1, get_file_dependencies, find_reachable_files_from_entries, parse_python_file from lib.Projectrans.DeclarationGenerator import DeclarationGenerator from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _ConfigMode from lib.includes import t from StubGen import PythonToStubConverter # 全局缓存目录（不被 --clean 清除），用于缓存库文件的 .pyi / .stub.ll / .doc.json _COMPILER_ROOT: str = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) _GLOBAL_CACHE_DIR: str = os.path.join(_COMPILER_ROOT, '.transpyc_cache') def _TryGlobalCache(Sha1: str, Ext: str, TempPath: str) -> bool: """检查全局缓存中是否存在指定文件，若存在则复制到 temp_dir 并返回 True。""" GlobalPath: str = os.path.join(_GLOBAL_CACHE_DIR, f"{Sha1}.{Ext}") if os.path.isfile(GlobalPath): try: os.makedirs(os.path.dirname(TempPath), exist_ok=True) shutil.copy2(GlobalPath, TempPath) return True except Exception: pass return False def _SaveToGlobalCache(Sha1: str, Ext: str, TempPath: str) -> None: """将文件保存到全局缓存供后续使用。""" GlobalPath: str = os.path.join(_GLOBAL_CACHE_DIR, f"{Sha1}.{Ext}") try: os.makedirs(_GLOBAL_CACHE_DIR, exist_ok=True) shutil.copy2(TempPath, GlobalPath) except Exception: pass class Phase1Generator: """阶段一：从源文件生成声明接口""" def __init__(self, src_root: str, temp_dir: str, include_dirs: list[str] | None = None, entry_files: list[str] | None = None, target_triple: str | None = None, target_datalayout: str | None = None) -> None: self.src_root: str = os.path.abspath(src_root) self.temp_dir: str = os.path.abspath(temp_dir) self.include_dirs: list[str] = include_dirs or [] self.sha1_map: dict[str, str] = {} self.include_py_map: dict[str, str] = {} self.entry_files: list[str] | None = entry_files self.target_triple: str | None = target_triple self.target_datalayout: str | None = target_datalayout os.makedirs(self.temp_dir, exist_ok=True) # 增量编译 manifest：{abs_path: {sha1, mtime, size}} # 未变更文件（mtime+size 匹配）跳过 open+read+sha1，仅需 stat self._manifest: dict[str, dict] = {} self._typedef_map_cache: dict[str, ast.AST] | None = None self._manifest_path: str = os.path.join(self.temp_dir, '_phase1_manifest.json') # includes 依赖图：{module_name: set[dep_module_name]}，用于 .stub.ll 缓存依赖校验 self._include_dep_graph: dict[str, set[str]] = {} # include 文件映射：{module_name: (src_path, rel_from_inc, includes_dir)} self._include_file_map: dict[str, tuple[str, str, str]] = {} self._load_manifest() def _load_manifest(self) -> None: """加载上次运行的 manifest（mtime+size → sha1 缓存）""" try: if os.path.isfile(self._manifest_path): with open(self._manifest_path, 'r', encoding='utf-8') as f: self._manifest = json.load(f) except Exception: self._manifest = {} def _save_manifest(self) -> None: """保存 manifest 供下次增量编译使用""" try: with open(self._manifest_path, 'w', encoding='utf-8') as f: json.dump(self._manifest, f) except Exception: pass def _get_sha1(self, src_path: str) -> str: """用 mtime+size 缓存加速 sha1 计算。 未变更文件（mtime+size 匹配 manifest）直接返回缓存的 sha1， 跳过 open+read+compute_sha1，仅需一次 stat 调用。 """ abs_path: str = os.path.abspath(src_path) try: st: os.stat_result = os.stat(abs_path) except OSError: return '' mtime: float = st.st_mtime size: int = st.st_size entry: dict = self._manifest.get(abs_path, {}) if entry.get('mtime') == mtime and entry.get('size') == size: sha1: str = entry.get('sha1', '') if sha1: return sha1 # mtime/size 不匹配，需读取内容重新计算 try: with open(abs_path, 'r', encoding='utf-8') as f: content: str = f.read() except Exception: return '' sha1 = compute_sha1(content) self._manifest[abs_path] = {'sha1': sha1, 'mtime': mtime, 'size': size} return sha1 def _get_needed_include_files(self, reachable_source_files: set[str]) -> list[tuple[str, str, str]]: """从可达源文件收集所有被引用（含传递依赖）的 include 文件""" include_file_map: dict[str, tuple[str, str, str]] = {} 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: str = os.path.join(root, fname) rel_from_inc: str = os.path.relpath(src_path, includes_dir) ModulePath: str = rel_from_inc.replace(os.sep, '.').replace('/', '.') module_name: str = os.path.splitext(ModulePath)[0] info: tuple[str, str, str] = (src_path, rel_from_inc, includes_dir) include_file_map[module_name] = info top_pkg: str = module_name.split('.')[0] # __init__.py 是包入口，应优先作为 top_pkg 的代表 # （参考 find_reachable_files_from_entries 的优先级逻辑）， # 否则若 os/path.py 先被遍历，'os' 会错误指向 path.py # 而非 os/__init__.py，导致包入口 SHA1 未注册到 sha1_map if module_name == f"{top_pkg}.__init__": include_file_map[top_pkg] = info elif top_pkg not in include_file_map: include_file_map[top_pkg] = info imported_from_src: set[str] = set() for src_path in reachable_source_files: deps: set[str] = get_file_dependencies(src_path, self.src_root) imported_from_src.update(deps) # 检测 dict/list 容器使用，自动添加 _dict/_list/json 依赖 for src_path in reachable_source_files: try: # 复用 parse_python_file 的 AST 缓存（get_file_dependencies 已解析过同一文件） content, tree = parse_python_file(src_path) if tree is None: continue for node in ast.walk(tree): if isinstance(node, ast.Subscript) and isinstance(node.value, ast.Name): if node.value.id == 'dict': imported_from_src.add('_dict') imported_from_src.add('json') break elif node.value.id == 'list': imported_from_src.add('_list') break if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): if node.func.id == 'dict': imported_from_src.add('_dict') imported_from_src.add('json') break elif node.func.id == 'list': imported_from_src.add('_list') break # 检测类型注解 d: dict / l: list (非泛型形式) if isinstance(node, ast.AnnAssign) and isinstance(node.annotation, ast.Name): if node.annotation.id == 'dict': imported_from_src.add('_dict') imported_from_src.add('json') break elif node.annotation.id == 'list': imported_from_src.add('_list') break except Exception: pass needed: set[str] = set() dep_graph: dict[str, set[str]] = {} queue: list[str] = list(imported_from_src) while queue: mod_name: str = queue.pop(0) if mod_name in needed: continue if mod_name in include_file_map: needed.add(mod_name) src_path: str = include_file_map[mod_name][0] includes_dir: str = include_file_map[mod_name][2] deps: set[str] = get_file_dependencies(src_path, includes_dir) dep_graph[mod_name] = deps for dep in deps: if dep in include_file_map and dep not in needed: queue.append(dep) needed_infos: list[tuple[str, str, str]] = [] for mod_name in sorted(needed): if mod_name in include_file_map: info: tuple[str, str, str] = include_file_map[mod_name] if info not in needed_infos: needed_infos.append(info) # 存储依赖图和文件映射，供 .stub.ll 缓存依赖校验使用 self._include_dep_graph = dep_graph self._include_file_map = include_file_map return needed_infos def run(self) -> None: """扫描源目录，只处理入口文件可达的 .py 文件（导入遍历）""" py_files: list[str] if self.entry_files: py_files = list(self.entry_files) else: main_py: str = 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: _vlog().warning("未找到入口文件或源文件") return reachable: set[str] 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) _vlog().info(f"找到 {len(reachable)} 个可达源文件（从入口遍历）") for i, src_path in enumerate(sorted(reachable), 1): rel: str = os.path.relpath(src_path, self.src_root) _vlog().info(f"[{i}/{len(reachable)}] 生成签名: {rel}") try: self._process_file_pyi(src_path, rel) except Exception as e: _vlog().error(f"生成签名失败: {e}") traceback.print_exc() needed_includes: list[tuple[str, str, str]] = self._get_needed_include_files(reachable) self._process_include_py_files_pyi(needed_includes) struct_names: set[str] enum_names: set[str] struct_sha1_map: dict[str, str] exception_names: set[str] struct_names, enum_names, struct_sha1_map, exception_names, class_def_map = self._build_struct_registry() _vlog().info(f"结构体注册表: {len(struct_names)} 个类型, {len(enum_names)} 个枚举") for name in sorted(struct_names): _vlog().debug(f" {name}") # 预先收集 typedef 映射，避免在 _process_file_stub 中对每个源文件重复扫描所有 .pyi self._typedef_map_cache: dict[str, ast.AST] = self._collect_typedef_map() for i, src_path in enumerate(sorted(reachable), 1): rel: str = 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, class_def_map=class_def_map) except Exception as e: _vlog().error(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, class_def_map=class_def_map) # 保存 manifest 供下次增量编译使用 self._save_manifest() _vlog().success(f"声明接口生成到: {self.temp_dir}") _vlog().info(f"SHA1 映射表 ({len(self.sha1_map)} 个文件):") for sha1, rel in sorted(self.sha1_map.items()): _vlog().debug(f" {sha1} -> {rel}") def _collect_include_py_files(self) -> list[tuple[str, str, str]]: include_py_files: list[tuple[str, str, str]] = [] 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: str = os.path.join(root, fname) rel_from_inc: str = 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[tuple[str, str, str]] | None = None) -> None: include_py_files: list[tuple[str, str, str]] 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 _vlog().info(f"处理 {len(include_py_files)} 个被引用的 Python 库文件") for src_path, rel_from_inc, includes_dir in include_py_files: ModulePath: str = rel_from_inc.replace(os.sep, '.').replace('/', '.') module_name: str = os.path.splitext(ModulePath)[0] try: # 增量优化：用 mtime+size 获取 sha1，未变更文件跳过 open+read sha1: str = self._get_sha1(src_path) if not sha1: continue self.sha1_map[sha1] = f"includes/{rel_from_inc}" top_module: str = 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: str = os.path.join(self.temp_dir, f"{sha1}.pyi") if os.path.isfile(sig_path): _vlog().info(f" 缓存命中: {rel_from_inc} -> {sha1}.pyi") continue # 全局缓存检查（不被 --clean 清除） if _TryGlobalCache(sha1, 'pyi', sig_path): _vlog().info(f" 缓存命中(全局): {rel_from_inc} -> {sha1}.pyi") continue # 缓存未命中，需读取 content 生成 .pyi with open(src_path, 'r', encoding='utf-8') as f: content: str = f.read() sig_content: str = PythonToStubConverter.convert(content, module_name) with open(sig_path, 'w', encoding='utf-8', newline='\n') as f: f.write(sig_content) _SaveToGlobalCache(sha1, 'pyi', sig_path) _vlog().info(f" 生成签名: {rel_from_inc} -> {sha1}.pyi") except Exception as e: _vlog().error(f"处理 include 文件失败 {rel_from_inc}: {e}") def _collect_typedef_map(self) -> dict[str, ast.AST]: typedef_map: dict[str, ast.AST] = {} 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: str = os.path.join(self.temp_dir, fname) try: with open(pyi_path, 'r', encoding='utf-8') as f: content: str = f.read() tree: ast.Module = ast.parse(content) for node in ast.iter_child_nodes(tree): if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): var_name: str = node.target.id is_typedef: bool = 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: if _ConfigMode == "strict": raise _vlog().warning(f"收集 typedef 映射失败: {_e}", "Exception") return typedef_map def _ComputeDepSha1Signature(self, module_name: str) -> str: """计算 module_name 的所有传递依赖的 SHA1 签名。 用于 .stub.ll 缓存依赖校验：当依赖的 include SHA1 变化时， 当前 include 的 .stub.ll 缓存应失效（因为类型引用的 SHA1 前缀会变化）。 """ if not self._include_dep_graph or not self._include_file_map: return '' Result: set[str] = set() Visited: set[str] = set() Queue: list[str] = list(self._include_dep_graph.get(module_name, set())) while Queue: DepMod: str = Queue.pop(0) if DepMod in Visited: continue Visited.add(DepMod) if DepMod in self._include_file_map: DepSrc: str = self._include_file_map[DepMod][0] DepSha1: str = self._get_sha1(DepSrc) if DepSha1: Result.add(DepSha1) Queue.extend(self._include_dep_graph.get(DepMod, set())) return '|'.join(sorted(Result)) def _ReadStubDep(self, stub_path: str) -> str: """读取 .stub.ll 的依赖签名 sidecar，不存在返回空字符串。""" DepPath: str = stub_path + '.dep' try: with open(DepPath, 'r', encoding='utf-8') as f: return f.read() except Exception: return '' def _WriteStubDep(self, stub_path: str, dep_sig: str) -> None: """写入 .stub.ll 的依赖签名 sidecar 到 temp_dir 和全局缓存。""" # temp_dir DepPath: str = stub_path + '.dep' try: with open(DepPath, 'w', encoding='utf-8') as f: f.write(dep_sig) except Exception: pass # 全局缓存 Sha1: str = os.path.basename(stub_path).replace('.stub.ll', '') GlobalDepPath: str = os.path.join(_GLOBAL_CACHE_DIR, f"{Sha1}.stub.ll.dep") try: os.makedirs(_GLOBAL_CACHE_DIR, exist_ok=True) with open(GlobalDepPath, 'w', encoding='utf-8') as f: f.write(dep_sig) except Exception: pass def _process_include_py_files_stub(self, struct_names: set[str], needed_includes: list[tuple[str, str, str]] | None = None, enum_names: set[str] | None = None, struct_sha1_map: dict[str, str] | None = None, exception_names: set[str] | None = None, class_def_map: dict[str, ast.ClassDef] | None = None) -> None: include_py_files: list[tuple[str, str, str]] 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: dict[str, ast.AST] = self._collect_typedef_map() for src_path, rel_from_inc, includes_dir in include_py_files: ModulePath: str = rel_from_inc.replace(os.sep, '.').replace('/', '.') module_name: str = os.path.splitext(ModulePath)[0] try: # 增量优化：用 mtime+size 获取 sha1，未变更文件跳过 open+read sha1: str = self._get_sha1(src_path) if not sha1: continue # 计算依赖签名（用于 .stub.ll 缓存依赖校验） # 当依赖的 include SHA1 变化时，.stub.ll 中类型引用的 SHA1 前缀会变化，缓存应失效 DepSig: str = self._ComputeDepSha1Signature(module_name) stub_path: str = os.path.join(self.temp_dir, f"{sha1}.stub.ll") if os.path.isfile(stub_path): # 依赖校验：temp_dir 中的 .stub.ll 可能是上次编译残留，引用旧 SHA1 if self._ReadStubDep(stub_path) == DepSig: _vlog().info(f" 缓存命中: {rel_from_inc} -> {sha1}.stub.ll") continue else: _vlog().info(f" 缓存失效(依赖变更): {rel_from_inc} -> {sha1}.stub.ll") try: os.remove(stub_path) except Exception: pass # 全局缓存检查（不被 --clean 清除） if _TryGlobalCache(sha1, 'stub.ll', stub_path): # 从全局缓存复制 sidecar .dep 文件 GlobalDepPath: str = os.path.join(_GLOBAL_CACHE_DIR, f"{sha1}.stub.ll.dep") TempDepPath: str = stub_path + '.dep' if os.path.isfile(GlobalDepPath): try: shutil.copy2(GlobalDepPath, TempDepPath) except Exception: pass # 依赖校验：全局缓存中的 .stub.ll 可能引用旧 SHA1 if self._ReadStubDep(stub_path) == DepSig: _vlog().info(f" 缓存命中(全局): {rel_from_inc} -> {sha1}.stub.ll") continue else: _vlog().info(f" 缓存失效(依赖变更): {rel_from_inc} -> {sha1}.stub.ll") try: os.remove(stub_path) except Exception: pass sig_path: str = os.path.join(self.temp_dir, f"{sha1}.pyi") if not os.path.isfile(sig_path): _TryGlobalCache(sha1, 'pyi', sig_path) with open(sig_path, 'r', encoding='utf-8') as f: sig_content: str = 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, class_def_map=class_def_map) _SaveToGlobalCache(sha1, 'stub.ll', stub_path) # 写入依赖签名 sidecar（temp_dir 和全局缓存） self._WriteStubDep(stub_path, DepSig) _vlog().info(f" 生成声明: {rel_from_inc} -> {sha1}.stub.ll") except Exception as e: _vlog().error(f"处理 include 文件失败 {rel_from_inc}: {e}") def _extract_docstrings(self, content: str) -> dict[str, str]: """从源码 AST 提取所有函数/结构体/方法的首条裸字符串字面量 docstring。 - 顶层 FunctionDef: key = 函数名 - 顶层 ClassDef: key = 类名；同时遍历方法，key = "ClassName.method" - docstring 必须是 body[0] 为 ast.Expr + ast.Constant(str) - 空字符串不计入 """ result: dict[str, str] = {} try: tree = ast.parse(content) except SyntaxError: return result for node in tree.body: if isinstance(node, ast.FunctionDef): doc = self._get_first_docstring(node.body) if doc is not None and doc != '': result[node.name] = doc elif isinstance(node, ast.ClassDef): doc = self._get_first_docstring(node.body) if doc is not None and doc != '': result[node.name] = doc for sub in node.body: if isinstance(sub, ast.FunctionDef): m_doc = self._get_first_docstring(sub.body) if m_doc is not None and m_doc != '': result[f"{node.name}.{sub.name}"] = m_doc return result def _get_first_docstring(self, body: list[ast.stmt]) -> str | None: """返回 body 首条裸字符串字面量，否则 None""" if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str): return body[0].value.value return None def _process_file_pyi(self, src_path: str, rel_path: str) -> None: # 增量优化：先用 mtime+size 获取 sha1，未变更文件跳过 open+read sha1: str = self._get_sha1(src_path) if not sha1: return self.sha1_map[sha1] = rel_path sig_path: str = os.path.join(self.temp_dir, f"{sha1}.pyi") if os.path.isfile(sig_path): _vlog().info(f" -> {sha1}.pyi (缓存)") # 缓存命中时补全缺失的 .doc.json doc_path: str = os.path.join(self.temp_dir, f"{sha1}.doc.json") if not os.path.isfile(doc_path): if not _TryGlobalCache(sha1, 'doc.json', doc_path): # doc.json 缺失时需读 content 提取 docstring with open(src_path, 'r', encoding='utf-8') as f: content: str = f.read() docs: dict[str, str] = self._extract_docstrings(content) try: with open(doc_path, 'w', encoding='utf-8', newline='\n') as f: json.dump(docs, f, ensure_ascii=False) _SaveToGlobalCache(sha1, 'doc.json', doc_path) except Exception as e: _vlog().warning(f"补写 {sha1}.doc.json 失败: {e}") return # 全局缓存检查 if _TryGlobalCache(sha1, 'pyi', sig_path): _vlog().info(f" -> {sha1}.pyi (全局缓存)") doc_path: str = os.path.join(self.temp_dir, f"{sha1}.doc.json") if not os.path.isfile(doc_path): _TryGlobalCache(sha1, 'doc.json', doc_path) return # 缓存未命中，需读取 content 生成 .pyi with open(src_path, 'r', encoding='utf-8') as f: content = f.read() module_name: str = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') sig_content: str = PythonToStubConverter.convert(content, module_name) with open(sig_path, 'w', encoding='utf-8', newline='\n') as f: f.write(sig_content) _SaveToGlobalCache(sha1, 'pyi', sig_path) _vlog().info(f" -> {sha1}.pyi (签名)") # 提取 docstring 写入 {sha1}.doc.json，供 Phase2 跨模块 __doc__ 加载 docs: dict[str, str] = self._extract_docstrings(content) doc_path: str = os.path.join(self.temp_dir, f"{sha1}.doc.json") try: with open(doc_path, 'w', encoding='utf-8', newline='\n') as f: json.dump(docs, f, ensure_ascii=False) _SaveToGlobalCache(sha1, 'doc.json', doc_path) except Exception as e: _vlog().warning(f"写入 {sha1}.doc.json 失败: {e}") def _process_file_stub(self, src_path: str, rel_path: str, struct_names: set[str], enum_names: set[str] | None = None, struct_sha1_map: dict[str, str] | None = None, exception_names: set[str] | None = None, class_def_map: dict[str, ast.ClassDef] | None = None) -> None: # 增量优化：用 mtime+size 获取 sha1，未变更文件跳过 open+read sha1: str = self._get_sha1(src_path) if not sha1: return stub_path: str = os.path.join(self.temp_dir, f"{sha1}.stub.ll") if os.path.isfile(stub_path): _vlog().info(f" -> {sha1}.stub.ll (缓存)") return sig_path: str = os.path.join(self.temp_dir, f"{sha1}.pyi") with open(sig_path, 'r', encoding='utf-8') as f: sig_content: str = f.read() typedef_map: dict[str, ast.AST] = self._typedef_map_cache or 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, class_def_map=class_def_map) _vlog().info(f" -> {sha1}.stub.ll (声明)") def _build_struct_registry(self) -> tuple[set[str], set[str], dict[str, str], set[str], dict[str, ast.ClassDef]]: struct_names: set[str] = set() enum_names: set[str] = set() exception_names: set[str] = set() struct_sha1_map: dict[str, str] = {} class_def_map: dict[str, ast.ClassDef] = {} # 预先收集所有 CType 子类名称（来自 t 模块和别名映射）， # 用于在 struct 注册表中跳过"类型强转"标记类。 ctype_marker_names: set[str] = set() for attr_name in dir(t): t_cls: type | None = getattr(t, attr_name, None) if isinstance(t_cls, type) and issubclass(t_cls, t.CType) and t_cls is not t.CType: if not issubclass(t_cls, (t.CEnum, t.REnum, t.CUnion, t.CStruct)): ctype_marker_names.add(t_cls.__name__) ctype_marker_names.add(attr_name) valid_sha1_keys: set[str] = set(self.sha1_map.keys()) for fname in os.listdir(self.temp_dir): if not fname.endswith('.pyi'): continue sha1_key: str = fname.replace('.pyi', '') if sha1_key not in valid_sha1_keys: continue pyi_path: str = os.path.join(self.temp_dir, fname) try: with open(pyi_path, 'r', encoding='utf-8') as f: content: str = f.read() tree: ast.Module = ast.parse(content) for node in ast.iter_child_nodes(tree): if isinstance(node, ast.ClassDef): is_enum: bool = False is_renum: bool = False is_exception: bool = False is_ctype_marker: bool = False if node.bases: for base in node.bases: if isinstance(base, ast.Attribute) and hasattr(base, 'attr'): if base.attr == 'REnum': is_renum = True break elif base.attr in ('CEnum', 'Enum'): is_enum = True break elif base.attr == 'Exception' or base.attr in exception_names: is_exception = True break elif base.attr in ctype_marker_names: is_ctype_marker = True break elif isinstance(base, ast.Name) and hasattr(base, 'id'): if base.id == 'REnum': is_renum = True break elif base.id in ('CEnum', 'Enum'): is_enum = True break elif base.id == 'Exception' or base.id in exception_names: is_exception = True break elif base.id in ctype_marker_names: is_ctype_marker = True break # 也检查装饰器形式 @t.CEnum / @t.REnum if not is_enum and not is_renum and not is_exception and not is_ctype_marker and hasattr(node, 'decorator_list') and node.decorator_list: for deco in node.decorator_list: deco_attr: str | None = None if isinstance(deco, ast.Attribute) and hasattr(deco, 'attr'): deco_attr = deco.attr elif isinstance(deco, ast.Name) and hasattr(deco, 'id'): deco_attr = deco.id if deco_attr == 'REnum': is_renum = True break elif deco_attr in ('CEnum', 'Enum'): is_enum = True break # CType 及其子类是"类型强转"标记，不加入 struct_names / enum_names / exception_names， # 也不加入 class_def_map（避免下游误将其作为结构体处理）。 if is_ctype_marker: continue # REnum 是结构体类型（带 __tag + payload），需加入 struct_names 以便 # DeclarationGenerator._get_type_str 返回 %Name* 而非 i32 if is_renum: struct_names.add(node.name) struct_sha1_map[node.name] = sha1_key elif 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 # 收集所有类（含枚举/异常）的 ClassDef 节点，供 DeclarationGenerator # 跨模块字段展平使用（如 exprs.py 的 Name(AST) 需查找 base.py 的 AST） class_def_map[node.name] = node except Exception as _e: if _ConfigMode == "strict": raise _vlog().warning(f"构建结构体/枚举注册表失败: {_e}", "Exception") return struct_names, enum_names, struct_sha1_map, exception_names, class_def_map def _generate_decl_ll(self, pyi_content: str, ll_path: str, src_path: str, struct_names: set[str] | None = None, enum_names: set[str] | None = None, module_sha1: str | None = None, struct_sha1_map: dict[str, str] | None = None, exception_names: set[str] | None = None, typedef_map: dict[str, ast.AST] | None = None, class_def_map: dict[str, ast.ClassDef] | None = None) -> None: try: decl_gen: DeclarationGenerator = 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, class_def_map=class_def_map) decl_ll: str = 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: _vlog().warning(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")