Files
TransPyC/Projectrans.py
2026-06-16 16:09:42 +08:00

3743 lines
186 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.
#!/usr/bin/env python3
"""
ProjectTrans.py - 工程级 TransPyC 翻译器
两阶段翻译流程:
阶段一:从源文件生成声明接口 <SHA1>.pyi仅有签名和 <SHA1>.stub.ll仅有声明
阶段二:使用声明接口翻译源文件,嵌入 .stub.ll 声明,生成含代码的 .ll 并编译为 .obj
SHA1 = SHA1(source file content)
"""
import sys
import os
import re
import hashlib
import shutil
from typing import List
import subprocess
import tempfile
import ast
import traceback
import json
from lib.core.Handles.HandlesBase import CTypeInfo
from lib.includes import t
from pathlib import Path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import TransPyC
from StubGen import PythonToStubConverter
from lib.core.logger import Logger, LogLevel, set_logger
# 创建全局日志实例
logger = Logger(
name="TransPyC",
log_file="build.log",
level=LogLevel.INFO,
use_colors=True
)
set_logger(logger)
def compute_sha1(content: str) -> str:
return hashlib.sha1(content.encode('utf-8')).hexdigest()[:16]
def _check_annotation_for_export(annotation) -> bool:
"""递归检查类型注解中是否包含 CExport"""
if annotation is None:
return False
if isinstance(annotation, ast.Attribute):
if hasattr(annotation, 'attr') and annotation.attr == 'CExport':
return True
elif isinstance(annotation, ast.BinOp) and isinstance(annotation.op, ast.BitOr):
return _check_annotation_for_export(annotation.left) or _check_annotation_for_export(annotation.right)
elif isinstance(annotation, ast.Name):
return annotation.id == 'CExport'
return False
def sha1_file(filepath: str) -> str:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
return compute_sha1(content)
def get_file_dependencies(filepath: str, src_root: str) -> set:
"""解析 Python 文件的导入依赖"""
dependencies = set()
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
tree = ast.parse(content)
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
module = alias.name
dependencies.add(module)
elif isinstance(node, ast.ImportFrom):
if node.module:
module = node.module
if node.level > 0:
# from .module import name → 依赖 pkg.module模块本身
pkg_dir = os.path.dirname(filepath)
pkg_rel = os.path.relpath(pkg_dir, src_root).replace(os.sep, '.').replace('/', '.')
if pkg_rel == '.':
pkg_rel = ''
if pkg_rel:
dep_module = f'{pkg_rel}.{module}'
else:
dep_module = module
dependencies.add(dep_module)
else:
if not module.startswith('_'):
dependencies.add(module)
elif node.level > 0 and node.names:
# from . import name → 依赖 pkg.name子模块
for alias in node.names:
pkg_dir = os.path.dirname(filepath)
pkg_rel = os.path.relpath(pkg_dir, src_root).replace(os.sep, '.').replace('/', '.')
if pkg_rel == '.':
pkg_rel = ''
dep_module = f'{pkg_rel}.{alias.name}' if pkg_rel else alias.name
dependencies.add(dep_module)
except Exception:
pass
return dependencies
def find_reachable_files_from_entries(src_root: str, entry_files: list) -> set:
"""从入口文件出发,找出所有可达的 .py 文件(通过导入关系)"""
all_files = {} # module_name -> filepath
for root, dirs, files in os.walk(src_root):
dirs[:] = [d for d in dirs if d not in ('__pycache__',)]
for file in files:
if file.endswith('.py'):
filepath = os.path.join(root, file)
rel = os.path.relpath(filepath, src_root)
module = os.path.splitext(rel)[0].replace(os.sep, '.').replace('/', '.')
if filepath.endswith('__init__.py'):
all_files[module] = filepath
parent_module = module.rsplit('.', 1)[0] if '.' in module else module
if parent_module not in all_files or not all_files[parent_module].endswith('__init__.py'):
all_files[parent_module] = filepath
elif module not in all_files:
all_files[module] = filepath
top_module = module.split('.')[0]
if top_module not in all_files:
all_files[top_module] = filepath
reachable = set()
queue = list(entry_files)
while queue:
current = queue.pop(0)
if current in reachable:
continue
reachable.add(current)
deps = get_file_dependencies(current, src_root)
for dep in deps:
if dep in all_files and all_files[dep] not in reachable:
queue.append(all_files[dep])
else:
for key, val in all_files.items():
if key.endswith('.' + dep) or key == dep:
if val not in reachable:
queue.append(val)
break
return reachable
def topological_sort_files(py_files: list, src_root: str) -> list:
"""对 Python 文件进行拓扑排序,确保依赖模块先被处理"""
# 构建模块名到文件路径的映射
module_to_file = {}
for filepath in py_files:
rel = os.path.relpath(filepath, src_root)
module = os.path.splitext(rel)[0].replace(os.sep, '.').replace('/', '.')
module_to_file[module] = filepath
# 也添加顶层模块名
top_module = module.split('.')[0]
if top_module not in module_to_file:
module_to_file[top_module] = filepath
# 构建依赖图
graph = {f: set() for f in py_files}
for filepath in py_files:
deps = get_file_dependencies(filepath, src_root)
for dep in deps:
if dep in module_to_file:
dep_file = module_to_file[dep]
if dep_file != filepath: # 避免自依赖
graph[filepath].add(dep_file)
# 拓扑排序 (Kahn's algorithm)
in_degree = {f: 0 for f in py_files}
for f, deps in graph.items():
for dep in deps:
if dep in in_degree:
in_degree[f] += 1
queue = [f for f, degree in in_degree.items() if degree == 0]
sorted_files = []
while queue:
# 按字母顺序处理同级别的文件,确保确定性
queue.sort()
current = queue.pop(0)
sorted_files.append(current)
for f, deps in graph.items():
if current in deps:
in_degree[f] -= 1
if in_degree[f] == 0:
queue.append(f)
# 如果有环,添加剩余文件
if len(sorted_files) < len(py_files):
remaining = [f for f in py_files if f not in sorted_files]
sorted_files.extend(remaining)
return sorted_files
class Phase1Generator:
"""阶段一:从源文件生成声明接口"""
def __init__(self, src_root: str, temp_dir: str, include_dirs: list = None, entry_files: list = None, target_triple: str = None, target_datalayout: str = None):
self.src_root = os.path.abspath(src_root)
self.temp_dir = os.path.abspath(temp_dir)
self.include_dirs = include_dirs or []
self.sha1_map: dict[str, str] = {}
self.include_py_map: dict[str, str] = {}
self.entry_files = entry_files
self.target_triple = target_triple
self.target_datalayout = target_datalayout
os.makedirs(self.temp_dir, exist_ok=True)
def _get_needed_include_files(self, reachable_source_files: set) -> list:
"""从可达源文件收集所有被引用(含传递依赖)的 include 文件"""
include_file_map = {}
for includes_dir in self.include_dirs:
if not os.path.isdir(includes_dir):
continue
for root, dirs, files in os.walk(includes_dir):
dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__']
for fname in files:
if fname.endswith('.py') or fname.endswith('.pyi'):
src_path = os.path.join(root, fname)
rel_from_inc = os.path.relpath(src_path, includes_dir)
module_path = rel_from_inc.replace(os.sep, '.').replace('/', '.')
module_name = os.path.splitext(module_path)[0]
info = (src_path, rel_from_inc, includes_dir)
include_file_map[module_name] = info
top_pkg = module_name.split('.')[0]
if top_pkg not in include_file_map:
include_file_map[top_pkg] = info
imported_from_src = set()
for src_path in reachable_source_files:
deps = get_file_dependencies(src_path, self.src_root)
imported_from_src.update(deps)
needed = set()
queue = list(imported_from_src)
while queue:
mod_name = queue.pop(0)
if mod_name in needed:
continue
if mod_name in include_file_map:
needed.add(mod_name)
src_path = include_file_map[mod_name][0]
includes_dir = include_file_map[mod_name][2]
deps = get_file_dependencies(src_path, includes_dir)
for dep in deps:
if dep in include_file_map and dep not in needed:
queue.append(dep)
needed_infos = []
for mod_name in sorted(needed):
if mod_name in include_file_map:
info = include_file_map[mod_name]
if info not in needed_infos:
needed_infos.append(info)
return needed_infos
def run(self):
"""扫描源目录,只处理入口文件可达的 .py 文件(导入遍历)"""
if self.entry_files:
py_files = list(self.entry_files)
else:
main_py = os.path.join(self.src_root, 'main.py')
if os.path.exists(main_py):
py_files = [main_py]
else:
py_files = []
for root, dirs, files in os.walk(self.src_root):
dirs[:] = [d for d in dirs if d not in ('__pycache__',)]
for file in files:
if file.endswith('.py'):
py_files.append(os.path.join(root, file))
if not py_files:
print("[阶段一] 未找到入口文件或源文件")
return
if self.entry_files:
reachable = find_reachable_files_from_entries(self.src_root, py_files)
else:
reachable = find_reachable_files_from_entries(self.src_root, py_files)
print(f"[阶段一] 找到 {len(reachable)} 个可达源文件(从入口遍历)")
for i, src_path in enumerate(sorted(reachable), 1):
rel = os.path.relpath(src_path, self.src_root)
print(f"[{i}/{len(reachable)}] 生成签名: {rel}")
try:
self._process_file_pyi(src_path, rel)
except Exception as e:
print(f" [错误] {e}")
traceback.print_exc()
needed_includes = self._get_needed_include_files(reachable)
self._process_include_py_files_pyi(needed_includes)
struct_names, enum_names, struct_sha1_map, exception_names = self._build_struct_registry()
print(f"\n[阶段一] 结构体注册表: {len(struct_names)} 个类型, {len(enum_names)} 个枚举")
for name in sorted(struct_names):
print(f" {name}")
for i, src_path in enumerate(sorted(reachable), 1):
rel = os.path.relpath(src_path, self.src_root)
try:
self._process_file_stub(src_path, rel, struct_names, enum_names=enum_names, struct_sha1_map=struct_sha1_map, exception_names=exception_names)
except Exception as e:
print(f" [错误] {e}")
traceback.print_exc()
self._process_include_py_files_stub(struct_names, needed_includes, enum_names=enum_names, struct_sha1_map=struct_sha1_map, exception_names=exception_names)
print(f"\n[阶段一完成] 声明接口生成到: {self.temp_dir}")
print(f"SHA1 映射表 ({len(self.sha1_map)} 个文件):")
for sha1, rel in sorted(self.sha1_map.items()):
print(f" {sha1} -> {rel}")
def _collect_include_py_files(self):
include_py_files = []
for includes_dir in self.include_dirs:
if not os.path.isdir(includes_dir):
continue
for root, dirs, files in os.walk(includes_dir):
dirs[:] = [d for d in dirs if not d.startswith('.') and d != '__pycache__']
for fname in files:
if fname.endswith('.py') or fname.endswith('.pyi'):
src_path = os.path.join(root, fname)
rel_from_inc = os.path.relpath(src_path, includes_dir)
include_py_files.append((src_path, rel_from_inc, includes_dir))
return include_py_files
def _process_include_py_files_pyi(self, needed_includes: list = None):
if needed_includes is not None:
include_py_files = needed_includes
else:
include_py_files = self._collect_include_py_files()
if not include_py_files:
return
print(f"\n[阶段一-includes] 处理 {len(include_py_files)} 个被引用的 Python 库文件")
for src_path, rel_from_inc, includes_dir in include_py_files:
module_path = rel_from_inc.replace(os.sep, '.').replace('/', '.')
module_name = os.path.splitext(module_path)[0]
try:
with open(src_path, 'r', encoding='utf-8') as f:
content = f.read()
sha1 = compute_sha1(content)
self.sha1_map[sha1] = f"includes/{rel_from_inc}"
top_module = rel_from_inc.split(os.sep)[0].split('/')[0]
if top_module not in self.include_py_map:
self.include_py_map[top_module] = sha1
self.include_py_map[module_name] = sha1
sig_path = os.path.join(self.temp_dir, f"{sha1}.pyi")
if os.path.isfile(sig_path):
print(f" 缓存命中: {rel_from_inc} -> {sha1}.pyi")
continue
sig_content = PythonToStubConverter.convert(content, module_name)
with open(sig_path, 'w', encoding='utf-8', newline='\n') as f:
f.write(sig_content)
print(f" 生成签名: {rel_from_inc} -> {sha1}.pyi")
except Exception as e:
print(f" [错误] {rel_from_inc}: {e}")
def _collect_typedef_map(self):
import ast as _ast
typedef_map = {}
if not os.path.isdir(self.temp_dir):
return typedef_map
for fname in os.listdir(self.temp_dir):
if not fname.endswith('.pyi'):
continue
pyi_path = os.path.join(self.temp_dir, fname)
try:
with open(pyi_path, 'r', encoding='utf-8') as f:
content = f.read()
tree = _ast.parse(content)
for node in _ast.iter_child_nodes(tree):
if isinstance(node, _ast.AnnAssign) and isinstance(node.target, _ast.Name):
var_name = node.target.id
is_typedef = False
if isinstance(node.annotation, _ast.Attribute) and hasattr(node.annotation, 'attr') and node.annotation.attr == 'CTypedef':
is_typedef = True
elif isinstance(node.annotation, _ast.Name) and node.annotation.id == 'CTypedef':
is_typedef = True
elif isinstance(node.annotation, _ast.BinOp) and isinstance(node.annotation.left, _ast.Attribute) and hasattr(node.annotation.left, 'attr') and node.annotation.left.attr == 'CTypedef':
is_typedef = True
if is_typedef and node.value and var_name not in typedef_map:
typedef_map[var_name] = node.value
except Exception:
pass
return typedef_map
def _process_include_py_files_stub(self, struct_names: set, needed_includes: list = None, enum_names: set = None, struct_sha1_map: dict = None, exception_names: set = None):
if needed_includes is not None:
include_py_files = needed_includes
else:
include_py_files = self._collect_include_py_files()
if not include_py_files:
return
typedef_map = self._collect_typedef_map()
for src_path, rel_from_inc, includes_dir in include_py_files:
module_path = rel_from_inc.replace(os.sep, '.').replace('/', '.')
module_name = os.path.splitext(module_path)[0]
try:
with open(src_path, 'r', encoding='utf-8') as f:
content = f.read()
sha1 = compute_sha1(content)
stub_path = os.path.join(self.temp_dir, f"{sha1}.stub.ll")
if os.path.isfile(stub_path):
print(f" 缓存命中: {rel_from_inc} -> {sha1}.stub.ll")
continue
sig_path = os.path.join(self.temp_dir, f"{sha1}.pyi")
with open(sig_path, 'r', encoding='utf-8') as f:
sig_content = f.read()
self._generate_decl_ll(sig_content, stub_path, src_path, struct_names, enum_names=enum_names, module_sha1=sha1, struct_sha1_map=struct_sha1_map, exception_names=exception_names, typedef_map=typedef_map)
print(f" 生成声明: {rel_from_inc} -> {sha1}.stub.ll")
except Exception as e:
print(f" [错误] {rel_from_inc}: {e}")
def _process_file_pyi(self, src_path: str, rel_path: str):
with open(src_path, 'r', encoding='utf-8') as f:
content = f.read()
sha1 = compute_sha1(content)
self.sha1_map[sha1] = rel_path
sig_path = os.path.join(self.temp_dir, f"{sha1}.pyi")
if os.path.isfile(sig_path):
print(f" -> {sha1}.pyi (缓存)")
return
module_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.')
sig_content = PythonToStubConverter.convert(content, module_name)
with open(sig_path, 'w', encoding='utf-8', newline='\n') as f:
f.write(sig_content)
print(f" -> {sha1}.pyi (签名)")
def _process_file_stub(self, src_path: str, rel_path: str, struct_names: set, enum_names: set = None, struct_sha1_map: dict = None, exception_names: set = None):
with open(src_path, 'r', encoding='utf-8') as f:
content = f.read()
sha1 = compute_sha1(content)
stub_path = os.path.join(self.temp_dir, f"{sha1}.stub.ll")
if os.path.isfile(stub_path):
print(f" -> {sha1}.stub.ll (缓存)")
return
sig_path = os.path.join(self.temp_dir, f"{sha1}.pyi")
with open(sig_path, 'r', encoding='utf-8') as f:
sig_content = f.read()
typedef_map = self._collect_typedef_map()
self._generate_decl_ll(sig_content, stub_path, src_path, struct_names, enum_names=enum_names, module_sha1=sha1, struct_sha1_map=struct_sha1_map, exception_names=exception_names, typedef_map=typedef_map)
print(f" -> {sha1}.stub.ll (声明)")
def _build_struct_registry(self):
struct_names = set()
enum_names = set()
exception_names = set()
struct_sha1_map = {}
valid_sha1_keys = set(self.sha1_map.keys())
for fname in os.listdir(self.temp_dir):
if not fname.endswith('.pyi'):
continue
sha1_key = fname.replace('.pyi', '')
if sha1_key not in valid_sha1_keys:
continue
pyi_path = os.path.join(self.temp_dir, fname)
try:
with open(pyi_path, 'r', encoding='utf-8') as f:
content = f.read()
tree = ast.parse(content)
for node in ast.iter_child_nodes(tree):
if isinstance(node, ast.ClassDef):
is_enum = False
is_exception = False
if node.bases:
for base in node.bases:
if isinstance(base, ast.Attribute) and hasattr(base, 'attr'):
if base.attr in ('CEnum', 'Enum'):
is_enum = True
break
elif base.attr == 'Exception' or base.attr in exception_names:
is_exception = True
break
elif isinstance(base, ast.Name) and hasattr(base, 'id'):
if base.id in ('CEnum', 'Enum'):
is_enum = True
break
elif base.id == 'Exception' or base.id in exception_names:
is_exception = True
break
if is_enum:
enum_names.add(node.name)
elif is_exception:
exception_names.add(node.name)
else:
struct_names.add(node.name)
struct_sha1_map[node.name] = sha1_key
except Exception:
pass
return struct_names, enum_names, struct_sha1_map, exception_names
def _generate_decl_ll(self, pyi_content: str, ll_path: str, src_path: str, struct_names: set = None, enum_names: set = None, module_sha1: str = None, struct_sha1_map: dict = None, exception_names: set = None, typedef_map: dict = None):
try:
decl_gen = DeclarationGenerator(struct_names=struct_names, enum_names=enum_names, module_sha1=module_sha1, target_triple=self.target_triple, target_datalayout=self.target_datalayout, struct_sha1_map=struct_sha1_map, exception_names=exception_names, typedef_map=typedef_map)
decl_ll = decl_gen.generate(pyi_content, src_path)
with open(ll_path, 'w', encoding='utf-8', newline='\n') as f:
f.write(decl_ll)
except Exception as e:
print(f" [警告] .ll 声明生成失败: {e}")
with open(ll_path, 'w', encoding='utf-8') as f:
f.write(f"; declaration for {os.path.basename(src_path)}\n; error: {e}\n")
class DeclarationGenerator:
"""从 .pyi AST 生成纯 LLVM IR 声明(不使用字符串操作)"""
def __init__(self, struct_names=None, enum_names=None, module_sha1=None, target_triple=None, target_datalayout=None, struct_sha1_map=None, exception_names=None, typedef_map=None):
import llvmlite.ir as ir
self.ir = ir
self.module = None
self.builder = None
self._define_constants = {}
self.struct_names = struct_names or set()
self.enum_names = enum_names or set()
self.module_sha1 = module_sha1
self.struct_sha1_map = struct_sha1_map or {}
self.exception_names = exception_names or set()
self.target_triple = target_triple or "x86_64-none-elf"
self.target_datalayout = target_datalayout or "e-m:e-p270:32:32-p271:32:32-p272:64:64-i64:64-f80:128-n8:16:32:64-S128"
self.typedef_map = typedef_map or {}
t.configure_platform(self.target_triple)
def generate(self, pyi_content: str, src_path: str) -> str:
import ast
tree = ast.parse(pyi_content)
self._define_constants = {}
self._pyi_tree = tree
global_typedef_map = self.typedef_map
self.typedef_map = {}
if global_typedef_map:
self.typedef_map.update(global_typedef_map)
for node in ast.iter_child_nodes(tree):
if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
var_name = node.target.id
if node.value and isinstance(node.value, ast.Constant):
self._define_constants[var_name] = node.value.value
elif node.value and isinstance(node.value, ast.Name):
self._define_constants[var_name] = node.value.id
is_typedef = False
if isinstance(node.annotation, ast.Attribute) and hasattr(node.annotation, 'attr') and node.annotation.attr == 'CTypedef':
is_typedef = True
elif isinstance(node.annotation, ast.Name) and node.annotation.id == 'CTypedef':
is_typedef = True
elif isinstance(node.annotation, ast.BinOp) and isinstance(node.annotation.left, ast.Attribute) and hasattr(node.annotation.left, 'attr') and node.annotation.left.attr == 'CTypedef':
is_typedef = True
if is_typedef:
if node.value:
self.typedef_map[var_name] = node.value
elif isinstance(node.annotation, ast.BinOp) and node.annotation.right:
self.typedef_map[var_name] = node.annotation.right
lines = []
lines.append('; ModuleID = "transpyc_decl"')
lines.append(f'target triple = "{self.target_triple}"')
lines.append(f'target datalayout = "{self.target_datalayout}"')
lines.append('')
for node in ast.iter_child_nodes(tree):
if isinstance(node, ast.FunctionDef):
if hasattr(node, 'type_params') and node.type_params:
continue
decl = self._generate_func_decl(node)
if decl:
lines.append(decl)
elif isinstance(node, ast.AnnAssign):
decl = self._generate_global_decl(node)
if decl:
lines.append(decl)
elif isinstance(node, ast.Assign):
decl = self._generate_global_assign_decl(node)
if decl:
lines.append(decl)
elif isinstance(node, ast.ClassDef):
if hasattr(node, 'type_params') and node.type_params:
continue
decls = self._generate_class_decl(node)
lines.extend(decls)
return '\n'.join(lines)
def _generate_func_decl(self, node: ast.FunctionDef) -> str:
"""生成函数声明"""
func_name = node.name
is_export = self._is_export_func(node)
if self.module_sha1 and not is_export:
func_name = f"{self.module_sha1}.{func_name}"
CReturnTypes = []
if node.decorator_list:
for decorator in node.decorator_list:
if isinstance(decorator, ast.Call) and isinstance(decorator.func, ast.Attribute):
if decorator.func.attr == 'CReturn':
for arg in decorator.args:
CReturnTypes.append(arg)
if node.returns:
if isinstance(node.returns, ast.Subscript) and isinstance(node.returns.value, ast.Name) and node.returns.value.id == 'tuple':
slice_node = node.returns.slice
if isinstance(slice_node, ast.Tuple):
for elt in slice_node.elts:
CReturnTypes.append(elt)
else:
CReturnTypes.append(slice_node)
if CReturnTypes:
elem_types = [self._get_type_str(rt, embedded=True) for rt in CReturnTypes]
ret_type = '{ ' + ', '.join(elem_types) + ' }'
else:
ret_type = self._get_type_str(node.returns)
if not ret_type or ret_type == 'void':
ret_type = 'void'
elif ret_type == 'i8*':
if node.returns is None:
ret_type = 'void'
params = []
for arg in node.args.args:
if arg.annotation:
arg_type = self._get_type_str(arg.annotation)
else:
arg_type = 'i8*'
if arg_type and arg_type != 'void':
params.append(arg_type)
param_str = ', '.join(params) if params else ''
if node.args.vararg:
param_str = param_str + ', ...' if param_str else '...'
if func_name[0].isdigit():
return f'declare {ret_type} @"{func_name}"({param_str})'
return f'declare {ret_type} @{func_name}({param_str})'
def _is_export_func(self, node: ast.FunctionDef) -> bool:
"""检查函数是否标记为 CExport"""
if not node.returns:
return False
return self._check_annotation_for_export(node.returns)
def _check_annotation_for_export(self, annotation) -> bool:
"""递归检查类型注解中是否包含 CExport 或 t.State"""
if isinstance(annotation, ast.Attribute):
if hasattr(annotation, 'attr') and annotation.attr in ('CExport', 'State'):
return True
elif isinstance(annotation, ast.BinOp) and isinstance(annotation.op, ast.BitOr):
return self._check_annotation_for_export(annotation.left) or self._check_annotation_for_export(annotation.right)
elif isinstance(annotation, ast.Name):
return annotation.id in ('CExport', 'State')
return False
def _generate_global_decl(self, node: ast.AnnAssign) -> str:
"""生成全局变量声明"""
if not isinstance(node.target, ast.Name):
return None
var_name = node.target.id
if isinstance(node.annotation, ast.Attribute) and hasattr(node.annotation, 'attr') and node.annotation.attr == 'CDefine':
return None
if isinstance(node.annotation, ast.Attribute) and hasattr(node.annotation, 'attr') and node.annotation.attr == 'CTypedef':
return None
if isinstance(node.annotation, ast.Name) and node.annotation.id == 'CTypedef':
return None
if isinstance(node.annotation, ast.BinOp) and isinstance(node.annotation.left, ast.Attribute) and hasattr(node.annotation.left, 'attr') and node.annotation.left.attr == 'CTypedef':
return None
if isinstance(node.annotation, ast.List):
for elt in node.annotation.elts:
if isinstance(elt, ast.Attribute) and hasattr(elt, 'attr') and elt.attr in ('CTypedef', 'Callable'):
return None
if isinstance(elt, ast.Subscript) and isinstance(elt.value, ast.Attribute) and isinstance(elt.value.value, ast.Name) and elt.value.value.id == 't' and elt.value.attr == 'Callable':
return None
var_type = self._get_type_str(node.annotation)
if var_type.startswith('[0 x ') and node.value and isinstance(node.value, ast.List):
elem_type = var_type[5:-1]
actual_count = len(node.value.elts)
if actual_count > 0:
var_type = f'[{actual_count} x {elem_type}]'
return f'@{var_name} = external global {var_type}'
def _generate_global_assign_decl(self, node: ast.Assign) -> str:
"""从无类型标注赋值推断并生成全局变量声明"""
if not node.targets or not isinstance(node.targets[0], ast.Name):
return None
var_name = node.targets[0].id
var_type = self._infer_type(node.value)
if var_type:
return f'@{var_name} = external global {var_type}'
return None
def _generate_class_decl(self, node: ast.ClassDef) -> List[str]:
"""生成结构体/类声明"""
decls = []
class_name = node.name
is_enum = False
is_exception = False
if node.bases:
for base in node.bases:
if isinstance(base, ast.Attribute) and hasattr(base, 'attr'):
if base.attr == 'CEnum' or base.attr == 'Enum':
is_enum = True
break
elif base.attr == 'Exception' or base.attr in self.exception_names:
is_exception = True
break
elif isinstance(base, ast.Name) and hasattr(base, 'id'):
if base.id == 'CEnum' or base.id == 'Enum':
is_enum = True
break
elif base.id == 'Exception' or base.id in self.exception_names:
is_exception = True
break
if is_exception:
return decls
if is_enum:
enum_values = {}
next_value = 0
for item in node.body:
if isinstance(item, ast.AnnAssign) and isinstance(item.target, ast.Name):
var_name = item.target.id
if item.value and isinstance(item.value, ast.Constant) and isinstance(item.value.value, int):
enum_values[var_name] = item.value.value
else:
enum_values[var_name] = next_value
next_value += 1
elif isinstance(item, ast.Assign) and len(item.targets) == 1 and isinstance(item.targets[0], ast.Name):
var_name = item.targets[0].id
if item.value and isinstance(item.value, ast.Constant) and isinstance(item.value.value, int):
enum_values[var_name] = item.value.value
next_value = item.value.value + 1
else:
enum_values[var_name] = next_value
next_value += 1
for var_name, value in enum_values.items():
decls.append(f'@__config_{class_name}_{var_name} = external global i32')
return decls
member_types = []
has_bitfield = False
total_bits = 0
has_methods = any(isinstance(item, ast.FunctionDef) for item in node.body)
is_cvtable = False
is_cpython_object = False
if hasattr(node, 'decorator_list') and node.decorator_list:
for decorator in node.decorator_list:
if isinstance(decorator, ast.Attribute):
if getattr(decorator.value, 'id', None) == 't':
if decorator.attr == 'CVTable':
is_cvtable = True
elif decorator.attr == 'Object':
is_cpython_object = True
elif isinstance(decorator, ast.Name):
if decorator.id == 'CVTable':
is_cvtable = True
elif decorator.id == 'Object':
is_cpython_object = True
if has_methods and not is_cpython_object:
is_cpython_object = True
has_parent_class = False
for base in node.bases:
base_name = None
if isinstance(base, ast.Attribute):
base_name = base.attr
elif isinstance(base, ast.Name):
base_name = base.id
elif isinstance(base, ast.Subscript):
if isinstance(base.value, ast.Attribute):
base_name = base.value.attr
elif isinstance(base.value, ast.Name):
base_name = base.value.id
if base_name and base_name not in ('Object', 'CVTable', 'CEnum', 'Enum', 'CStruct', 'CUnion'):
has_parent_class = True
break
if has_parent_class and not is_cvtable:
is_cvtable = True
base_has_vtable = False
for base in node.bases:
base_name = None
if isinstance(base, ast.Attribute):
base_name = base.attr
elif isinstance(base, ast.Name):
base_name = base.id
elif isinstance(base, ast.Subscript):
if isinstance(base.value, ast.Attribute):
base_name = base.value.attr
elif isinstance(base.value, ast.Name):
base_name = base.value.id
if base_name and base_name not in ('Object', 'CVTable', 'CEnum', 'Enum', 'CStruct', 'CUnion'):
for n in ast.iter_child_nodes(self._pyi_tree):
if isinstance(n, ast.ClassDef) and n.name == base_name:
if hasattr(n, 'decorator_list') and n.decorator_list:
for dec in n.decorator_list:
if isinstance(dec, ast.Attribute) and getattr(dec.value, 'id', None) == 't' and dec.attr == 'CVTable':
base_has_vtable = True
elif isinstance(dec, ast.Name) and dec.id == 'CVTable':
base_has_vtable = True
break
if base_has_vtable:
break
if has_methods and is_cvtable and not base_has_vtable:
member_types.append('i8*')
seen_member_names = set()
for base in node.bases:
base_name = None
if isinstance(base, ast.Attribute):
base_name = base.attr
elif isinstance(base, ast.Name):
base_name = base.id
elif isinstance(base, ast.Subscript):
if isinstance(base.value, ast.Attribute):
base_name = base.value.attr
elif isinstance(base.value, ast.Name):
base_name = base.value.id
if base_name and base_name not in ('Object', 'CVTable', 'CEnum', 'Enum', 'CStruct', 'CUnion'):
base_member_types, base_seen = self._get_inherited_members(base_name)
for mt in base_member_types:
member_types.append(mt)
seen_member_names.update(base_seen)
for item in node.body:
if isinstance(item, ast.AnnAssign) and isinstance(item.target, ast.Name):
seen_member_names.add(item.target.id)
init_members = []
for item in node.body:
if isinstance(item, ast.FunctionDef) and item.name == '__init__':
for stmt in item.body:
if (isinstance(stmt, ast.AnnAssign)
and isinstance(stmt.target, ast.Attribute)
and isinstance(stmt.target.value, ast.Name)
and stmt.target.value.id == 'self'):
init_members.append(stmt)
untyped_self_members = []
for m_name in [m.attr for m in init_members if isinstance(m.target, ast.Attribute)]:
seen_member_names.add(m_name)
for item in node.body:
if isinstance(item, ast.FunctionDef):
if hasattr(item, 'type_params') and item.type_params:
continue
for stmt in item.body:
if (isinstance(stmt, ast.Assign)
and len(stmt.targets) == 1
and isinstance(stmt.targets[0], ast.Attribute)
and isinstance(stmt.targets[0].value, ast.Name)
and stmt.targets[0].value.id == 'self'):
attr_name = stmt.targets[0].attr
if attr_name not in seen_member_names:
seen_member_names.add(attr_name)
untyped_self_members.append(stmt)
for item in list(node.body) + init_members + untyped_self_members:
if isinstance(item, ast.AnnAssign):
bit_width = self._get_bitfield_width(item.annotation)
if bit_width is not None:
has_bitfield = True
total_bits += bit_width
else:
var_type = self._get_type_str(item.annotation, embedded=True)
if (isinstance(item.annotation, ast.Subscript)
and isinstance(item.annotation.value, ast.Name)
and item.annotation.value.id == 'list'):
slice_node = item.annotation.slice
is_dynamic = False
if isinstance(slice_node, ast.Tuple) and len(slice_node.elts) == 2:
count_node = slice_node.elts[1]
if isinstance(count_node, ast.Constant) and count_node.value is None:
is_dynamic = True
elif not (isinstance(count_node, ast.Constant) and isinstance(count_node.value, int) and count_node.value > 0):
if not isinstance(count_node, (ast.Name, ast.BinOp)):
is_dynamic = True
elif not isinstance(slice_node, ast.Tuple):
is_dynamic = True
if is_dynamic:
elem_type = self._get_type_str(slice_node if not isinstance(slice_node, ast.Tuple) else slice_node.elts[0], embedded=True)
init_len = 0
if isinstance(item.value, ast.List):
init_len = len(item.value.elts)
elif isinstance(item.value, ast.Constant) and isinstance(item.value.value, str):
init_len = len(item.value.value) + 1
var_type = f'[{init_len} x {elem_type}]'
member_types.append(var_type)
elif isinstance(item, ast.Assign):
if (len(item.targets) == 1
and isinstance(item.targets[0], ast.Attribute)
and isinstance(item.targets[0].value, ast.Name)
and item.targets[0].value.id == 'self'):
attr_name = item.targets[0].attr
if attr_name in seen_member_names:
continue
InferredType = 'i32'
if item.value and isinstance(item.value, ast.Constant):
if isinstance(item.value.value, float):
InferredType = 'float'
elif isinstance(item.value.value, bool):
InferredType = 'i8'
elif isinstance(item.value.value, str):
InferredType = 'i8*'
member_types.append(InferredType)
else:
member_types.append('i32')
if has_bitfield:
if total_bits <= 8:
member_types.insert(0, 'i8')
elif total_bits <= 16:
member_types.insert(0, 'i16')
elif total_bits <= 32:
member_types.insert(0, 'i32')
else:
member_types.insert(0, 'i64')
struct_type_name = f'%"{self.module_sha1}.{class_name}"' if self.module_sha1 else f'%struct.{class_name}'
struct_decl = f'{struct_type_name} = type {{ {", ".join(member_types)} }}'
decls.append(struct_decl)
if is_cpython_object or is_cvtable:
new_func_name = f'{class_name}.__before_init__'
if self.module_sha1:
new_func_name = f"{self.module_sha1}.{new_func_name}"
if new_func_name[0].isdigit():
new_func_decl = f'declare void @"{new_func_name}"({struct_type_name}*)'
else:
new_func_decl = f'declare void @{new_func_name}({struct_type_name}*)'
decls.append(new_func_decl)
for item in node.body:
if isinstance(item, ast.FunctionDef):
if hasattr(item, 'type_params') and item.type_params:
continue
method_name = f'{class_name}.{item.name}'
if self.module_sha1:
method_name = f"{self.module_sha1}.{method_name}"
ret_type = self._get_type_str(item.returns) if item.returns else 'void'
if not ret_type:
ret_type = 'void'
params = []
for arg_idx, arg in enumerate(item.args.args):
if arg.annotation:
arg_type = self._get_type_str(arg.annotation)
elif arg_idx == 0 and arg.arg == 'self':
arg_type = f'{struct_type_name}*'
else:
arg_type = 'i8*'
params.append(arg_type)
param_str = ', '.join(params) if params else ''
if method_name[0].isdigit():
decls.append(f'declare {ret_type} @"{method_name}"({param_str})')
else:
decls.append(f'declare {ret_type} @{method_name}({param_str})')
return decls
def _get_inherited_members(self, base_name: str):
import ast
member_types = []
seen_names = set()
if not hasattr(self, '_pyi_tree') or self._pyi_tree is None:
return member_types, seen_names
base_node = None
for node in ast.iter_child_nodes(self._pyi_tree):
if isinstance(node, ast.ClassDef) and node.name == base_name:
base_node = node
break
if base_node is None:
return member_types, seen_names
base_decls = self._generate_class_decl(base_node)
for decl in base_decls:
if '= type {' in decl:
type_body = decl.split('= type {')[1].rstrip('}').strip()
if type_body:
for t in type_body.split(','):
t = t.strip()
if t:
member_types.append(t)
break
for item in base_node.body:
if isinstance(item, ast.AnnAssign) and isinstance(item.target, ast.Name):
seen_names.add(item.target.id)
elif isinstance(item, ast.Assign) and len(item.targets) == 1:
if isinstance(item.targets[0], ast.Attribute) and isinstance(item.targets[0].value, ast.Name) and item.targets[0].value.id == 'self':
seen_names.add(item.targets[0].attr)
for item in base_node.body:
if isinstance(item, ast.FunctionDef) and item.name == '__init__':
for stmt in item.body:
if isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Attribute) and isinstance(stmt.target.value, ast.Name) and stmt.target.value.id == 'self':
seen_names.add(stmt.target.attr)
elif isinstance(stmt, ast.Assign) and len(stmt.targets) == 1 and isinstance(stmt.targets[0], ast.Attribute) and isinstance(stmt.targets[0].value, ast.Name) and stmt.targets[0].value.id == 'self':
seen_names.add(stmt.targets[0].attr)
return member_types, seen_names
@staticmethod
def _ctype_name_to_llvm(name: str) -> str:
"""根据 CType 元属性自动推导 LLVM IR 类型
通过 CTypeRegistry.NameToLLVM 查询,利用 CType 的
position/IsSigned/Size 元属性自动推导,无需手动映射表。
"""
from lib.includes.t import CTypeRegistry
result = CTypeRegistry.NameToLLVM(name)
return result
def _get_type_str(self, annotation: ast.AST, embedded: bool = False) -> str:
import ast
from lib.includes.t import CTypeRegistry
if annotation is None:
return 'i8*'
non_struct_types = {'list', 'dict', 'tuple', 'set', 'array'}
if isinstance(annotation, ast.Name):
if annotation.id == 'None':
return 'void'
if annotation.id in ('str', 'bytes'):
return 'i8*'
llvm_type = CTypeRegistry.NameToLLVM(annotation.id)
if llvm_type is not None:
return llvm_type
resolved = CTypeRegistry.ResolveName(annotation.id)
if resolved is not None:
ctype_cls, ptr_level = resolved
base = CTypeRegistry.CTypeToLLVM(ctype_cls)
if ptr_level > 0:
if base == 'void':
return 'i8*'
if '*' in base:
return base
return f'{base}*'
return base
if annotation.id in non_struct_types:
return 'i32'
if annotation.id in self.enum_names:
return 'i32'
if annotation.id in self.struct_names:
sha1 = self.struct_sha1_map.get(annotation.id, self.module_sha1)
sname = f'%"{sha1}.{annotation.id}"' if sha1 else f'%struct.{annotation.id}'
if embedded:
return sname
else:
return f'{sname}*'
if annotation.id in self.typedef_map:
resolved = self._get_type_str(self.typedef_map[annotation.id], embedded=embedded)
return resolved
sha1 = self.struct_sha1_map.get(annotation.id, self.module_sha1)
sname = f'%"{sha1}.{annotation.id}"' if sha1 else f'%struct.{annotation.id}'
return f'{sname}*'
elif isinstance(annotation, ast.Attribute):
attr_name = annotation.attr if hasattr(annotation, 'attr') else ''
module_name = ''
if hasattr(annotation, 'value'):
if isinstance(annotation.value, ast.Name):
module_name = annotation.value.id
elif isinstance(annotation.value, ast.Attribute) and hasattr(annotation.value, 'attr'):
module_name = annotation.value.attr
if (module_name == 't' and attr_name == 'State') or attr_name == 'State':
return ''
if (module_name == 't' and attr_name == 'Callable') or attr_name == 'Callable':
return 'i8*'
llvm_type = CTypeRegistry.NameToLLVM(attr_name)
if llvm_type is not None:
return llvm_type
resolved = CTypeRegistry.ResolveName(attr_name)
if resolved is not None:
ctype_cls, ptr_level = resolved
base = CTypeRegistry.CTypeToLLVM(ctype_cls)
if ptr_level > 0:
if base == 'void':
return 'i8*'
if '*' in base:
return base
return f'{base}*'
return base
if attr_name in self.typedef_map:
return self._get_type_str(self.typedef_map[attr_name], embedded=embedded)
if attr_name in self.enum_names:
return 'i32'
if attr_name in self.struct_names:
sha1 = self.struct_sha1_map.get(attr_name, self.module_sha1)
sname = f'%"{sha1}.{attr_name}"' if sha1 else f'%struct.{attr_name}'
if embedded:
return sname
else:
return f'{sname}*'
if attr_name and attr_name[0].isupper() and attr_name not in non_struct_types:
sha1 = self.struct_sha1_map.get(attr_name, self.module_sha1)
sname = f'%"{sha1}.{attr_name}"' if sha1 else f'%struct.{attr_name}'
return f'{sname}*'
if attr_name and attr_name not in non_struct_types:
sha1 = self.struct_sha1_map.get(attr_name, self.module_sha1)
sname = f'%"{sha1}.{attr_name}"' if sha1 else f'%struct.{attr_name}'
return f'{sname}*'
return 'i32'
elif isinstance(annotation, ast.BinOp):
left_type = self._get_type_str(annotation.left, embedded=embedded)
right_type = self._get_type_str(annotation.right, embedded=embedded)
if left_type in ('', 'void'):
return right_type if right_type not in ('', 'void') else (left_type or right_type)
if right_type in ('', 'void'):
return left_type
if '*' in left_type:
return left_type
if '*' in right_type:
if right_type == 'i8*':
return self._type_to_llvm_ptr(left_type)
return right_type
return left_type
elif isinstance(annotation, ast.Subscript):
base = self._get_type_str(annotation.value)
if base == 'i8*' and isinstance(annotation.slice, ast.Constant):
return f'[{self._get_const_int(annotation.slice)} x i8]'
if isinstance(annotation.slice, ast.Constant) and isinstance(annotation.slice.value, int):
return f'[{annotation.slice.value} x {base}]'
if isinstance(annotation.value, ast.Name) and annotation.value.id == 'list':
slice_node = annotation.slice
if isinstance(slice_node, ast.Tuple) and len(slice_node.elts) == 2:
elem_type = self._get_type_str(slice_node.elts[0], embedded=True)
count_node = slice_node.elts[1]
array_count = self._get_const_int(count_node)
if array_count > 0:
return f'[{array_count} x {elem_type}]'
else:
return f'[0 x {elem_type}]'
elem_type = self._get_type_str(slice_node, embedded=True)
return f'[0 x {elem_type}]'
if isinstance(annotation.value, ast.Name) and annotation.value.id == 'tuple':
slice_node = annotation.slice
if isinstance(slice_node, ast.Tuple):
elem_types = [self._get_type_str(e, embedded=True) for e in slice_node.elts]
else:
elem_types = [self._get_type_str(slice_node, embedded=True)]
return '{ ' + ', '.join(elem_types) + ' }'
return f'{base}'
elif isinstance(annotation, ast.Constant):
if annotation.value is None:
return 'void'
if isinstance(annotation.value, int):
return 'i32'
if isinstance(annotation.value, str):
type_name = annotation.value
if type_name in self.enum_names:
return 'i32'
if type_name in self.struct_names:
sha1 = self.struct_sha1_map.get(type_name, self.module_sha1)
sname = f'%"{sha1}.{type_name}"' if sha1 else f'%struct.{type_name}'
if embedded:
return sname
else:
return f'{sname}*'
llvm_type = CTypeRegistry.NameToLLVM(type_name)
if llvm_type is not None:
return llvm_type
resolved = CTypeRegistry.ResolveName(type_name)
if resolved is not None:
ctype_cls, ptr_level = resolved
base = CTypeRegistry.CTypeToLLVM(ctype_cls)
if ptr_level > 0:
if base == 'void':
return 'i8*'
if '*' in base:
return base
return f'{base}*'
return base
sha1 = self.struct_sha1_map.get(type_name, self.module_sha1)
sname = f'%"{sha1}.{type_name}"' if sha1 else f'%struct.{type_name}'
return f'{sname}*'
return 'i8*'
elif isinstance(annotation, ast.Call):
if isinstance(annotation.func, ast.Name) and annotation.func.id == 'callable':
return 'i8*'
if isinstance(annotation.func, ast.Attribute) and annotation.func.attr == 'Callable':
return 'i8*'
return 'i32'
def _type_to_llvm_ptr(self, type_str: str) -> str:
if type_str.startswith('%struct.') and not type_str.endswith('*'):
return type_str + '*'
if type_str.startswith('%') and not type_str.endswith('*'):
return type_str + '*'
if type_str.endswith('*'):
return type_str
from lib.includes.t import CTypeRegistry
resolved = CTypeRegistry.ResolveName(type_str)
if resolved is not None:
ctype_cls, ptr_level = resolved
llvm_str = CTypeRegistry.CTypeToLLVM(ctype_cls)
if llvm_str:
return f'{llvm_str}{"*" * (ptr_level + 1)}'
cnameres = CTypeRegistry.CNameToClass(type_str)
if cnameres is not None:
llvm_str = CTypeRegistry.CTypeToLLVM(cnameres)
if llvm_str:
return f'{llvm_str}*'
return f'{type_str}*'
def _infer_type(self, value: ast.AST) -> str:
"""从值推断类型"""
import ast
if isinstance(value, ast.Constant):
if isinstance(value.value, int):
return 'i32'
elif isinstance(value.value, float):
return 'double'
elif isinstance(value.value, str):
return 'i8*'
elif isinstance(value.value, bool):
return 'i8'
elif isinstance(value, ast.List):
return 'i8*'
elif isinstance(value, ast.Dict):
return 'i8*'
elif isinstance(value, ast.Name):
return 'i32'
elif isinstance(value, ast.BinOp):
return self._infer_type(value.left)
elif isinstance(value, ast.Call):
return 'i8*'
return 'i32'
def _get_bitfield_width(self, annotation: ast.AST):
import ast
if isinstance(annotation, ast.BinOp) and isinstance(annotation.op, ast.BitOr):
right_width = self._get_bitfield_width(annotation.right)
if right_width is not None:
return right_width
return self._get_bitfield_width(annotation.left)
if isinstance(annotation, ast.Call):
if isinstance(annotation.func, ast.Attribute) and annotation.func.attr == 'Bit':
if annotation.args:
arg = annotation.args[0]
if isinstance(arg, ast.Constant) and isinstance(arg.value, int):
return arg.value
if isinstance(annotation, ast.Subscript):
if isinstance(annotation.value, ast.Attribute) and annotation.value.attr == 'Bit':
if isinstance(annotation.slice, ast.Constant) and isinstance(annotation.slice.value, int):
return annotation.slice.value
return None
def _get_const_int(self, node: ast.AST) -> int:
"""获取常量整数值,支持符号常量和简单表达式"""
import ast
if isinstance(node, ast.Constant) and isinstance(node.value, int):
return node.value
if isinstance(node, ast.Name):
if node.id in self._define_constants:
val = self._define_constants[node.id]
if isinstance(val, int):
return val
if isinstance(node, ast.BinOp):
left_val = self._get_const_int(node.left)
right_val = self._get_const_int(node.right)
if left_val and right_val:
if isinstance(node.op, ast.Add):
return left_val + right_val
if isinstance(node.op, ast.Sub):
return left_val - right_val
if isinstance(node.op, ast.Mult):
return left_val * right_val
if isinstance(node.op, ast.Div):
return left_val // right_val
if isinstance(node.op, ast.FloorDiv):
return left_val // right_val
return 0
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 Projectrans import Phase2Translator, 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=50)}')
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.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):
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_error_stack = 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._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._load_sha1_map()
def _load_sha1_map(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:
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_module_sha1_map = self._build_current_module_sha1_map(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:
pass
if os.path.isfile(out_path):
deps_changed = self._check_deps_changed(sha1, current_module_sha1_map)
if not deps_changed:
print(f"[{i}/{len(py_files)}] 跳过(缓存): {rel} -> {sha1}.ll")
continue
else:
if self._recombine_ll(sha1, current_module_sha1_map):
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_module_sha1_map):
print(f"[{i}/{len(py_files)}] 重组(缓存恢复): {rel} -> {sha1}.ll")
continue
need_translate.append((i, src_path, out_path, sha1, current_module_sha1_map))
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[:2000]}")
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_module_sha1_map=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_module_sha1_map(self, self_sha1: str) -> dict:
"""构建当前模块的依赖 SHA1 映射(与 _translate_file 中逻辑一致)"""
module_sha1_map = {}
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
module_sha1_map[inc_mod_name] = actual_sha1
module_sha1_map[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]
module_sha1_map[parent_pkg] = actual_sha1
continue
module_name = os.path.splitext(rel_path)[0].replace(os.sep, '.').replace('/', '.')
module_sha1_map[module_name] = sha1_key
short_name = module_name.split('.')[-1] if '.' in module_name else module_name
module_sha1_map[short_name] = sha1_key
return module_sha1_map
@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, module_sha1_map: dict):
"""保存依赖指纹到 .deps.json"""
deps_path = os.path.join(self.output_dir, f"{sha1}.deps.json")
deps = {}
for mod_name, dep_sha1 in module_sha1_map.items():
deps[mod_name] = dep_sha1
with open(deps_path, 'w', encoding='utf-8') as f:
json.dump(deps, f)
def _load_deps(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:
pass
return None
def _check_deps_changed(self, sha1: str, current_module_sha1_map: dict) -> bool:
"""检查依赖指纹是否变化"""
saved_deps = self._load_deps(sha1)
if saved_deps is None:
return True
for mod_name, dep_sha1 in current_module_sha1_map.items():
if saved_deps.get(mod_name) != dep_sha1:
return True
for mod_name, dep_sha1 in saved_deps.items():
if current_module_sha1_map.get(mod_name) != dep_sha1:
return True
return False
def _recombine_ll(self, sha1: str, current_module_sha1_map: dict) -> bool:
"""依赖变化时重新组合 .ll更新 .stub.ll 中的 SHA1 前缀,拼接 .stub.ll + .text.ll"""
saved_deps = self._load_deps(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_module_sha1_map.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_module_sha1_map)
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_module_sha1_map: dict = None):
"""翻译单个源文件,嵌入相关 .stub.ll 声明"""
with open(src_path, 'r', encoding='utf-8') as f:
code = f.read()
sha1 = compute_sha1(code)
if prebuilt_module_sha1_map is not None:
module_sha1_map = prebuilt_module_sha1_map
else:
module_sha1_map = self._build_current_module_sha1_map(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._module_sha1_map = module_sha1_map
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:
pass
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)
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__')]
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, '_error_stack', [])
if error_stack:
self._last_error_stack = 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, module_sha1_map)
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._load_deps(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:
pass
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:
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:
pass
# 拓扑排序 (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_module_sha1_map = self._build_current_module_sha1_map(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_module_sha1_map[mod_full] = sha1
include_module_sha1_map[module_name] = sha1
except ValueError:
pass
include_module_sha1_map[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_module_sha1_map=include_module_sha1_map)
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_module_sha1_map)
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_module_sha1_map):
"""编译 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:
pass # 签名生成失败不阻塞编译
# 注册到 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:
pass
# 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:
pass
# 注册到 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:
pass
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._module_sha1_map = {}
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)
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__')]
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:
pass
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:
pass
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:
pass
break
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
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:
cmd = [self.linker_cmd] + self.linker_flags + ['-o', link_output] + all_link_files
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)
def save_sha1_map(temp_dir: str, sha1_map: dict[str, str]):
"""将 SHA1 映射表保存到文件"""
map_path = os.path.join(temp_dir, '_sha1_map.txt')
with open(map_path, 'w', encoding='utf-8') as f:
for sha1, rel in sorted(sha1_map.items()):
f.write(f"{sha1}:{rel}\n")
def load_sha1_map(temp_dir: str) -> dict[str, str]:
"""从文件加载 SHA1 映射表"""
map_path = os.path.join(temp_dir, '_sha1_map.txt')
result = {}
if os.path.exists(map_path):
with open(map_path, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if ':' in line:
sha1, rel = line.split(':', 1)
result[sha1] = rel
return result
def load_project_config(project_file: str) -> dict:
"""从 project.json 加载配置"""
import json
with open(project_file, 'r', encoding='utf-8') as f:
return json.load(f)
def resolve_paths(config: dict, project_file: str) -> dict:
"""将配置中的相对路径转换为绝对路径"""
project_dir = os.path.dirname(os.path.abspath(project_file))
result = dict(config)
def resolve(p):
if p is None:
return None
if os.path.isabs(p):
return p
return os.path.normpath(os.path.join(project_dir, p))
if 'source_dir' in result:
result['source_dir'] = resolve(result.get('source_dir'))
elif 'sources' in result:
resolved_sources = []
for src in result['sources']:
if '*' in src or '?' in src:
import glob as glob_module
for matched in glob_module.glob(os.path.join(project_dir, src), recursive=True):
if os.path.isfile(matched):
resolved_sources.append(matched)
else:
resolved_sources.append(resolve(src))
result['sources'] = [s for s in resolved_sources if os.path.isfile(s)]
result['temp_dir'] = resolve(config.get('temp_dir'))
result['output_dir'] = resolve(config.get('output_dir'))
if 'includes' in result:
result['includes'] = [resolve(inc) for inc in result['includes']]
result['_project_dir'] = project_dir
return result
def main():
import argparse
parser = argparse.ArgumentParser(
description='TransPyC 工程翻译器 - 基于 project.json 的两阶段翻译',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''
示例:
# 从 project.json 读取配置并执行(默认查找当前目录)
python ProjectTrans.py
# 指定 project.json 文件
python ProjectTrans.py --project ./TestProject/project.json
# 使用命令行参数覆盖 project.json 配置
python ProjectTrans.py --src ./src --temp ./decl --output ./out --phase all
# 仅运行阶段一
python ProjectTrans.py --phase 1
# 仅运行阶段二
python ProjectTrans.py --phase 2
# 清理临时目录
python ProjectTrans.py --clean
'''
)
parser.add_argument('--project', help='project.json 路径(默认查找当前目录)')
parser.add_argument('--src', help='源文件目录(覆盖 project.json')
parser.add_argument('--temp', help='声明接口临时目录(覆盖 project.json')
parser.add_argument('--output', help='输出目录(覆盖 project.json')
parser.add_argument('--phase', choices=['1', '2', 'all'], help='阶段: 1=生成声明, 2=翻译+编译, all=全部')
parser.add_argument('--cc', help='LLVM 编译器命令(覆盖 project.json')
parser.add_argument('--clean', action='store_true', help='清理 output 和 temp 目录')
args = parser.parse_args()
project_file = args.project
if not project_file:
candidates = ['project.json', './project.json']
for c in candidates:
if os.path.exists(c):
project_file = c
break
if project_file and os.path.exists(project_file):
print(f"[配置] 读取: {project_file}")
config = load_project_config(project_file)
project_dir = os.path.dirname(os.path.abspath(project_file))
print(f"[配置] 项目目录: {project_dir}")
else:
print("[错误] 未找到 project.json请使用 --project 指定")
sys.exit(1)
resolved = resolve_paths(config, project_file)
src = args.src or resolved.get('source_dir') or resolved.get('sources')
temp_dir = args.temp or resolved.get('temp_dir')
output_dir = args.output or resolved.get('output_dir')
phase = args.phase or 'all'
cc_cmd = args.cc or resolved.get('compiler', {}).get('cmd', 'llc')
cc_flags = resolved.get('compiler', {}).get('flags', ['-filetype=obj'])
slice_level = resolved.get('options', {}).get('slice_level', 3)
project_include_dirs = resolved.get('includes', [])
target_triple = resolved.get('target', {}).get('triple')
target_datalayout = resolved.get('target', {}).get('datalayout')
if isinstance(src, list):
if len(src) == 1 and os.path.isdir(src[0]):
src = src[0]
else:
src = src[0] if len(src) == 1 else os.path.dirname(os.path.commonpath(src)) if all(os.path.exists(p) for p in src) else src[0]
print(f"[配置] 源目录: {src}")
print(f"[配置] 临时目录: {temp_dir}")
print(f"[配置] 输出目录: {output_dir}")
print(f"[配置] 阶段: {phase}")
print(f"[配置] 编译器: {cc_cmd} {' '.join(cc_flags)}")
if target_triple:
print(f"[配置] 目标三元组: {target_triple}")
if target_datalayout:
print(f"[配置] 数据布局: {target_datalayout}")
if args.clean:
if output_dir and os.path.exists(output_dir):
for f in os.listdir(output_dir):
fpath = os.path.join(output_dir, f)
try:
if os.path.isfile(fpath) or os.path.islink(fpath):
os.remove(fpath)
elif os.path.isdir(fpath):
shutil.rmtree(fpath, ignore_errors=True)
except:
pass
print(f"[清理] output: 已清空 {output_dir}")
if temp_dir and os.path.exists(temp_dir):
for f in os.listdir(temp_dir):
fpath = os.path.join(temp_dir, f)
try:
if os.path.isfile(fpath) or os.path.islink(fpath):
os.remove(fpath)
elif os.path.isdir(fpath):
shutil.rmtree(fpath, ignore_errors=True)
except:
pass
print(f"[清理] temp: 已清空 {temp_dir}")
print("[清理] 完成")
if (not phase or phase == 'all') and not args.project:
return
if phase in ('1', 'all'):
if isinstance(src, list):
print("[错误] 阶段一需要指定源目录,不支持文件列表")
sys.exit(1)
all_include_dirs = list(project_include_dirs) if project_include_dirs else []
for d in Phase2Translator.INCLUDE_DIRS:
if os.path.isdir(d):
d = os.path.abspath(d)
if d not in [os.path.abspath(x) for x in all_include_dirs]:
all_include_dirs.append(d)
gen = Phase1Generator(src, temp_dir, include_dirs=all_include_dirs, target_triple=target_triple, target_datalayout=target_datalayout)
gen.run()
save_sha1_map(temp_dir, gen.sha1_map)
if phase in ('2', 'all'):
if not output_dir:
print("[错误] 阶段二需要指定 --output 目录")
sys.exit(1)
if not os.path.exists(temp_dir):
print(f"[错误] 声明目录不存在,请先运行阶段一: {temp_dir}")
sys.exit(1)
if isinstance(src, list):
print("[错误] 阶段二需要指定源目录,不支持文件列表")
sys.exit(1)
linker_cmd = resolved.get('linker', {}).get('cmd')
linker_flags = resolved.get('linker', {}).get('flags', [])
linker_output = resolved.get('linker', {}).get('output')
target_triple = resolved.get('target', {}).get('triple')
target_datalayout = resolved.get('target', {}).get('datalayout')
trans = Phase2Translator(src, temp_dir, output_dir, compile_cmd=cc_cmd, compile_flags=cc_flags, linker_cmd=linker_cmd, linker_flags=linker_flags, linker_output=linker_output, include_dirs=project_include_dirs, target_triple=target_triple, target_datalayout=target_datalayout)
trans.sha1_map = load_sha1_map(temp_dir)
trans.slice_level = slice_level
trans.run()
if __name__ == '__main__':
main()