from __future__ import annotations import os import sys import json import glob as glob_module import argparse import shutil import subprocess from typing import Any from lib.core.VLogger import get_logger as _vlog from lib.Projectrans.Phase1Generator import Phase1Generator from lib.Projectrans.Phase2Translator import Phase2Translator # ============================================================ # 全局 SHA1 映射内存存储器 # # Phase1 生成 sha1_map 后写入此处(save_sha1_map), # Phase2 从此处读取(Load_sha1_map / Phase2Translator._LoadSha1Map)。 # _sha1_map.txt 文件仅作为人类可读输出,不再作为机器分析数据源。 # ============================================================ _g_sha1_map_store: dict[str, str] = {} def get_sha1_map_store() -> dict[str, str]: """返回全局 SHA1 映射内存存储器(供 Phase2Translator 直接访问)""" return _g_sha1_map_store def clear_sha1_map_store() -> None: """清空全局 SHA1 映射内存存储器""" _g_sha1_map_store.clear() def save_sha1_map(temp_dir: str, sha1_map: dict[str, str]) -> None: """将 SHA1 映射表保存到内存存储器,并输出人类可读的 _sha1_map.txt 机器分析使用内存存储器(_g_sha1_map_store), _sha1_map.txt 仅作为人类可读输出,不再被机器读取。 """ # 写入内存存储器(机器分析数据源) _g_sha1_map_store.clear() for sha1, rel in sha1_map.items(): _g_sha1_map_store[sha1] = rel # 写入文件(仅人类可读输出) map_path: str = 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 映射表 temp_dir 参数保留用于签名兼容性,但不再读取文件。 机器分析使用内存存储器(_g_sha1_map_store)。 """ result: dict[str, str] = {} for sha1, rel in _g_sha1_map_store.items(): result[sha1] = rel return result def Load_project_config(project_file: str) -> dict[str, Any]: """从 project.json 加载配置""" with open(project_file, 'r', encoding='utf-8') as f: return json.load(f) def resolve_paths(config: dict[str, Any], project_file: str) -> dict[str, Any]: """将配置中的相对路径转换为绝对路径""" project_dir: str = os.path.dirname(os.path.abspath(project_file)) result: dict[str, Any] = dict(config) def resolve(p: str | None) -> str | None: 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: list[str] = [] for src in result['sources']: if '*' in src or '?' in src: 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)] # 直接使用 output_dir / temp_dir(与新版本 TPV 一致,不再支持 build_dir) result['temp_dir'] = resolve(config.get('temp_dir', './temp')) result['output_dir'] = resolve(config.get('output_dir', './output')) if 'includes' in result: result['includes'] = [resolve(inc) for inc in result['includes']] result['_project_dir'] = project_dir return result def main() -> None: parser: argparse.ArgumentParser = 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 # 编译并立即执行 python Projectrans.py --project ./TestProject/project.json --clean --run # 删除 includes.binary 预编译缓存并重新编译所有 includes python Projectrans.py --project ./TestProject/project.json --rebuild-includes ''' ) 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 目录') parser.add_argument('--run', action='store_true', help='编译成功后立即执行生成的可执行文件') parser.add_argument('--rebuild-includes', action='store_true', help='删除 includes.binary 预编译缓存并重新编译所有 includes') parser.add_argument('--clear-cache', action='store_true', help='清除 .transpyc_cache 全局缓存(.pyi/.stub.ll/.doc.json)') args: argparse.Namespace = parser.parse_args() project_file: str | None = args.project if not project_file: candidates: list[str] = ['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): _vlog().info(f"读取配置文件: {project_file}") config: dict[str, Any] = Load_project_config(project_file) project_dir: str = os.path.dirname(os.path.abspath(project_file)) _vlog().info(f"项目目录: {project_dir}") else: _vlog().error("未找到 project.json,请使用 --project 指定") sys.exit(1) resolved: dict[str, Any] = resolve_paths(config, project_file) src: str | list[str] | None = args.src or resolved.get('source_dir') or resolved.get('sources') temp_dir: str | None = args.temp or resolved.get('temp_dir') output_dir: str | None = args.output or resolved.get('output_dir') phase: str = args.phase or 'all' cc_cmd: str = args.cc or resolved.get('compiler', {}).get('cmd', 'llc') cc_flags: list[str] = resolved.get('compiler', {}).get('flags', ['-filetype=obj']) slice_level: int = resolved.get('options', {}).get('slice_level', 3) project_include_dirs: list[str] = resolved.get('includes', []) target_triple: str | None = resolved.get('target', {}).get('triple') target_datalayout: str | None = 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] _vlog().info(f"源目录: {src}") _vlog().info(f"临时目录: {temp_dir}") _vlog().info(f"输出目录: {output_dir}") _vlog().info(f"阶段: {phase}") _vlog().info(f"编译器: {cc_cmd} {' '.join(cc_flags)}") if target_triple: _vlog().info(f"目标三元组: {target_triple}") if target_datalayout: _vlog().info(f"数据布局: {target_datalayout}") if args.clean: if output_dir and os.path.exists(output_dir): for f in os.listdir(output_dir): fpath: str = 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 _vlog().info(f"output: 已清空 {output_dir}") if temp_dir and os.path.exists(temp_dir): for f in os.listdir(temp_dir): fpath: str = 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 _vlog().info(f"temp: 已清空 {temp_dir}") _vlog().info("清理完成") if (not phase or phase == 'all') and not args.project: return if args.clear_cache: # 清除 .transpyc_cache 全局缓存 cache_dir: str = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), '.transpyc_cache') if os.path.isdir(cache_dir): shutil.rmtree(cache_dir, ignore_errors=True) os.makedirs(cache_dir, exist_ok=True) _vlog().info(f"已清除全局缓存: {cache_dir}") else: _vlog().info(f"全局缓存目录不存在: {cache_dir}") if phase in ('1', 'all'): if isinstance(src, list): _vlog().error("阶段一需要指定源目录,不支持文件列表") sys.exit(1) all_include_dirs: list[str] = 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 = 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: _vlog().error("阶段二需要指定 --output 目录") sys.exit(1) if not os.path.exists(temp_dir): _vlog().error(f"声明目录不存在,请先运行阶段一: {temp_dir}") sys.exit(1) if isinstance(src, list): _vlog().error("阶段二需要指定源目录,不支持文件列表") sys.exit(1) linker_cmd: str | None = resolved.get('linker', {}).get('cmd') linker_flags: list[str] = resolved.get('linker', {}).get('flags', []) linker_output: str | None = resolved.get('linker', {}).get('output') target_triple = resolved.get('target', {}).get('triple') target_datalayout = resolved.get('target', {}).get('datalayout') force_recompile_includes: bool = bool(resolved.get('options', {}).get('force_recompile_includes', False)) if args.rebuild_includes: force_recompile_includes = True all_inc_dirs: list[str] = list(project_include_dirs) + list(Phase2Translator.INCLUDE_DIRS) seen_dirs: set[str] = set() for inc_dir in all_inc_dirs: abs_inc: str = os.path.abspath(inc_dir) if abs_inc in seen_dirs: continue seen_dirs.add(abs_inc) if not os.path.isdir(abs_inc): continue inc_parent: str = os.path.dirname(abs_inc) inc_base: str = os.path.basename(abs_inc) precompiled_dir: str = os.path.join(inc_parent, inc_base + '.binary') if os.path.isdir(precompiled_dir): cleared: int = 0 for fname in os.listdir(precompiled_dir): fpath: str = os.path.join(precompiled_dir, fname) try: if os.path.isfile(fpath) or os.path.islink(fpath): os.remove(fpath) cleared += 1 except: pass _vlog().info(f"includes.binary: 已清空 {precompiled_dir} ({cleared} 个文件)") trans: Phase2Translator = 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, force_recompile_includes=force_recompile_includes) trans.sha1_map = Load_sha1_map(temp_dir) trans.slice_level = slice_level trans.run() if args.run and linker_output and not linker_output.endswith('.bin'): exe_path: str = os.path.join(output_dir, linker_output) if os.path.isfile(exe_path): _vlog().info(f"\n[运行] 执行: {exe_path}") try: result: subprocess.CompletedProcess = subprocess.run([exe_path], cwd=output_dir) _vlog().info(f"[运行] 退出码: {result.returncode}") except Exception as e: _vlog().error(f"[运行] 执行失败: {e}") sys.exit(1) else: _vlog().error(f"[运行] 找不到可执行文件: {exe_path}") sys.exit(1)