refactor: 提取 is_vp_or_py_at/JoinPath/read_file_to_buf 公共工具,消除 IncludesScanner/Phase2 重复代码

This commit is contained in:
TermiNexus committed 2026-08-08 00:02:39 +08:00
1 parent e2bdfc1c4e
commit 96e045889e
3 files changed
+155 -75

No files matched your search

+118 -3
View File
@@ -7,6 +7,7 @@ import viperlib
import hashlib
import w32.win32file
import w32.win32base
import w32.fileio as fileio
# ============================================================
@@ -91,6 +92,122 @@ def compute_sha1(pool: memhub.MemBuddy | t.CPtr, content: str) -> str:
return bytes_to_hex(pool, digest, 8, 0, 0)
# ============================================================
# is_vp_or_py_at - 检查路径段是否以 .vp/.py 后缀结尾
# ============================================================
def is_vp_or_py_at(p: str, start: t.CSizeT, plen: t.CSizeT) -> int:
"""检查 p[start, start+plen) 是否以 .vp/.py 后缀结尾(1=是,0=否)"""
if p is None or plen <= 3:
return 0
base: t.CSizeT = start + plen - 3
if p[base] == '.':
if p[base + 1] == 'v' and p[base + 2] == 'p':
return 1
if p[base + 1] == 'p' and p[base + 2] == 'y':
return 1
return 0
# ============================================================
# JoinPath - 拼接两个路径段为 a/b
# ============================================================
def JoinPath(a: str, b: str) -> str:
"""拼接路径段为 "a/b"(stdlib.malloc 分配,调用方负责 free),失败返回 None"""
if a is None or b is None:
return None
a_len: t.CSizeT = string.strlen(a)
b_len: t.CSizeT = string.strlen(b)
total_len: t.CSizeT = a_len + b_len + 2
buf: str = stdlib.malloc(total_len)
if buf is None:
return None
viperlib.snprintf(buf, total_len, "%s/%s", a, b)
return buf
# ============================================================
# read_file_to_buf - 读取整个文件到缓冲区(null 终止)
# ============================================================
def read_file_to_buf(path: str, buf_size: t.CSizeT) -> bytes:
"""读取文件到 stdlib.malloc 缓冲区并保证 null 终止
返回缓冲区(调用方负责 free);打开失败/空文件/分配失败返回 None。
合并自 Phase2/StubMerger/IncludesScanner 的重复"打开+读取+终止"序列。
"""
if path is None or buf_size <= 0:
return None
sf: fileio.File | t.CPtr = fileio.File(path, fileio.MODE.R)
if sf.closed:
return None
buf: bytes = stdlib.malloc(buf_size)
if buf is None:
sf.close()
return None
br: t.CInt64T = sf.read_all(buf, buf_size)
sf.close()
if br <= 0:
stdlib.free(buf)
return None
if br < buf_size:
buf[br] = 0
else:
buf[buf_size - 1] = 0
return buf
# ============================================================
# PathToModuleName - 文件路径转模块名
# ============================================================
def PathToModuleName(path: str, strip_includes_prefix: int) -> str:
"""将文件路径转换为模块名(stdlib.malloc 分配,调用方负责 free)
strip_includes_prefix=1 时跳过 includes/ 前缀(如 includes/ast/__init__.vp → ast)。
特殊处理 __init__.vp:pkg/__init__.vp → pkg(而非 pkg.__init__)。
合并自 StubMerger._PathToModuleName / IncludesScanner 内联模块名计算。
"""
if path is None:
return None
plen: t.CSizeT = string.strlen(path)
# 跳过 includes/ 或 includes\ 前缀(Windows 反斜杠兼容)
prefix_len: t.CSizeT = 9
path_start: t.CSizeT = 0
if strip_includes_prefix != 0 and plen > prefix_len and string.strncmp(path, "includes", 8) == 0:
psep_ch: t.CChar = path[8]
if psep_ch == '/' or psep_ch == '\\':
path_start = prefix_len
plen = plen - prefix_len
# 去掉 .vp 或 .py 后缀(.vp 为 Viper 源,.py 向后兼容)
if is_vp_or_py_at(path, path_start, plen) != 0:
plen -= 3
if plen == 0:
return None
# 检查是否以 /__init__ 或 \__init__ 结尾(包的 __init__.vp)
# 同时支持 / 和 \ 两种分隔符(TPC 写入的路径可能混合使用)
if plen > 9:
tail_off: t.CSizeT = path_start + plen - 9
sep_char: t.CChar = path[tail_off]
if (sep_char == '/' or sep_char == '\\') \
and path[tail_off + 1] == '_' and path[tail_off + 2] == '_' \
and path[tail_off + 3] == 'i' and path[tail_off + 4] == 'n' and path[tail_off + 5] == 'i' \
and path[tail_off + 6] == 't' and path[tail_off + 7] == '_' and path[tail_off + 8] == '_':
# 去掉 分隔符+__init__ 后缀,plen 就是包名部分长度
plen = tail_off - path_start
if plen == 0:
return None
# 复制并替换 / 为 .
result: str = stdlib.malloc(plen + 1)
if result is None:
return None
for ri in range(plen):
ch: t.CChar = path[path_start + ri]
if ch == '/' or ch == '\\':
result[ri] = '.'
else:
result[ri] = ch
result[plen] = '\0'
return result
# ============================================================
# CleanDir - 递归删除目录中所有文件和子目录(保留 dir_path 本身)
#
@@ -131,10 +248,8 @@ def CleanDir(dir_path: str) -> int:
elif fname[1] == '.' and fname[2] == '\0':
fname = None
if fname is not None:
fname_len: t.CSizeT = string.strlen(fname)
full_path: bytes = stdlib.malloc(dir_len + fname_len + 2)
full_path: bytes = JoinPath(dir_path, fname)
if full_path is not None:
viperlib.snprintf(full_path, dir_len + fname_len + 2, "%s/%s", dir_path, fname)
# 检查是否为目录(dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY)
is_dir: t.CUInt32T = find_data.dwFileAttributes & w32.win32file.FILE_ATTRIBUTE_DIRECTORY
if is_dir != 0:
+18 -42
View File
@@ -6,7 +6,6 @@ import stdio
import stdlib
import w32.win32file as win32file
import w32.win32base as win32base
import w32.fileio as fileio
import hashlib
import viperlib
import lib.Projectrans.Utils as Utils
@@ -91,21 +90,10 @@ def add_file_entry(result: ScanResult | t.CPtr,
if result.Count >= result.Capacity:
return 1
# 计算模块名:rel_path 中的 / 替换为 .,去掉 .vp/.py 扩展名
rel_len: t.CSizeT = string.strlen(rel_path)
mod_buf: str = stdlib.malloc(rel_len + 1)
# 计算模块名:rel_path 中 / 替换为 .,去掉 .vp/.py 扩展名(__init__.vp 特判由 Utils 统一处理)
mod_buf: str = Utils.PathToModuleName(rel_path, 0)
if mod_buf is None:
return 1
string.strcpy(mod_buf, rel_path)
# 替换 / 为 .
for i in range(rel_len):
ch: t.CChar = mod_buf[i]
if ch == '/' or ch == '\\':
mod_buf[i] = '.'
# 去掉 .vp 或 .py 扩展名(.vp 为 Viper 源,.py 向后兼容)
if rel_len >= 3:
if (mod_buf[rel_len - 3] == '.' and mod_buf[rel_len - 2] == 'v' and mod_buf[rel_len - 1] == 'p') or (mod_buf[rel_len - 3] == '.' and mod_buf[rel_len - 2] == 'p' and mod_buf[rel_len - 1] == 'y'):
mod_buf[rel_len - 3] = '\0'
# 获取条目地址
entry_size: t.CSizeT = FileEntry.__sizeof__()
@@ -130,7 +118,7 @@ def add_file_entry(result: ScanResult | t.CPtr,
entry.Sha1 = sha1_buf
# 复制相对路径
rel_buf: str = stdlib.malloc(rel_len + 1)
rel_buf: str = stdlib.malloc(string.strlen(rel_path) + 1)
if rel_buf is None:
return 1
string.strcpy(rel_buf, rel_path)
@@ -156,30 +144,17 @@ def compute_file_sha1(pool: memhub.MemBuddy | t.CPtr,
if pool is None or file_path is None:
return None
# 打开文件
f: fileio.File | t.CPtr = fileio.File(file_path, fileio.MODE.R)
if f.closed:
return None
# 分配读取缓冲区(128KB,足够大多数 .vp/.py 文件)
# 使用 stdlib.malloc 避免 mbuddy 池耗尽(每文件 128KB,60+ 文件会耗尽 16MB 池)
# 读取文件内容(stdlib.malloc 分配,避免 mbuddy 池耗尽)
BUF_SIZE: t.CSizeT = 131072
buf: bytes = stdlib.malloc(BUF_SIZE)
buf: bytes = Utils.read_file_to_buf(file_path, BUF_SIZE)
if buf is None:
f.close()
return None
# 读取文件内容
bytes_read: t.CInt64T = f.read_all(buf, BUF_SIZE)
f.close()
if bytes_read <= 0:
stdlib.free(buf)
return None
# 原地去除 \r(CRLF → LF),与 Projectrans.py 文本模式读取一致
buf_len: t.CSizeT = string.strlen(buf)
write_pos: t.CSizeT = 0
read_pos: t.CSizeT = 0
while read_pos < bytes_read:
while read_pos < buf_len:
ch: t.CChar = buf[read_pos]
if ch != '\r':
buf[write_pos] = ch
@@ -266,22 +241,24 @@ def scan_directory_recursive(pool: memhub.MemBuddy | t.CPtr,
# 构造完整路径
fname_len: t.CSizeT = string.strlen(fname)
full_path: bytes = stdlib.malloc(root_len + fname_len + 2)
full_path: bytes = Utils.JoinPath(root_dir, fname)
if full_path is None:
break
viperlib.snprintf(full_path, root_len + fname_len + 2, "%s/%s", root_dir, fname)
# 构造相对路径
prefix_len: t.CSizeT = 0
if rel_prefix is not None:
prefix_len = string.strlen(rel_prefix)
rel_path: bytes = stdlib.malloc(prefix_len + fname_len + 2)
rel_path: bytes = None
if rel_prefix is not None and prefix_len > 0:
rel_path = Utils.JoinPath(rel_prefix, fname)
else:
rel_path = stdlib.malloc(fname_len + 1)
if rel_path is None:
break
string.strcpy(rel_path, fname)
if rel_path is None:
break
if rel_prefix is not None and prefix_len > 0:
viperlib.snprintf(rel_path, prefix_len + fname_len + 2, "%s/%s", rel_prefix, fname)
else:
string.strcpy(rel_path, fname)
if is_dir != 0:
# 递归扫描子目录
@@ -289,9 +266,8 @@ def scan_directory_recursive(pool: memhub.MemBuddy | t.CPtr,
else:
# 检查是否为 .vp 或 .py 文件(.vp 为 Viper 源,.py 向后兼容)
is_py: int = 0
if fname_len >= 3:
if (fname[fname_len - 3] == '.' and fname[fname_len - 2] == 'v' and fname[fname_len - 1] == 'p') or (fname[fname_len - 3] == '.' and fname[fname_len - 2] == 'p' and fname[fname_len - 1] == 'y'):
is_py = 1
if Utils.is_vp_or_py_at(fname, 0, fname_len) != 0:
is_py = 1
if is_py != 0:
# 跳过 __pycache__ 目录下的文件
+19 -30
View File
@@ -111,44 +111,33 @@ def _ScanDirForPyFiles(mb: memhub.MemBuddy | t.CPtr,
is_dir = 1
if is_dir != 0:
# 递归扫描子目录
sub_dir: bytes = stdlib.malloc(dir_len + fname_len + 2)
sub_dir: bytes = Utils.JoinPath(dir_path, fname)
if sub_dir is not None:
viperlib.snprintf(sub_dir, dir_len + fname_len + 2, "%s/%s", dir_path, fname)
file_count = _ScanDirForPyFiles(mb, sub_dir, entries, entry_size, file_count, max_files)
stdlib.free(sub_dir)
else:
# 检查是否是 .vp 或 .py 文件(.vp 为 Viper 源,.py 向后兼容)
if fname_len > 3:
if (fname[fname_len - 3] == '.' and fname[fname_len - 2] == 'v' and fname[fname_len - 1] == 'p') or (fname[fname_len - 3] == '.' and fname[fname_len - 2] == 'p' and fname[fname_len - 1] == 'y'):
if Utils.is_vp_or_py_at(fname, 0, fname_len) != 0:
if file_count < max_files:
full_path: bytes = stdlib.malloc(dir_len + fname_len + 2)
full_path: bytes = Utils.JoinPath(dir_path, fname)
if full_path is not None:
viperlib.snprintf(full_path, dir_len + fname_len + 2, "%s/%s", dir_path, fname)
sf: fileio.File | t.CPtr = fileio.File(full_path, fileio.MODE.R)
if not sf.closed:
sbuf: bytes = stdlib.malloc(SRC_BUF_SIZE)
if sbuf is not None:
br: LONG = sf.read_all(sbuf, SRC_BUF_SIZE)
sf.close()
if br > 0:
if br < SRC_BUF_SIZE:
sbuf[br] = 0
else:
sbuf[SRC_BUF_SIZE - 1] = 0
sha1_val: str = Utils.compute_sha1(mb, sbuf)
if sha1_val is not None:
ea: t.CUInt64T = t.CUInt64T(entries) + file_count * entry_size
ent: SrcFileEntry | t.CPtr = (SrcFileEntry | t.CPtr)(t.CVoid(ea, t.CPtr))
if ent is not None:
ent.Path = full_path
ent.Sha1 = sha1_val
file_count += 1
fb: t.CChar | t.CPtr = VLogger.fmt_buf()
if fb is not None:
viperlib.snprintf(fb, 1024, "%s (sha1=%s)", fname, sha1_val)
VLogger.info(fb, "project")
stdlib.free(sbuf)
# full_path 不释放:ent.Path 引用它
sbuf: bytes = Utils.read_file_to_buf(full_path, SRC_BUF_SIZE)
if sbuf is not None:
sha1_val: str = Utils.compute_sha1(mb, sbuf)
if sha1_val is not None:
ea: t.CUInt64T = t.CUInt64T(entries) + file_count * entry_size
ent: SrcFileEntry | t.CPtr = (SrcFileEntry | t.CPtr)(t.CVoid(ea, t.CPtr))
if ent is not None:
ent.Path = full_path
ent.Sha1 = sha1_val
file_count += 1
fb: t.CChar | t.CPtr = VLogger.fmt_buf()
if fb is not None:
viperlib.snprintf(fb, 1024, "%s (sha1=%s)", fname, sha1_val)
VLogger.info(fb, "project")
stdlib.free(sbuf)
# full_path 不释放:ent.Path 引用它
if w32.win32file.FindNextFileA(handle, find_data) == 0:
break