import sys import os import re import shutil import subprocess import ast import traceback import json from lib.core.Handles.HandlesBase import CTypeInfo from lib.includes import t from lib.Projectrans.Utils import compute_sha1, _check_annotation_for_export, find_reachable_files_from_entries, topological_sort_files import TransPyC from StubGen import PythonToStubConverter def _parallel_translate_worker(src_path, out_path, src_root, temp_dir, output_dir, include_dirs, triple, datalayout, sha1_map, sig_files, stub_files, include_py_map, slice_level, shared_sym_pickle_path, struct_sha1_map): try: import pickle from lib.Projectrans import Phase2Translator from lib.Projectrans.Utils import compute_sha1 trans = Phase2Translator(src_root, temp_dir, output_dir, compile_cmd='llc', include_dirs=include_dirs, target_triple=triple, target_datalayout=datalayout) trans.sha1_map = sha1_map trans.sig_files = sig_files trans.stub_files = stub_files trans.include_py_map = include_py_map trans.slice_level = slice_level trans.struct_sha1_map = struct_sha1_map if shared_sym_pickle_path and os.path.exists(shared_sym_pickle_path): with open(shared_sym_pickle_path, 'rb') as f: shared_data = pickle.load(f) trans._shared_symbol_table = shared_data['symbol_table'] trans._shared_source_module_sig_files = shared_data['source_module_sig_files'] trans._shared_all_dc = shared_data['all_dc'] trans._shared_export_extern_funcs = shared_data['export_extern_funcs'] trans.inline_func_symbols = shared_data['inline_func_symbols'] trans.function_default_args = shared_data['function_default_args'] trans._shared_generic_class_templates = shared_data.get('generic_class_templates', {}) else: trans._precollect_inline_symbols() trans._build_shared_symbol_data() trans._translate_file(src_path, out_path) return ('ok', src_path, '') except Exception as e: import traceback return ('error', src_path, f'{e}\n{traceback.format_exc()}') # limit class Phase2Translator: """阶段二:使用声明接口翻译源文件""" INCLUDE_LIB_EXTENSIONS = ('.dll', '.ll', '.so', '.o', '.a', '.lib') INCLUDE_SRC_EXTENSIONS = ('.c', '.cpp', '.cc', '.cxx', '.m', '.mm') INCLUDE_DIRS = [ os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), 'includes'), r"d:\Users\TermiNexus\Desktop\TransPyC\includes", ] def __init__(self, src_root: str, temp_dir: str, output_dir: str, compile_cmd: str = 'llc', compile_flags: list = None, linker_cmd: str = None, linker_flags: list = None, linker_output: str = None, include_dirs: list = None, entry_files: list = None, target_triple: str = None, target_datalayout: str = None, startup=None): self.src_root = os.path.abspath(src_root) self.temp_dir = os.path.abspath(temp_dir) self.output_dir = os.path.abspath(output_dir) self.compile_cmd = compile_cmd self.compile_flags = compile_flags or ['-filetype=obj'] self.triple = target_triple self.datalayout = target_datalayout if not self.triple: for i, flag in enumerate(self.compile_flags): if flag.startswith('-mtriple='): self.triple = flag[len('-mtriple='):] elif flag == '-mtriple' and i + 1 < len(self.compile_flags): self.triple = self.compile_flags[i + 1] self.linker_cmd = linker_cmd self.linker_flags = linker_flags or [] self.linker_output = linker_output self.slice_level = 3 self.sha1_map: dict[str, str] = {} self.sig_files: dict[str, str] = {} self.stub_files: dict[str, str] = {} self.include_py_map: dict[str, str] = {} self.used_includes: set = set() self._last_ErrorStack = None self.function_default_args: dict = {} self.inline_func_symbols: dict = {} self.extra_link_files: list = [] self.extra_compile_files: list = [] self.extra_py_files: list = [] self._include_sha1s: set = set() self.entry_files = entry_files self.startup = startup # None | True | dict (如 {"__main": "main"}) self._shared_symbol_table = None self._shared_source_module_sig_files = {} self._shared_all_dc = {} self._shared_export_extern_funcs = set() self._stub_decls_cache = {} default_dirs = [d for d in self.INCLUDE_DIRS if os.path.isdir(d)] if include_dirs: seen = set() self.include_dirs = [] for d in list(include_dirs) + default_dirs: d = os.path.abspath(d) if d not in seen and os.path.isdir(d): self.include_dirs.append(d) seen.add(d) else: self.include_dirs = default_dirs self._LoadSha1Map() def _LoadSha1Map(self): """从 temp_dir 加载 SHA1 映射、签名文件和 stub 声明文件列表""" if not os.path.exists(self.temp_dir): print(f"[错误] 声明目录不存在: {self.temp_dir}") return for fname in os.listdir(self.temp_dir): fpath = os.path.join(self.temp_dir, fname) if fname.startswith('_'): continue if fname.endswith('.pyi'): sha1 = fname[:-4] self.sig_files[sha1] = fpath elif fname.endswith('.stub.ll'): sha1 = fname[:-8] self.stub_files[sha1] = fpath sha1_file_path = os.path.join(self.temp_dir, '_sha1_map.txt') if os.path.exists(sha1_file_path): with open(sha1_file_path, 'r', encoding='utf-8') as f: for line in f: line = line.strip() if ':' in line: sha1, rel = line.split(':', 1) self.sha1_map[sha1] = rel if rel.startswith('includes/'): module_name = os.path.splitext(os.path.basename(rel))[0] self.include_py_map[module_name] = sha1 else: for sha1, stub_path in self.stub_files.items(): self.sha1_map[sha1] = sha1 def _build_struct_sha1_map(self): 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: tree = ast.parse(f.read()) except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"解析 .pyi 签名文件失败: {_e}", "Exception") continue for node in ast.walk(tree): if isinstance(node, ast.ClassDef): struct_sha1_map[node.name] = sha1_key return struct_sha1_map def run(self): """扫描源目录,翻译并生成 .ll 含代码文件(导入遍历)""" 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 reachable = find_reachable_files_from_entries(self.src_root, py_files) py_files = list(reachable) py_files = topological_sort_files(py_files, self.src_root) print(f"[阶段二] 找到 {len(py_files)} 个可达源文件(已按依赖排序)") print("[阶段二] 处理顺序:") for i, f in enumerate(py_files[:10], 1): rel = os.path.relpath(f, self.src_root) print(f" {i}. {rel}") os.makedirs(self.output_dir, exist_ok=True) valid_sha1s = set(self.sha1_map.keys()) old_stub_count = len(self.stub_files) old_sig_count = len(self.sig_files) self.stub_files = {k: v for k, v in self.stub_files.items() if k in valid_sha1s} self.sig_files = {k: v for k, v in self.sig_files.items() if k in valid_sha1s} if old_stub_count != len(self.stub_files) or old_sig_count != len(self.sig_files): print(f"[阶段二] 过滤旧文件: stub {old_stub_count}->{len(self.stub_files)}, sig {old_sig_count}->{len(self.sig_files)}") active_sha1s = set() self._precollect_inline_symbols() self._build_shared_symbol_data() self.struct_sha1_map = self._build_struct_sha1_map() need_translate = [] for i, src_path in enumerate(py_files, 1): rel = os.path.relpath(src_path, self.src_root) with open(src_path, 'r', encoding='utf-8') as f: content = f.read() sha1 = compute_sha1(content) active_sha1s.add(sha1) out_path = os.path.join(self.output_dir, f"{sha1}.ll") current_ModuleSha1Map = self._build_current_ModuleSha1Map(sha1) try: tree = ast.parse(content) for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: self.used_includes.add(alias.name.split('.')[0]) elif isinstance(node, ast.ImportFrom): if node.module: self.used_includes.add(node.module.split('.')[0]) except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"收集源文件导入的 include 模块失败: {_e}", "Exception") if os.path.isfile(out_path): deps_changed = self._check_deps_changed(sha1, current_ModuleSha1Map) if not deps_changed: print(f"[{i}/{len(py_files)}] 跳过(缓存): {rel} -> {sha1}.ll") continue else: if self._recombine_ll(sha1, current_ModuleSha1Map): print(f"[{i}/{len(py_files)}] 重组(依赖变化): {rel} -> {sha1}.ll") continue print(f"[{i}/{len(py_files)}] 重译(依赖变化): {rel} -> {sha1}.ll") else: stub_cache = os.path.join(self.output_dir, f"{sha1}.stub.ll") text_cache = os.path.join(self.output_dir, f"{sha1}.text.ll") if os.path.isfile(stub_cache) and os.path.isfile(text_cache): if self._recombine_ll(sha1, current_ModuleSha1Map): print(f"[{i}/{len(py_files)}] 重组(缓存恢复): {rel} -> {sha1}.ll") continue need_translate.append((i, src_path, out_path, sha1, current_ModuleSha1Map)) if need_translate: print(f"\n[阶段二] 需要翻译 {len(need_translate)} 个文件") import time t_start = time.time() n_workers = min(os.cpu_count() or 4, len(need_translate), 8) if n_workers > 1 and len(need_translate) > 1: from concurrent.futures import ProcessPoolExecutor, as_completed import pickle print(f" 并行翻译 (workers={n_workers})") shared_pickle_path = os.path.join(self.temp_dir, '_shared_sym.pkl') try: shared_data = { 'symbol_table': self._shared_symbol_table, 'source_module_sig_files': self._shared_source_module_sig_files, 'all_dc': self._shared_all_dc, 'export_extern_funcs': self._shared_export_extern_funcs, 'inline_func_symbols': self.inline_func_symbols, 'function_default_args': self.function_default_args, 'generic_class_templates': self._shared_generic_class_templates, } with open(shared_pickle_path, 'wb') as f: pickle.dump(shared_data, f, protocol=pickle.HIGHEST_PROTOCOL) print(f" 共享符号表已序列化 ({os.path.getsize(shared_pickle_path)//1024}KB)") except Exception as e: print(f" [警告] 符号表序列化失败({e}),worker将自行构建") shared_pickle_path = '' errors = [] with ProcessPoolExecutor(max_workers=n_workers) as executor: futures = {} for i, src_path, out_path, sha1, msm in need_translate: rel = os.path.relpath(src_path, self.src_root) future = executor.submit( _parallel_translate_worker, src_path, out_path, self.src_root, self.temp_dir, self.output_dir, self.include_dirs, self.triple, self.datalayout, self.sha1_map, self.sig_files, self.stub_files, self.include_py_map, self.slice_level, shared_pickle_path, self.struct_sha1_map ) futures[future] = (i, rel) done_count = 0 for future in as_completed(futures): i, rel = futures[future] done_count += 1 try: status, _, err = future.result() if status == 'ok': print(f" [{done_count}/{len(need_translate)}] 完成: {rel}") else: print(f" [错误] {rel}: {err}") errors.append((rel, err)) except Exception as e: print(f" [错误] {rel}: {e}") errors.append((rel, str(e))) if errors: print(f"\n[编译终止] {len(errors)} 个文件翻译失败") sys.exit(1) else: for idx, (i, src_path, out_path, sha1, msm) in enumerate(need_translate): rel = os.path.relpath(src_path, self.src_root) t0 = time.time() try: self._translate_file(src_path, out_path, prebuilt_ModuleSha1Map=msm) t1 = time.time() print(f" [{idx+1}/{len(need_translate)}] {rel} ({t1-t0:.1f}s)") except Exception as e: print(f" [错误] {rel}: {e}") traceback.print_exc() print(f"\n[编译终止] 翻译失败") sys.exit(1) t_end = time.time() print(f" 翻译总耗时: {t_end-t_start:.1f}s") print(f"\n[阶段二完成] .ll 文件生成到: {self.output_dir}") self._scan_include_libraries() self._compile_include_py_files(active_sha1s) self._compile_ll_files(active_sha1s) self._compile_include_sources() if self.linker_cmd: self._link_obj_files(active_sha1s) def _build_current_ModuleSha1Map(self, self_sha1: str) -> dict: """构建当前模块的依赖 SHA1 映射(与 _translate_file 中逻辑一致)""" ModuleSha1Map = {} for sha1_key, sig_path in self.sig_files.items(): if sha1_key == self_sha1: continue rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): inc_mod_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') inc_mod_name = inc_mod_name[len('includes/'):] inc_short_name = inc_mod_name.split('.')[-1] if '.' in inc_mod_name else inc_mod_name actual_sha1 = sha1_key for inc_dir in self.include_dirs: if os.path.isdir(inc_dir): py_path = os.path.join(inc_dir, rel_path[len('includes/'):].replace(os.sep, '/')) py_path = os.path.splitext(py_path)[0] + '.py' if os.path.isfile(py_path): with open(py_path, 'r', encoding='utf-8') as f: py_sha1 = compute_sha1(f.read()) actual_sha1 = py_sha1 break ModuleSha1Map[inc_mod_name] = actual_sha1 ModuleSha1Map[inc_short_name] = actual_sha1 # Also add parent package name (e.g., 'hashlib' for 'hashlib.__init__') if inc_short_name == '__init__' and '.' in inc_mod_name: parent_pkg = inc_mod_name.rsplit('.', 1)[0] ModuleSha1Map[parent_pkg] = actual_sha1 continue module_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') ModuleSha1Map[module_name] = sha1_key short_name = module_name.split('.')[-1] if '.' in module_name else module_name ModuleSha1Map[short_name] = sha1_key return ModuleSha1Map @staticmethod def _split_ll(ll_content: str): """将 .ll 内容拆分为 stub(声明)和 text(代码)两部分 stub: target, 注释, %type = type, declare, @xxx = external global text: define ... { ... }, @xxx = global/constant (带初始化器) """ import re stub_lines = [] text_lines = [] in_define = False brace_depth = 0 module_sha1 = None defined_names = set() for line in ll_content.splitlines(True): stripped = line.strip() if stripped.startswith('define '): dm = re.match(r'define\s+[^@]*@"?([a-f0-9]+)\.', stripped) if dm and module_sha1 is None: module_sha1 = dm.group(1) dm2 = re.match(r'define\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm2: defined_names.add(dm2.group(1)) in_global_def = False global_var_name = None global_type_parts = [] global_brace_depth = 0 for line in ll_content.splitlines(True): stripped = line.strip() if in_global_def: text_lines.append(line) global_type_parts.append(stripped) global_brace_depth += stripped.count('{') - stripped.count('}') if global_brace_depth <= 0: full_type = ' '.join(global_type_parts).strip() depth = 0 type_end = len(full_type) for ci, cc in enumerate(full_type): if cc in ('{', '['): depth += 1 elif cc in ('}', ']'): depth -= 1 elif depth == 0 and cc == ' ': type_end = ci break var_type = full_type[:type_end].strip() if var_type: stub_lines.append(f'{global_var_name} = external global {var_type}\n') in_global_def = False global_var_name = None global_type_parts = [] global_brace_depth = 0 continue if in_define: text_lines.append(line) brace_depth += stripped.count('{') - stripped.count('}') if brace_depth <= 0: in_define = False continue if stripped.startswith('define '): text_lines.append(line) in_define = True brace_depth = stripped.count('{') - stripped.count('}') decl_line = re.sub(r'^define\s+', 'declare ', stripped) decl_line = re.sub(r'\b(linkonce_odr|weak_odr|linkonce|weak|common|appending|internal|private)\s+', '', decl_line) decl_line = re.sub(r'\balwaysinline\s+', '', decl_line) paren_pos = decl_line.rfind(')') if paren_pos >= 0: decl_line = decl_line[:paren_pos + 1] stub_lines.append(decl_line + '\n') continue if stripped == '{' and text_lines and not in_define: prev_stripped = text_lines[-1].strip() if text_lines else '' if prev_stripped.startswith('define ') or prev_stripped.endswith('noredzone') or prev_stripped.endswith(')'): text_lines.append(line) brace_depth = 1 in_define = True if prev_stripped.startswith('define '): decl_line = re.sub(r'^define\s+', 'declare ', prev_stripped) decl_line = re.sub(r'\b(linkonce_odr|weak_odr|linkonce|weak|common|appending|internal|private)\s+', '', decl_line) decl_line = re.sub(r'\balwaysinline\s+', '', decl_line) paren_pos = decl_line.rfind(')') if paren_pos >= 0: decl_line = decl_line[:paren_pos + 1] stub_lines.append(decl_line + '\n') continue if stripped.startswith('@') and ' global ' in stripped and 'external global' not in stripped: text_lines.append(line) if ' internal ' not in stripped and ' private ' not in stripped: name_match = re.match(r'(@\S+)', stripped) if name_match: var_name = name_match.group(1) global_pos = stripped.find(' global ') after_global = stripped[global_pos + 8:] depth = 0 type_end = len(after_global) for ci, cc in enumerate(after_global): if cc in ('{', '['): depth += 1 elif cc in ('}', ']'): depth -= 1 elif depth == 0 and cc == ' ': type_end = ci break var_type = after_global[:type_end].strip() brace_depth_gv = 0 for cc in var_type: if cc == '{': brace_depth_gv += 1 elif cc == '}': brace_depth_gv -= 1 if brace_depth_gv > 0: in_global_def = True global_var_name = var_name global_type_parts = [var_type] global_brace_depth = brace_depth_gv elif var_type: stub_lines.append(f'{var_name} = external global {var_type}\n') continue if stripped.startswith('@') and ' constant ' in stripped: text_lines.append(line) continue if stripped.startswith('declare'): dm = re.match(r'declare\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm: fname = dm.group(1) if fname in defined_names: stub_lines.append(line) continue dot_pos = fname.find('.') if dot_pos > 0: fname_sha1 = fname[:dot_pos] if module_sha1 and fname_sha1 != module_sha1 and len(fname_sha1) >= 16: continue if re.match(r'^[a-f0-9]{16}\.', fname): continue stub_lines.append(line) continue stub_lines.append(line) return ''.join(stub_lines), ''.join(text_lines) def _save_deps(self, sha1: str, ModuleSha1Map: dict): """保存依赖指纹到 .deps.json""" deps_path = os.path.join(self.output_dir, f"{sha1}.deps.json") deps = {} for mod_name, dep_sha1 in ModuleSha1Map.items(): deps[mod_name] = dep_sha1 with open(deps_path, 'w', encoding='utf-8') as f: json.dump(deps, f) def _LoadDeps(self, sha1: str) -> dict: """加载依赖指纹""" deps_path = os.path.join(self.output_dir, f"{sha1}.deps.json") if os.path.isfile(deps_path): try: with open(deps_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"加载依赖指纹 JSON 失败: {_e}", "Exception") return None def _check_deps_changed(self, sha1: str, current_ModuleSha1Map: dict) -> bool: """检查依赖指纹是否变化""" saved_deps = self._LoadDeps(sha1) if saved_deps is None: return True for mod_name, dep_sha1 in current_ModuleSha1Map.items(): if saved_deps.get(mod_name) != dep_sha1: return True for mod_name, dep_sha1 in saved_deps.items(): if current_ModuleSha1Map.get(mod_name) != dep_sha1: return True return False def _recombine_ll(self, sha1: str, current_ModuleSha1Map: dict) -> bool: """依赖变化时重新组合 .ll:更新 .stub.ll 中的 SHA1 前缀,拼接 .stub.ll + .text.ll""" saved_deps = self._LoadDeps(sha1) if saved_deps is None: return False stub_path = os.path.join(self.output_dir, f"{sha1}.stub.ll") text_path = os.path.join(self.output_dir, f"{sha1}.text.ll") ll_path = os.path.join(self.output_dir, f"{sha1}.ll") if not os.path.isfile(stub_path) or not os.path.isfile(text_path): return False with open(stub_path, 'r', encoding='utf-8') as f: stub_content = f.read() with open(text_path, 'r', encoding='utf-8') as f: text_content = f.read() sha1_replacements = {} for mod_name, old_sha1 in saved_deps.items(): new_sha1 = current_ModuleSha1Map.get(mod_name) if new_sha1 and old_sha1 != new_sha1: sha1_replacements[old_sha1] = new_sha1 if sha1_replacements: for old_sha1, new_sha1 in sha1_replacements.items(): stub_content = stub_content.replace(old_sha1, new_sha1) text_content = text_content.replace(old_sha1, new_sha1) combined = stub_content if not combined.endswith('\n'): combined += '\n' combined += text_content with open(ll_path, 'w', encoding='utf-8', newline='\n') as f: f.write(combined) with open(stub_path, 'w', encoding='utf-8', newline='\n') as f: f.write(stub_content) with open(text_path, 'w', encoding='utf-8', newline='\n') as f: f.write(text_content) self._save_deps(sha1, current_ModuleSha1Map) obj_path = os.path.join(self.output_dir, f"{sha1}.obj") if os.path.isfile(obj_path): os.remove(obj_path) return True def _translate_file(self, src_path: str, out_path: str, prebuilt_ModuleSha1Map: dict = None): """翻译单个源文件,嵌入相关 .stub.ll 声明""" with open(src_path, 'r', encoding='utf-8') as f: code = f.read() sha1 = compute_sha1(code) if prebuilt_ModuleSha1Map is not None: ModuleSha1Map = prebuilt_ModuleSha1Map else: ModuleSha1Map = self._build_current_ModuleSha1Map(sha1) trans = TransPyC.TransPyC(code=code, triple=self.triple, datalayout=self.datalayout) trans.translator.CurrentFile = src_path trans.SliceLevel = self.slice_level trans.translator.SliceLevel = self.slice_level trans.translator.SliceCount = 0 trans.translator.SliceInfos = [] trans.translator.LlvmGen = None trans.translator._module_sha1 = sha1 trans.translator._ModuleSha1Map = ModuleSha1Map trans.translator._struct_sha1_map = self.struct_sha1_map trans.translator._global_function_default_args = self.function_default_args trans.translator._temp_dir = self.temp_dir if hasattr(self, '_shared_symbol_table') and self._shared_symbol_table is not None: trans.translator.SymbolTable = self._shared_symbol_table # 确保共享符号表的 translator 指向当前 translator,以便 import 别名解析正确 if hasattr(self._shared_symbol_table, 'translator'): self._shared_symbol_table.translator = trans.translator trans.translator._source_module_sig_files = self._shared_source_module_sig_files trans.translator._all_define_constants = self._shared_all_dc trans.translator._export_extern_funcs = self._shared_export_extern_funcs if hasattr(self, '_shared_generic_class_templates') and self._shared_generic_class_templates: if not hasattr(trans.translator.ClassHandler, '_generic_class_templates'): trans.translator.ClassHandler._generic_class_templates = {} trans.translator.ClassHandler._generic_class_templates.update(self._shared_generic_class_templates) else: export_extern_funcs = set() for sha1_key, sig_path in self.sig_files.items(): if sha1_key == sha1: continue try: with open(sig_path, 'r', encoding='utf-8') as f: sig_content = f.read() sig_tree = ast.parse(sig_content) for node in ast.iter_child_nodes(sig_tree): if isinstance(node, ast.FunctionDef): if _check_annotation_for_export(node.returns): export_extern_funcs.add(node.name) except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"收集导出外部函数声明失败: {_e}", "Exception") if export_extern_funcs: trans.translator._export_extern_funcs = export_extern_funcs for includes_dir in self.include_dirs: if os.path.isdir(includes_dir): for pyi_file in os.listdir(includes_dir): if pyi_file.endswith('.pyi'): pyi_path = os.path.join(includes_dir, pyi_file) module_name = os.path.splitext(pyi_file)[0] trans.translator.SymbolTable.LoadModuleSymbols(pyi_path, module_name, lineno=0) for py_file in os.listdir(includes_dir): if py_file.endswith('.py') and not py_file.startswith('_'): module_name = os.path.splitext(py_file)[0] sha1_key = self.include_py_map.get(module_name) if sha1_key and sha1_key in self.sig_files: sig_path = self.sig_files[sha1_key] trans.translator.SymbolTable.LoadModuleSymbols(sig_path, module_name, lineno=0) # 收集需要重导出的包,等所有模块符号加载完后再处理 _pending_reexports = [] for sha1_key, sig_path in self.sig_files.items(): rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): mod_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') mod_name = mod_name[len('includes/'):] trans.translator.SymbolTable.LoadModuleSymbols(sig_path, mod_name, lineno=0) if mod_name.endswith('.__init__'): package_name = mod_name[:-len('.__init__')] _pending_reexports.append((sig_path, package_name)) # 所有 includes 模块符号已加载,现在处理包重导出 for sig_path, package_name in _pending_reexports: self._register_package_reexports(trans, sig_path, package_name) for sha1_key, sig_path in self.sig_files.items(): if sha1_key == sha1: continue rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): continue module_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') trans.translator.SymbolTable.LoadModuleSymbols(sig_path, module_name, lineno=0) trans.translator._source_module_sig_files[module_name] = sig_path short_name = module_name.split('.')[-1] if '.' in module_name else module_name if short_name != module_name: trans.translator._source_module_sig_files[short_name] = sig_path all_dc = {} for sha1_key, stub_path in self.stub_files.items(): if sha1_key == sha1: continue if os.path.exists(stub_path): try: with open(stub_path, 'r', encoding='utf-8') as f: stub_content = f.read() for line in stub_content.splitlines(): line = line.strip() if line.startswith('@') and '= global' in line and 'i32' in line and not line.startswith('@llvm'): var_name = line.split('=')[0].strip().lstrip('@') val_part = line.split('i32')[-1].strip().rstrip(']').lstrip('[').lstrip('i32') try: val = int(val_part) all_dc[var_name] = val except: pass except: pass for sha1_key, pyi_path in self.sig_files.items(): if sha1_key == sha1: continue if os.path.exists(pyi_path): try: import ast as _ast with open(pyi_path, 'r', encoding='utf-8') as f: pyi_content = f.read() tree = _ast.parse(pyi_content) for node in _ast.iter_child_nodes(tree): if isinstance(node, _ast.AnnAssign) and isinstance(node.target, _ast.Name): ann_str = _ast.dump(node.annotation) if 'CDefine' in ann_str and node.value: val = None if isinstance(node.value, _ast.Constant): val = node.value.value elif isinstance(node.value, _ast.Call) and isinstance(node.value.func, _ast.Name) and node.value.func.id == 'float' and node.value.args and isinstance(node.value.args[0], _ast.Constant): val = node.value.args[0].value if val is not None: all_dc[f"{node.target.id}"] = val except: pass if all_dc: trans.translator._all_define_constants = all_dc for sym_name, sym_info in self.inline_func_symbols.items(): if sym_name not in trans.translator.SymbolTable: trans.translator.SymbolTable[sym_name] = sym_info else: existing = trans.translator.SymbolTable[sym_name] if not getattr(existing, 'InlineBody', None) and getattr(sym_info, 'InlineBody', None): existing.IsInline = sym_info.IsInline existing.InlineBody = sym_info.InlineBody existing.InlineParams = sym_info.InlineParams try: result = trans.Convert( OutputFilename=out_path, SourceFilename=src_path, target='llvm' ) except Exception as e: error_stack = getattr(trans.translator, '_ErrorStack', []) if error_stack: self._last_ErrorStack = error_stack chain_lines = [] for entry in reversed(error_stack): if len(entry) >= 3: exc_msg, line_info = entry[1], entry[2] chain_lines.append(" %s\n %s" % (line_info, exc_msg)) if chain_lines: enriched = str(e) + "\n" + "\n".join(chain_lines) raise type(e)(enriched) from e node_info = "" if hasattr(trans.translator, 'LlvmGen') and trans.translator.LlvmGen: node_info = trans.translator.LlvmGen._get_node_info() if node_info: enriched = "%s\n %s" % (str(e), node_info.strip()) raise type(e)(enriched) from e raise if result and isinstance(result, str): for func_name, func_def in trans.translator.FunctionDefCache.items(): if hasattr(func_def, 'args') and hasattr(func_def.args, 'defaults') and func_def.args.defaults: mangled = trans.translator.LlvmGen._mangle_func_name(func_name) if trans.translator.LlvmGen else func_name defaults = [] for d in func_def.args.defaults: if isinstance(d, ast.Constant): defaults.append(d.value) else: defaults.append(None) self.function_default_args[mangled] = defaults self.function_default_args[func_name] = defaults for sym_name, sym_info in trans.translator.SymbolTable.items(): if isinstance(sym_info, CTypeInfo) and getattr(sym_info, 'IsInline', False) and getattr(sym_info, 'InlineBody', None): self.inline_func_symbols[sym_name] = sym_info # stub声明现在通过_dso.ll编译时拼接依赖模块stub.ll提供 # stub_decls = self._collect_stub_decls_for_module(sha1) # if stub_decls: # result = self._inject_stub_decls(result, stub_decls) out_dir = os.path.dirname(out_path) if out_dir: os.makedirs(out_dir, exist_ok=True) with open(out_path, 'w', encoding='utf-8', newline='\n') as f: f.write(result) stub_content, text_content = self._split_ll(result) base = os.path.splitext(out_path)[0] with open(base + '.stub.ll', 'w', encoding='utf-8', newline='\n') as f: f.write(stub_content) with open(base + '.text.ll', 'w', encoding='utf-8', newline='\n') as f: f.write(text_content) self._save_deps(sha1, ModuleSha1Map) print(f" -> {os.path.basename(out_path)}") else: stub_path = self.stub_files.get(sha1) if stub_path and os.path.exists(stub_path): out_dir = os.path.dirname(out_path) if out_dir: os.makedirs(out_dir, exist_ok=True) with open(stub_path, 'r', encoding='utf-8') as f: stub_content = f.read() with open(out_path, 'w', encoding='utf-8', newline='\n') as f: f.write(stub_content) print(f" -> {os.path.basename(out_path)} (仅声明)") def _register_package_reexports(self, trans, init_pyi_path, package_name): """解析 __init__.pyi 中的 from .xxx import yyy 或 from __xxx import yyy,在包级别注册导出符号""" try: with open(init_pyi_path, 'r', encoding='utf-8') as f: content = f.read() tree = ast.parse(content) for node in ast.walk(tree): if isinstance(node, ast.ImportFrom) and node.module: sub_module = node.module.lstrip('.') for alias in node.names: symbol_name = alias.name exported_name = alias.asname if alias.asname else symbol_name source_keys = [ f"{package_name}.{sub_module}.{symbol_name}", f"{package_name}.__{sub_module}.{symbol_name}" if not sub_module.startswith('__') else None, symbol_name, ] target_key = f"{package_name}.{exported_name}" if target_key not in trans.translator.SymbolTable: for source_key in source_keys: if source_key and source_key in trans.translator.SymbolTable: trans.translator.SymbolTable[target_key] = trans.translator.SymbolTable[source_key] break if exported_name not in trans.translator.SymbolTable: for source_key in source_keys: if source_key and source_key in trans.translator.SymbolTable: trans.translator.SymbolTable[exported_name] = trans.translator.SymbolTable[source_key] break except Exception as e: print(f" [警告] 解析包重导出失败 {init_pyi_path}: {e}") def _collect_stub_decls(self, self_sha1: str) -> str: """收集 includes 目录中模块的 .stub.ll 声明内容,去重""" seen_symbols = set() decl_lines = [] for sha1_key, stub_path in self.stub_files.items(): if sha1_key == self_sha1: continue rel_path = self.sha1_map.get(sha1_key, sha1_key) if not rel_path.startswith('includes/'): continue if os.path.exists(stub_path): with open(stub_path, 'r', encoding='utf-8') as f: content = f.read() for line in content.splitlines(): stripped = line.strip() if not stripped: continue if stripped.startswith(';'): continue if stripped.startswith('target '): continue if stripped.startswith('source_filename'): continue if stripped.startswith('@') and '=' in stripped: name = stripped.split('=', 1)[0].strip() if name in seen_symbols: continue seen_symbols.add(name) elif stripped.startswith('%') and '= type' in stripped: name = stripped.split('=', 1)[0].strip() if name in seen_symbols: continue seen_symbols.add(name) elif stripped.startswith('declare'): parts = stripped.split('@', 1) if len(parts) > 1: name = '@' + parts[1].split('(', 1)[0].strip() if name in seen_symbols: continue seen_symbols.add(name) decl_lines.append(line) if decl_lines: result = '\n'.join(decl_lines) self._stub_decls_cache[self_sha1] = result return result self._stub_decls_cache[self_sha1] = '' return '' def _collect_stub_decls_for_module(self, self_sha1: str) -> str: """收集特定模块需要的 .stub.ll 声明(包括结构体类型定义、函数声明和 typedef)""" if self_sha1 in self._stub_decls_cache: return self._stub_decls_cache[self_sha1] import re seen_symbols = set() seen_func_symbols = set() decl_lines = [] non_struct_types = {'UINT', 'INT', 'BYTE', 'BYTEPTR', 'WORD', 'DWORD', 'QWORD', 'INT8', 'INT16', 'INT32', 'INT64', 'UINT8', 'UINT16', 'UINT32', 'UINT64', 'INTPTR', 'UINTPTR', 'SIZE_T', 'SSIZE_T', 'PTRDIFF_T', 'CType', 'CVolatile', 'CEnum', 'CUnion', 'CStruct', 'atomic_int', 'atomic_uint', 'atomic_long', 'atomic_ulong', 'atomic_int8', 'atomic_uint8', 'atomic_int16', 'atomic_uint16', 'atomic_int32', 'atomic_uint32', 'atomic_int64', 'atomic_uint64', 'atomic_ptr', 'atomic_flag', 'typedef', 'list', 'dict', 'tuple', 'set', 'str', 'bytes', 'array'} struct_names = set() struct_source_sha1 = {} def _extract_struct_name(type_def_str): m = re.match(r'%"?((?:[a-f0-9]+\.)?[\w.]+)"?\s*=\s*type', type_def_str) if m: raw = m.group(1) if '.' in raw: return raw.split('.', 1)[1], raw.split('.', 1)[0] return raw, None return None, None def _replace_struct_refs(type_str): def replace_struct_name(name): if name in struct_names: name_sha1 = struct_source_sha1.get(name, '') return f'%"{name_sha1}.{name}"' if name_sha1 else None return None def replace_struct_match(match): name = match.group(1) replaced = replace_struct_name(name) return replaced if replaced else match.group(0) def replace_sha1_struct_match(match): name = match.group(2) replaced = replace_struct_name(name) return replaced if replaced else match.group(0) result = re.sub(r'%struct\.(\w+)', replace_struct_match, type_str) result = re.sub(r'%"?([a-f0-9]+)\.(\w+)"?', replace_sha1_struct_match, result) return result for sha1_key, stub_path in self.stub_files.items(): if os.path.exists(stub_path): with open(stub_path, 'r', encoding='utf-8') as f: content = f.read() for line in content.splitlines(): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: clean_name, sha1_part = _extract_struct_name(stripped) if clean_name and clean_name not in non_struct_types: struct_names.add(clean_name) if clean_name not in struct_source_sha1: struct_source_sha1[clean_name] = sha1_part or sha1_key for sha1_key, stub_path in self.stub_files.items(): if sha1_key == self_sha1: continue if os.path.exists(stub_path): with open(stub_path, 'r', encoding='utf-8') as f: content = f.read() for line in content.splitlines(): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: clean_name, sha1_part = _extract_struct_name(stripped) if not clean_name or clean_name in non_struct_types: continue if clean_name in seen_symbols: continue parts = stripped.split('= type', 1) struct_type_body = parts[1].strip() if len(parts) > 1 else '' if 'opaque' in struct_type_body: seen_symbols.add(clean_name) name_sha1 = struct_source_sha1.get(clean_name, sha1_key) decl_lines.append(f'%"{name_sha1}.{clean_name}" = type opaque') continue if clean_name in struct_names: struct_type_body = _replace_struct_refs(struct_type_body) name_sha1 = struct_source_sha1.get(clean_name, sha1_key) new_def = f'%"{name_sha1}.{clean_name}" = type {struct_type_body}' seen_symbols.add(clean_name) decl_lines.append(new_def) if decl_lines: return '\n'.join(decl_lines) return '' def _inject_stub_decls(self, ll_content: str, stub_decls: str) -> str: """将 stub 声明注入到 LLVM IR 模块头之后,替换 opaque 类型为具体定义""" def normalize_type_name(name): name = name.strip() if name.startswith('%'): return '%' + name[1:].strip().strip('"') return name existing_symbols = {} # normalized_name -> (line_index, full_line) for i, line in enumerate(ll_content.splitlines()): stripped = line.strip() if stripped.startswith('@') and '=' in stripped: name = stripped.split('=', 1)[0].strip() existing_symbols[name] = (i, line) elif stripped.startswith('%') and '= type' in stripped: name = normalize_type_name(stripped.split('=', 1)[0].strip()) existing_symbols[name] = (i, line) elif stripped.startswith('declare') or stripped.startswith('define'): parts = stripped.split('@', 1) if len(parts) > 1: name = '@' + parts[1].split('(', 1)[0].strip() existing_symbols[name] = (i, line) lines = ll_content.splitlines() replacements = {} # line_index -> new_line new_decls = [] # 新类型定义(不在现有输出中的) def normalize_type_name(name): name = name.strip() if name.startswith('%'): return '%' + name[1:].strip().strip('"') return name for line in stub_decls.splitlines(): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: name = normalize_type_name(stripped.split('=', 1)[0].strip()) if name in existing_symbols: idx, existing_line = existing_symbols[name] existing_type_body = existing_line.strip().split('= type', 1)[1].strip() if '= type' in existing_line else '' stub_type_body = stripped.split('= type', 1)[1].strip() if '= type' in stripped else '' if 'opaque' in existing_line and 'opaque' not in stripped: replacements[idx] = line else: new_decls.append(line) # 应用替换 for idx, new_line in replacements.items(): lines[idx] = new_line # 追加新类型定义到末尾(去重处理) if new_decls: # 过滤掉已经在 Translator 输出中存在的类型定义 existing_types = set() for line in lines: stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: type_name = normalize_type_name(stripped.split('=', 1)[0].strip()) existing_types.add(type_name) filtered_decls = [] for line in new_decls: stripped = line.strip() if '= type' in stripped: type_name = normalize_type_name(stripped.split('=', 1)[0].strip()) if type_name not in existing_types: filtered_decls.append(line) existing_types.add(type_name) if filtered_decls: lines.append('') lines.extend(filtered_decls) # 确保所有类型定义在全局变量之前,LLVM 要求在常量初始化器中使用类型之前必须先定义 type_def_indices = [] first_global_idx = None for i, line in enumerate(lines): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: if first_global_idx is not None: type_def_indices.append(i) elif stripped.startswith('@') and ' global' in stripped: if first_global_idx is None: first_global_idx = i if type_def_indices and first_global_idx is not None: type_defs = [lines[i] for i in reversed(type_def_indices)] for i in reversed(type_def_indices): del lines[i] insert_pos = first_global_idx if type_def_indices[0] < first_global_idx: insert_pos = first_global_idx - len(type_def_indices) for j, td in enumerate(type_defs): lines.insert(insert_pos, td) result = '\n'.join(lines) return result def _compile_ll_files(self, active_sha1s: set): """将 output_dir 中属于 active_sha1s 的 .ll 文件编译为 .obj(跳过已编译过的 include 文件)""" stale_files = [] for root, dirs, files in os.walk(self.output_dir): for file in files: fpath = os.path.join(root, file) for ext in ('.ll', '.obj', '.stub.ll', '.text.ll', '.deps.json', '_dso.ll'): if file.endswith(ext): sha1 = file[:-(len(ext))] if sha1 not in active_sha1s: stale_files.append(fpath) break if stale_files: for fpath in stale_files: try: os.remove(fpath) except: pass print(f"[清理] 删除 {len(stale_files)} 个过时文件") ll_files = [] for root, dirs, files in os.walk(self.output_dir): for file in files: if file.endswith('.ll') and not file.endswith('.stub.ll') and not file.endswith('.text.ll'): sha1 = file[:-3] if sha1 in active_sha1s: obj_path = os.path.join(root, file[:-3] + '.obj') if not os.path.isfile(obj_path): ll_files.append(os.path.join(root, file)) if not ll_files: print("[编译] 无 .ll 文件需要重新编译") else: print(f"\n[编译] 找到 {len(ll_files)} 个 .ll 文件需要编译") has_inline = False all_ll_files = [] for root, dirs, files in os.walk(self.output_dir): for file in files: if file.endswith('.ll') and not file.endswith('.stub.ll') and not file.endswith('.text.ll'): sha1 = file[:-3] if sha1 in active_sha1s and sha1 not in self._include_sha1s: all_ll_files.append(os.path.join(root, file)) for ll_path in all_ll_files: try: with open(ll_path, 'r', encoding='utf-8') as f: content = f.read() if 'alwaysinline' in content: has_inline = True break except: pass if has_inline: llvm_link = shutil.which('llvm-link') or shutil.which('llvm-link.exe') opt_cmd = shutil.which('opt') or shutil.which('opt.exe') if llvm_link and opt_cmd: print(" [内联] 检测到 alwaysinline 函数,执行跨模块内联优化...") for ll_path in all_ll_files: try: with open(ll_path, 'r', encoding='utf-8') as f: content = f.read() type_defs = [] type_def_indices = set() for i, line in enumerate(content.splitlines()): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: type_defs.append(line) type_def_indices.add(i) if type_def_indices: lines = content.splitlines(True) content_no_types = ''.join(line for i, line in enumerate(lines) if i not in type_def_indices) header_end = 0 for i, line in enumerate(content_no_types.splitlines()): stripped = line.strip() if stripped.startswith(';') or stripped.startswith('target ') or stripped.startswith('source_filename') or not stripped: header_end = i + 1 continue break lines2 = content_no_types.splitlines(True) fixed = ''.join(lines2[:header_end]) + '\n'.join(type_defs) + '\n' + ''.join(lines2[header_end:]) with open(ll_path, 'w', encoding='utf-8') as f: f.write(fixed) except: pass merged_ll = os.path.join(self.output_dir, '_merged.ll') optimized_ll = os.path.join(self.output_dir, '_merged_opt.ll') try: link_cmd = [llvm_link] + all_ll_files + ['-o', merged_ll] result = subprocess.run(link_cmd, capture_output=True, text=True) if result.returncode != 0: print(f" [警告] llvm-link 失败: {result.stderr.strip()},回退到单独编译") has_inline = False else: opt_result_cmd = [opt_cmd, '-passes=always-inline', merged_ll, '-S', '-o', optimized_ll] result = subprocess.run(opt_result_cmd, capture_output=True, text=True) if result.returncode != 0: print(f" [警告] opt 失败: {result.stderr.strip()},回退到单独编译") has_inline = False else: print(" [内联] 跨模块内联优化完成") with open(optimized_ll, 'r', encoding='utf-8') as f: opt_content = f.read() import re opt_content = re.sub(r'\b(define|declare)\s+(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external)\s+', r'\1 \2 dso_local ', opt_content) opt_content = re.sub(r'\b(define|declare)\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external|dso_local|@)', r'\1 dso_local ', opt_content) opt_content = re.sub(r'(@[\w.]+)\s*=\s*(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external)\s+(global|constant)', r'\1 = \2 dso_local \3', opt_content) opt_content = re.sub(r'(@[\w.]+)\s*=\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external|dso_local)(global|constant)', r'\1 = linkonce_odr \2', opt_content) with open(optimized_ll, 'w', encoding='utf-8') as f: f.write(opt_content) merged_obj = os.path.join(self.output_dir, '_merged.obj') try: cmd = [self.compile_cmd] + self.compile_flags + ['-o', merged_obj, optimized_ll] result = subprocess.run(cmd, capture_output=True, text=True, cwd=self.output_dir) if result.returncode == 0: print(f" [成功] 合并模块编译完成") for ll_path in ll_files: obj_path = ll_path[:-3] + '.obj' sha1_obj = os.path.join(self.output_dir, os.path.basename(ll_path)[:-3] + '.obj') if os.path.isfile(sha1_obj): os.remove(sha1_obj) return else: print(f" [警告] 合并模块编译失败: {result.stderr.strip()},回退到单独编译") has_inline = False except Exception as e: print(f" [警告] 合并模块编译异常: {e},回退到单独编译") has_inline = False except FileNotFoundError: print(f" [警告] 找不到 llvm-link 或 opt,回退到单独编译") has_inline = False else: print(" [警告] 未找到 llvm-link/opt 工具,回退到单独编译(alwaysinline 可能不生效)") from concurrent.futures import ThreadPoolExecutor, as_completed import re def _compile_one(ll_path): rel = os.path.relpath(ll_path, self.output_dir) obj_path = ll_path[:-3] + '.obj' if os.path.isfile(obj_path): return (rel, 'cached', '') try: with open(ll_path, 'r', encoding='utf-8') as f: ll_content = f.read() sha1 = os.path.basename(ll_path)[:-3] deps = self._LoadDeps(sha1) stub_prefix = '' if deps: existing_symbols = {} opaque_line_indices = {} existing_declares = set() existing_declare_lines = {} for i, line in enumerate(ll_content.splitlines()): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: m = re.match(r'%"?((?:[a-f0-9]+\.)?[\w.]+)"?\s*=', stripped) if m: name = m.group(1) if '= type opaque' in stripped: existing_symbols[name] = 'opaque' opaque_line_indices[name] = i else: existing_symbols[name] = 'full' elif stripped.startswith('declare'): dm = re.match(r'declare\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm: fname = dm.group(1) existing_declares.add(fname) existing_declare_lines[fname] = i seen_stub_sha1 = set() replaced_opaque_names = set() replaced_declare_names = set() all_stub_lines = [] stub_type_map = {} stub_declare_names = set() stub_declare_lines = {} existing_defines = set() existing_global_defs = set() stub_global_defs = set() for line in ll_content.splitlines(): stripped = line.strip() if stripped.startswith('define'): dm = re.match(r'define\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm: existing_defines.add(dm.group(1)) elif stripped.startswith('@') and (' global ' in stripped or ' constant ' in stripped): gm = re.match(r'(@\S+)', stripped) if gm: existing_global_defs.add(gm.group(1)) for dep_name, dep_sha1 in deps.items(): if dep_sha1 in seen_stub_sha1: continue seen_stub_sha1.add(dep_sha1) stub_path = os.path.join(self.output_dir, f"{dep_sha1}.stub.ll") if not os.path.isfile(stub_path): stub_path = os.path.join(self.temp_dir, f"{dep_sha1}.stub.ll") if os.path.isfile(stub_path): with open(stub_path, 'r', encoding='utf-8') as sf: stub_content = sf.read() for line in stub_content.splitlines(): stripped = line.strip() if stripped.startswith(';') or stripped.startswith('target ') or stripped.startswith('source_filename') or not stripped: continue if stripped.startswith('declare'): dm = re.match(r'declare\s+[^@]*@"?([^"\s(]+)"?', stripped) if dm: fname = dm.group(1) if re.search(r'%("[^"]+"|[a-f0-9]+\.[A-Za-z_]\w*)\s+%', stripped): continue if fname in existing_defines: continue if fname in existing_declares or fname in stub_declare_names: existing_decl_line = stub_declare_lines.get(fname) if existing_decl_line is not None: has_struct = '%' in stripped existing_has_struct = '%' in existing_decl_line if existing_decl_line else False if has_struct and not existing_has_struct: for i, l in enumerate(all_stub_lines): if l is not None and l.strip().startswith('declare'): em = re.match(r'declare\s+[^@]*@"?([^"\s(]+)"?', l.strip()) if em and em.group(1) == fname: all_stub_lines[i] = line break continue elif fname in existing_declares: has_struct = '%' in stripped if has_struct: replaced_declare_names.add(fname) stub_declare_names.add(fname) stub_declare_lines[fname] = line all_stub_lines.append(line) continue continue stub_declare_names.add(fname) stub_declare_lines[fname] = line all_stub_lines.append(line) continue if stripped.startswith('%') and '= type' in stripped: m = re.match(r'%"?((?:[a-f0-9]+\.)?[\w.]+)"?\s*=', stripped) if m: name = m.group(1) is_opaque = '= type opaque' in stripped if name in existing_symbols: if existing_symbols[name] == 'opaque' and not is_opaque: replaced_opaque_names.add(name) existing_symbols[name] = 'full' idx = len(all_stub_lines) all_stub_lines.append(line) stub_type_map[name] = (is_opaque, idx) continue if name in stub_type_map: prev_opaque, prev_idx = stub_type_map[name] if prev_opaque and not is_opaque: all_stub_lines[prev_idx] = None idx = len(all_stub_lines) all_stub_lines.append(line) stub_type_map[name] = (is_opaque, idx) continue idx = len(all_stub_lines) all_stub_lines.append(line) stub_type_map[name] = (is_opaque, idx) else: all_stub_lines.append(line) elif stripped.startswith('@') and (' global ' in stripped or ' constant ' in stripped): gm = re.match(r'(@\S+)', stripped) if gm and (gm.group(1) in existing_global_defs or gm.group(1) in stub_global_defs): continue if gm: stub_global_defs.add(gm.group(1)) all_stub_lines.append(line) else: all_stub_lines.append(line) filtered = [l for l in all_stub_lines if l is not None] if filtered: stub_prefix = '\n'.join(filtered) + '\n' lines_to_remove = set() if replaced_opaque_names: for name in replaced_opaque_names: if name in opaque_line_indices: lines_to_remove.add(opaque_line_indices[name]) if replaced_declare_names: for fname in replaced_declare_names: if fname in existing_declare_lines: lines_to_remove.add(existing_declare_lines[fname]) if lines_to_remove: ll_lines = ll_content.splitlines(True) ll_content = ''.join(line for i, line in enumerate(ll_lines) if i not in lines_to_remove) type_defs_in_ll = [] type_def_line_indices = set() for i, line in enumerate(ll_content.splitlines()): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: type_defs_in_ll.append(line) type_def_line_indices.add(i) if type_def_line_indices: ll_lines = ll_content.splitlines(True) ll_content = ''.join(line for i, line in enumerate(ll_lines) if i not in type_def_line_indices) if type_defs_in_ll: stub_prefix += '\n'.join(type_defs_in_ll) + '\n' if stub_prefix: header_end = 0 for i, line in enumerate(ll_content.splitlines()): stripped = line.strip() if stripped.startswith(';') or stripped.startswith('target ') or stripped.startswith('source_filename') or not stripped: header_end = i + 1 continue break ll_lines = ll_content.splitlines(True) ll_content = ''.join(ll_lines[:header_end]) + stub_prefix + ''.join(ll_lines[header_end:]) ll_content = re.sub(r'\b(define|declare)\s+(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external)\s+', r'\1 \2 dso_local ', ll_content) ll_content = re.sub(r'\b(define|declare)\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external|dso_local|@)', r'\1 dso_local ', ll_content) ll_content = re.sub(r'(@[\w.]+)\s*=\s*(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external)\s+(global|constant)', r'\1 = \2 dso_local \3', ll_content) ll_content = re.sub(r'(@[\w.]+)\s*=\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external|dso_local)(global|constant)', r'\1 = dso_local \2', ll_content) final_type_defs = [] final_type_def_indices = set() for i, line in enumerate(ll_content.splitlines()): stripped = line.strip() if stripped.startswith('%') and '= type' in stripped: final_type_defs.append(line) final_type_def_indices.add(i) if final_type_def_indices: final_lines = ll_content.splitlines(True) ll_content_no_types = ''.join(line for i, line in enumerate(final_lines) if i not in final_type_def_indices) header_end = 0 for i, line in enumerate(ll_content_no_types.splitlines()): stripped = line.strip() if stripped.startswith(';') or stripped.startswith('target ') or stripped.startswith('source_filename') or not stripped: header_end = i + 1 continue break final_lines2 = ll_content_no_types.splitlines(True) ll_content = ''.join(final_lines2[:header_end]) + '\n'.join(final_type_defs) + '\n' + ''.join(final_lines2[header_end:]) dso_ll_path = ll_path[:-3] + '_dso.ll' with open(dso_ll_path, 'w', encoding='utf-8') as f: f.write(ll_content) cmd = [self.compile_cmd] + self.compile_flags + ['-o', obj_path, dso_ll_path] result = subprocess.run(cmd, capture_output=True, text=True, cwd=os.path.dirname(ll_path) or '.') if result.returncode == 0: return (rel, 'ok', '') else: return (rel, 'error', result.stderr.strip()) except FileNotFoundError: return (rel, 'error', f'找不到编译器: {self.compile_cmd}') except Exception as e: return (rel, 'error', str(e)) need_compile = [] for ll_path in ll_files: obj_path = ll_path[:-3] + '.obj' if not os.path.isfile(obj_path): need_compile.append(ll_path) if need_compile: n_workers = min(os.cpu_count() or 4, len(need_compile), 8) print(f" 并行编译 {len(need_compile)} 个文件 (workers={n_workers})") errors = [] with ThreadPoolExecutor(max_workers=n_workers) as executor: futures = {executor.submit(_compile_one, ll_path): ll_path for ll_path in need_compile} for future in as_completed(futures): rel, status, err = future.result() if status == 'ok': print(f" [成功] {rel}") elif status == 'error': print(f" [错误] {rel}: {err}") errors.append((rel, err)) elif status == 'cached': pass if errors: print(f"\n[编译终止] {len(errors)} 个文件编译失败") sys.exit(1) def _scan_include_libraries(self): """扫描 includes 目录,检测被使用的模块对应的库文件和源文件""" if not self.used_includes: return for includes_dir in self.include_dirs: if not os.path.isdir(includes_dir): continue for module_name in self.used_includes: module_dir = os.path.join(includes_dir, module_name) if os.path.isdir(module_dir): self._scan_dir_for_libs(module_dir, module_name) for root, dirs, files in os.walk(module_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if fname.endswith('.py'): py_path = os.path.join(root, fname) if py_path not in self.extra_py_files: self.extra_py_files.append(py_path) print(f" [includes] 发现 Python 包文件: {os.path.relpath(py_path, includes_dir)}") for ext in self.INCLUDE_LIB_EXTENSIONS: lib_path = os.path.join(includes_dir, module_name + ext) if os.path.isfile(lib_path): self.extra_link_files.append(lib_path) print(f" [includes] 发现库文件: {os.path.relpath(lib_path, includes_dir)}") for ext in self.INCLUDE_SRC_EXTENSIONS: src_path = os.path.join(includes_dir, module_name + ext) if os.path.isfile(src_path): self.extra_compile_files.append(src_path) print(f" [includes] 发现源文件: {os.path.relpath(src_path, includes_dir)}") py_path = os.path.join(includes_dir, module_name + '.py') if os.path.isfile(py_path): self.extra_py_files.append(py_path) print(f" [includes] 发现 Python 源文件: {os.path.relpath(py_path, includes_dir)}") discovered = set() for ef in list(self.extra_py_files): self._discover_transitive_deps(ef, includes_dir, discovered) if self.extra_link_files: print(f"\n[includes] 需要链接的库文件: {len(self.extra_link_files)} 个") if self.extra_compile_files: print(f"[includes] 需要编译的源文件: {len(self.extra_compile_files)} 个") def _scan_dir_for_libs(self, directory: str, module_name: str): """递归扫描目录中的库文件和源文件""" for root, dirs, files in os.walk(directory): for fname in files: fpath = os.path.join(root, fname) _, ext = os.path.splitext(fname) ext = ext.lower() if ext in self.INCLUDE_LIB_EXTENSIONS: self.extra_link_files.append(fpath) print(f" [includes] 发现库文件: {os.path.relpath(fpath, directory)}") elif ext in self.INCLUDE_SRC_EXTENSIONS: self.extra_compile_files.append(fpath) print(f" [includes] 发现源文件: {os.path.relpath(fpath, directory)}") def _discover_transitive_deps(self, py_path: str, includes_dir: str, discovered: set): """递归发现 Python 文件的间接依赖""" if py_path in discovered: return discovered.add(py_path) try: with open(py_path, 'r', encoding='utf-8') as f: content = f.read() import ast as _ast tree = _ast.parse(content) for node in _ast.walk(tree): if isinstance(node, _ast.Import): for alias in node.names: mod = alias.name.split('.')[0] self._try_add_include_dep(mod, includes_dir, discovered) elif isinstance(node, _ast.ImportFrom): if node.module and node.level == 0: mod = node.module.split('.')[0] self._try_add_include_dep(mod, includes_dir, discovered) except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"发现 include 传递依赖失败: {_e}", "Exception") def _try_add_include_dep(self, module_name: str, includes_dir: str, discovered: set): """尝试将模块添加为 include 依赖""" py_path = os.path.join(includes_dir, module_name + '.py') if os.path.isfile(py_path) and py_path not in self.extra_py_files: self.extra_py_files.append(py_path) print(f" [includes] 发现间接依赖: {os.path.relpath(py_path, includes_dir)}") self._discover_transitive_deps(py_path, includes_dir, discovered) def _is_decl_only_file(self, src_path: str) -> bool: try: with open(src_path, 'r', encoding='utf-8') as f: content = f.read() import ast as _ast tree = _ast.parse(content) for node in _ast.walk(tree): if isinstance(node, _ast.ClassDef): return False if isinstance(node, _ast.FunctionDef): body = node.body if len(body) == 1: stmt = body[0] if isinstance(stmt, _ast.Expr) and isinstance(stmt.value, _ast.Constant) and stmt.value.value is ...: continue if isinstance(stmt, _ast.Pass): continue return False return True except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"判断是否为声明文件失败: {_e}", "Exception") return False def _sort_include_files_by_deps(self, file_list: list) -> list: """按依赖关系对 include 文件进行拓扑排序,确保被依赖的文件先编译""" import ast as _ast # 构建文件名到路径的映射 name_to_path = {} for fpath in file_list: basename = os.path.splitext(os.path.basename(fpath))[0] name_to_path[basename] = fpath # 也支持包内模块: vpsdk/window -> path for fpath in file_list: # 尝试从 includes 目录推断模块全名 for includes_dir in self.include_dirs: try: rel = os.path.relpath(fpath, includes_dir) mod_name = os.path.splitext(rel)[0].replace(os.sep, '.').replace('/', '.') name_to_path[mod_name] = fpath except ValueError: pass # 分析每个文件的 import 依赖 deps = {} # path -> set of paths it depends on for fpath in file_list: deps[fpath] = set() try: with open(fpath, '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.Import): for alias in node.names: mod = alias.name.split('.')[0] if mod in name_to_path and name_to_path[mod] != fpath: deps[fpath].add(name_to_path[mod]) elif isinstance(node, _ast.ImportFrom): if node.module and node.level == 0: mod = node.module.split('.')[0] if mod in name_to_path and name_to_path[mod] != fpath: deps[fpath].add(name_to_path[mod]) except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"分析 include 文件导入依赖失败: {_e}", "Exception") # 拓扑排序 (Kahn's algorithm) in_degree = {fpath: len(deps[fpath]) for fpath in file_list} # 反向邻接表:如果 A 依赖 B,则 B -> A reverse_adj = {fpath: [] for fpath in file_list} for fpath in file_list: for dep in deps[fpath]: if dep in reverse_adj: reverse_adj[dep].append(fpath) result = [] queue = [fpath for fpath in file_list if in_degree[fpath] == 0] # 对队列排序以保持确定性 queue.sort(key=lambda x: x) while queue: current = queue.pop(0) result.append(current) for neighbor in reverse_adj[current]: in_degree[neighbor] -= 1 if in_degree[neighbor] == 0: queue.append(neighbor) queue.sort(key=lambda x: x) # 如果有循环依赖,把剩余文件也加入 for fpath in file_list: if fpath not in result: result.append(fpath) return result def _compile_include_py_files(self, active_sha1s: set): """通过 TransPyC 翻译器编译 includes 目录中发现的 .py 文件""" if not self.extra_py_files: return # 按依赖关系排序 include 文件,确保被依赖的文件先编译 sorted_files = self._sort_include_files_by_deps(self.extra_py_files) print(f"\n[includes 编译] 找到 {len(sorted_files)} 个 Python 源文件") # 构建 include 文件的模块 SHA1 映射 # 必须在编译前预构建完整的映射,否则后续文件引用前面文件时会用错误的 SHA1 include_ModuleSha1Map = self._build_current_ModuleSha1Map(None) # 预先为所有 include 文件计算 SHA1 并添加到映射中 for src_path in sorted_files: module_name = os.path.splitext(os.path.basename(src_path))[0] with open(src_path, 'r', encoding='utf-8') as f: py_content = f.read() sha1 = compute_sha1(py_content) self.include_py_map[module_name] = sha1 active_sha1s.add(sha1) self._include_sha1s.add(sha1) # 将 include 文件的模块名映射到其 SHA1 for includes_dir in self.include_dirs: try: rel = os.path.relpath(src_path, includes_dir) mod_full = os.path.splitext(rel)[0].replace(os.sep, '.').replace('/', '.') include_ModuleSha1Map[mod_full] = sha1 include_ModuleSha1Map[module_name] = sha1 except ValueError: pass include_ModuleSha1Map[module_name] = sha1 ll_path = os.path.join(self.output_dir, f"{sha1}.ll") obj_path = os.path.join(self.output_dir, f"{sha1}.obj") if self._is_decl_only_file(src_path): print(f" 跳过(声明文件): {module_name}.py") self._collect_inline_symbols(src_path) continue if os.path.isfile(obj_path): print(f" 跳过(缓存): {module_name}.py -> {sha1}.obj") self._collect_inline_symbols(src_path) self.extra_link_files.append(obj_path) continue print(f" 翻译: {os.path.basename(src_path)} -> {sha1}.ll") try: self._translate_file(src_path, ll_path, prebuilt_ModuleSha1Map=include_ModuleSha1Map) if os.path.isfile(ll_path): print(f" 编译: {sha1}.ll -> {sha1}.obj") with open(ll_path, 'r', encoding='utf-8') as f: ll_content = f.read() import re ll_content = re.sub(r'\b(define|declare)\s+(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external)\s+', r'\1 \2 dso_local ', ll_content) ll_content = re.sub(r'\b(define|declare)\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external|dso_local)', r'\1 dso_local ', ll_content) ll_content = re.sub(r'(@[\w.]+)\s*=\s*(private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external)\s+(global|constant)', r'\1 = \2 dso_local \3', ll_content) ll_content = re.sub(r'(@[\w.]+)\s*=\s+(?!private|internal|available_externally|linkonce|weak|common|appending|extern_weak|linkonce_odr|weak_odr|external|dso_local)(global|constant)', r'\1 = dso_local \2', ll_content) stub_decls = [] for other_sha1, other_stub in self.stub_files.items(): if other_sha1 == sha1: continue if os.path.exists(other_stub): try: with open(other_stub, 'r', encoding='utf-8') as sf: for sline in sf: sl = sline.strip() if sl.startswith('%') and '= type' in sl and 'opaque' not in sl: struct_name_match = re.match(r'%"?([\w.]+)"?\s*=', sl) if struct_name_match: stub_decls.append(sl) except: pass if stub_decls: for sdecl in stub_decls: sname_match = re.match(r'%"?([\w.]+)"?\s*=\s*type', sdecl) if sname_match: sname = sname_match.group(1) short = sname.split('.')[-1] if '.' in sname else sname ll_content = re.sub( r'%"?' + re.escape(sname) + r'"?\s*=\s*type\s*opaque', sdecl, ll_content) if short != sname: ll_content = re.sub( r'%"?' + re.escape(short) + r'"?\s*=\s*type\s*opaque', sdecl, ll_content) dso_ll_path = ll_path[:-3] + '_dso.ll' with open(dso_ll_path, 'w', encoding='utf-8') as f: f.write(ll_content) cmd = [self.compile_cmd] + self.compile_flags + ['-o', obj_path, dso_ll_path] result = subprocess.run( cmd, capture_output=True, text=True, cwd=self.output_dir ) if result.returncode == 0: print(f" [成功]") self.extra_link_files.append(obj_path) # 编译成功后,生成 .pyi 签名文件和 .stub.ll 文件 # 并注册到符号表和签名映射中,以便后续 include 文件可以引用 self._register_compiled_include(src_path, sha1, ll_content, include_ModuleSha1Map) else: print(f" [警告] 编译失败: {result.stderr.strip()}") for ext in ['.ll', '_dso.ll']: p = ll_path[:-3] + ext if os.path.isfile(p): os.remove(p) else: print(f" [错误] 翻译未生成 .ll 文件") except Exception as e: import traceback for ext in ['.ll', '_dso.ll']: p = ll_path[:-3] + ext if os.path.isfile(p): os.remove(p) print(f" [错误] 翻译异常: {e}") traceback.print_exc() print(f"\n[编译终止] includes 文件翻译失败") sys.exit(1) def _register_compiled_include(self, src_path, sha1, ll_content, include_ModuleSha1Map): """编译 include 文件成功后,生成签名和 stub 文件并注册到符号表中""" module_name = os.path.splitext(os.path.basename(src_path))[0] # 确定完整模块名(如 vpsdk.window) mod_full = module_name for includes_dir in self.include_dirs: try: rel = os.path.relpath(src_path, includes_dir) mod_full = os.path.splitext(rel)[0].replace(os.sep, '.').replace('/', '.') break except ValueError: pass # 1. 生成 .pyi 签名文件(如果还没有的话) sig_path = os.path.join(self.temp_dir, f"{sha1}.pyi") if not os.path.isfile(sig_path): try: with open(src_path, 'r', encoding='utf-8') as f: py_content = f.read() sig_content = PythonToStubConverter.convert(py_content, mod_full) os.makedirs(self.temp_dir, exist_ok=True) with open(sig_path, 'w', encoding='utf-8', newline='\n') as f: f.write(sig_content) except Exception as e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"签名生成失败不阻塞编译: {e}", "Exception") # 注册到 sig_files 和 _source_module_sig_files if os.path.isfile(sig_path): self.sig_files[sha1] = sig_path self.sha1_map[sha1] = f"includes/{mod_full.replace('.', os.sep)}.py" if hasattr(self, '_shared_source_module_sig_files') and self._shared_source_module_sig_files is not None: self._shared_source_module_sig_files[mod_full] = sig_path self._shared_source_module_sig_files[module_name] = sig_path # 将签名信息加载到共享符号表中 if hasattr(self, '_shared_symbol_table') and self._shared_symbol_table is not None: try: self._shared_symbol_table.LoadModuleSymbols(sig_path, mod_full, lineno=0) except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"加载模块签名到共享符号表失败: {_e}", "Exception") # 2. 生成 .stub.ll 文件 stub_path = os.path.join(self.output_dir, f"{sha1}.stub.ll") if not os.path.isfile(stub_path) and ll_content: try: stub_content, _ = self._split_ll(ll_content) with open(stub_path, 'w', encoding='utf-8', newline='\n') as f: f.write(stub_content) except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"生成 .stub.ll 文件失败: {_e}", "Exception") # 注册到 stub_files if os.path.isfile(stub_path): self.stub_files[sha1] = stub_path elif os.path.isfile(sig_path): # 如果 stub 文件不存在,也尝试从 temp 目录查找 temp_stub = os.path.join(self.temp_dir, f"{sha1}.stub.ll") if os.path.isfile(temp_stub): self.stub_files[sha1] = temp_stub def _precollect_inline_symbols(self): """在主翻译循环之前,扫描includes目录中used_includes相关的模块收集内联函数AST""" if not self.used_includes: return for includes_dir in self.include_dirs: if not os.path.isdir(includes_dir): continue for module_name in self.used_includes: module_dir = os.path.join(includes_dir, module_name) if os.path.isdir(module_dir): for root, dirs, files in os.walk(module_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if fname.endswith('.py'): src_path = os.path.join(root, fname) self._collect_inline_symbols(src_path) py_path = os.path.join(includes_dir, module_name + '.py') if os.path.isfile(py_path): self._collect_inline_symbols(py_path) for ef in self.extra_py_files: self._collect_inline_symbols(ef) def _collect_inline_symbols(self, src_path: str): """解析源文件,收集内联函数的AST信息到inline_func_symbols""" try: with open(src_path, 'r', encoding='utf-8') as f: code = f.read() tree = ast.parse(code) module_name = os.path.splitext(os.path.basename(src_path))[0] for node in ast.iter_child_nodes(tree): if isinstance(node, ast.FunctionDef) and node.returns: is_inline = False if isinstance(node.returns, ast.BinOp) and isinstance(node.returns.op, ast.BitOr): for part in [node.returns.left, node.returns.right]: if isinstance(part, ast.Attribute) and part.attr == 'CInline': is_inline = True break if isinstance(part, ast.Name) and part.id == 'CInline': is_inline = True break elif isinstance(node.returns, ast.Attribute) and node.returns.attr == 'CInline': is_inline = True elif isinstance(node.returns, ast.Name) and node.returns.id == 'CInline': is_inline = True if is_inline: func_name = node.name sym_key = f"{module_name}.{func_name}" info = CTypeInfo() info.Name = func_name info.IsFunction = True info.IsInline = True info.InlineBody = node.body info.InlineParams = [arg.arg for arg in node.args.args] info.Storage = t.CInline() self.inline_func_symbols[func_name] = info self.inline_func_symbols[sym_key] = info except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"预收集内联函数符号失败: {_e}", "Exception") def _build_shared_symbol_data(self): """一次性构建所有文件共享的符号表数据,避免每个文件重复加载""" import copy import time t0 = time.time() print("[缓存] 构建共享符号表数据...") base_trans = TransPyC.TransPyC(code="pass", triple=self.triple, datalayout=self.datalayout) base_trans.translator.LlvmGen = None base_trans.translator._ModuleSha1Map = {} base_trans.translator._global_function_default_args = self.function_default_args for includes_dir in self.include_dirs: if os.path.isdir(includes_dir): for pyi_file in os.listdir(includes_dir): if pyi_file.endswith('.pyi'): pyi_path = os.path.join(includes_dir, pyi_file) module_name = os.path.splitext(pyi_file)[0] base_trans.translator.SymbolTable.LoadModuleSymbols(pyi_path, module_name, lineno=0) for py_file in os.listdir(includes_dir): if py_file.endswith('.py') and not py_file.startswith('_'): module_name = os.path.splitext(py_file)[0] sha1_key = self.include_py_map.get(module_name) if sha1_key and sha1_key in self.sig_files: sig_path = self.sig_files[sha1_key] base_trans.translator.SymbolTable.LoadModuleSymbols(sig_path, module_name, lineno=0) # 收集需要重导出的包,等所有模块符号加载完后再处理 _pending_reexports = [] for sha1_key, sig_path in self.sig_files.items(): rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): mod_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') mod_name = mod_name[len('includes/'):] base_trans.translator.SymbolTable.LoadModuleSymbols(sig_path, mod_name, lineno=0) if mod_name.endswith('.__init__'): package_name = mod_name[:-len('.__init__')] _pending_reexports.append((sig_path, package_name)) # 所有 includes 模块符号已加载,现在处理包重导出 for sig_path, package_name in _pending_reexports: self._register_package_reexports(base_trans, sig_path, package_name) for sha1_key, sig_path in self.sig_files.items(): rel_path = self.sha1_map.get(sha1_key, sha1_key) if rel_path.startswith('includes/'): continue module_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.') base_trans.translator.SymbolTable.LoadModuleSymbols(sig_path, module_name, lineno=0) base_trans.translator._source_module_sig_files[module_name] = sig_path short_name = module_name.split('.')[-1] if '.' in module_name else module_name if short_name != module_name: base_trans.translator._source_module_sig_files[short_name] = sig_path all_dc = {} for sha1_key, stub_path in self.stub_files.items(): if os.path.exists(stub_path): try: with open(stub_path, 'r', encoding='utf-8') as f: stub_content = f.read() for line in stub_content.splitlines(): line = line.strip() if line.startswith('@') and '= global' in line and 'i32' in line and not line.startswith('@llvm'): var_name = line.split('=')[0].strip().lstrip('@') val_part = line.split('i32')[-1].strip().rstrip(']').lstrip('[').lstrip('i32') try: val = int(val_part) all_dc[var_name] = val except: pass except: pass for sha1_key, pyi_path in self.sig_files.items(): if os.path.exists(pyi_path): try: with open(pyi_path, 'r', encoding='utf-8') as f: pyi_content = f.read() tree = ast.parse(pyi_content) for node in ast.iter_child_nodes(tree): if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): ann_str = ast.dump(node.annotation) if 'CDefine' in ann_str and node.value: val = None if isinstance(node.value, ast.Constant): val = node.value.value elif isinstance(node.value, ast.Call) and isinstance(node.value.func, ast.Name) and node.value.func.id == 'float' and node.value.args and isinstance(node.value.args[0], ast.Constant): val = node.value.args[0].value if val is not None: all_dc[f"{node.target.id}"] = val except: pass export_extern_funcs = set() for sha1_key, sig_path in self.sig_files.items(): try: with open(sig_path, 'r', encoding='utf-8') as f: sig_content = f.read() sig_tree = ast.parse(sig_content) for node in ast.iter_child_nodes(sig_tree): if isinstance(node, ast.FunctionDef): if _check_annotation_for_export(node.returns): export_extern_funcs.add(node.name) except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"收集共享导出外部函数失败: {_e}", "Exception") for sym_name, sym_info in self.inline_func_symbols.items(): if sym_name not in base_trans.translator.SymbolTable: base_trans.translator.SymbolTable[sym_name] = sym_info else: existing = base_trans.translator.SymbolTable[sym_name] if not getattr(existing, 'InlineBody', None) and getattr(sym_info, 'InlineBody', None): existing.IsInline = sym_info.IsInline existing.InlineBody = sym_info.InlineBody existing.InlineParams = sym_info.InlineParams self._shared_symbol_table = base_trans.translator.SymbolTable self._shared_source_module_sig_files = dict(base_trans.translator._source_module_sig_files) self._shared_all_dc = all_dc self._shared_export_extern_funcs = export_extern_funcs generic_class_templates = {} for includes_dir in self.include_dirs: if os.path.isdir(includes_dir): abs_includes = os.path.join(self.src_root, includes_dir) if not os.path.isabs(includes_dir) else includes_dir if not os.path.isdir(abs_includes): abs_includes = includes_dir for py_file in os.listdir(abs_includes): if py_file.endswith('.py') and not py_file.startswith('_'): py_path = os.path.join(abs_includes, py_file) try: with open(py_path, 'r', encoding='utf-8') as f: py_code = f.read() py_tree = ast.parse(py_code) for node in py_tree.body: if isinstance(node, ast.ClassDef) and hasattr(node, 'type_params') and node.type_params: type_params = [tp.name for tp in node.type_params] generic_class_templates[node.name] = { 'node': node, 'type_params': type_params, } except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"收集泛型类模板失败: {_e}", "Exception") self._shared_generic_class_templates = generic_class_templates t1 = time.time() print(f"[缓存] 共享符号表构建完成 ({t1-t0:.2f}s, {len(self._shared_symbol_table)} 符号, {len(generic_class_templates)} 泛型模板)") def _compile_include_sources(self): """扫描项目目录中的汇编存根,编译 includes 目录中发现的 C/C++ 源文件""" extra_sources = list(self.extra_compile_files) scan_dirs = [] if self.src_root and os.path.isdir(self.src_root): scan_dirs.append(self.src_root) project_dir = os.path.dirname(self.output_dir) if project_dir and os.path.isdir(project_dir) and project_dir not in scan_dirs: scan_dirs.append(project_dir) for scan_dir in scan_dirs: for root, dirs, files in os.walk(scan_dir): dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__'] for fname in files: if fname.endswith(('.S', '.s', '.c', '.cpp')): if fname.startswith('test_'): continue fpath = os.path.join(root, fname) if fpath not in extra_sources: extra_sources.append(fpath) if not extra_sources: return cc_map = { '.c': 'gcc', '.cpp': 'g++', '.cc': 'g++', '.cxx': 'g++', '.m': 'clang', '.mm': 'clang++', '.s': 'gcc', '.S': 'gcc', } print(f"\n[includes 编译] 找到 {len(extra_sources)} 个源文件") for src_path in extra_sources: _, ext = os.path.splitext(src_path) ext = ext.lower() obj_path = os.path.join(self.output_dir, os.path.splitext(os.path.basename(src_path))[0] + '.obj') if ext in ('.s', '.S') and any('oformat' in f for f in self.linker_flags): cc = 'clang' compile_src = src_path extra_flags = ['-c', '-target', 'x86_64-none-elf', '-o', obj_path, compile_src] else: cc = cc_map.get(ext, 'gcc') extra_flags = ['-c', '-o', obj_path, src_path] print(f" 编译: {os.path.basename(src_path)} -> {os.path.basename(obj_path)}") try: cmd = [cc] + extra_flags result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: print(f" [成功]") self.extra_link_files.append(obj_path) else: print(f" [错误] 编译失败: {result.stderr.strip()}") print(f"\n[编译终止] C/C++ 源文件编译失败,立即终止编译。") sys.exit(1) except FileNotFoundError: print(f" [错误] 找不到编译器: {cc}") print(f"\n[编译终止] 找不到编译器,立即终止编译。") sys.exit(1) except Exception as e: print(f" [错误] {e}") print(f"\n[编译终止] 编译异常,立即终止编译。") sys.exit(1) def _strip_debug_sections(self, exe_path: str): """剥离 .exe 中的调试段(跳过裸机二进制)""" if not os.path.exists(exe_path): return ext = os.path.splitext(exe_path)[1].lower() if ext not in ('.exe', '.dll', '.obj', '.o', '.elf'): print(f" [剥离] 跳过({ext} 格式无需剥离)") return import shutil for tool in ['llvm-strip', 'strip']: strip_cmd = shutil.which(tool) if strip_cmd: try: result = subprocess.run( [strip_cmd, '--strip-debug', exe_path], capture_output=True, text=True ) if result.returncode == 0: print(f" [剥离] 调试段已移除 ({tool})") else: print(f" [剥离] 警告: {result.stderr.strip()}") except Exception as _e: from lib.core.VLogger import get_logger as _vlog from lib.constants.config import mode as _config_mode if _config_mode == "strict": raise _vlog().warning(f"剥离可执行文件调试段失败: {_e}", "Exception") break def _generate_startup_obj(self): """当 project.json 中配置了 startup 时,生成符号别名 shim。 startup 配置格式: true -> 等价于 {"__main": "main"} {"__main": "main", "_start": "main"} -> 为每个别名生成调用目标函数的包装 生成的 .ll 会被编译为 _startup.obj 并加入链接列表。 """ if not self.startup: return None # 归一化配置 if self.startup is True: aliases = {"__main": "main"} elif isinstance(self.startup, dict): aliases = self.startup else: return None # 构建 LLVM IR decls = [] defs = [] for alias_name, target_name in aliases.items(): # 声明目标函数 decls.append(f"declare i32 @{target_name}()") # 定义别名函数:调用目标函数并返回其返回值 defs.append( f"define dso_local i32 @{alias_name}() {{\n" f"entry:\n" f" %r = call i32 @{target_name}()\n" f" ret i32 %r\n" f"}}\n" ) ll_content = "; startup shim (auto-generated)\n" + "\n".join(decls) + "\n\n" + "\n".join(defs) + "\n" ll_path = os.path.join(self.output_dir, '_startup.ll') obj_path = os.path.join(self.output_dir, '_startup.obj') with open(ll_path, 'w', encoding='utf-8', newline='\n') as f: f.write(ll_content) # 编译为 .obj cmd = [self.compile_cmd] + self.compile_flags + ['-o', obj_path, ll_path] result = subprocess.run(cmd, capture_output=True, text=True, cwd=self.output_dir or '.') if result.returncode != 0: print(f" [startup] 编译失败: {result.stderr.strip()}") return None print(f" [startup] 生成符号别名: {aliases}") return obj_path def _link_obj_files(self, active_sha1s: set): """将 active_sha1s 对应的 .obj 文件和 includes 库文件链接为可执行文件""" obj_files = [] merged_obj = os.path.join(self.output_dir, '_merged.obj') if os.path.isfile(merged_obj): obj_files.append(merged_obj) else: for root, dirs, files in os.walk(self.output_dir): for file in files: if file.endswith('.obj'): sha1 = file[:-4] if sha1 in active_sha1s and sha1 not in self._include_sha1s: obj_files.append(os.path.join(root, file)) if not obj_files and not self.extra_link_files: print("[链接] 无文件可链接") return # 生成 startup 符号别名 shim startup_obj = self._generate_startup_obj() if startup_obj: obj_files.append(startup_obj) print(f"\n[链接] 找到 {len(obj_files)} 个 .obj 文件") all_link_files = obj_files + self.extra_link_files seen = set() all_link_files = [f for f in all_link_files if not (f in seen or seen.add(f))] if self.extra_link_files: print(f"[链接] 额外库文件: {len(self.extra_link_files)} 个") for lib in self.extra_link_files: print(f" - {os.path.basename(lib)}") if not self.linker_cmd: print("[链接] 未配置链接器") return exe_path = os.path.join(self.output_dir, self.linker_output) if self.linker_output else os.path.join(self.output_dir, 'a.exe') is_binary = self.linker_output and self.linker_output.endswith('.bin') # 检查是否直接输出二进制(使用 --oformat binary / -oformat binary 标志) is_direct_binary = is_binary and any('oformat' in f for f in self.linker_flags) if is_binary and not is_direct_binary: link_output = exe_path[:-4] + '_pe.exe' else: link_output = exe_path # 自动复制或生成 linker.ld 到输出目录 linker_ld_src = os.path.join(os.path.dirname(self.output_dir), 'linker.ld') linker_ld_dst = os.path.join(self.output_dir, 'linker.ld') if os.path.isfile(linker_ld_src) and not os.path.isfile(linker_ld_dst): import shutil shutil.copy2(linker_ld_src, linker_ld_dst) print(f" [链接] 使用链接脚本: linker.ld") elif not os.path.isfile(linker_ld_dst) and any('-T' in f or '-T' in f.replace(',', ' ') for f in self.linker_flags): default_ld = """ENTRY(_start) SECTIONS { . = 0x100000; .text : { *(.text.startup) *(.text) *(.text.*) } .rodata : { *(.rodata) *(.rodata.*) } .data : { *(.data) *(.data.*) } .bss : { *(.bss) *(.bss.*) } /DISCARD/ : { *(.comment) *(.note) *(.eh_frame) *(.eh_frame_hdr) *(.reloc) *(.rela) *(.debug*) } } """ with open(linker_ld_dst, 'w', encoding='utf-8', newline='\n') as f: f.write(default_ld) print(f" [链接] 自动创建默认链接脚本: linker.ld") try: # 将 linker_flags 拆分为非库标志和库标志(-l/-L),库标志放在目标文件之后 # 这样 Unix 风格链接器才能正确解析库中的符号 # -Wl, 选项是链接器全局选项,放在目标文件之前 non_lib_flags = [] lib_flags = [] i = 0 while i < len(self.linker_flags): flag = self.linker_flags[i] if flag in ('-l', '-L') and i + 1 < len(self.linker_flags): # -l xxx / -L xxx 分开的形式 lib_flags.extend([flag, self.linker_flags[i + 1]]) i += 2 elif flag.startswith('-l') or flag.startswith('-L'): # -lxxx / -Lxxx 合并的形式 lib_flags.append(flag) i += 1 else: # 其他所有标志(包括 -Wl, 选项)放在目标文件之前 non_lib_flags.append(flag) i += 1 cmd = [self.linker_cmd] + non_lib_flags + ['-o', link_output] + all_link_files + lib_flags print(f" 执行: {' '.join(cmd)}") result = subprocess.run( cmd, capture_output=True, text=True, cwd=self.output_dir or '.' ) if result.returncode == 0: if is_binary and not is_direct_binary: import shutil objcopy = shutil.which('llvm-objcopy') or shutil.which('objcopy') if objcopy: conv = subprocess.run( [objcopy, '-O', 'binary', link_output, exe_path], capture_output=True, text=True ) if conv.returncode == 0: os.remove(link_output) print(f" [成功] 生成裸机二进制: {exe_path}") file_size = os.path.getsize(exe_path) print(f" [大小] {file_size} 字节") else: print(f" [错误] 二进制转换失败: {conv.stderr.strip()}") sys.exit(1) else: print(" [错误] 找不到 llvm-objcopy,无法生成裸机二进制") sys.exit(1) else: print(f" [成功] 生成: {exe_path}") file_size = os.path.getsize(exe_path) print(f" [大小] {file_size} 字节") if not is_binary: self._strip_debug_sections(exe_path) else: print(f" [错误] 链接失败: {result.stderr.strip()}") print(f"\n[编译终止] 链接失败,立即终止编译。") sys.exit(1) except FileNotFoundError: print(f" [错误] 找不到链接器: {self.linker_cmd}") print(f"\n[编译终止] 找不到链接器,立即终止编译。") sys.exit(1) except Exception as e: print(f" [错误] {e}") print(f"\n[编译终止] 链接异常,立即终止编译。") sys.exit(1)