zzy
ce8031b21f
- Add support for # (stringify) and ## (concatenation) operators in macro replacement lists - Implement scc_pp_expand_string_unsafe() to process operator tokens during macro expansion - Add helper macros to identify blank, stringify, and concatenation tokens in replacement lists - Include missing headers (ctype.h, string.h) for character handling functions - Update object macro expansion to use new string expansion function instead of simple concatenation - Improve whitespace handling in macro replacement parsing to prevent interference with operator processing
1166 lines
36 KiB
Python
1166 lines
36 KiB
Python
"""cbuild.py - 优化的轻量C构建系统"""
|
||
|
||
from abc import ABC, abstractmethod
|
||
import tomllib
|
||
import subprocess
|
||
from pathlib import Path
|
||
from dataclasses import asdict, dataclass
|
||
from enum import Enum, auto
|
||
import logging
|
||
import hashlib
|
||
from graphlib import TopologicalSorter
|
||
import shutil
|
||
import argparse
|
||
import time
|
||
import sys
|
||
import json
|
||
import re
|
||
import concurrent.futures
|
||
import os
|
||
import locale
|
||
|
||
|
||
class ColorFormatter(logging.Formatter):
|
||
"""颜色格式化器"""
|
||
|
||
COLORS = {
|
||
logging.DEBUG: "\x1b[36m", # 青色
|
||
logging.INFO: "\x1b[32m", # 绿色
|
||
logging.WARNING: "\x1b[33m", # 黄色
|
||
logging.ERROR: "\x1b[31m", # 红色
|
||
logging.CRITICAL: "\x1b[31;4m", # 红色加下划线
|
||
}
|
||
|
||
def format(self, record):
|
||
color = self.COLORS.get(record.levelno, "\x1b[0m")
|
||
reset = "\x1b[0m"
|
||
fmt = f"{color}%(name)s - %(levelname)s:{reset} %(message)s"
|
||
if record.levelno in (logging.DEBUG, logging.CRITICAL):
|
||
fmt += " - (%(filename)s:%(lineno)d)"
|
||
formatter = logging.Formatter(fmt)
|
||
return formatter.format(record)
|
||
|
||
|
||
def setup_logger() -> logging.Logger:
|
||
"""设置日志记录器"""
|
||
inner_logger = logging.getLogger(__name__)
|
||
handler = logging.StreamHandler()
|
||
handler.setFormatter(ColorFormatter())
|
||
inner_logger.addHandler(handler)
|
||
return inner_logger
|
||
|
||
|
||
logger = setup_logger()
|
||
|
||
|
||
# 简单的翻译系统
|
||
class Translator:
|
||
"""简单的翻译器"""
|
||
|
||
def __init__(self, lang=None):
|
||
def detect_language() -> str:
|
||
(name, _) = locale.getlocale()
|
||
if name is None:
|
||
return "en_US"
|
||
lower_lang = name.lower()
|
||
# 检查是否包含中文相关关键词
|
||
chinese_keywords = [
|
||
"chinese",
|
||
"china",
|
||
"zh",
|
||
"cn",
|
||
]
|
||
for keyword in chinese_keywords:
|
||
if keyword in lower_lang:
|
||
return "zh_CN"
|
||
return "en_US"
|
||
|
||
self.lang: str = lang if lang is not None else detect_language()
|
||
self.translations = {
|
||
"zh_CN": {
|
||
"build_complete": "完成 {mode} 目标，耗时 {time} (使用 {jobs} 线程)",
|
||
"clean_directory": "已清理构建目录: {path}",
|
||
"cleaned_files": "清理了 {count} 个文件，总大小 {size}",
|
||
"no_build_dir": "没有找到构建目录，无需清理",
|
||
"no_targets": "没有可运行的目标",
|
||
"running_test": "运行测试: {name}",
|
||
"no_tests": "没有找到匹配的测试",
|
||
"test_passed": " ✓ 测试 {name} 通过",
|
||
"test_failed": " ✗ 测试 {name} 失败",
|
||
"test_timeout": " ✗ 测试 {name} 超时",
|
||
"test_error": " ✗ 测试 {name} 运行异常: {error}",
|
||
"test_results": "测试结果: {passed} 通过, {failed} 失败",
|
||
"project_initialized": "项目 '{name}' 初始化完成！",
|
||
"project_structure": "项目结构:",
|
||
"usage": "使用以下命令构建项目:",
|
||
"build_command": " cbuild build",
|
||
"run_command": " cbuild run",
|
||
"test_command": " cbuild test",
|
||
"dependency_tree": "dependency tree:",
|
||
"skipping": "跳过 {file}",
|
||
"executing_command": "执行命令: {cmd}",
|
||
"command_failed": "命令执行失败 [{code}]: {cmd}",
|
||
"compiling_file": "编译文件 {file} 时出错: {error}",
|
||
"linking_target": "链接目标 {target} 时出错: {error}",
|
||
"regex_filter": "使用正则表达式过滤测试: '{pattern}'，匹配到 {count} 个测试",
|
||
"invalid_regex": "无效的正则表达式 '{pattern}': {error}",
|
||
},
|
||
"en_US": {
|
||
"build_complete": "Finished {mode} target(s) in {time} (using {jobs} threads)",
|
||
"clean_directory": "Cleaned build directory: {path}",
|
||
"cleaned_files": "Cleaned {count} files, total size {size}",
|
||
"no_build_dir": "No build directory found, nothing to clean",
|
||
"no_targets": "No target to run",
|
||
"running_test": "Running test: {name}",
|
||
"no_tests": "No matching tests found",
|
||
"test_passed": " ✓ Test {name} passed",
|
||
"test_failed": " ✗ Test {name} failed",
|
||
"test_timeout": " ✗ Test {name} timed out",
|
||
"test_error": " ✗ Test {name} error: {error}",
|
||
"test_results": "Test results: {passed} passed, {failed} failed",
|
||
"project_initialized": "Project '{name}' initialized!",
|
||
"project_structure": "Project structure:",
|
||
"usage": "Use the following commands to build the project:",
|
||
"build_command": " cbuild build",
|
||
"run_command": " cbuild run",
|
||
"test_command": " cbuild test",
|
||
"dependency_tree": "dependency tree:",
|
||
"skipping": "Skipping {file}",
|
||
"executing_command": "Executing command: {cmd}",
|
||
"command_failed": "Command failed [{code}]: {cmd}",
|
||
"compiling_file": "Error compiling file {file}: {error}",
|
||
"linking_target": "Error linking target {target}: {error}",
|
||
"regex_filter": "Filtering tests with regex: '{pattern}', matched {count} tests",
|
||
"invalid_regex": "Invalid regex '{pattern}': {error}",
|
||
},
|
||
}
|
||
|
||
def translate(self, key, **kwargs):
|
||
"""翻译键值"""
|
||
lang_dict = self.translations.get(self.lang, self.translations["zh_CN"])
|
||
template = lang_dict.get(key)
|
||
if template is None:
|
||
return key
|
||
if kwargs:
|
||
try:
|
||
return template.format(**kwargs)
|
||
except KeyError:
|
||
return template
|
||
return template
|
||
|
||
def set_language(self, lang):
|
||
"""设置语言"""
|
||
if lang in self.translations:
|
||
self.lang = lang
|
||
|
||
|
||
# 全局翻译器实例
|
||
translator = Translator(None)
|
||
|
||
|
||
@dataclass
|
||
class Dependency:
|
||
"""依赖配置"""
|
||
|
||
name: str
|
||
path: str
|
||
version: str = "0.0.0"
|
||
optional: bool = False
|
||
|
||
|
||
@dataclass
|
||
class Feature:
|
||
"""特性配置"""
|
||
|
||
name: str
|
||
description: str = ""
|
||
dependencies: list[str] | None = None
|
||
|
||
def __post_init__(self):
|
||
if self.dependencies is None:
|
||
self.dependencies = []
|
||
|
||
|
||
class PackageConfig:
|
||
"""包配置类"""
|
||
|
||
CONFIG_FILE = "cbuild.toml"
|
||
|
||
def __init__(self, config_path: Path):
|
||
package_file = config_path / self.CONFIG_FILE
|
||
if not package_file.exists():
|
||
raise ValueError(f"配置文件 {package_file} 不存在")
|
||
|
||
with open(package_file, "rb") as f:
|
||
self.config = tomllib.load(f).get("package", {})
|
||
self.path = config_path
|
||
|
||
def __getitem__(self, key):
|
||
return self.config.get(key)
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
"""包的名称"""
|
||
return self.config.get("name", "[unnamed package]")
|
||
|
||
@property
|
||
def version(self) -> str:
|
||
"""包的版本号 eg. 0.1.0"""
|
||
return self.config.get("version", "0.0.0")
|
||
|
||
@property
|
||
def dependencies(self) -> list[Dependency]:
|
||
"""获取当前的直接依赖"""
|
||
deps = []
|
||
for dep in self.config.get("dependencies", []):
|
||
if isinstance(dep, dict):
|
||
deps.append(
|
||
Dependency(
|
||
name=dep.get("name", ""),
|
||
path=dep.get("path", ""),
|
||
version=dep.get("version", "0.0.0"),
|
||
optional=dep.get("optional", False),
|
||
)
|
||
)
|
||
return deps
|
||
|
||
|
||
class DependencyResolver:
|
||
"""依赖解析器"""
|
||
|
||
def __init__(self, root_package: PackageConfig):
|
||
self.root = root_package
|
||
self.deps: dict[str, PackageConfig] = {}
|
||
self.graph = TopologicalSorter()
|
||
self.dep_map: dict[str, list[str]] = {}
|
||
self._resolved = False
|
||
|
||
def resolve(self):
|
||
"""解析所有依赖"""
|
||
if self._resolved:
|
||
return
|
||
|
||
queue = [self.root]
|
||
visited = set()
|
||
|
||
while queue:
|
||
pkg = queue.pop(0)
|
||
name = pkg.name
|
||
|
||
if name in visited:
|
||
continue
|
||
visited.add(name)
|
||
|
||
self.deps[name] = pkg
|
||
self.dep_map.setdefault(name, [])
|
||
|
||
for dep in pkg.dependencies:
|
||
dep_path = Path(pkg.path / dep.path)
|
||
dep_pkg = PackageConfig(dep_path)
|
||
dep_name = dep_pkg.name
|
||
|
||
self.graph.add(name, dep_name)
|
||
self.dep_map[name].append(dep_name)
|
||
|
||
if dep_name not in visited and dep_name not in self.deps:
|
||
queue.append(dep_pkg)
|
||
|
||
self.graph.prepare()
|
||
self._resolved = True
|
||
|
||
def print_tree(self):
|
||
"""打印依赖树"""
|
||
self.resolve()
|
||
print("dependency tree:")
|
||
|
||
def _print(pkg_name, prefix="", is_last=True):
|
||
pkg = self.deps[pkg_name]
|
||
connector = "└── " if is_last else "├── "
|
||
if pkg_name == self.root.name:
|
||
print(f"{pkg.name} v{pkg.version}")
|
||
child_prefix = ""
|
||
else:
|
||
print(f"{prefix}{connector}{pkg.name} v{pkg.version}")
|
||
child_prefix = prefix + (" " if is_last else "│ ")
|
||
|
||
deps = self.dep_map.get(pkg_name, [])
|
||
for i, dep_name in enumerate(deps):
|
||
_print(dep_name, child_prefix, i == len(deps) - 1)
|
||
|
||
_print(self.root.name)
|
||
|
||
def get_all_packages(self) -> list[PackageConfig]:
|
||
"""获取所有包"""
|
||
if not self._resolved:
|
||
self.resolve()
|
||
return list(self.deps.values())
|
||
|
||
|
||
class BuildMode(Enum):
|
||
"""构建模式"""
|
||
|
||
TEST = "test"
|
||
DEV = "dev"
|
||
DEBUG = "debug"
|
||
RELEASE = "release"
|
||
NONE = "none"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BuildPaths:
|
||
"""构建路径"""
|
||
|
||
root: Path
|
||
src: Path
|
||
include: Path
|
||
tests: Path
|
||
output: Path
|
||
objects: Path
|
||
main_bin: Path
|
||
main_lib: Path
|
||
extra_bins: Path
|
||
|
||
@classmethod
|
||
def create(cls, root: Path, mode: BuildMode = BuildMode.DEV):
|
||
"""创建构建路径"""
|
||
src = root / "src"
|
||
output = root / "build"
|
||
|
||
# 根据模式确定输出目录
|
||
if mode == BuildMode.RELEASE:
|
||
output = output / "release"
|
||
elif mode == BuildMode.DEBUG:
|
||
output = output / "debug"
|
||
elif mode == BuildMode.TEST:
|
||
output = output / "test"
|
||
else:
|
||
output = output / "dev"
|
||
|
||
return cls(
|
||
root=root,
|
||
src=src,
|
||
include=root / "include",
|
||
tests=root / "tests",
|
||
output=output,
|
||
objects=output / "obj",
|
||
main_bin=src / "main.c",
|
||
main_lib=src / "lib.c",
|
||
extra_bins=src / "bin",
|
||
)
|
||
|
||
|
||
class TargetType(Enum):
|
||
"""目标类型"""
|
||
|
||
MAIN_EXEC = auto()
|
||
EXEC = auto()
|
||
TEST_EXEC = auto()
|
||
STATIC_LIB = auto()
|
||
|
||
|
||
@dataclass
|
||
class Target:
|
||
"""构建目标"""
|
||
|
||
name: str
|
||
type: TargetType
|
||
source: Path
|
||
object: Path
|
||
output: Path
|
||
|
||
|
||
class BuildContext:
|
||
"""构建上下文"""
|
||
|
||
def __init__(self, package: PackageConfig, mode: BuildMode = BuildMode.DEV):
|
||
self.package = package
|
||
self.paths = BuildPaths.create(package.path, mode)
|
||
self.resolver = DependencyResolver(package)
|
||
|
||
def get_sources(self, pattern: str = "**/*.c") -> list[Path]:
|
||
"""获取源文件"""
|
||
if self.paths.src.exists():
|
||
return list(self.paths.src.glob(pattern))
|
||
return []
|
||
|
||
def get_bin_sources(self) -> list[Path]:
|
||
"""获取额外的可执行文件源"""
|
||
if self.paths.extra_bins.exists():
|
||
return list(self.paths.extra_bins.glob("*.c"))
|
||
return []
|
||
|
||
def get_test_sources(self) -> list[Path]:
|
||
"""获取测试源文件"""
|
||
if self.paths.tests.exists():
|
||
return list(self.paths.tests.glob("test_*.c"))
|
||
return []
|
||
|
||
def get_includes(self) -> list[Path]:
|
||
"""获取包含路径"""
|
||
includes = []
|
||
for pkg in self.resolver.get_all_packages():
|
||
inc_path = pkg.path / "include"
|
||
if inc_path.exists():
|
||
includes.append(inc_path)
|
||
return includes
|
||
|
||
def get_targets(self) -> list[Target]:
|
||
"""获取所有构建目标"""
|
||
targets = []
|
||
ext = ".exe" if sys.platform == "win32" else ""
|
||
|
||
# 主可执行文件
|
||
if self.paths.main_bin.exists():
|
||
targets.append(
|
||
Target(
|
||
name=self.package.name,
|
||
type=TargetType.MAIN_EXEC,
|
||
source=self.paths.main_bin,
|
||
object=self.get_object_path(self.paths.main_bin),
|
||
output=self.paths.output / f"{self.package.name}{ext}",
|
||
)
|
||
)
|
||
|
||
# 静态库
|
||
if self.paths.main_lib.exists():
|
||
targets.append(
|
||
Target(
|
||
name=f"lib{self.package.name}",
|
||
type=TargetType.STATIC_LIB,
|
||
source=self.paths.main_lib,
|
||
object=self.get_object_path(self.paths.main_lib),
|
||
output=self.paths.output / f"lib{self.package.name}.a",
|
||
)
|
||
)
|
||
|
||
# 额外可执行文件
|
||
for src in self.get_bin_sources():
|
||
targets.append(
|
||
Target(
|
||
name=src.stem,
|
||
type=TargetType.EXEC,
|
||
source=src,
|
||
object=self.get_object_path(src),
|
||
output=self.paths.output / f"{src.stem}{ext}",
|
||
)
|
||
)
|
||
|
||
# 测试文件
|
||
for test in self.get_test_sources():
|
||
targets.append(
|
||
Target(
|
||
name=test.stem,
|
||
type=TargetType.TEST_EXEC,
|
||
source=test,
|
||
object=self.get_object_path(test),
|
||
output=self.paths.output / f"{test.stem}{ext}",
|
||
)
|
||
)
|
||
|
||
return targets
|
||
|
||
def get_object_path(self, source: Path) -> Path:
|
||
"""获取对象文件路径"""
|
||
self.paths.objects.mkdir(parents=True, exist_ok=True)
|
||
|
||
try:
|
||
rel_path = source.relative_to(self.paths.root)
|
||
safe_name = str(rel_path).replace("/", "_").replace("\\", "_")
|
||
return self.paths.objects / Path(safe_name).with_suffix(".o")
|
||
except ValueError:
|
||
# 源文件不在项目目录中
|
||
path_hash = hashlib.md5(str(source.absolute()).encode()).hexdigest()[:8]
|
||
return self.paths.objects / f"{source.stem}_{path_hash}.o"
|
||
|
||
def get_compile_objects(self) -> list[tuple[Path, Path]]:
|
||
"""获取需要编译的源文件和目标文件"""
|
||
objects = []
|
||
|
||
# 获取所有包的源文件（排除主文件）
|
||
for pkg in self.resolver.get_all_packages():
|
||
ctx = BuildContext(pkg)
|
||
for src in ctx.get_sources():
|
||
if src in (ctx.paths.main_bin, ctx.paths.main_lib):
|
||
continue
|
||
if ctx.paths.extra_bins.exists() and src.is_relative_to(
|
||
ctx.paths.extra_bins
|
||
):
|
||
continue
|
||
objects.append((src, self.get_object_path(src)))
|
||
|
||
return objects
|
||
|
||
|
||
@dataclass
|
||
class CacheEntry:
|
||
"""缓存条目"""
|
||
|
||
mtime: float
|
||
hash: str
|
||
obj_path: str
|
||
|
||
|
||
class BuildCache:
|
||
"""构建缓存"""
|
||
|
||
def __init__(self, build_path: Path):
|
||
self.cache_dir = build_path / "cache"
|
||
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
||
self.cache_file = self.cache_dir / "cache.json"
|
||
self.cache = {}
|
||
|
||
def load(self):
|
||
"""加载缓存"""
|
||
if not self.cache_file.exists():
|
||
return
|
||
try:
|
||
with open(self.cache_file, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
self.cache = {k: CacheEntry(**v) for k, v in data.items()}
|
||
except OSError, json.JSONDecodeError, TypeError:
|
||
self.cache = {}
|
||
|
||
def save(self):
|
||
"""保存缓存"""
|
||
with open(self.cache_file, "w", encoding="utf-8") as f:
|
||
json.dump({k: asdict(v) for k, v in self.cache.items()}, f)
|
||
|
||
def needs_rebuild(self, source: Path, obj: Path) -> bool:
|
||
"""检查是否需要重新构建"""
|
||
key = str(source)
|
||
|
||
if key not in self.cache:
|
||
return True
|
||
|
||
entry = self.cache[key]
|
||
|
||
# 检查文件是否更新
|
||
if source.stat().st_mtime > entry.mtime:
|
||
return True
|
||
|
||
# 检查对象文件是否存在
|
||
if not obj.exists():
|
||
return True
|
||
|
||
# 检查哈希
|
||
with open(source, "rb") as f:
|
||
current_hash = hashlib.md5(f.read()).hexdigest()
|
||
|
||
return current_hash != entry.hash
|
||
|
||
def update(self, source: Path, obj: Path):
|
||
"""更新缓存"""
|
||
with open(source, "rb") as f:
|
||
file_hash = hashlib.md5(f.read()).hexdigest()
|
||
|
||
self.cache[str(source)] = CacheEntry(
|
||
mtime=source.stat().st_mtime, hash=file_hash, obj_path=str(obj.absolute())
|
||
)
|
||
|
||
|
||
class Compiler(ABC):
|
||
"""编译器抽象类"""
|
||
|
||
def __init__(self):
|
||
self.recorded = []
|
||
self.recording = False
|
||
|
||
def enable_recording(self, enable=True):
|
||
"""启用命令记录"""
|
||
self.recording = enable
|
||
if enable:
|
||
self.recorded.clear()
|
||
|
||
def record(self, cmd):
|
||
"""记录命令"""
|
||
if self.recording:
|
||
self.recorded.append(" ".join(cmd) if isinstance(cmd, list) else cmd)
|
||
|
||
def run(self, cmd):
|
||
"""运行命令"""
|
||
self.record(cmd)
|
||
logger.debug("执行命令: %s", cmd)
|
||
try:
|
||
subprocess.run(cmd, check=True)
|
||
except subprocess.CalledProcessError as e:
|
||
logger.fatal("命令执行失败 [%d]: %s", e.returncode, e.cmd)
|
||
raise
|
||
|
||
@abstractmethod
|
||
def get_flags(self, mode: BuildMode) -> list[str]:
|
||
"""获取编译标志"""
|
||
|
||
@abstractmethod
|
||
def compile(
|
||
self, source: Path, output: Path, includes: list[Path], flags: list[str]
|
||
):
|
||
"""编译源文件"""
|
||
|
||
@abstractmethod
|
||
def link(self, objects: list[Path], output: Path, flags: list[str]):
|
||
"""链接对象文件"""
|
||
|
||
|
||
class GccCompiler(Compiler):
|
||
"""GCC编译器"""
|
||
|
||
def get_flags(self, mode: BuildMode) -> list[str]:
|
||
flags = {
|
||
BuildMode.TEST: [
|
||
"-DTEST_MODE",
|
||
"-O0",
|
||
"-g",
|
||
"--coverage",
|
||
"-fprofile-update=atomic",
|
||
"-Wall",
|
||
"-Wextra",
|
||
],
|
||
BuildMode.DEV: ["-DDEV_MODE", "-O0", "-g", "-Wall", "-Wextra"],
|
||
BuildMode.DEBUG: [
|
||
"-DDEBUG_MODE",
|
||
"-O0",
|
||
"-g",
|
||
"-Wall",
|
||
"-Wextra",
|
||
"-Werror",
|
||
],
|
||
BuildMode.RELEASE: ["-O2", "-flto"],
|
||
BuildMode.NONE: [],
|
||
}
|
||
return flags.get(mode, [])
|
||
|
||
def compile(
|
||
self, source: Path, output: Path, includes: list[Path], flags: list[str]
|
||
):
|
||
cmd = ["gcc"] + flags + ["-c", str(source), "-o", str(output)]
|
||
cmd += [f"-I{inc}" for inc in includes]
|
||
self.run(cmd)
|
||
|
||
def link(self, objects: list[Path], output: Path, flags: list[str]):
|
||
cmd = ["gcc"] + flags + ["-o", str(output)] + [str(obj) for obj in objects]
|
||
self.run(cmd)
|
||
|
||
|
||
class ClangCompiler(Compiler):
|
||
"""Clang编译器"""
|
||
|
||
def get_flags(self, mode: BuildMode) -> list[str]:
|
||
flags = {
|
||
BuildMode.TEST: [
|
||
"-DTEST_MODE",
|
||
"-O0",
|
||
"-g",
|
||
"--coverage",
|
||
"-fprofile-update=atomic",
|
||
"-Wall",
|
||
"-Wextra",
|
||
],
|
||
BuildMode.DEV: ["-DDEV_MODE", "-O0", "-g", "-Wall", "-Wextra"],
|
||
BuildMode.DEBUG: [
|
||
"-DDEBUG_MODE",
|
||
"-O0",
|
||
"-g",
|
||
"-Wall",
|
||
"-Wextra",
|
||
"-Werror",
|
||
],
|
||
BuildMode.RELEASE: ["-O2"],
|
||
BuildMode.NONE: [],
|
||
}
|
||
return flags.get(mode, [])
|
||
|
||
def compile(
|
||
self, source: Path, output: Path, includes: list[Path], flags: list[str]
|
||
):
|
||
cmd = ["clang"] + flags + ["-c", str(source), "-o", str(output)]
|
||
cmd += [f"-I{inc}" for inc in includes]
|
||
self.run(cmd)
|
||
|
||
def link(self, objects: list[Path], output: Path, flags: list[str]):
|
||
cmd = ["clang"] + flags + ["-o", str(output)] + [str(obj) for obj in objects]
|
||
self.run(cmd)
|
||
|
||
|
||
class DummyCompiler(Compiler):
|
||
"""虚拟编译器（用于测试）"""
|
||
|
||
def get_flags(self, mode: BuildMode) -> list[str]:
|
||
return []
|
||
|
||
def compile(
|
||
self, source: Path, output: Path, includes: list[Path], flags: list[str]
|
||
):
|
||
pass
|
||
|
||
def link(self, objects: list[Path], output: Path, flags: list[str]):
|
||
pass
|
||
|
||
|
||
class PackageBuilder:
|
||
"""包构建器"""
|
||
|
||
def __init__(
|
||
self,
|
||
package_path: Path,
|
||
compiler: Compiler,
|
||
mode: BuildMode = BuildMode.DEV,
|
||
jobs: int = 0, # 0 表示自动检测 CPU 核心数
|
||
):
|
||
self.package = PackageConfig(package_path)
|
||
self.context = BuildContext(self.package, mode)
|
||
self.compiler = compiler
|
||
self.cache = BuildCache(self.context.paths.output)
|
||
self.mode = mode
|
||
self.flags = self.compiler.get_flags(mode)
|
||
self.jobs = jobs if jobs > 0 else self._get_cpu_count()
|
||
|
||
def _get_cpu_count(self) -> int:
|
||
"""获取 CPU 核心数"""
|
||
return os.cpu_count() or 1
|
||
|
||
def _compile(self, source: Path, output: Path):
|
||
"""编译单个源文件"""
|
||
if self.cache.needs_rebuild(source, output):
|
||
self.compiler.compile(
|
||
source, output, self.context.get_includes(), self.flags
|
||
)
|
||
self.cache.update(source, output)
|
||
else:
|
||
logger.debug(translator.translate("skipping", file=str(source)))
|
||
|
||
def _parallel_compile(self, compile_tasks: list[tuple[Path, Path]]) -> list[Path]:
|
||
"""并行编译多个源文件"""
|
||
if not compile_tasks:
|
||
return []
|
||
|
||
# 如果只有一个文件或线程数为1，使用串行编译
|
||
if len(compile_tasks) == 1 or self.jobs == 1:
|
||
objects = []
|
||
for src, obj in compile_tasks:
|
||
self._compile(src, obj)
|
||
objects.append(obj)
|
||
return objects
|
||
|
||
# 使用线程池并行编译
|
||
objects = []
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=self.jobs) as executor:
|
||
# 提交所有编译任务
|
||
future_to_task = {}
|
||
for src, obj in compile_tasks:
|
||
future = executor.submit(self._compile, src, obj)
|
||
future_to_task[future] = (src, obj)
|
||
|
||
# 等待所有任务完成
|
||
for future in concurrent.futures.as_completed(future_to_task):
|
||
src, obj = future_to_task[future]
|
||
try:
|
||
future.result() # 检查是否有异常
|
||
objects.append(obj)
|
||
except Exception as e:
|
||
logger.error("编译文件 %s 时出错: %s", src, e)
|
||
raise
|
||
|
||
return objects
|
||
|
||
def _parallel_link(self, link_tasks: list[tuple[Target, list[Path]]]):
|
||
"""并行链接多个目标"""
|
||
if not link_tasks:
|
||
return
|
||
|
||
# 如果只有一个链接任务或线程数为1，使用串行链接
|
||
if len(link_tasks) == 1 or self.jobs == 1:
|
||
for target, objects in link_tasks:
|
||
objects.append(target.object)
|
||
self.compiler.link(objects, target.output, self.flags)
|
||
return
|
||
|
||
# 使用线程池并行链接
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=self.jobs) as executor:
|
||
# 提交所有链接任务
|
||
future_to_task = {}
|
||
for target, objects in link_tasks:
|
||
objects.append(target.object)
|
||
future = executor.submit(
|
||
self.compiler.link, objects, target.output, self.flags
|
||
)
|
||
future_to_task[future] = target
|
||
|
||
# 等待所有任务完成
|
||
for future in concurrent.futures.as_completed(future_to_task):
|
||
target = future_to_task[future]
|
||
try:
|
||
future.result() # 检查是否有异常
|
||
except Exception as e:
|
||
logger.error("链接目标 %s 时出错: %s", target.name, e)
|
||
raise
|
||
|
||
def build(self, target_types: list[TargetType]):
|
||
"""构建包"""
|
||
start = time.time()
|
||
self.cache.load()
|
||
|
||
# 确保输出目录存在
|
||
self.context.paths.output.mkdir(parents=True, exist_ok=True)
|
||
self.context.paths.objects.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 获取需要编译的文件列表
|
||
compile_tasks = self.context.get_compile_objects()
|
||
|
||
# 使用线程池并行编译
|
||
objects = self._parallel_compile(compile_tasks)
|
||
|
||
# 构建目标
|
||
exec_types = {TargetType.MAIN_EXEC, TargetType.EXEC, TargetType.TEST_EXEC}
|
||
link_tasks = []
|
||
|
||
for target in self.context.get_targets():
|
||
if target.type not in target_types:
|
||
continue
|
||
if target.type not in exec_types:
|
||
continue
|
||
|
||
self._compile(target.source, target.object)
|
||
link_tasks.append((target, objects.copy()))
|
||
|
||
# 并行链接
|
||
self._parallel_link(link_tasks)
|
||
|
||
self.cache.save()
|
||
elapsed = time.time() - start
|
||
time_str = f"{elapsed:.2f}s" if elapsed < 60 else f"{elapsed / 60:.1f}m"
|
||
logger.info(
|
||
"完成 %s 目标，耗时 %s (使用 %d 线程)", self.mode.value, time_str, self.jobs
|
||
)
|
||
|
||
def run(self):
|
||
"""运行主程序"""
|
||
targets = [
|
||
t for t in self.context.get_targets() if t.type == TargetType.MAIN_EXEC
|
||
]
|
||
if len(targets) != 1:
|
||
logger.error("没有可运行的目标")
|
||
return
|
||
subprocess.run(targets[0].output, check=False)
|
||
|
||
def clean(self):
|
||
"""清理构建产物"""
|
||
if self.context.paths.output.exists():
|
||
total_size = 0
|
||
file_count = 0
|
||
for file in self.context.paths.output.rglob("*"):
|
||
if file.is_file():
|
||
file_count += 1
|
||
total_size += file.stat().st_size
|
||
shutil.rmtree(self.context.paths.output)
|
||
logger.info("已清理构建目录: %s", self.context.paths.output)
|
||
logger.info(
|
||
"清理了 %d 个文件，总大小 %s", file_count, self._format_size(total_size)
|
||
)
|
||
else:
|
||
logger.info("没有找到构建目录，无需清理")
|
||
|
||
def tree(self):
|
||
"""打印依赖树"""
|
||
self.context.resolver.print_tree()
|
||
|
||
def tests(self, filter_str: str = "", timeout: int = 30):
|
||
"""运行测试"""
|
||
targets = [
|
||
t for t in self.context.get_targets() if t.type == TargetType.TEST_EXEC
|
||
]
|
||
|
||
if filter_str:
|
||
try:
|
||
pattern = re.compile(filter_str)
|
||
targets = [t for t in targets if pattern.search(t.name)]
|
||
logger.debug(
|
||
"使用正则表达式过滤测试: '%s'，匹配到 %d 个测试",
|
||
filter_str,
|
||
len(targets),
|
||
)
|
||
except re.error as e:
|
||
logger.error("无效的正则表达式 '%s': %s", filter_str, e)
|
||
return
|
||
|
||
if not targets:
|
||
logger.info("没有找到匹配的测试")
|
||
return
|
||
|
||
passed = failed = 0
|
||
for target in targets:
|
||
logger.info("运行测试: %s", target.name)
|
||
try:
|
||
result = subprocess.run(target.output, check=True, timeout=timeout)
|
||
if result.returncode == 0:
|
||
print(f" ✓ 测试 {target.name} 通过")
|
||
passed += 1
|
||
else:
|
||
print(f" ✗ 测试 {target.name} 失败")
|
||
failed += 1
|
||
except subprocess.TimeoutExpired:
|
||
print(f" ✗ 测试 {target.name} 超时")
|
||
failed += 1
|
||
except subprocess.SubprocessError as e:
|
||
print(f" ✗ 测试 {target.name} 运行异常: {e}")
|
||
failed += 1
|
||
|
||
print(f"\n测试结果: {passed} 通过, {failed} 失败")
|
||
|
||
def _format_size(self, size_bytes):
|
||
"""格式化文件大小"""
|
||
if size_bytes == 0:
|
||
return "0B"
|
||
units = ["B", "KB", "MB", "GB"]
|
||
i = 0
|
||
while size_bytes >= 1024 and i < len(units) - 1:
|
||
size_bytes /= 1024.0
|
||
i += 1
|
||
return (
|
||
f"{size_bytes:.1f}{units[i]}" if i > 0 else f"{int(size_bytes)}{units[i]}"
|
||
)
|
||
|
||
|
||
def init_project(project_path: Path, name: str | None = None, is_lib: bool = False):
|
||
"""初始化新项目"""
|
||
if name is None:
|
||
name = project_path.name
|
||
name = str(name) # 确保 name 是字符串
|
||
|
||
# 创建目录
|
||
(project_path / "src").mkdir(parents=True, exist_ok=True)
|
||
(project_path / "include").mkdir(parents=True, exist_ok=True)
|
||
(project_path / "tests").mkdir(parents=True, exist_ok=True)
|
||
|
||
# 创建 cbuild.toml
|
||
toml_content = f"""[package]
|
||
name = "{name}"
|
||
version = "0.1.0"
|
||
authors = []
|
||
description = ""
|
||
|
||
# dependencies = []
|
||
# features = {{}}
|
||
# default_features = []
|
||
"""
|
||
(project_path / "cbuild.toml").write_text(toml_content, encoding="utf-8")
|
||
|
||
# 创建源文件
|
||
if is_lib:
|
||
(project_path / "src" / "lib.c").write_text(
|
||
"""#include <stdio.h>
|
||
|
||
void hello_from_lib() {
|
||
printf("Hello from library!\\n");
|
||
}
|
||
""",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
header_content = f"""#ifndef __{name.upper()}_H__
|
||
#define __{name.upper()}_H__
|
||
|
||
void hello_from_lib();
|
||
|
||
#endif
|
||
"""
|
||
(project_path / "include" / f"{name}.h").write_text(
|
||
header_content, encoding="utf-8"
|
||
)
|
||
else:
|
||
(project_path / "src" / "main.c").write_text(
|
||
"""#include <stdio.h>
|
||
|
||
int main() {
|
||
printf("Hello, World!\\n");
|
||
return 0;
|
||
}
|
||
""",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
# 创建测试文件
|
||
(project_path / "tests" / f"test_{name}.c").write_text(
|
||
"""#include <stdio.h>
|
||
|
||
void test_example() {
|
||
printf("Test passed!\\n");
|
||
}
|
||
|
||
int main() {
|
||
test_example();
|
||
return 0;
|
||
}
|
||
""",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
logger.info("项目 '%s' 初始化完成！", name)
|
||
logger.info("项目结构:")
|
||
logger.info(" %s/", project_path)
|
||
logger.info(" ├── cbuild.toml")
|
||
logger.info(" ├── src/")
|
||
logger.info(" │ └── %s", "lib.c" if is_lib else "main.c")
|
||
logger.info(" ├── include/")
|
||
if is_lib:
|
||
logger.info(" │ └── %s.h", name)
|
||
logger.info(" └── tests/")
|
||
logger.info(" └── test_%s.c", name)
|
||
|
||
|
||
def create_parser():
|
||
"""创建命令行解析器"""
|
||
parser = argparse.ArgumentParser(description="轻量C构建系统", prog="cbuild")
|
||
parser.add_argument("--verbose", "-v", action="store_true", help="详细输出")
|
||
parser.add_argument("--path", "-p", default=".", help="项目路径")
|
||
|
||
subparsers = parser.add_subparsers(dest="command", required=True, metavar="COMMAND")
|
||
|
||
def add_common_args(subparser):
|
||
subparser.add_argument(
|
||
"--compiler", "-c", choices=["gcc", "clang"], default="gcc", help="编译器"
|
||
)
|
||
subparser.add_argument("--record", "-r", action="store_true", help="记录命令")
|
||
subparser.add_argument(
|
||
"--jobs", "-j", type=int, default=0, help="并行编译任务数 (0=自动检测)"
|
||
)
|
||
mode_group = subparser.add_mutually_exclusive_group()
|
||
mode_group.add_argument(
|
||
"--debug",
|
||
action="store_const",
|
||
dest="mode",
|
||
const=BuildMode.DEBUG,
|
||
help="调试模式",
|
||
)
|
||
mode_group.add_argument(
|
||
"--release",
|
||
action="store_const",
|
||
dest="mode",
|
||
const=BuildMode.RELEASE,
|
||
help="发布模式",
|
||
)
|
||
mode_group.add_argument(
|
||
"--dev",
|
||
action="store_const",
|
||
dest="mode",
|
||
const=BuildMode.DEV,
|
||
help="开发模式",
|
||
)
|
||
mode_group.add_argument(
|
||
"--test",
|
||
action="store_const",
|
||
dest="mode",
|
||
const=BuildMode.TEST,
|
||
help="测试模式",
|
||
)
|
||
|
||
# build 命令
|
||
build_parser = subparsers.add_parser("build", help="构建项目")
|
||
add_common_args(build_parser)
|
||
build_parser.set_defaults(mode=BuildMode.DEV)
|
||
|
||
# run 命令
|
||
run_parser = subparsers.add_parser("run", help="运行项目")
|
||
add_common_args(run_parser)
|
||
run_parser.set_defaults(mode=BuildMode.DEV)
|
||
|
||
# test 命令
|
||
test_parser = subparsers.add_parser("test", help="运行测试")
|
||
add_common_args(test_parser)
|
||
test_parser.add_argument("--timeout", "-t", type=int, default=3, help="测试时间")
|
||
test_parser.add_argument("--filter", default="", help="过滤测试")
|
||
test_parser.set_defaults(mode=BuildMode.TEST)
|
||
|
||
# clean 命令
|
||
clean_parser = subparsers.add_parser("clean", help="清理构建产物")
|
||
add_common_args(clean_parser)
|
||
clean_parser.add_argument(
|
||
"-a", "--all", action="store_true", default=True, help="清理所有模式"
|
||
)
|
||
|
||
# tree 命令
|
||
tree_parser = subparsers.add_parser("tree", help="显示依赖树")
|
||
add_common_args(tree_parser)
|
||
|
||
# init 命令
|
||
init_parser = subparsers.add_parser("init", help="初始化项目")
|
||
init_parser.add_argument("--name", "-n", help="项目名称")
|
||
init_parser.add_argument("--lib", action="store_true", help="创建库项目")
|
||
|
||
return parser
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
parser = create_parser()
|
||
args = parser.parse_args()
|
||
|
||
# 设置日志级别
|
||
logger.setLevel(logging.DEBUG if args.verbose else logging.INFO)
|
||
|
||
# 处理 init 命令
|
||
if args.command == "init":
|
||
init_project(Path(args.path), args.name, args.lib)
|
||
return
|
||
|
||
# 选择编译器
|
||
compiler_map = {
|
||
"gcc": GccCompiler(),
|
||
"clang": ClangCompiler(),
|
||
}
|
||
compiler = compiler_map.get(args.compiler, GccCompiler())
|
||
|
||
if hasattr(args, "record") and args.record:
|
||
compiler.enable_recording()
|
||
|
||
# 获取构建模式
|
||
mode = getattr(args, "mode", BuildMode.DEV)
|
||
|
||
# 获取线程数
|
||
jobs = getattr(args, "jobs", 0)
|
||
|
||
# 创建构建器
|
||
builder = PackageBuilder(Path(args.path), compiler, mode, jobs)
|
||
|
||
# 执行命令
|
||
if args.command == "build":
|
||
builder.build(
|
||
[
|
||
TargetType.MAIN_EXEC,
|
||
TargetType.EXEC,
|
||
TargetType.TEST_EXEC,
|
||
TargetType.STATIC_LIB,
|
||
]
|
||
)
|
||
elif args.command == "run":
|
||
builder.build([TargetType.MAIN_EXEC])
|
||
builder.run()
|
||
elif args.command == "test":
|
||
builder.build([TargetType.TEST_EXEC])
|
||
builder.tests(args.filter, args.timeout)
|
||
elif args.command == "clean":
|
||
if hasattr(args, "all") and args.all:
|
||
# 清理所有模式
|
||
for clean_mode in [
|
||
BuildMode.DEV,
|
||
BuildMode.DEBUG,
|
||
BuildMode.RELEASE,
|
||
BuildMode.TEST,
|
||
]:
|
||
clean_builder = PackageBuilder(Path(args.path), compiler, clean_mode)
|
||
clean_builder.clean()
|
||
else:
|
||
builder.clean()
|
||
elif args.command == "tree":
|
||
builder.tree()
|
||
|
||
# 输出记录的命令
|
||
if hasattr(args, "record") and args.record:
|
||
cmds = compiler.recorded
|
||
logger.info("记录的命令 (%d 条):", len(cmds))
|
||
for cmd in cmds:
|
||
print(cmd)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|