feat(archive): add pkgmgr archive subcommand for task-tracked spec dirs
Walks a directory of numbered NNN-topic.md files, promotes every file with zero unchecked task-list items into the directorys README under the ## Archive section, then deletes the source file. Keeps spec directories (typically docs/requirements) short and focused on open work. The action ships as pkgmgr.actions.archive with four leaf modules (discovery, inspect, readme, workflow) and is wired into the CLI as pkgmgr archive [DIR] [--readme PATH] [--dry-run] [--include-template]. Extracted verbatim from cli/contributing/requirements/archive in infinito-nexus-core so every kpmx-managed repository can rely on the same archival convention without copy-pasting helpers. Twenty unit tests cover discovery, inspection, README merge, and end-to-end workflow paths. Also: realign tests/unit/pkgmgr/cli/commands/test_release.py with the new run_release(retry=False) signature shipped in v1.14.0. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
1 parent
a4a5b661b9
commit
a37b9ed8a7
16 files changed
+729
No files matched your search
@@ -0,0 +1,31 @@
|
||||
"""Archive fully-checked Markdown files into a README index, then delete them.
|
||||
|
||||
The archive action walks a directory for numbered ``NNN-topic.md`` files
|
||||
(default pattern ``^\\d{3}-[^/]+\\.md$``), promotes every file with zero
|
||||
unchecked ``- [ ]`` task-list markers into a ``## Archive`` index inside
|
||||
the directory README, and deletes the per-file source. Useful for
|
||||
keeping ``docs/requirements/`` (or any other task-tracked spec folder)
|
||||
short and focused on open work.
|
||||
|
||||
The module was extracted from
|
||||
``cli/contributing/requirements/archive`` in infinito-nexus-core so
|
||||
every kpmx-managed repository can rely on the same archival convention
|
||||
without copy-pasting the helpers.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .discovery import iter_archivable_files
|
||||
from .inspect import count_unchecked_items, extract_h1
|
||||
from .readme import existing_archive_entries, merge_archive_section
|
||||
from .workflow import ArchivePlan, run_archive
|
||||
|
||||
__all__ = [
|
||||
"ArchivePlan",
|
||||
"count_unchecked_items",
|
||||
"existing_archive_entries",
|
||||
"extract_h1",
|
||||
"iter_archivable_files",
|
||||
"merge_archive_section",
|
||||
"run_archive",
|
||||
]
|
||||
@@ -0,0 +1,53 @@
|
||||
"""Locate archivable Markdown files under a target directory."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
|
||||
DEFAULT_FILENAME_PATTERN = re.compile(r"^\d{3}-[^/]+\.md$")
|
||||
TEMPLATE_FILENAME = "000-template.md"
|
||||
|
||||
|
||||
def iter_archivable_files(
|
||||
directory: Path,
|
||||
*,
|
||||
include_template: bool = False,
|
||||
pattern: re.Pattern[str] = DEFAULT_FILENAME_PATTERN,
|
||||
template_filename: str = TEMPLATE_FILENAME,
|
||||
) -> list[Path]:
|
||||
"""Return all files in *directory* whose name matches *pattern*, sorted.
|
||||
|
||||
``000-template.md`` (or whatever *template_filename* matches) is
|
||||
excluded unless *include_template* is true. The check is filename
|
||||
based; nested directories are not traversed.
|
||||
"""
|
||||
if not directory.is_dir():
|
||||
return []
|
||||
files: list[Path] = []
|
||||
for path in sorted(directory.iterdir()):
|
||||
if not path.is_file() or not pattern.match(path.name):
|
||||
continue
|
||||
if not include_template and path.name == template_filename:
|
||||
continue
|
||||
files.append(path)
|
||||
return files
|
||||
|
||||
|
||||
def filter_archivable_files(
|
||||
paths: Iterable[Path],
|
||||
*,
|
||||
include_template: bool = False,
|
||||
pattern: re.Pattern[str] = DEFAULT_FILENAME_PATTERN,
|
||||
template_filename: str = TEMPLATE_FILENAME,
|
||||
) -> list[Path]:
|
||||
"""Same predicate as :func:`iter_archivable_files`, applied to an iterable."""
|
||||
result: list[Path] = []
|
||||
for path in paths:
|
||||
if not path.is_file() or not pattern.match(path.name):
|
||||
continue
|
||||
if not include_template and path.name == template_filename:
|
||||
continue
|
||||
result.append(path)
|
||||
return result
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Parse a single Markdown file: H1 heading and task-list completeness."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
H1_RE = re.compile(r"^#\s+(?P<title>\S.*?)\s*$")
|
||||
UNCHECKED_TASK_RE = re.compile(r"^\s*[-*+]\s+\[\s\]\s")
|
||||
|
||||
|
||||
def extract_h1(path: Path) -> str | None:
|
||||
"""Return the first H1 title in *path* or ``None`` if there is none."""
|
||||
try:
|
||||
with path.open(encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
match = H1_RE.match(line.rstrip("\n"))
|
||||
if match:
|
||||
return match.group("title")
|
||||
except OSError:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def count_unchecked_items(path: Path) -> int:
|
||||
"""Return the number of ``- [ ]`` task-list markers anywhere in *path*.
|
||||
|
||||
A non-zero count means the file is not yet fully complete and MUST
|
||||
NOT be archived.
|
||||
"""
|
||||
try:
|
||||
with path.open(encoding="utf-8") as fh:
|
||||
return sum(1 for line in fh if UNCHECKED_TASK_RE.match(line))
|
||||
except OSError:
|
||||
return 0
|
||||
@@ -0,0 +1,76 @@
|
||||
"""Read and update the ``## Archive`` section of a directory README."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
ARCHIVE_HEADING = "## Archive"
|
||||
LIST_ITEM_RE = re.compile(r"^\s*-\s+(?P<body>\S.*)$")
|
||||
|
||||
|
||||
def existing_archive_entries(readme_text: str) -> set[str]:
|
||||
"""Return the deduplicated set of list-item bodies under ``## Archive``."""
|
||||
lines = readme_text.splitlines()
|
||||
in_archive = False
|
||||
entries: set[str] = set()
|
||||
for line in lines:
|
||||
stripped = line.rstrip()
|
||||
if stripped == ARCHIVE_HEADING:
|
||||
in_archive = True
|
||||
continue
|
||||
if in_archive and stripped.startswith("## "):
|
||||
break
|
||||
if not in_archive:
|
||||
continue
|
||||
match = LIST_ITEM_RE.match(stripped)
|
||||
if match:
|
||||
entries.add(match.group("body").strip())
|
||||
return entries
|
||||
|
||||
|
||||
def merge_archive_section(readme_text: str, new_entries: list[str]) -> str:
|
||||
"""Return ``readme_text`` with *new_entries* appended under ``## Archive``.
|
||||
|
||||
Existing entries are preserved verbatim. If the section is missing it
|
||||
is created at the end of the document.
|
||||
"""
|
||||
if not new_entries:
|
||||
return readme_text
|
||||
|
||||
lines = readme_text.splitlines()
|
||||
archive_index = next(
|
||||
(i for i, line in enumerate(lines) if line.rstrip() == ARCHIVE_HEADING),
|
||||
None,
|
||||
)
|
||||
|
||||
if archive_index is None:
|
||||
suffix = [""] if (lines and lines[-1] != "") else []
|
||||
suffix.append(ARCHIVE_HEADING)
|
||||
suffix.append("")
|
||||
suffix.extend(f"- {entry}" for entry in new_entries)
|
||||
merged = lines + suffix
|
||||
return "\n".join(merged) + "\n"
|
||||
|
||||
section_end = next(
|
||||
(i for i in range(archive_index + 1, len(lines)) if lines[i].startswith("## ")),
|
||||
len(lines),
|
||||
)
|
||||
|
||||
body_start = archive_index + 1
|
||||
while body_start < section_end and not lines[body_start].strip():
|
||||
body_start += 1
|
||||
|
||||
last_item = body_start - 1
|
||||
for i in range(body_start, section_end):
|
||||
if LIST_ITEM_RE.match(lines[i]):
|
||||
last_item = i
|
||||
|
||||
insertion_point = (last_item + 1) if last_item >= body_start else body_start
|
||||
if insertion_point == body_start and body_start == archive_index + 1:
|
||||
new_block = ["", *[f"- {entry}" for entry in new_entries]]
|
||||
else:
|
||||
new_block = [f"- {entry}" for entry in new_entries]
|
||||
|
||||
merged = lines[:insertion_point] + new_block + lines[insertion_point:]
|
||||
trailing = "\n" if readme_text.endswith("\n") else ""
|
||||
return "\n".join(merged) + trailing
|
||||
@@ -0,0 +1,115 @@
|
||||
"""Orchestrator for archiving fully-checked Markdown files."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from .discovery import iter_archivable_files
|
||||
from .inspect import count_unchecked_items, extract_h1
|
||||
from .readme import existing_archive_entries, merge_archive_section
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ArchivePlan:
|
||||
"""Outcome of an archive analysis run.
|
||||
|
||||
Attributes:
|
||||
archived: ``(source_path, title)`` for every file that was (or
|
||||
would be) archived. Order matches the original directory
|
||||
listing.
|
||||
skipped_incomplete: ``(source_path, unchecked_count)`` for files
|
||||
that still hold ``- [ ]`` markers.
|
||||
skipped_without_h1: files that had no H1 heading to use as title.
|
||||
new_entries: titles that will be appended to the README index.
|
||||
existing_entries: titles already present in the README index.
|
||||
"""
|
||||
|
||||
archived: list[tuple[Path, str]]
|
||||
skipped_incomplete: list[tuple[Path, int]]
|
||||
skipped_without_h1: list[Path]
|
||||
new_entries: list[str]
|
||||
existing_entries: set[str]
|
||||
|
||||
|
||||
def _bucket_files(
|
||||
files: list[Path],
|
||||
) -> tuple[
|
||||
list[tuple[Path, str]],
|
||||
list[tuple[Path, int]],
|
||||
list[Path],
|
||||
]:
|
||||
plan: list[tuple[Path, str]] = []
|
||||
skipped_incomplete: list[tuple[Path, int]] = []
|
||||
skipped_without_h1: list[Path] = []
|
||||
for path in files:
|
||||
unchecked = count_unchecked_items(path)
|
||||
if unchecked > 0:
|
||||
skipped_incomplete.append((path, unchecked))
|
||||
continue
|
||||
title = extract_h1(path)
|
||||
if title is None:
|
||||
skipped_without_h1.append(path)
|
||||
continue
|
||||
plan.append((path, title))
|
||||
return plan, skipped_incomplete, skipped_without_h1
|
||||
|
||||
|
||||
def _dedupe_titles(
|
||||
plan: list[tuple[Path, str]], already_archived: set[str]
|
||||
) -> list[str]:
|
||||
new_entries: list[str] = []
|
||||
for _path, title in plan:
|
||||
if title in already_archived or title in new_entries:
|
||||
continue
|
||||
new_entries.append(title)
|
||||
return new_entries
|
||||
|
||||
|
||||
def run_archive(
|
||||
directory: Path,
|
||||
readme_path: Path,
|
||||
*,
|
||||
dry_run: bool = False,
|
||||
include_template: bool = False,
|
||||
) -> ArchivePlan:
|
||||
"""Walk *directory* and archive every fully-checked file into *readme_path*.
|
||||
|
||||
Returns an :class:`ArchivePlan` describing the outcome. When
|
||||
``dry_run`` is true no files are deleted and the README is not
|
||||
rewritten — the plan still reflects what *would* happen.
|
||||
|
||||
Raises ``FileNotFoundError`` if *directory* or *readme_path* does
|
||||
not exist.
|
||||
"""
|
||||
if not directory.is_dir():
|
||||
raise FileNotFoundError(f"Archive directory not found: {directory}")
|
||||
if not readme_path.is_file():
|
||||
raise FileNotFoundError(f"README not found: {readme_path}")
|
||||
|
||||
files = iter_archivable_files(directory, include_template=include_template)
|
||||
readme_text = readme_path.read_text(encoding="utf-8")
|
||||
already_archived = existing_archive_entries(readme_text)
|
||||
|
||||
archived, skipped_incomplete, skipped_without_h1 = _bucket_files(files)
|
||||
new_entries = _dedupe_titles(archived, already_archived)
|
||||
|
||||
if not dry_run and new_entries:
|
||||
merged_text = merge_archive_section(readme_text, new_entries)
|
||||
if merged_text != readme_text:
|
||||
readme_path.write_text(merged_text, encoding="utf-8")
|
||||
|
||||
if not dry_run:
|
||||
for path, _title in archived:
|
||||
try:
|
||||
path.unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
return ArchivePlan(
|
||||
archived=archived,
|
||||
skipped_incomplete=skipped_incomplete,
|
||||
skipped_without_h1=skipped_without_h1,
|
||||
new_entries=new_entries,
|
||||
existing_entries=already_archived,
|
||||
)
|
||||
@@ -1,3 +1,4 @@
|
||||
from .archive import handle_archive
|
||||
from .repos import handle_repos_command
|
||||
from .config import handle_config
|
||||
from .tools import handle_tools_command
|
||||
@@ -10,6 +11,7 @@ from .branch import handle_branch
|
||||
from .mirror import handle_mirror_command
|
||||
|
||||
__all__ = [
|
||||
"handle_archive",
|
||||
"handle_repos_command",
|
||||
"handle_config",
|
||||
"handle_tools_command",
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from pkgmgr.actions.archive import ArchivePlan, run_archive
|
||||
from pkgmgr.cli.context import CLIContext
|
||||
|
||||
|
||||
def _print_summary(
|
||||
directory: Path, readme: Path, plan: ArchivePlan, dry_run: bool
|
||||
) -> None:
|
||||
print(f"[archive] Directory: {directory}")
|
||||
print(f"[archive] README: {readme}")
|
||||
print(f"[archive] Files to process: {len(plan.archived)}")
|
||||
print(f"[archive] Skipped (incomplete): {len(plan.skipped_incomplete)}")
|
||||
print(f"[archive] New archive entries: {len(plan.new_entries)}")
|
||||
print(f"[archive] Dry-run: {dry_run}")
|
||||
|
||||
|
||||
def _print_skips(plan: ArchivePlan, cwd: Path) -> None:
|
||||
if plan.skipped_incomplete:
|
||||
print(
|
||||
"[archive] SKIP: files with unchecked `- [ ]` items "
|
||||
"(not archived, not deleted):"
|
||||
)
|
||||
for path, count in plan.skipped_incomplete:
|
||||
suffix = "s" if count != 1 else ""
|
||||
rel = _rel_or_abs(path, cwd)
|
||||
print(f" - {rel} ({count} unchecked item{suffix})")
|
||||
if plan.skipped_without_h1:
|
||||
print("[archive] WARN: skipped files without an H1 heading:")
|
||||
for path in plan.skipped_without_h1:
|
||||
print(f" - {_rel_or_abs(path, cwd)}")
|
||||
|
||||
|
||||
def _print_actions(plan: ArchivePlan, cwd: Path, dry_run: bool) -> None:
|
||||
verb = "would archive" if dry_run else "archived"
|
||||
rm_verb = "would delete" if dry_run else "deleted"
|
||||
for path, title in plan.archived:
|
||||
rel = _rel_or_abs(path, cwd)
|
||||
print(f"[archive] {verb}: {rel} -> '{title}'")
|
||||
if not dry_run:
|
||||
print(f"[archive] {rm_verb}: {rel}")
|
||||
|
||||
|
||||
def _rel_or_abs(path: Path, cwd: Path) -> str:
|
||||
try:
|
||||
return path.resolve().relative_to(cwd).as_posix()
|
||||
except ValueError:
|
||||
return path.as_posix()
|
||||
|
||||
|
||||
def handle_archive(args, _ctx: CLIContext) -> None:
|
||||
directory = Path(args.directory).resolve()
|
||||
readme = (
|
||||
Path(args.readme).resolve()
|
||||
if args.readme
|
||||
else (directory / "README.md").resolve()
|
||||
)
|
||||
|
||||
try:
|
||||
plan = run_archive(
|
||||
directory=directory,
|
||||
readme_path=readme,
|
||||
dry_run=args.dry_run,
|
||||
include_template=args.include_template,
|
||||
)
|
||||
except FileNotFoundError as exc:
|
||||
print(f"[archive] ERROR: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
cwd = Path.cwd().resolve()
|
||||
_print_summary(directory, readme, plan, args.dry_run)
|
||||
_print_skips(plan, cwd)
|
||||
_print_actions(plan, cwd, args.dry_run)
|
||||
@@ -10,6 +10,7 @@ from pkgmgr.core.repository.selected import get_selected_repos
|
||||
from pkgmgr.core.repository.dir import get_repo_dir
|
||||
|
||||
from pkgmgr.cli.commands import (
|
||||
handle_archive,
|
||||
handle_repos_command,
|
||||
handle_tools_command,
|
||||
handle_release,
|
||||
@@ -60,6 +61,10 @@ def dispatch_command(args, ctx: CLIContext) -> None:
|
||||
if maybe_handle_proxy(args, ctx):
|
||||
return
|
||||
|
||||
if args.command == "archive":
|
||||
handle_archive(args, ctx)
|
||||
return
|
||||
|
||||
commands_with_selection = {
|
||||
"install",
|
||||
"update",
|
||||
|
||||
@@ -4,6 +4,7 @@ import argparse
|
||||
|
||||
from pkgmgr.cli.proxy import register_proxy_commands
|
||||
|
||||
from .archive_cmd import add_archive_subparser
|
||||
from .branch_cmd import add_branch_subparsers
|
||||
from .changelog_cmd import add_changelog_subparser
|
||||
from .common import SortedSubParsersAction
|
||||
@@ -65,6 +66,7 @@ def create_parser(description_text: str) -> argparse.ArgumentParser:
|
||||
|
||||
add_make_subparsers(subparsers)
|
||||
add_mirror_subparsers(subparsers)
|
||||
add_archive_subparser(subparsers)
|
||||
|
||||
register_proxy_commands(subparsers)
|
||||
return parser
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
|
||||
|
||||
def add_archive_subparser(subparsers: argparse._SubParsersAction) -> None:
|
||||
"""Register the archive subcommand.
|
||||
|
||||
Walks a directory of numbered ``NNN-topic.md`` files, promotes every
|
||||
file whose ``- [ ]`` checklist is fully checked into the directorys
|
||||
README ``## Archive`` section, then deletes the source file.
|
||||
"""
|
||||
parser = subparsers.add_parser(
|
||||
"archive",
|
||||
help=(
|
||||
"Archive fully-checked Markdown spec files (NNN-topic.md) "
|
||||
"into the directorys README and delete them."
|
||||
),
|
||||
description=(
|
||||
"Walk DIR for files that match the numbered "
|
||||
"NNN-topic.md naming, promote every file with zero "
|
||||
"unchecked `- [ ]` items into READMEs ## Archive section, "
|
||||
"then delete the source file. Files with unchecked items are "
|
||||
"skipped. Use --dry-run to preview without writing."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"directory",
|
||||
nargs="?",
|
||||
default="docs/requirements",
|
||||
help=(
|
||||
"Directory to scan for archivable Markdown files. "
|
||||
"Defaults to docs/requirements (relative to the current "
|
||||
"working directory)."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--readme",
|
||||
default=None,
|
||||
help=(
|
||||
"Path to the README that holds the ## Archive index. "
|
||||
"Defaults to <directory>/README.md."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Print planned changes without modifying or deleting anything.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--include-template",
|
||||
action="store_true",
|
||||
help=(
|
||||
"Also archive and delete 000-template.md. Off by default "
|
||||
"because contributor guides typically reference it."
|
||||
),
|
||||
)
|
||||
Reference in new issue
Block a user