mirror of
https://github.com/MemeMeow-Studio/MemeMeow.git
synced 2026-10-16 20:31:33 +08:00
infstellar
bd37ae07df
- Refactor collection renaming logic to maintain state integrity. - Update UI elements for better accessibility and user experience in CollectionsWorkspace. - Implement responsive design for collection details and asset layout. - Add comprehensive tests for CollectionsWorkspace and LibraryWorkspace components. - Introduce sync_env script for managing .env files with safety and structure. - Create tests for sync_env functionality to ensure correct behavior and error handling.
232 lines
9.1 KiB
Python
232 lines
9.1 KiB
Python
"""项目维护脚本：按 ``.env.example`` 的结构同步本地 ``.env``。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import re
|
||
import stat
|
||
import tempfile
|
||
from collections.abc import Sequence
|
||
from io import StringIO
|
||
from pathlib import Path
|
||
|
||
from dotenv.parser import Binding, parse_stream
|
||
|
||
|
||
def _parse_bindings(content: str, source: Path) -> list[Binding]:
|
||
"""解析 dotenv 文本并拒绝语法错误，供合并前的数据校验调用。"""
|
||
bindings = list(parse_stream(StringIO(content)))
|
||
errors = [str(binding.original.line) for binding in bindings if binding.error]
|
||
if errors:
|
||
lines = ", ".join(errors)
|
||
raise ValueError(f"{source} 存在无法解析的 dotenv 内容（起始行：{lines}）")
|
||
return bindings
|
||
|
||
|
||
def _assignment_parts(binding: Binding) -> tuple[str, str, str] | None:
|
||
"""拆出赋值语句的前缀、原始值和后缀，避免改变引号及变量插值语义。"""
|
||
text = binding.original.string
|
||
quote: str | None = None
|
||
equals_at: int | None = None
|
||
for index, character in enumerate(text):
|
||
if character in {"'", '"'}:
|
||
escaped = index > 0 and text[index - 1] == "\\"
|
||
if not escaped and (quote is None or quote == character):
|
||
quote = None if quote == character else character
|
||
elif character == "=" and quote is None:
|
||
equals_at = index
|
||
break
|
||
elif character in {"\r", "\n"} and quote is None and text[:index].strip():
|
||
break
|
||
if equals_at is None:
|
||
return None
|
||
|
||
value_start = equals_at + 1
|
||
while value_start < len(text) and text[value_start].isspace() and text[value_start] not in {"\r", "\n"}:
|
||
value_start += 1
|
||
|
||
if value_start < len(text) and text[value_start] in {"'", '"'}:
|
||
delimiter = text[value_start]
|
||
value_end = value_start + 1
|
||
while value_end < len(text):
|
||
if text[value_end] == delimiter:
|
||
backslashes = 0
|
||
cursor = value_end - 1
|
||
while cursor >= value_start and text[cursor] == "\\":
|
||
backslashes += 1
|
||
cursor -= 1
|
||
if backslashes % 2 == 0:
|
||
value_end += 1
|
||
break
|
||
value_end += 1
|
||
else:
|
||
line_end = len(text)
|
||
for newline in ("\r", "\n"):
|
||
position = text.find(newline, value_start)
|
||
if position >= 0:
|
||
line_end = min(line_end, position)
|
||
unquoted = text[value_start:line_end]
|
||
comment = re.search(r"[^\S\r\n]+#", unquoted)
|
||
value_end = value_start + (comment.start() if comment else len(unquoted.rstrip()))
|
||
return text[:value_start], text[value_start:value_end], text[value_end:]
|
||
|
||
|
||
def _binding_head(binding: Binding) -> tuple[str, str]:
|
||
"""拆出模板字段名及其后缀，供无等号声明和普通赋值相互转换。"""
|
||
match = re.match(r"\s*(?:export[^\S\r\n]+)?(?:'[^']+'|[^=#\s]+)", binding.original.string)
|
||
if match is None:
|
||
raise ValueError(f"无法定位 dotenv 字段：{binding.key}")
|
||
return binding.original.string[: match.end()], binding.original.string[match.end() :]
|
||
|
||
|
||
def _without_leading_blank_lines(text: str) -> str:
|
||
"""移除解析器归入字段的前置空行，供模板外字段统一追加到文件尾部。"""
|
||
return re.sub(r"\A(?:[^\S\r\n]*(?:\r\n|\n|\r))+", "", text)
|
||
|
||
|
||
def _newline_for(content: str) -> str:
|
||
"""选择模板使用的换行符，供追加字段时保持文件风格一致。"""
|
||
return "\r\n" if "\r\n" in content else "\n"
|
||
|
||
|
||
def _read_text(path: Path) -> str:
|
||
"""读取 dotenv 原文且不转换换行符，供同步过程保留模板的文件风格。"""
|
||
with path.open("r", encoding="utf-8", newline="") as handle:
|
||
return handle.read()
|
||
|
||
|
||
def merge_env_text(example: str, current: str, *, example_path: Path, env_path: Path) -> str:
|
||
"""以模板结构合并 dotenv 文本，返回可直接写入目标文件的完整内容。"""
|
||
example_bindings = _parse_bindings(example, example_path)
|
||
current_bindings = _parse_bindings(current, env_path)
|
||
|
||
current_by_key: dict[str, Binding] = {}
|
||
for binding in current_bindings:
|
||
if binding.key is not None:
|
||
# 字典保留字段首次出现的位置，同时让最后一次定义提供最终生效值。
|
||
current_by_key[binding.key] = binding
|
||
|
||
template_keys = {binding.key for binding in example_bindings if binding.key is not None}
|
||
output: list[str] = []
|
||
for binding in example_bindings:
|
||
source = current_by_key.get(binding.key) if binding.key is not None else None
|
||
if source is None:
|
||
output.append(binding.original.string)
|
||
continue
|
||
|
||
template_parts = _assignment_parts(binding)
|
||
source_parts = _assignment_parts(source)
|
||
if template_parts is None and source_parts is None:
|
||
output.append(binding.original.string)
|
||
continue
|
||
if template_parts is None:
|
||
head, suffix = _binding_head(binding)
|
||
output.append(f"{head}={source_parts[1]}{suffix}")
|
||
continue
|
||
prefix, _, suffix = template_parts
|
||
if source_parts is None:
|
||
head, _ = _binding_head(binding)
|
||
output.append(f"{head}{suffix}")
|
||
else:
|
||
output.append(f"{prefix}{source_parts[1]}{suffix}")
|
||
|
||
extras = [
|
||
_without_leading_blank_lines(binding.original.string)
|
||
for key, binding in current_by_key.items()
|
||
if key not in template_keys
|
||
]
|
||
if not extras:
|
||
return "".join(output)
|
||
|
||
newline = _newline_for(example)
|
||
merged = "".join(output)
|
||
if merged and not merged.endswith(("\n", "\r")):
|
||
merged += newline
|
||
if merged and not merged.endswith(newline * 2):
|
||
merged += newline
|
||
for extra in extras:
|
||
merged += extra
|
||
if not merged.endswith(("\n", "\r")):
|
||
merged += newline
|
||
return merged
|
||
|
||
|
||
def _validate_target(target: Path) -> None:
|
||
"""确认写入目标不会跟随符号链接或覆盖非普通文件。"""
|
||
if target.is_symlink():
|
||
raise ValueError(f"拒绝写入符号链接：{target}")
|
||
if target.exists() and not target.is_file():
|
||
raise ValueError(f"写入目标不是普通文件：{target}")
|
||
for parent in (target.parent, *target.parent.parents):
|
||
if parent == parent.parent:
|
||
break
|
||
if parent.is_symlink():
|
||
raise ValueError(f"拒绝通过符号链接目录写入：{parent}")
|
||
|
||
|
||
def _atomic_write(target: Path, content: str) -> None:
|
||
"""以 ``0600`` 权限原子替换目标，避免密钥短暂暴露或留下半写文件。"""
|
||
_validate_target(target)
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
descriptor, temporary_name = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent)
|
||
temporary = Path(temporary_name)
|
||
try:
|
||
os.fchmod(descriptor, stat.S_IRUSR | stat.S_IWUSR)
|
||
with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as handle:
|
||
handle.write(content)
|
||
handle.flush()
|
||
os.fsync(handle.fileno())
|
||
os.replace(temporary, target)
|
||
try:
|
||
directory_descriptor = os.open(target.parent, os.O_RDONLY)
|
||
try:
|
||
os.fsync(directory_descriptor)
|
||
finally:
|
||
os.close(directory_descriptor)
|
||
except OSError:
|
||
pass
|
||
except Exception:
|
||
temporary.unlink(missing_ok=True)
|
||
raise
|
||
|
||
|
||
def sync_env(example_path: str | Path = ".env.example", env_path: str | Path = ".env") -> Path:
|
||
"""读取模板和已有配置、完成合并并原子写回，供 CLI 或其他维护脚本调用。"""
|
||
example = Path(example_path).expanduser()
|
||
target = Path(env_path).expanduser()
|
||
if example.resolve() == target.resolve():
|
||
raise ValueError("模板与输出不能是同一个文件")
|
||
if not example.is_file():
|
||
raise FileNotFoundError(f"dotenv 模板不存在：{example}")
|
||
_validate_target(target)
|
||
example_text = _read_text(example)
|
||
current_text = _read_text(target) if target.exists() else ""
|
||
merged = merge_env_text(example_text, current_text, example_path=example, env_path=target)
|
||
_atomic_write(target, merged)
|
||
return target
|
||
|
||
|
||
def _parser() -> argparse.ArgumentParser:
|
||
"""构造命令行解析器，供入口函数和 CLI 测试复用。"""
|
||
parser = argparse.ArgumentParser(description="以 .env.example 为蓝本同步 .env")
|
||
parser.add_argument("--example", default=".env.example", help="dotenv 模板路径（默认：.env.example）")
|
||
parser.add_argument("--env", default=".env", help="dotenv 输出路径（默认：.env）")
|
||
return parser
|
||
|
||
|
||
def main(argv: Sequence[str] | None = None) -> int:
|
||
"""执行 dotenv 同步命令，并仅输出不含配置值的结果摘要。"""
|
||
parser = _parser()
|
||
arguments = parser.parse_args(argv)
|
||
try:
|
||
target = sync_env(arguments.example, arguments.env)
|
||
except (OSError, ValueError) as error:
|
||
parser.error(str(error))
|
||
print(f"已根据 {arguments.example} 生成 {target}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|