GVSDS/TransPyV
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

实现了 TPV 的 pyi 生成逻辑（部分），删除了直接拷贝自 CPython（TPC） 版本的死代码

Browse Source
This commit is contained in:
TermiNexus
2026-07-22 19:47:23 +08:00
parent a99d2a5ad9
commit 751dc72d61
19 changed files with 641 additions and 1436 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
App/lib/core/Phase1.py
View File

@@ -19,6 +19,7 @@ import lib.core.Handles.HandlesImports as HandlesImports
import lib.core.IncludesScanner as IncludesScanner
import lib.core.StubMerger as StubMerger
import lib.Projectrans.Config as Config
import lib.StubGen.Converter as StubConverter
# 全局 mbuddy 指针
_mbuddy: memhub.MemManager | t.CPtr
@@ -26,6 +27,10 @@ _mbuddy: memhub.MemManager | t.CPtr
# 源代码缓冲区大小（1MB）
SRC_BUF_SIZE: t.CDefine = 1048576
# pyi 缓冲区大小（256KB）
PYI_BUF_SIZE: t.CSizeT = 262144
# ============================================================
# RunPhase1 - Phase1: 扫描 includes 目录，按需翻译并生成 stub
#
@@ -73,9 +78,6 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
if set_count < 0:
stdio.printf("[Phase1] 无法加载 _sha1_map.txt，跳过 Phase1\n")
return 1
stdio.printf("[Phase1] includes SHA1 集合: %d 个\n", set_count)
stdio.fflush(0)
# 构建模块 SHA1 映射（供跨模块函数调用名混淆使用）
td_len_p1map: t.CSizeT = string.strlen(temp_dir)
p1_sha1_arr: bytes = stdlib.malloc(StubMerger.MAX_INCLUDES * 17)
@@ -98,7 +100,6 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
# 不调用 resolve_annotation_type，避免 list[...] 等不支持的语法触发 crash。
# 生成 .deps.txt 供依赖图按需翻译使用。
# ============================================================
stdio.printf("[Phase1a-pre] 扫描 import 依赖\n")
p1a_registered: int = 0
p1a_skipped: int = 0
p1a_failed: int = 0
@@ -196,8 +197,6 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
stdlib.free(src_buf_a)
stdio.printf("[Phase1a-pre] 完成: 扫描=%d 跳过=%d 失败=%d\n", p1a_registered, p1a_skipped, p1a_failed)
# ============================================================
# 依赖图按需翻译：构建可达 SHA1 集合
#
@@ -216,16 +215,13 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
reachable_count = StubMerger._BuildReachableSha1Set(mb, Config.SourceDir, temp_dir, reachable_set)
if reachable_count > 0:
use_reachable = 1
stdio.printf("[Phase1b] 使用可达 SHA1 集合过滤: %d 个\n", reachable_count)
# 用可达集合重新生成 _sha1_map.txt（按图求索的最终产物）
# Phase B+ 只遍历这些条目，避免编译不需要的 includes（如 asm.py）
StubMerger.WriteIncludesSha1Map(mb, temp_dir, result, reachable_set, reachable_count)
# 重新加载 sha1_set，使后续 Phase 1a-pre/1a/1b 的过滤也使用可达集合
string.memset(sha1_set, 0, StubMerger.MAX_INCLUDES_SHA1 * 17)
set_count = StubMerger._load_includes_sha1_set(mb, temp_dir, sha1_set)
stdio.printf("[Phase1b] _sha1_map.txt 已重写为可达集合: %d 个\n", set_count)
else:
stdio.printf("[Phase1b] 可达 SHA1 集合构建失败，回退到全量集合\n")
stdlib.free(reachable_set)
reachable_set = None
@@ -237,28 +233,18 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
# Phase 1b 全量翻译时 struct 已注册，走 existing 路径只翻译方法体。
# 只处理可达文件，避免翻译不需要的 includes（如 Test 不依赖 ast 模块）。
# ============================================================
stdio.printf("[Phase1a] 注册可达文件 struct/enum/union\n")
stdio.fflush(0)
p1a_reg: int = 0
p1a_skp: int = 0
p1a_fl: int = 0
for i in range(result.Count):
stdio.printf("[Phase1a] iter=%d/%d\n", i, result.Count)
stdio.fflush(0)
entry_addr_r: t.CUInt64T = t.CUInt64T(result.Entries) + i * entry_size
stdio.printf("[Phase1a] iter=%d entry_addr=%d\n", i, entry_addr_r)
stdio.fflush(0)
entry_r: IncludesScanner.FileEntry | t.CPtr = (IncludesScanner.FileEntry | t.CPtr)(t.CVoid(entry_addr_r, t.CPtr))
stdio.printf("[Phase1a] iter=%d entry_r=%d\n", i, entry_r)
stdio.fflush(0)
if entry_r is None:
p1a_fl += 1
continue
sha1_r: str = entry_r.Sha1
stdio.printf("[Phase1a] iter=%d sha1=%s\n", i, sha1_r)
stdio.fflush(0)
if sha1_r is None:
p1a_fl += 1
continue
@@ -275,8 +261,6 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
# 读取文件内容
file_path_r: str = entry_r.Path
stdio.printf("[Phase1a] 处理: sha1=%s path=%s\n", sha1_r, file_path_r)
stdio.fflush(0)
f_r: fileio.File | t.CPtr = fileio.File(file_path_r, fileio.MODE.R)
if f_r.closed:
p1a_fl += 1
@@ -325,67 +309,18 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
HandlesType.set_current_file(file_path_r)
HandlesType.clear_cdefine_constants()
HandlesStruct.reset_visible_structs(mb, 0)
stdio.printf("[Phase1a] 翻译开始: sha1=%s\n", sha1_r)
stdio.fflush(0)
ret_r: int = tr_r.translate(tree_r)
stdio.printf("[Phase1a] 翻译完成: sha1=%s ret=%d\n", sha1_r, ret_r)
stdio.fflush(0)
if ret_r != 0:
p1a_fl += 1
else:
p1a_reg += 1
# 释放 Translator 的 C malloc 资源
stdio.printf("[Phase1a] 释放开始: sha1=%s\n", sha1_r)
stdio.fflush(0)
if tr_r._global_names is not None:
stdio.printf("[Phase1a] free(_global_names) 前: sha1=%s\n", sha1_r)
stdio.fflush(0)
stdlib.free(tr_r._global_names)
stdio.printf("[Phase1a] free(_global_names) 后: sha1=%s\n", sha1_r)
stdio.fflush(0)
if tr_r._nonlocal_names is not None:
stdio.printf("[Phase1a] free(_nonlocal_names) 前: sha1=%s\n", sha1_r)
stdio.fflush(0)
stdlib.free(tr_r._nonlocal_names)
stdio.printf("[Phase1a] free(_nonlocal_names) 后: sha1=%s\n", sha1_r)
stdio.fflush(0)
stdio.printf("[Phase1a] free(src_buf_r) 前: sha1=%s\n", sha1_r)
stdio.fflush(0)
stdlib.free(src_buf_r)
stdio.printf("[Phase1a] free(src_buf_r) 后: sha1=%s\n", sha1_r)
stdio.fflush(0)
stdio.printf("[Phase1a] 循环已退出\n")
stdio.fflush(0)
test_val: int = 42
stdio.printf("[Phase1a] test_val=%d p1a_reg=%d p1a_skp=%d p1a_fl=%d\n", test_val, p1a_reg, p1a_skp, p1a_fl)
stdio.fflush(0)
stdio.printf("[Phase1a] done simple\n")
stdio.fflush(0)
stdio.printf("[Phase1a] done ascii reg=%d skp=%d fl=%d\n", p1a_reg, p1a_skp, p1a_fl)
stdio.fflush(0)
test_ptr: bytes = stdlib.malloc(64)
if test_ptr is not None:
stdio.printf("[Phase1a] malloc test ok ptr=%d\n", test_ptr)
stdio.fflush(0)
stdlib.free(test_ptr)
stdio.printf("[Phase1a] free test ok\n")
stdio.fflush(0)
stdio.printf("[Phase1a] 测试中文=%d\n", 42)
stdio.fflush(0)
stdio.printf("[Phase1a] 完成: 注册=%d 跳过=%d 失败=%d\n", 15, 78, 0)
stdio.fflush(0)
stdio.printf("[Phase1a] before_var_check ok\n")
stdio.fflush(0)
stdio.printf("[Phase1a] 完成: 注册=%d\n", p1a_reg)
stdio.fflush(0)
stdio.printf("[Phase1a] 完成: 注册=%d 跳过=%d\n", p1a_reg, p1a_skp)
stdio.fflush(0)
stdio.printf("[Phase1a] 完成: 注册=%d 跳过=%d 失败=%d\n", p1a_reg, p1a_skp, p1a_fl)
stdio.fflush(0)
stdio.printf("[Phase1a] ALL_DIAG_OK\n")
stdio.fflush(0)
# ============================================================
# Phase 1b: 全量翻译（struct 已注册，走 existing 路径翻译方法体）
@@ -395,25 +330,15 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
skipped: int = 0
failed: int = 0
stdio.printf("[Phase1b] 开始: Count=%d entry_size=%d\n", result.Count, entry_size)
stdio.fflush(0)
for i in range(result.Count):
stdio.printf("[Phase1b] iter=%d/%d\n", i, result.Count)
stdio.fflush(0)
# 获取 entry
entry_addr: t.CUInt64T = t.CUInt64T(result.Entries) + i * entry_size
stdio.printf("[Phase1b] iter=%d entry_addr=%d\n", i, entry_addr)
stdio.fflush(0)
entry: IncludesScanner.FileEntry | t.CPtr = (IncludesScanner.FileEntry | t.CPtr)(t.CVoid(entry_addr, t.CPtr))
stdio.printf("[Phase1b] iter=%d entry=%d\n", i, entry)
stdio.fflush(0)
if entry is None:
failed += 1
continue
sha1: str = entry.Sha1
stdio.printf("[Phase1b] iter=%d sha1=%s\n", i, sha1)
stdio.fflush(0)
if sha1 is None:
failed += 1
continue
@@ -424,41 +349,21 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
in_set = StubMerger._is_in_sha1_set(sha1, reachable_set, reachable_count)
else:
in_set = StubMerger._is_in_sha1_set(sha1, sha1_set, set_count)
stdio.printf("[Phase1b] iter=%d in_set=%d use_reachable=%d\n", i, in_set, use_reachable)
stdio.fflush(0)
if in_set == 0:
skipped += 1
continue
# 构造 stub 路径: {temp_dir}/{sha1}.stub.ll
stdio.printf("[Phase1b] iter=%d strlen 前\n", i)
stdio.fflush(0)
dir_len: t.CSizeT = string.strlen(temp_dir)
sha1_len: t.CSizeT = string.strlen(sha1)
stdio.printf("[Phase1b] iter=%d strlen 后 dir_len=%d sha1_len=%d\n", i, dir_len, sha1_len)
stdio.fflush(0)
stub_path: bytes = stdlib.malloc(dir_len + sha1_len + 16)
if stub_path is None:
failed += 1
continue
stdio.printf("[Phase1b] iter=%d snprintf 前 stub_path=%d\n", i, stub_path)
stdio.fflush(0)
viperlib.snprintf(stub_path, dir_len + sha1_len + 16, "%s/%s.stub.ll", temp_dir, sha1)
stdio.printf("[Phase1b] iter=%d snprintf 后\n", i)
stdio.fflush(0)
# 检查 stub 是否已存在（按需翻译：跳过已存在的）
stdio.printf("[Phase1b] iter=%d File() 前 stub_path=%d\n", i, stub_path)
stdio.fflush(0)
sf: fileio.File | t.CPtr = fileio.File(stub_path, fileio.MODE.R)
stdio.printf("[Phase1b] iter=%d File() 后 sf=%d\n", i, sf)
stdio.fflush(0)
if sf is None:
stdio.printf("[Phase1b] iter=%d sf is None, 翻译\n", i)
stdio.fflush(0)
else:
stdio.printf("[Phase1b] iter=%d sf.closed=%d\n", i, sf.closed)
stdio.fflush(0)
if not sf.closed:
sf.close()
# 检查 text.ll 是否也存在（虚表扫描需要 text.ll）
@@ -473,67 +378,37 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
continue
stdlib.free(text_path)
# text.ll 不存在，需要重新翻译
stdio.printf("[Phase1b] iter=%d RelPath 前\n", i)
stdio.fflush(0)
rp: str = entry.RelPath
stdio.printf("[Phase1b] iter=%d RelPath=%s\n", i, rp)
stdio.fflush(0)
stdio.printf("[Phase1] text.ll 不存在，重新翻译: %s (sha1=%s)\n", rp, sha1)
else:
# stub 不存在，需要翻译
stdio.printf("[Phase1b] iter=%d else 分支 RelPath 前\n", i)
stdio.fflush(0)
rp: str = entry.RelPath
stdio.printf("[Phase1b] iter=%d else RelPath=%s\n", i, rp)
stdio.fflush(0)
stdio.printf("[Phase1] 翻译: %s (sha1=%s)\n", rp, sha1)
stdio.fflush(0)
stdio.printf("[Phase1b] iter=%d free(stub_path) 前\n", i)
stdio.fflush(0)
stdlib.free(stub_path)
stdio.printf("[Phase1b] iter=%d free(stub_path) 后\n", i)
stdio.fflush(0)
# 读取文件内容
stdio.printf("[Phase1b] iter=%d entry.Path 前\n", i)
stdio.fflush(0)
file_path: str = entry.Path
stdio.printf("[Phase1b] iter=%d file_path=%s\n", i, file_path)
stdio.fflush(0)
stdio.printf("[Phase1b] iter=%d File(file_path) 前\n", i)
stdio.fflush(0)
f: fileio.File | t.CPtr = fileio.File(file_path, fileio.MODE.R)
stdio.printf("[Phase1b] iter=%d File(file_path) 后 f=%d\n", i, f)
stdio.fflush(0)
if f is None:
stdio.printf("[Phase1] 无法打开(None): %s\n", file_path)
stdio.fflush(0)
failed += 1
continue
stdio.printf("[Phase1b] iter=%d f.closed=%d\n", i, f.closed)
stdio.fflush(0)
if f.closed:
stdio.printf("[Phase1] 无法打开: %s\n", file_path)
stdio.fflush(0)
failed += 1
continue
stdio.printf("[Phase1b] iter=%d malloc(src_buf) 前\n", i)
stdio.fflush(0)
src_buf: bytes = stdlib.malloc(SRC_BUF_SIZE)
stdio.printf("[Phase1b] iter=%d malloc(src_buf) 后=%d\n", i, src_buf)
stdio.fflush(0)
if src_buf is None:
f.close()
failed += 1
continue
stdio.printf("[Phase1b] iter=%d read_all 前\n", i)
stdio.fflush(0)
bytes_read: LONG = f.read_all(src_buf, SRC_BUF_SIZE)
stdio.printf("[Phase1b] iter=%d read_all 后=%d\n", i, bytes_read)
stdio.fflush(0)
f.close()
if bytes_read <= 0:
stdio.printf("[Phase1] 读取失败: %s\n", file_path)
@@ -546,34 +421,15 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
else:
src_buf[SRC_BUF_SIZE - 1] = 0
stdio.printf("[Phase1b] iter=%d AST 前\n", i)
stdio.fflush(0)
# 解析 AST
stdio.printf("[Phase1b] iter=%d new_lexer 前\n", i)
stdio.fflush(0)
lx: ast.Lexer | t.CPtr = ast.new_lexer(mb)
stdio.printf("[Phase1b] iter=%d new_lexer 后=%d\n", i, lx)
stdio.fflush(0)
if lx is None:
stdlib.free(src_buf)
failed += 1
continue
stdio.printf("[Phase1b] iter=%d _lexer_init 前\n", i)
stdio.fflush(0)
ast._lexer_init(lx, src_buf, mb)
stdio.printf("[Phase1b] iter=%d _lexer_init 后\n", i)
stdio.fflush(0)
stdio.printf("[Phase1b] iter=%d tokenize 前\n", i)
stdio.fflush(0)
tokens: ast.Token | t.CPtr = ast.tokenize(lx)
stdio.printf("[Phase1b] iter=%d tokenize 后=%d\n", i, tokens)
stdio.fflush(0)
stdio.printf("[Phase1b] iter=%d parse_tokens 前\n", i)
stdio.fflush(0)
tree: ast.AST | t.CPtr = ast.parse_tokens(mb, tokens)
stdio.printf("[Phase1b] iter=%d parse_tokens 后=%d\n", i, tree)
stdio.fflush(0)
if tree is None:
stdio.printf("[Phase1] AST 解析失败: %s\n", file_path)
stdio.fflush(0)
@@ -581,32 +437,33 @@ def RunPhase1(mb: memhub.MemBuddy | t.CPtr, includes_dir: str, temp_dir: str,
failed += 1
continue
# 生成 .pyi 存根文件（直接遍历 AST，不依赖 PythonToStubConverter）
pyi_buf: bytes = stdlib.malloc(PYI_BUF_SIZE)
if pyi_buf is not None:
pyi_pos: t.CSizeT = StubConverter._GeneratePyiFromAst(mb, tree, entry.RelPath, pyi_buf, PYI_BUF_SIZE)
if pyi_pos > 0:
pyi_path: bytes = stdlib.malloc(dir_len + 32)
if pyi_path is not None:
viperlib.snprintf(pyi_path, dir_len + 32, "%s/%s.pyi", temp_dir, sha1)
pf: fileio.File | t.CPtr = fileio.File(pyi_path, fileio.MODE.W)
if not pf.closed:
pf.write(pyi_buf, pyi_pos)
pf.close()
stdlib.free(pyi_path)
stdlib.free(pyi_buf)
# 翻译 AST → LLVM IR
stdio.printf("[Phase1b] iter=%d Translator() 前\n", i)
stdio.fflush(0)
tr: HandlesTranslator.Translator | t.CPtr = HandlesTranslator.Translator()
stdio.printf("[Phase1b] iter=%d Translator() 后=%d\n", i, tr)
stdio.fflush(0)
if tr is None:
stdlib.free(src_buf)
failed += 1
continue
stdio.printf("[Phase1b] iter=%d ModuleSha1 前\n", i)
stdio.fflush(0)
tr.ModuleSha1 = sha1
stdio.printf("[Phase1b] iter=%d CurrentPackage 前\n", i)
stdio.fflush(0)
tr.CurrentPackage = HandlesImports.compute_package_from_relpath(mb, entry.RelPath)
stdio.printf("[Phase1b] iter=%d set_current_file 前\n", i)
stdio.fflush(0)
HandlesType.set_current_file(file_path)
HandlesType.clear_cdefine_constants()
HandlesStruct.reset_visible_structs(mb, 0)
stdio.printf("[Phase1b] iter=%d translate 前\n", i)
stdio.fflush(0)
ret: int = tr.translate(tree)
stdio.printf("[Phase1b] iter=%d translate 后=%d\n", i, ret)
stdio.fflush(0)
if ret != 0:
stdio.printf("[Phase1] 翻译失败: %s\n", file_path)
stdlib.free(src_buf)