GVSDS/TransPyC
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e02c867edf6adf584fdb0a348f37bb1112b8e944
TransPyC/lib/Projectrans/Phase2Translator.py

2424 lines
125 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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)
Reference in New Issue View Git Blame Copy Permalink