"""push で送るファイルの選定（除外規則・追加指定・サーバ側の削除対象の算出）． 既定では，Windows 専用の仮想環境・計算結果・キャッシュ・スイープの一時生成物を除いた プロジェクト全体を送る．gitignore 対象でも実測参照データ（lightsim/・reference_measurement/・ REFERENCE/）はサーバでの評価に要るので送る． """ from __future__ import annotations import fnmatch import os import re from collections.abc import Callable, Collection, Iterable, Iterator, Sequence from dataclasses import dataclass from pathlib import Path, PurePosixPath # どの階層にあっても送らないディレクトリ名（.venv は Windows 用でサーバでは使えない） EXCLUDED_DIR_NAMES = frozenset({".git", ".venv", "__pycache__", ".pytest_cache", ".ruff_cache"}) # プロジェクト直下でだけ除外するエントリ（サーバ側の計算結果を上書きしないため output/ は送らない） EXCLUDED_ROOT_ENTRIES = frozenset({"output"}) # プロジェクトルート基準のパターン（スイープ生成物．約 450 MB あり必要時だけ --include で送る） EXCLUDED_GLOBS = ("configs/_tmp_*",) # push --prune でサーバ側の古いファイルを消してよいディレクトリ（コードのみ．output/・configs/ は # サーバ側で生成されたファイルがありうるので対象外） PRUNE_DIRS = ("src", "tests", "scripts") # シーン YAML の `path: C:\...` / `- path: 'D:/...'` のような Windows 絶対パス _WINDOWS_ABS_PATH_RE = re.compile(r"^\s*(?:-\s*)?path:\s*['\"]?[A-Za-z]:[\\/]", re.MULTILINE) @dataclass(frozen=True) class PushEntry: """送信するファイル 1 件． Attributes: path: 手元の実ファイル． arcname: リモートのプロジェクトルート基準の相対パス（POSIX 区切り）． size: バイト数（進捗表示用）． """ path: Path arcname: str size: int def find_project_root(start: Path) -> Path: """start から上方向にたどり，SmTIAS-LightSim のプロジェクトルートを探す． Args: start: 探索を始めるディレクトリ（通常はカレントディレクトリ）． Returns: pyproject.toml と src/sm_tias_lightsim/ を持つディレクトリ． Raises: ValueError: 見つからない場合． """ start = start.resolve() for directory in (start, *start.parents): if (directory / "pyproject.toml").is_file() and ( directory / "src" / "sm_tias_lightsim" ).is_dir(): return directory msg = ( f"{start} から上にプロジェクトルート（pyproject.toml と src/sm_tias_lightsim/）が" "見つかりません．リポジトリ内で実行するか --local-dir / --dest で指定してください．" ) raise ValueError(msg) def _is_cache_path(rel_path: str) -> bool: """パスのどこかに除外ディレクトリ名（キャッシュ・.git 等）を含むか．""" return any(part in EXCLUDED_DIR_NAMES for part in PurePosixPath(rel_path).parts) def is_excluded(rel_path: str) -> bool: """プロジェクトルート基準の相対パスが既定の push から除外されるかを返す． Args: rel_path: POSIX 区切りの相対パス（ファイルでもディレクトリでもよい）． Returns: 除外対象なら True． """ parts = PurePosixPath(rel_path).parts if not parts: return False if _is_cache_path(rel_path) or parts[0] in EXCLUDED_ROOT_ENTRIES: return True prefixes = ("/".join(parts[:i]) for i in range(1, len(parts) + 1)) return any( fnmatch.fnmatchcase(prefix, pattern) for prefix in prefixes for pattern in EXCLUDED_GLOBS ) def _walk_files(root: Path, start: Path, skip: Callable[[str], bool]) -> Iterator[PushEntry]: """start 配下のファイルを，skip で除外しつつ決定的な順で列挙する．""" for dirpath, dirnames, filenames in os.walk(start): dir_rel = Path(dirpath).relative_to(root).as_posix() prefix = "" if dir_rel == "." else f"{dir_rel}/" dirnames[:] = sorted(d for d in dirnames if not skip(prefix + d)) for filename in sorted(filenames): rel = prefix + filename path = Path(dirpath) / filename if skip(rel) or not path.is_file(): continue yield PushEntry(path=path, arcname=rel, size=path.stat().st_size) def resolve_extra_path(root: Path, text: str) -> tuple[Path, str]: """--include / --only で指定されたパスを検証し，実体と相対パスを返す． Args: root: プロジェクトルート（解決済み）． text: 指定されたパス（相対ならプロジェクトルート基準）． Returns: (実体の絶対パス, プロジェクトルート基準の POSIX 相対パス)． Raises: ValueError: プロジェクト外・プロジェクト全体・output/ 配下・存在しない場合． """ candidate = Path(text) if not candidate.is_absolute(): candidate = root / candidate resolved = candidate.resolve() try: rel = resolved.relative_to(root) except ValueError as exc: raise ValueError(f"プロジェクト外のパスは送れません: {text}") from exc if not rel.parts: msg = "プロジェクト全体は --include / --only ではなく通常の push で送ってください．" raise ValueError(msg) if rel.parts[0] in EXCLUDED_ROOT_ENTRIES: msg = f"output/ 配下は push できません（サーバ側の計算結果を上書きしうるため）: {text}" raise ValueError(msg) if not resolved.exists(): raise ValueError(f"パスが存在しません: {text}") return resolved, rel.as_posix() def collect_push_entries( root: Path, *, include_base: bool, extra_paths: Sequence[str] = () ) -> list[PushEntry]: """送信するファイルの一覧を作る． Args: root: プロジェクトルート． include_base: True なら既定の除外規則を適用したプロジェクト全体を含める． extra_paths: 追加で送るパス（configs/_tmp_* 等）．既定の除外規則のうち グロブ（configs/_tmp_*）は適用しないが，キャッシュ類は除く． Returns: arcname 順に並べた送信対象（同じ arcname は 1 件にまとめる）． Raises: ValueError: extra_paths に不正なパスがある場合． """ root = root.resolve() entries: dict[str, PushEntry] = {} if include_base: for entry in _walk_files(root, root, is_excluded): entries[entry.arcname] = entry for text in extra_paths: path, rel = resolve_extra_path(root, text) if path.is_file(): entries[rel] = PushEntry(path=path, arcname=rel, size=path.stat().st_size) continue for entry in _walk_files(root, path, _is_cache_path): entries[entry.arcname] = entry return [entries[name] for name in sorted(entries)] def find_yaml_with_windows_paths(entries: Iterable[PushEntry]) -> list[str]: """メッシュ等の path に Windows の絶対パスを含む YAML を返す（サーバでは読めない）． configs/_tmp_*/ の古い生成物は絶対パスで書かれているため，送る前に警告する目的． 読めないファイルは判定対象外にする（警告用の補助チェックで，送信自体は妨げない）． Args: entries: 送信対象． Returns: 該当する YAML の arcname 一覧． """ hits: list[str] = [] for entry in entries: if not entry.arcname.endswith((".yaml", ".yml")): continue try: text = entry.path.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError): continue if _WINDOWS_ABS_PATH_RE.search(text): hits.append(entry.arcname) return hits def compute_prune_targets( remote_files: Iterable[str], local_arcnames: Collection[str], prune_dirs: Sequence[str] = PRUNE_DIRS, ) -> list[str]: """サーバにあって手元に無い（＝削除・改名された）ファイルを求める． Args: remote_files: サーバの prune_dirs 配下のファイル一覧（プロジェクトルート基準）． local_arcnames: 手元から送ったファイルの arcname 集合． prune_dirs: 削除してよいディレクトリ． Returns: 削除対象の相対パス（ソート済み）．キャッシュ類は含めない． Raises: RuntimeError: prune_dirs 外・絶対パス・`..` を含むパスを受け取った場合 （リモートの一覧コマンドの不変条件違反．誤削除を防ぐため中断する）． """ local = set(local_arcnames) targets: list[str] = [] for rel in remote_files: path = PurePosixPath(rel) if ( path.is_absolute() or ".." in path.parts or len(path.parts) < 2 or path.parts[0] not in prune_dirs ): raise RuntimeError(f"想定外のリモートパスを受信しました（削除を中止）: {rel!r}") normalized = path.as_posix() if _is_cache_path(normalized) or normalized in local: continue targets.append(normalized) return sorted(targets)