Private
Public Access
botched the chronology, going to rewrite the track.
This commit is contained in:
@@ -133,12 +133,107 @@ def _repo_root(start: Path) -> Path:
|
||||
return start.parent
|
||||
|
||||
|
||||
def _git_log(folder_relpath: str, *args: str) -> str:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "log", *args, "--", folder_relpath],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=_GIT_TIMEOUT,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return ""
|
||||
return result.stdout
|
||||
except (subprocess.SubprocessError, OSError):
|
||||
return ""
|
||||
|
||||
|
||||
def _git_first_line(folder_relpath: str, *args: str) -> str:
|
||||
out = _git_log(folder_relpath, *args)
|
||||
stripped = out.strip()
|
||||
if not stripped:
|
||||
return ""
|
||||
return stripped.splitlines()[0]
|
||||
|
||||
|
||||
def _repo_root(start: Path) -> Path:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "rev-parse", "--show-toplevel"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
check=False,
|
||||
cwd=str(start),
|
||||
)
|
||||
if result.returncode == 0 and result.stdout.strip():
|
||||
return Path(result.stdout.strip())
|
||||
except (subprocess.SubprocessError, OSError):
|
||||
pass
|
||||
return start.parent
|
||||
|
||||
|
||||
def _parse_state_phase(state_path: Path) -> str:
|
||||
if not state_path.is_file():
|
||||
return "no-state-toml"
|
||||
try:
|
||||
for line in state_path.read_text(encoding="utf-8").splitlines():
|
||||
if line.startswith("current_phase"):
|
||||
v = line.split("=", 1)[1].strip().split("#")[0].strip().strip('"')
|
||||
return v
|
||||
except (subprocess.SubprocessError, OSError, Exception):
|
||||
pass
|
||||
return "?"
|
||||
|
||||
|
||||
def _last_commit_date(folder_relpath: str) -> str:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "log", "-1", "--format=%ad", "--date=short", "--", folder_relpath],
|
||||
capture_output=True, text=True, timeout=_GIT_TIMEOUT, check=False,
|
||||
)
|
||||
return result.stdout.strip()
|
||||
except (subprocess.SubprocessError, OSError):
|
||||
return "never"
|
||||
|
||||
|
||||
def _classify_status(folder_link: str, current: str, track_id: str) -> str:
|
||||
"""Per-row manual review classification (FR6 hard gate).
|
||||
|
||||
Logic (per user directive 2026-06-20):
|
||||
- PLACEHOLDER tracks: keep as is
|
||||
- archive/ folder: default to Completed (the work was done and archived; metadata status may be stale)
|
||||
- tracks/ folder + state_phase=complete OR chrono in {completed, complete, shipped}: Completed
|
||||
- tracks/ folder + everything else: keep original chrono status (in flight)
|
||||
- Abandoned is reserved for explicit user marking; the script does NOT auto-mark.
|
||||
|
||||
Note: "Completed" (not "Shipped") is the canonical term per user directive 2026-06-20.
|
||||
This is a side-project, not a shipped product.
|
||||
"""
|
||||
if "PLACEHOLDER" in track_id:
|
||||
return current
|
||||
if "contingency" in current.lower():
|
||||
return current
|
||||
is_archive = folder_link.startswith("conductor/archive/")
|
||||
is_tracks = folder_link.startswith("conductor/tracks/")
|
||||
if is_archive:
|
||||
return "Completed"
|
||||
folder = Path(folder_link)
|
||||
state_phase = _parse_state_phase(folder / "state.toml") if is_tracks else "?"
|
||||
chrono_lower = current.lower()
|
||||
is_completed = chrono_lower in {"completed", "complete", "shipped"} or state_phase in {"complete", '"complete"'}
|
||||
if is_tracks and is_completed:
|
||||
return "Completed"
|
||||
return current
|
||||
|
||||
|
||||
def walk_track_folders(root: Path) -> list[dict]:
|
||||
repo_root: Path = _repo_root(root)
|
||||
rows: list[dict] = []
|
||||
for parent_dir, default_status in (
|
||||
(root / "tracks", "Active"),
|
||||
(root / "archive", "Shipped"),
|
||||
(root / "archive", "Completed"),
|
||||
):
|
||||
if not parent_dir.is_dir():
|
||||
continue
|
||||
@@ -156,8 +251,8 @@ def walk_track_folders(root: Path) -> list[dict]:
|
||||
else:
|
||||
first_commit = _git_first_line(folder_relpath, "--reverse", "--format=%aI")
|
||||
date = first_commit[:10] if first_commit else ""
|
||||
status: str = default_status
|
||||
metadata_path = folder / "metadata.json"
|
||||
status: str = default_status
|
||||
if metadata_path.is_file():
|
||||
try:
|
||||
data = json.loads(metadata_path.read_text(encoding="utf-8"))
|
||||
@@ -166,6 +261,7 @@ def walk_track_folders(root: Path) -> list[dict]:
|
||||
status = meta_status
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
status = _classify_status(folder_relpath, status, track_id)
|
||||
summary: str = extract_summary(folder)
|
||||
init_sha: str = _git_first_line(folder_relpath, "--reverse", "--format=%h")
|
||||
end_sha: str = _git_first_line(folder_relpath, "-1", "--format=%h")
|
||||
|
||||
Reference in New Issue
Block a user