Пара плагинов с намеренно проведённой границей: av-dev-tasks отвечает за то, что делаем и в каком порядке, av-dev-pipeline — за то, как ведём одну задачу. Зависимости между ними нет: управление задачами работает и с ручным исполнением, пайплайн — на проекте с любым учётом задач. - av-dev-tasks — преемник av-dev-backlog: цели вместо приоритетов, спринт под одну цель с заморозкой набора, различение вопроса и блокера, каденция «вопросы — разбор — переоценка — набор». Раскладка docs/tasks с items/, PLAN.md, BACKLOG.md, SPRINT.md, REJECTED.md; проверенное из av-dev-backlog перенесено, не переписано. - av-dev-pipeline — вынос того, что лежало копиями в healthlog и jellybit (3628 строк) и уже разошлось: цикл SDD, конвейер ревью с обязательным триажем, прогон нескольких задач разом. Проектная специфика вынесена в файл-бриф, charter'ы несут метод. Коммит фиксирует состояние на момент ревью: три прохода нашли блокирующие дефекты (нет шага, заводящего бриф; git rebase на занятой worktree ветке; sprint drop пишет наполовину) — они чинятся следующими коммитами. Сохранено как база, от которой видно правки.
1309 lines
64 KiB
Python
Executable File
1309 lines
64 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Детерминированный инструмент управления задачами: файлы против индексов.
|
||
|
||
Преемник backlog.py. Разница по существу одна: **порядка нет, есть цель**.
|
||
Приоритеты и секция-как-уровень заменены на цель (`goal:<слаг>` тегом) и на
|
||
спринт — замороженный набор задач под одну цель. Индексов теперь четыре, и
|
||
задача живёт ровно в одном из них за раз.
|
||
|
||
Раскладка (все имена настраиваются, см. init и `.tasks.json`):
|
||
|
||
<tasks>/
|
||
items/ задачи и цели файлами, <slug>.md
|
||
PLAN.md оглавление целей: линия (упорядоченная) и кусты
|
||
BACKLOG.md что можно взять — только задачи, целей здесь нет
|
||
SPRINT.md текущий спринт: цель, набор, дата
|
||
REJECTED.md ушедшее БЕЗ реализации, с причиной и датой
|
||
|
||
Источник истины — файл задачи в items/. Индексы производны: расходятся —
|
||
неправ индекс. Исключение одно и оно намеренное: **в каком индексе лежит
|
||
задача — знают индексы**, потому что «в спринте» это свойство спринта, а не
|
||
задачи; поля-состояния в файле нет, а рассогласование ловит check.
|
||
|
||
Тип — ключевое слово (goal | idea | epic | task), по-английски, как и прочие
|
||
токены команд. Обычная задача (task) префикса не несёт, остальные кодируются
|
||
префиксом `[goal]`/`[idea]`/`[epic]` в заголовке H1. Отдельного поля типа нет.
|
||
|
||
`[goal]` и `[epic]` — разные вещи: цель постоянна и живёт, пока живёт
|
||
направление; эпик временен — это задача, которая не мерджится целиком, её
|
||
разбирают, и он исчезает.
|
||
|
||
Использование:
|
||
tasks.py init [--dir DIR] [--sections …] [--plan-sections …] [--items …]
|
||
tasks.py check [--dir DIR] [--fix]
|
||
tasks.py list [--dir DIR] [--stale] [--section S] [--type T] [--tag T]
|
||
[--goal S] [--index backlog|sprint|plan|all] [--questions]
|
||
tasks.py add --slug S --title T [--type goal|idea|epic] [--section S]
|
||
[--goal G] [--hook H] [--reason R] [--tag a,b]
|
||
tasks.py edit S [--title T] [--hook H] [--type T] [--goal G]
|
||
[--add-tag a,b] [--rm-tag c,d]
|
||
tasks.py move S --section S [--reason R] [--after S | --first]
|
||
tasks.py close S (--reason R | --implemented)
|
||
tasks.py sprint start --goal S [--date ГГГГ-ММ-ДД]
|
||
tasks.py sprint take S [S …]
|
||
tasks.py sprint drop S [S …] --reason R
|
||
tasks.py sprint close [--dissolve --reason R]
|
||
|
||
Тело задачи (контекст, критерии, вопросы, ссылки) остаётся агенту — add кладёт
|
||
заголовок, мета-строку и шаблон-плейсхолдер; агент дописывает редактором.
|
||
|
||
Границы безопасности: слаг — только латиница kebab-case (traversal невозможен),
|
||
--dir обязан быть внутри рабочего каталога, в заголовок/хук/причину не пролезет
|
||
перевод строки, `·` в причине запрещён (это разделитель мета-полей).
|
||
|
||
Язык не зашит инструментально: секции сопоставляются с заголовками индексов как
|
||
есть, имена служебных файлов и заголовков настраиваются. Текст задач — русский.
|
||
"""
|
||
|
||
import argparse
|
||
import datetime
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
CONFIG_NAME = ".tasks.json"
|
||
|
||
DEFAULTS = {
|
||
"items": "items",
|
||
"backlog": "BACKLOG.md",
|
||
"plan": "PLAN.md",
|
||
"sprint": "SPRINT.md",
|
||
"rejected": "REJECTED.md",
|
||
"sprint_section": "Набор",
|
||
"criteria_heading": "Критерии приёмки",
|
||
"questions_heading": "Вопросы",
|
||
}
|
||
|
||
DEFAULT_SECTIONS = "ядро,инфра"
|
||
DEFAULT_PLAN_SECTIONS = "линия,кусты"
|
||
|
||
META_FIELD = re.compile(r"^\*\*(.+?):\*\*\s*(.*)$")
|
||
INDEX_ENTRY = re.compile(r"^- \[(.+?)\]\((.+?\.md)\)\s*(?:—\s*(.*))?$")
|
||
SECTION = re.compile(r"^##\s+(.+?)\s*$")
|
||
TYPE_PREFIX = re.compile(r"^\[(.+?)\]\s*(.*)$")
|
||
SLUG_RE = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*")
|
||
SLUG = re.compile(SLUG_RE.pattern + r"\.md")
|
||
GOAL_LINE = re.compile(r"^\*\*(?:Цель|Goal):\*\*\s*\[(.+?)\]\((.+?\.md)\)")
|
||
DATE_RE = re.compile(r"\d{4}-\d{2}-\d{2}")
|
||
# Строка кладбища: - ГГГГ-ММ-ДД `slug` — текст
|
||
REJECTED_ENTRY = re.compile(r"^- \d{4}-\d{2}-\d{2} `[a-z0-9-]+` — .+")
|
||
|
||
GOAL = "goal"
|
||
TYPES = ("goal", "idea", "epic") # непустые типы-ключевые слова, префикс [..] в H1
|
||
PLAIN_TYPE = "task" # обычная задача — без префикса
|
||
TAKEABLE = (PLAIN_TYPE,) # что вообще можно взять в спринт
|
||
QUESTION_TAG = "question"
|
||
GOAL_TAG = "goal:"
|
||
STALE_DAYS = 180 # порог «залежалась» для метрики здоровья в check
|
||
BODY_PLACEHOLDER = "<!-- "
|
||
|
||
|
||
# --- Валидация недоверенного ввода (аргументы могут прийти из текста задачи) ---
|
||
|
||
def bad_line(value: str | None, field: str) -> str | None:
|
||
"""Однострочность: перевод строки/управляющий символ ломает индекс и файл."""
|
||
if value is not None and (any(c in value for c in "\n\r") or any(ord(c) < 32 for c in value)):
|
||
return f"{field}: перевод строки или управляющий символ запрещён"
|
||
return None
|
||
|
||
|
||
def bad_slug(slug: str) -> str | None:
|
||
if not SLUG_RE.fullmatch(slug):
|
||
return f"слаг «{slug}» — только латиница kebab-case (без ../, точек, слэшей)"
|
||
return None
|
||
|
||
|
||
def bad_reason(reason: str | None) -> str | None:
|
||
if reason is None:
|
||
return None
|
||
if (e := bad_line(reason, "причина")):
|
||
return e
|
||
if "·" in reason:
|
||
return "причина: символ · зарезервирован под разделитель мета-полей"
|
||
return None
|
||
|
||
|
||
def bad_tags(raw: str | None) -> str | None:
|
||
if raw is None:
|
||
return None
|
||
if (e := bad_line(raw, "теги")):
|
||
return e
|
||
for t in split_tags(raw):
|
||
if not re.fullmatch(r"[a-z0-9]+(?:[-:.][a-z0-9]+)*", t):
|
||
return f"тег «{t}» — латиница, цифры и разделители -:. (например goal:merge-robustness)"
|
||
return None
|
||
|
||
|
||
def split_tags(raw: str | None) -> list[str]:
|
||
return [t.strip().lower() for t in (raw or "").split(",") if t.strip()]
|
||
|
||
|
||
def dir_within_cwd(root: Path) -> bool:
|
||
try:
|
||
root.resolve().relative_to(Path.cwd().resolve())
|
||
return True
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
# --- Атомарная запись: падение посреди write не оставит усечённый индекс ---
|
||
|
||
def write_atomic(path: Path, text: str) -> None:
|
||
tmp = path.with_name(path.name + ".tmp")
|
||
tmp.write_text(text, encoding="utf-8")
|
||
os.replace(tmp, path)
|
||
|
||
|
||
class Layout:
|
||
"""Каталог задач и имена его частей. Всё настраивается: у соседнего проекта
|
||
может быть другой подкаталог и другие имена индексов, а семантика та же."""
|
||
|
||
def __init__(self, root: Path, cfg: dict):
|
||
self.root = root
|
||
self.cfg = {**DEFAULTS, **cfg}
|
||
self.items = root / self.cfg["items"]
|
||
|
||
def index(self, kind: str) -> Path:
|
||
return self.root / self.cfg[kind]
|
||
|
||
def name(self, kind: str) -> str:
|
||
return self.cfg[kind]
|
||
|
||
@property
|
||
def indexes(self) -> tuple[str, ...]:
|
||
return ("backlog", "sprint", "plan")
|
||
|
||
|
||
def load_config(root: Path) -> dict:
|
||
path = root / CONFIG_NAME
|
||
if not path.is_file():
|
||
return {}
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as e:
|
||
sys.exit(f"{path}: не разбирается как JSON — {e}")
|
||
if not isinstance(data, dict):
|
||
sys.exit(f"{path}: ожидался объект с настройками")
|
||
unknown = set(data) - set(DEFAULTS)
|
||
if unknown:
|
||
sys.exit(f"{path}: неизвестные ключи: {', '.join(sorted(unknown))}"
|
||
f" (известны: {', '.join(sorted(DEFAULTS))})")
|
||
return data
|
||
|
||
|
||
def resolve_layout(explicit: str | None) -> Layout:
|
||
"""Каталог задач для команд, кроме init. Явный --dir обязан быть внутри cwd."""
|
||
def looks_like_tasks(p: Path) -> bool:
|
||
if (p / CONFIG_NAME).is_file():
|
||
return True
|
||
cfg = {**DEFAULTS}
|
||
return (p / cfg["backlog"]).is_file()
|
||
|
||
if explicit:
|
||
root = Path(explicit)
|
||
if not dir_within_cwd(root):
|
||
sys.exit(f"--dir вне рабочего каталога: {explicit}")
|
||
if not looks_like_tasks(root):
|
||
sys.exit(f"задач нет в «{explicit}»; новый проект — tasks.py init --dir {explicit}")
|
||
return Layout(root, load_config(root))
|
||
for candidate in ("docs/tasks", "tasks", "doc/tasks", "docs/backlog"):
|
||
p = Path(candidate)
|
||
if looks_like_tasks(p):
|
||
return Layout(p, load_config(p))
|
||
sys.exit("каталог задач не найден, укажи --dir"
|
||
" (искал: docs/tasks, tasks, doc/tasks, docs/backlog)")
|
||
|
||
|
||
# --- Чтение индексов ---
|
||
|
||
def read_lines(path: Path) -> list[str]:
|
||
return path.read_text(encoding="utf-8").splitlines() if path.is_file() else []
|
||
|
||
|
||
def parse_entries(lines: list[str]) -> tuple[dict[str, dict], list[str]]:
|
||
"""Строки индекса по имени файла + порядок секций.
|
||
|
||
Дубли имени файла тут схлопываются (побеждает последний) — их отдельно
|
||
ловит index_lint, поэтому опираться на этот dict как на полноту нельзя.
|
||
"""
|
||
entries: dict[str, dict] = {}
|
||
sections: list[str] = []
|
||
section = None
|
||
for num, line in enumerate(lines, 1):
|
||
m = SECTION.match(line)
|
||
if m:
|
||
section = m.group(1)
|
||
sections.append(section)
|
||
continue
|
||
m = INDEX_ENTRY.match(line)
|
||
if m:
|
||
title, target, hook = m.group(1), m.group(2), (m.group(3) or "").strip()
|
||
entries[os.path.basename(target)] = {
|
||
"title": title, "section": section, "hook": hook,
|
||
"line": num, "target": target}
|
||
return entries, sections
|
||
|
||
|
||
def index_lint(lines: list[str], label: str) -> list[str]:
|
||
"""Структурные дефекты индекса, которых схлопнутый dict не видит: битые
|
||
строки-пункты, дубли на один файл, задачи до первой секции."""
|
||
errors: list[str] = []
|
||
section = None
|
||
seen: dict[str, int] = {}
|
||
for num, line in enumerate(lines, 1):
|
||
if (m := SECTION.match(line)):
|
||
section = m.group(1)
|
||
continue
|
||
if not line.startswith("- ["):
|
||
continue
|
||
m = INDEX_ENTRY.match(line)
|
||
if not m:
|
||
errors.append(f"{label}:{num}: строка-пункт не по формату"
|
||
f" «- [Заголовок](items/slug.md) — хук»")
|
||
continue
|
||
target = os.path.basename(m.group(2))
|
||
if section is None:
|
||
errors.append(f"{label}:{num}: {target} стоит до первой секции")
|
||
if target in seen:
|
||
errors.append(f"{label}:{num}: дубль строки для {target}"
|
||
f" (первая — строка {seen[target]})")
|
||
else:
|
||
seen[target] = num
|
||
return errors
|
||
|
||
|
||
# --- Чтение файлов задач ---
|
||
|
||
def body_sections(text: str) -> dict[str, str]:
|
||
"""Содержимое разделов `## …` тела — по нижнему регистру заголовка.
|
||
|
||
Комментарии-плейсхолдеры вырезаются: раздел, в котором остался только
|
||
подсказочный `<!-- … -->`, считается пустым, иначе шаблон `add` сам себя
|
||
засчитывал бы за заполненные критерии.
|
||
"""
|
||
text = re.sub(r"<!--.*?-->", "", text, flags=re.S)
|
||
out: dict[str, list[str]] = {}
|
||
cur = None
|
||
for line in text.splitlines()[1:]:
|
||
if (m := SECTION.match(line)):
|
||
cur = m.group(1).strip().lower()
|
||
out.setdefault(cur, [])
|
||
continue
|
||
if cur is not None:
|
||
out[cur].append(line)
|
||
return {k: "\n".join(v).strip() for k, v in out.items()}
|
||
|
||
|
||
def parse_task(path: Path) -> dict:
|
||
text = path.read_text(encoding="utf-8")
|
||
lines = text.splitlines()
|
||
title = lines[0].removeprefix("#").strip() if lines and lines[0].startswith("#") else ""
|
||
kind, bare = PLAIN_TYPE, title
|
||
if (m := TYPE_PREFIX.match(title)):
|
||
kind, bare = m.group(1).strip().lower(), m.group(2).strip()
|
||
# Мета-строка — первая непустая строка после заголовка (task-format.md).
|
||
# Поля разделены `·`, порядок свободный: секция распознаётся, где бы она ни
|
||
# стояла. Причина не должна содержать `·` — это разделитель.
|
||
meta = next((ln.strip() for ln in lines[1:] if ln.strip()), "")
|
||
section, reason, tags = "", "", []
|
||
if META_FIELD.match(meta):
|
||
for chunk in meta.split("·"):
|
||
f = META_FIELD.match(chunk.strip())
|
||
if not f:
|
||
continue
|
||
key, value = f.group(1).strip().lower(), f.group(2).strip()
|
||
if key in ("секция", "section"):
|
||
section, _, reason = (p.strip() for p in value.partition("—"))
|
||
section = section.rstrip(".,").lower()
|
||
elif key in ("теги", "tags"):
|
||
tags = [t.strip().lower() for t in value.split(",") if t.strip()]
|
||
goal = next((t[len(GOAL_TAG):] for t in tags if t.startswith(GOAL_TAG)), "")
|
||
return {"title": title, "bare": bare, "type": kind, "section": section,
|
||
"reason": reason, "tags": tags, "goal": goal, "path": path,
|
||
"text": text, "body": body_sections(text)}
|
||
|
||
|
||
def tasks_of(lay: Layout) -> dict[str, dict]:
|
||
if not lay.items.is_dir():
|
||
return {}
|
||
return {p.name: parse_task(p) for p in sorted(lay.items.glob("*.md"))}
|
||
|
||
|
||
def home_index(task: dict) -> str:
|
||
"""Индекс, которому задача принадлежит по типу. Спринт — исключение: туда
|
||
задача попадает не по типу, а решением набора."""
|
||
return "plan" if task["type"] == GOAL else "backlog"
|
||
|
||
|
||
def touched_map(lay: Layout) -> dict[str, str]:
|
||
"""Дата последнего коммита каждого файла задач — одним вызовом git. Ключ —
|
||
имя файла (в items/ имена уникальны). Нет git / нет истории → пустая карта,
|
||
вызывающий подставит «—»."""
|
||
try:
|
||
out = subprocess.run(["git", "log", "--format=%as", "--name-only", "--", str(lay.items)],
|
||
capture_output=True, text=True).stdout
|
||
except FileNotFoundError:
|
||
return {}
|
||
dates: dict[str, str] = {}
|
||
cur = None
|
||
for line in out.splitlines():
|
||
if not line.strip():
|
||
continue
|
||
if DATE_RE.fullmatch(line):
|
||
cur = line # лог новейшие сверху → первая дата и есть последняя правка
|
||
elif cur:
|
||
dates.setdefault(os.path.basename(line), cur)
|
||
return dates
|
||
|
||
|
||
# --- Спринт ---
|
||
|
||
def sprint_goal(lay: Layout) -> tuple[str, str]:
|
||
"""Слаг и заголовок цели текущего спринта; ('', '') — спринта нет."""
|
||
for line in read_lines(lay.index("sprint")):
|
||
if (m := GOAL_LINE.match(line.strip())):
|
||
return Path(m.group(2)).stem, m.group(1)
|
||
return "", ""
|
||
|
||
|
||
def sprint_header(lay: Layout, goal_slug: str, goal_title: str, date: str) -> list[str]:
|
||
link = f"{lay.cfg['items']}/{goal_slug}.md"
|
||
return ["# Спринт", "",
|
||
f"**Цель:** [{goal_title}]({link}) · **Начат:** {date}", ""]
|
||
|
||
|
||
def empty_sprint(lay: Layout) -> str:
|
||
return ("# Спринт\n\n"
|
||
"Спринта нет. Цель называет человек, набор собирает агент:\n"
|
||
"`tasks.py sprint start --goal <слаг>`.\n\n"
|
||
f"## {lay.cfg['sprint_section']}\n")
|
||
|
||
|
||
# --- check ---
|
||
|
||
def check(lay: Layout, fix: bool = False) -> int:
|
||
if fix:
|
||
fixed = apply_fixes(lay)
|
||
for line in fixed:
|
||
print(f"ПОЧИНЕНО {line}")
|
||
if fixed:
|
||
print()
|
||
|
||
idx = {k: parse_entries(read_lines(lay.index(k))) for k in lay.indexes}
|
||
entries = {k: v[0] for k, v in idx.items()}
|
||
sections = {k: v[1] for k, v in idx.items()}
|
||
tasks = tasks_of(lay)
|
||
errors: list[str] = []
|
||
notes: list[str] = []
|
||
label = {k: lay.name(k) for k in lay.indexes}
|
||
|
||
known = {k: {s.lower() for s in sections[k]} for k in lay.indexes}
|
||
goal_slugs = {p[:-3] for p, t in tasks.items() if t["type"] == GOAL}
|
||
goal_of_sprint, _ = sprint_goal(lay)
|
||
crit = lay.cfg["criteria_heading"].lower()
|
||
ques = lay.cfg["questions_heading"].lower()
|
||
|
||
for name, task in tasks.items():
|
||
where = [k for k in lay.indexes if name in entries[k]]
|
||
home = home_index(task)
|
||
|
||
if not SLUG.fullmatch(name):
|
||
errors.append(f"{name}: слаг не kebab-case латиницей")
|
||
if not task["title"]:
|
||
errors.append(f"{name}: нет заголовка H1")
|
||
if task["type"] not in TYPES and task["type"] != PLAIN_TYPE:
|
||
notes.append(f"{name}: тип «{task['type']}» вне словаря"
|
||
f" ({'/'.join(TYPES)} или без префикса)")
|
||
|
||
# 1. Задача живёт ровно в одном индексе за раз.
|
||
allowed = {home} | ({"sprint"} if home == "backlog" else set())
|
||
if not where:
|
||
errors.append(f"{name}: нет ни в одном индексе (ожидался {label[home]})")
|
||
elif len(where) > 1:
|
||
errors.append(f"{name}: сразу в нескольких индексах"
|
||
f" ({', '.join(label[k] for k in where)}) — задача живёт в одном")
|
||
elif where[0] not in allowed:
|
||
errors.append(f"{name}: лежит в {label[where[0]]}, а по типу"
|
||
f" «{task['type']}» её место в {label[home]}")
|
||
|
||
place = where[0] if len(where) == 1 else None
|
||
entry = entries[place][name] if place else None
|
||
|
||
# 2. Секция файла — истина, секция индекса производна. В спринте секция
|
||
# файла хранит, куда задачу вернуть, и с «Набором» не сверяется.
|
||
if not task["section"]:
|
||
errors.append(f"{name}: нет поля **Секция:** в мета-строке")
|
||
elif task["section"] not in known[home]:
|
||
errors.append(f"{name}: секция «{task['section']}» не совпадает ни с одной"
|
||
f" секцией {label[home]} ({', '.join(sections[home])})")
|
||
elif place and place != "sprint" and entry["section"] \
|
||
and entry["section"].lower() != task["section"]:
|
||
errors.append(f"{name}: секция в файле «{task['section']}»,"
|
||
f" а в {label[place]} — «{entry['section']}»")
|
||
|
||
if entry and entry["title"] != task["title"]:
|
||
errors.append(f"{name}: заголовок разошёлся\n"
|
||
f" файл: {task['title']}\n"
|
||
f" индекс: {entry['title']}")
|
||
if entry and not entry["hook"]:
|
||
notes.append(f"{name}: строка индекса без хука — по ней не выбрать задачу")
|
||
|
||
# 3. Цель. Связь однонаправленна: задача несёт goal:<слаг>, перечень
|
||
# задач цели выводится `list --goal`, а не хранится.
|
||
if task["type"] != GOAL:
|
||
if not task["goal"]:
|
||
if task["type"] == "idea":
|
||
notes.append(f"{name}: идея без цели — цель проставляется,"
|
||
f" когда идея становится задачей")
|
||
else:
|
||
errors.append(f"{name}: нет тега {GOAL_TAG}<слаг> — задача вне цели"
|
||
f" не попадёт ни в один спринт")
|
||
elif task["goal"] not in goal_slugs:
|
||
errors.append(f"{name}: тег {GOAL_TAG}{task['goal']} указывает на цель,"
|
||
f" которой нет в {lay.cfg['items']}/")
|
||
|
||
# 4. Спринт: задача с открытым вопросом в набор не берётся.
|
||
if place == "sprint":
|
||
if QUESTION_TAG in task["tags"]:
|
||
errors.append(f"{name}: в спринте, но с тегом «{QUESTION_TAG}» —"
|
||
f" вопрос разбирается до взятия")
|
||
if task["type"] not in TAKEABLE:
|
||
errors.append(f"{name}: в спринте тип «{task['type']}»"
|
||
f" — берутся только задачи")
|
||
if goal_of_sprint and task["goal"] and task["goal"] != goal_of_sprint:
|
||
errors.append(f"{name}: цель задачи «{task['goal']}» не цель спринта"
|
||
f" «{goal_of_sprint}» — набор служит одной цели")
|
||
if crit not in task["body"] or not task["body"].get(crit):
|
||
errors.append(f"{name}: в спринте без раздела «{lay.cfg['criteria_heading']}»"
|
||
f" — принимать будет не по чему")
|
||
|
||
# 5. Тег «question» и раздел вопросов — один факт, два места.
|
||
has_q_section = bool(task["body"].get(ques))
|
||
if has_q_section and QUESTION_TAG not in task["tags"]:
|
||
notes.append(f"{name}: раздел «{lay.cfg['questions_heading']}» есть,"
|
||
f" а тега «{QUESTION_TAG}» нет")
|
||
if QUESTION_TAG in task["tags"] and not has_q_section:
|
||
notes.append(f"{name}: тег «{QUESTION_TAG}» есть, а раздела"
|
||
f" «{lay.cfg['questions_heading']}» нет — снять тег?")
|
||
|
||
if BODY_PLACEHOLDER in task["text"]:
|
||
notes.append(f"{name}: тело не дописано (остался плейсхолдер add)")
|
||
|
||
for kind in lay.indexes:
|
||
for name, entry in entries[kind].items():
|
||
if name not in tasks:
|
||
errors.append(f"{label[kind]}:{entry['line']}: ссылка на несуществующий {name}")
|
||
errors += index_lint(read_lines(lay.index(kind)), label[kind])
|
||
|
||
if goal_of_sprint and goal_of_sprint + ".md" not in tasks:
|
||
errors.append(f"{label['sprint']}: цель «{goal_of_sprint}» не найдена"
|
||
f" в {lay.cfg['items']}/")
|
||
if not goal_of_sprint and entries["sprint"]:
|
||
errors.append(f"{label['sprint']}: набор есть, а цель не названа")
|
||
|
||
rejected = lay.index("rejected")
|
||
if rejected.is_file():
|
||
for num, line in enumerate(read_lines(rejected), 1):
|
||
if line.startswith("- ") and not REJECTED_ENTRY.match(line):
|
||
errors.append(f"{label.get('rejected', lay.name('rejected'))}:{num}:"
|
||
f" строка не по формату «- ГГГГ-ММ-ДД `slug` — …»")
|
||
|
||
print(f"задачи: {lay.root}, файлов {len(tasks)}, строк индексов "
|
||
+ ", ".join(f"{label[k]} {len(entries[k])}" for k in lay.indexes))
|
||
health(lay, tasks, entries, sections)
|
||
for e in errors:
|
||
print(f"ОШИБКА {e}")
|
||
for n in notes:
|
||
print(f"замечание {n}")
|
||
if errors:
|
||
print(f"\nрасхождений: {len(errors)}")
|
||
return 1
|
||
print("\nиндексы согласованы" + (f", замечаний: {len(notes)}" if notes else ""))
|
||
return 0
|
||
|
||
|
||
def health(lay: Layout, tasks: dict, entries: dict, sections: dict) -> None:
|
||
"""Метрики здоровья: размер секций, спринт, открытые вопросы, залежалость,
|
||
цели без задач. Механизирует то, что иначе держится на дисциплине."""
|
||
backlog = [t for n, t in tasks.items() if n in entries["backlog"]]
|
||
by_section = {s.lower(): 0 for s in sections["backlog"]}
|
||
for t in backlog:
|
||
if t["section"] in by_section:
|
||
by_section[t["section"]] += 1
|
||
if by_section:
|
||
print(" беклог: " + ", ".join(f"{s} {by_section[s.lower()]}" for s in sections["backlog"]))
|
||
|
||
goal_slug, goal_title = sprint_goal(lay)
|
||
if goal_slug:
|
||
print(f" спринт: цель «{goal_slug}», задач {len(entries['sprint'])}")
|
||
else:
|
||
print(" спринт: не начат")
|
||
|
||
questions = [n for n, t in tasks.items() if QUESTION_TAG in t["tags"]]
|
||
if questions:
|
||
print(f" с открытым вопросом: {len(questions)}"
|
||
f" — в спринт не берутся, разбор первым шагом сессии")
|
||
|
||
goals = {n[:-3]: t for n, t in tasks.items() if t["type"] == GOAL}
|
||
if goals:
|
||
counts = {g: sum(1 for t in tasks.values() if t["goal"] == g) for g in goals}
|
||
empty = [g for g, c in counts.items() if c == 0]
|
||
print(f" цели: {len(goals)}, из них без открытых задач {len(empty)}"
|
||
+ (f" ({', '.join(sorted(empty))})" if empty else ""))
|
||
|
||
dates = touched_map(lay)
|
||
if not dates:
|
||
return
|
||
cutoff = (datetime.date.today() - datetime.timedelta(days=STALE_DAYS)).isoformat()
|
||
stale = sum(1 for t in tasks.values() if (d := dates.get(t["path"].name)) and d < cutoff)
|
||
if stale:
|
||
print(f" залежалось (>{STALE_DAYS} дней без правки): {stale}"
|
||
f" — переоценка просрочена, начни с `list --stale`")
|
||
|
||
|
||
# --- list ---
|
||
|
||
def list_tasks(lay: Layout, a: argparse.Namespace) -> int:
|
||
tasks = tasks_of(lay)
|
||
entries = {k: parse_entries(read_lines(lay.index(k)))[0] for k in lay.indexes}
|
||
order = {}
|
||
for k in lay.indexes:
|
||
for i, s in enumerate(parse_entries(read_lines(lay.index(k)))[1]):
|
||
order.setdefault(s.lower(), i)
|
||
|
||
def place(name: str) -> str:
|
||
for k in lay.indexes:
|
||
if name in entries[k]:
|
||
return k
|
||
return "—"
|
||
|
||
rows = []
|
||
for name, t in tasks.items():
|
||
t["place"] = place(name)
|
||
if a.index and a.index != "all" and t["place"] != a.index:
|
||
continue
|
||
if a.section and t["section"] != a.section.lower():
|
||
continue
|
||
if a.type and t["type"] != a.type.lower():
|
||
continue
|
||
if a.tag and a.tag.lower() not in t["tags"]:
|
||
continue
|
||
if a.goal and t["goal"] != a.goal.lower():
|
||
continue
|
||
if a.questions and QUESTION_TAG not in t["tags"]:
|
||
continue
|
||
rows.append(t)
|
||
|
||
if a.stale:
|
||
dates = touched_map(lay)
|
||
for t in rows:
|
||
t["touched"] = dates.get(t["path"].name, "—")
|
||
rows.sort(key=lambda t: (t["touched"] == "—", t["touched"]))
|
||
else:
|
||
rows.sort(key=lambda t: (order.get(t["section"], 99), t["path"].name))
|
||
|
||
for t in rows:
|
||
touched = f"{t.get('touched', ''):<11}" if a.stale else ""
|
||
kind = "" if t["type"] == PLAIN_TYPE else f"[{t['type']}] "
|
||
goal = f" →{t['goal']}" if t["goal"] and not a.goal else ""
|
||
flag = " ?" if QUESTION_TAG in t["tags"] else " "
|
||
print(f"{touched}{t['place']:<8}{t['section']:<8}{flag} "
|
||
f"{t['path'].stem:<44} {kind}{t['bare']}{goal}")
|
||
print(f"\nвсего: {len(rows)}")
|
||
return 0
|
||
|
||
|
||
# --- Мутации: правят файл и индексы заодно, рассогласовать их вручную нельзя ---
|
||
|
||
def fail(msg: str) -> int:
|
||
print(f"ошибка: {msg}", file=sys.stderr)
|
||
return 2
|
||
|
||
|
||
def build_meta(section: str, reason: str, tags: list[str]) -> str:
|
||
s = f"**Секция:** {section}"
|
||
if reason:
|
||
s += f" — {reason}"
|
||
if tags:
|
||
s += " · **Теги:** " + ", ".join(tags)
|
||
return s
|
||
|
||
|
||
def save_index(lay: Layout, kind: str, lines: list[str]) -> None:
|
||
write_atomic(lay.index(kind), "\n".join(lines) + "\n")
|
||
|
||
|
||
def section_headers(lines: list[str]) -> list[tuple[int, str]]:
|
||
return [(i, m.group(1)) for i, l in enumerate(lines) if (m := SECTION.match(l))]
|
||
|
||
|
||
def find_section(lines: list[str], name: str) -> tuple[int | None, str]:
|
||
for i, s in section_headers(lines):
|
||
if s.lower() == name.lower():
|
||
return i, s
|
||
return None, ""
|
||
|
||
|
||
def find_entry_index(lines: list[str], slug: str) -> int | None:
|
||
for i, l in enumerate(lines):
|
||
m = INDEX_ENTRY.match(l)
|
||
if m and Path(m.group(2)).name == f"{slug}.md":
|
||
return i
|
||
return None
|
||
|
||
|
||
def locate(lay: Layout, slug: str) -> tuple[str | None, list[str], int | None]:
|
||
"""В каком индексе лежит строка задачи: (вид, строки индекса, номер строки)."""
|
||
for kind in lay.indexes:
|
||
lines = read_lines(lay.index(kind))
|
||
ei = find_entry_index(lines, slug)
|
||
if ei is not None:
|
||
return kind, lines, ei
|
||
return None, [], None
|
||
|
||
|
||
def insert_entry(lines: list[str], section: str, entry: str,
|
||
after: str | None = None, first: bool = False) -> None:
|
||
"""Вставляет строку в секцию: по умолчанию в конец, --after <слаг> — следом
|
||
за указанной строкой, --first — первой. Порядок нужен только упорядоченной
|
||
части плана; в беклоге он значения не имеет."""
|
||
hi, _ = find_section(lines, section)
|
||
end = next((j for j in range(hi + 1, len(lines)) if SECTION.match(lines[j])), len(lines))
|
||
if first:
|
||
ins = hi + 1
|
||
while ins < end and not lines[ins].strip():
|
||
ins += 1
|
||
elif after:
|
||
ai = find_entry_index(lines[hi:end], after)
|
||
if ai is None:
|
||
raise KeyError(after)
|
||
ins = hi + ai + 1
|
||
else:
|
||
ins = end
|
||
while ins - 1 > hi and not lines[ins - 1].strip():
|
||
ins -= 1
|
||
lines.insert(ins, entry)
|
||
|
||
|
||
def update_meta(path: Path, section: str | None, new_reason: str | None,
|
||
tags: list[str] | None) -> bool:
|
||
"""Хирургически правит мета-строку файла, сохраняя нераспознанные поля и
|
||
регистр. False — мета-строки нет (тогда правку делать нельзя)."""
|
||
flines = path.read_text(encoding="utf-8").splitlines()
|
||
mi = next((i for i in range(1, len(flines)) if flines[i].strip()), None)
|
||
if mi is None or not META_FIELD.match(flines[mi].strip()):
|
||
return False
|
||
chunks = flines[mi].split("·")
|
||
seen_section = seen_tags = False
|
||
for idx, chunk in enumerate(chunks):
|
||
f = META_FIELD.match(chunk.strip())
|
||
if not f:
|
||
continue
|
||
key = f.group(1).strip().lower()
|
||
lead = chunk[:len(chunk) - len(chunk.lstrip())]
|
||
trail = chunk[len(chunk.rstrip()):]
|
||
if key in ("секция", "section") and section is not None:
|
||
_, _, old_reason = (p.strip() for p in f.group(2).partition("—"))
|
||
reason = new_reason if new_reason is not None else old_reason
|
||
field = f"**Секция:** {section}" + (f" — {reason}" if reason else "")
|
||
chunks[idx] = lead + field + trail
|
||
seen_section = True
|
||
elif key in ("теги", "tags") and tags is not None:
|
||
seen_tags = True
|
||
if tags:
|
||
chunks[idx] = lead + "**Теги:** " + ", ".join(tags) + trail
|
||
else:
|
||
chunks[idx] = None # тегов не осталось — поле убираем
|
||
chunks = [c for c in chunks if c is not None]
|
||
if tags and not seen_tags:
|
||
chunks.append(" **Теги:** " + ", ".join(tags))
|
||
if section is not None and not seen_section:
|
||
return False
|
||
write_atomic(path, "\n".join((*flines[:mi], "·".join(chunks), *flines[mi + 1:])) + "\n")
|
||
return True
|
||
|
||
|
||
def entry_line(lay: Layout, title: str, slug: str, hook: str) -> str:
|
||
link = f"{lay.cfg['items']}/{slug}.md"
|
||
return f"- [{title}]({link})" + (f" — {hook}" if hook else "")
|
||
|
||
|
||
def body_template(kind: str, lay: Layout) -> str:
|
||
if kind == GOAL:
|
||
return ("<!-- зачем эта цель: какое состояние продукта она создаёт -->\n\n"
|
||
"## Завершение\n\n"
|
||
"<!-- по чему видно, что цель достигнута; задачи здесь НЕ перечисляются"
|
||
" — перечень даёт `tasks.py list --goal <слаг>` -->\n")
|
||
if kind == "idea":
|
||
return ("<!-- что за идея, откуда взялась, чем может быть полезна;"
|
||
" критериев у идеи нет — потому она и идея -->\n")
|
||
return ("<!-- задача в одной фразе: что станет наблюдаемо иначе -->\n\n"
|
||
f"## {lay.cfg['criteria_heading']}\n\n"
|
||
"<!-- 2–5 проверяемых утверждений, у каждого назван оракул -->\n\n"
|
||
"## Рамки\n\n"
|
||
"<!-- одна строка: чего касаться нельзя, что перезапускается,"
|
||
" что считается необратимым -->\n")
|
||
|
||
|
||
def cmd_add(lay: Layout, a: argparse.Namespace) -> int:
|
||
for err in (bad_slug(a.slug), bad_line(a.title, "заголовок"), bad_line(a.hook, "хук"),
|
||
bad_tags(a.tag), bad_reason(a.reason), bad_slug(a.goal) if a.goal else None):
|
||
if err:
|
||
return fail(err)
|
||
if not a.title.strip():
|
||
return fail("пустой заголовок")
|
||
kind = (a.type or "").strip().lower()
|
||
path = lay.items / f"{a.slug}.md"
|
||
if path.exists():
|
||
return fail(f"{path.name} уже существует — дедуп: допиши в него, а не заводи новый")
|
||
|
||
target = "plan" if kind == GOAL else "backlog"
|
||
lines = read_lines(lay.index(target))
|
||
if not lines:
|
||
return fail(f"нет индекса {lay.name(target)} — прогони tasks.py init")
|
||
if find_entry_index(lines, a.slug) is not None:
|
||
return fail(f"строка в {lay.name(target)} для {a.slug} уже есть")
|
||
section = a.section or (section_headers(lines)[0][1] if section_headers(lines) else "")
|
||
hi, section = find_section(lines, section)
|
||
if hi is None:
|
||
avail = ", ".join(n for _, n in section_headers(lines))
|
||
return fail(f"нет секции «{a.section}» в {lay.name(target)} (есть: {avail})")
|
||
|
||
tags = split_tags(a.tag)
|
||
if a.goal:
|
||
tags = [t for t in tags if not t.startswith(GOAL_TAG)] + [f"{GOAL_TAG}{a.goal}"]
|
||
if not (lay.items / f"{a.goal}.md").exists():
|
||
print(f" внимание: цели {a.goal}.md нет — заведи её (--type goal) или поправь тег")
|
||
if kind != GOAL and not any(t.startswith(GOAL_TAG) for t in tags):
|
||
print(" без цели: задача вне цели не попадёт ни в один спринт —"
|
||
f" проставь `tasks.py edit {a.slug} --goal <слаг>`")
|
||
if QUESTION_TAG in tags:
|
||
print(f" тег «{QUESTION_TAG}»: не забудь раздел «{lay.cfg['questions_heading']}» в теле")
|
||
|
||
title_full = f"[{kind}] {a.title}" if kind else a.title
|
||
meta = build_meta(section, a.reason or "", tags)
|
||
lay.items.mkdir(parents=True, exist_ok=True)
|
||
write_atomic(path, f"# {title_full}\n\n{meta}\n\n{body_template(kind, lay)}")
|
||
insert_entry(lines, section, entry_line(lay, title_full, a.slug, a.hook or ""))
|
||
save_index(lay, target, lines)
|
||
|
||
print(f"создано: {lay.cfg['items']}/{a.slug}.md,"
|
||
f" строка в {lay.name(target)} (секция «{section}»); допиши тело редактором")
|
||
if not a.hook:
|
||
print(f" без хука — задай: tasks.py edit {a.slug} --hook …")
|
||
warn_rejected(lay, a.slug, a.title)
|
||
return 0
|
||
|
||
|
||
def warn_rejected(lay: Layout, slug: str, title: str) -> None:
|
||
"""Дедупликация против ушедшего без реализации: та же задача возвращается
|
||
через квартал тем же текстом, и кладбище — единственный её след."""
|
||
path = lay.index("rejected")
|
||
if not path.is_file():
|
||
return
|
||
words = {w for w in re.findall(r"[^\W\d_]{5,}", title.lower())}
|
||
for line in read_lines(path):
|
||
if not REJECTED_ENTRY.match(line):
|
||
continue
|
||
low = line.lower()
|
||
if f"`{slug}`" in low or (words and len(words & set(re.findall(r"[^\W\d_]{5,}", low))) >= 2):
|
||
print(f" похоже на ушедшее без реализации — покажи пользователю, что изменилось:\n"
|
||
f" {line.strip()}")
|
||
|
||
|
||
def cmd_edit(lay: Layout, a: argparse.Namespace) -> int:
|
||
for err in (bad_slug(a.slug), bad_line(a.title, "заголовок"), bad_line(a.hook, "хук"),
|
||
bad_tags(a.add_tag), bad_tags(a.rm_tag), bad_slug(a.goal) if a.goal else None):
|
||
if err:
|
||
return fail(err)
|
||
if all(v is None for v in (a.title, a.hook, a.type, a.goal, a.add_tag, a.rm_tag)):
|
||
return fail("нечего менять: дай --title, --hook, --type, --goal, --add-tag или --rm-tag")
|
||
path = lay.items / f"{a.slug}.md"
|
||
if not path.exists():
|
||
return fail(f"{a.slug}.md не найден в {lay.cfg['items']}/")
|
||
kind_index, lines, ei = locate(lay, a.slug)
|
||
if ei is None:
|
||
return fail(f"строки индекса для {a.slug} нет — прогони check --fix")
|
||
|
||
task = parse_task(path)
|
||
if a.title is not None and not a.title.strip():
|
||
return fail("пустой заголовок")
|
||
bare = a.title if a.title is not None else task["bare"]
|
||
kind = task["type"] if a.type is None else a.type.strip().lower()
|
||
if (kind == GOAL) != (task["type"] == GOAL):
|
||
return fail("тип goal не меняется на месте: цель и задача живут в разных индексах —"
|
||
" заведи новую запись и закрой старую с причиной")
|
||
prefix = "" if kind in ("", PLAIN_TYPE) else f"[{kind}] "
|
||
h1 = f"{prefix}{bare}"
|
||
|
||
flines = path.read_text(encoding="utf-8").splitlines()
|
||
if not flines or not flines[0].startswith("#"):
|
||
return fail(f"{a.slug}.md без заголовка H1 — прогони check")
|
||
flines[0] = f"# {h1}"
|
||
write_atomic(path, "\n".join(flines) + "\n")
|
||
|
||
tags = list(task["tags"])
|
||
if a.rm_tag is not None:
|
||
drop = set(split_tags(a.rm_tag))
|
||
tags = [t for t in tags if t not in drop]
|
||
if a.add_tag is not None:
|
||
for t in split_tags(a.add_tag):
|
||
if t not in tags:
|
||
tags.append(t)
|
||
if a.goal is not None:
|
||
tags = [t for t in tags if not t.startswith(GOAL_TAG)] + [f"{GOAL_TAG}{a.goal}"]
|
||
if not (lay.items / f"{a.goal}.md").exists():
|
||
print(f" внимание: цели {a.goal}.md нет — заведи её или поправь тег")
|
||
if tags != task["tags"] and not update_meta(path, None, None, tags):
|
||
return fail(f"{a.slug}.md без мета-строки — прогони check и почини")
|
||
|
||
m = INDEX_ENTRY.match(lines[ei])
|
||
hook = a.hook if a.hook is not None else (m.group(3) or "").strip()
|
||
lines[ei] = entry_line(lay, h1, a.slug, hook)
|
||
save_index(lay, kind_index, lines)
|
||
|
||
changed = [n for n, v in (("заголовок", a.title), ("хук", a.hook), ("тип", a.type),
|
||
("цель", a.goal), ("теги", a.add_tag or a.rm_tag))
|
||
if v is not None]
|
||
print(f"{a.slug}: обновлено ({', '.join(changed)})")
|
||
if QUESTION_TAG in tags and QUESTION_TAG not in task["tags"] and kind_index == "sprint":
|
||
print(" задача в спринте, а вопрос открыт: либо остаток есть и вопрос ждёт сессии,"
|
||
f" либо задача выходит — `tasks.py sprint drop {a.slug} --reason …`")
|
||
return 0
|
||
|
||
|
||
def cmd_move(lay: Layout, a: argparse.Namespace) -> int:
|
||
for err in (bad_slug(a.slug), bad_reason(a.reason), bad_slug(a.after) if a.after else None):
|
||
if err:
|
||
return fail(err)
|
||
path = lay.items / f"{a.slug}.md"
|
||
if not path.exists():
|
||
return fail(f"{a.slug}.md не найден в {lay.cfg['items']}/")
|
||
kind_index, lines, ei = locate(lay, a.slug)
|
||
if ei is None:
|
||
return fail(f"строки индекса для {a.slug} нет — прогони check --fix")
|
||
if kind_index == "sprint":
|
||
return fail(f"{a.slug} в спринте: секция — свойство беклога."
|
||
f" Сперва верни задачу: tasks.py sprint drop {a.slug} --reason …")
|
||
hi, section = find_section(lines, a.section)
|
||
if hi is None:
|
||
avail = ", ".join(n for _, n in section_headers(lines))
|
||
return fail(f"нет секции «{a.section}» в {lay.name(kind_index)} (есть: {avail})")
|
||
if not update_meta(path, section, a.reason, None):
|
||
return fail(f"{a.slug}.md без мета-строки **Секция:** — прогони check и почини")
|
||
entry = lines.pop(ei)
|
||
try:
|
||
insert_entry(lines, section, entry, a.after, a.first)
|
||
except KeyError:
|
||
return fail(f"--after {a.after}: такой строки в секции «{section}» нет")
|
||
save_index(lay, kind_index, lines)
|
||
print(f"{a.slug}: перенесено в «{section}» ({lay.name(kind_index)})")
|
||
return 0
|
||
|
||
|
||
def cmd_close(lay: Layout, a: argparse.Namespace) -> int:
|
||
for err in (bad_slug(a.slug), bad_reason(a.reason)):
|
||
if err:
|
||
return fail(err)
|
||
path = lay.items / f"{a.slug}.md"
|
||
if not path.exists():
|
||
return fail(f"{a.slug}.md не найден в {lay.cfg['items']}/")
|
||
kind_index, lines, ei = locate(lay, a.slug)
|
||
if ei is None:
|
||
return fail(f"строки индекса для {a.slug} нет — прогони check --fix")
|
||
task = parse_task(path)
|
||
if task["type"] == GOAL:
|
||
open_tasks = [n for n, t in tasks_of(lay).items() if t["goal"] == a.slug]
|
||
if open_tasks:
|
||
return fail(f"у цели {a.slug} осталось открытых задач: {len(open_tasks)}"
|
||
f" ({', '.join(sorted(t[:-3] for t in open_tasks))})."
|
||
f" Цель закрыта, когда не осталось её задач")
|
||
if a.reason:
|
||
reason = a.reason.rstrip()
|
||
dot = "" if reason.endswith((".", "!", "?")) else "."
|
||
date = datetime.date.today().isoformat()
|
||
bullet = (f"- {date} `{a.slug}` — {task['title']}. Причина: {reason}{dot}"
|
||
f" Была секция: {task['section'] or '—'}.")
|
||
rej = lay.index("rejected")
|
||
prev = rej.read_text(encoding="utf-8") if rej.exists() else "# Ушедшее без реализации\n"
|
||
if not prev.endswith("\n"):
|
||
prev += "\n"
|
||
write_atomic(rej, prev + bullet + "\n")
|
||
# Порядок: индекс без строки → потом unlink. Обратный порядок оставил бы
|
||
# ссылку в никуда, если бы unlink упал.
|
||
lines.pop(ei)
|
||
save_index(lay, kind_index, lines)
|
||
path.unlink()
|
||
print(f"{a.slug}: {'записано в ' + lay.name('rejected') + ' + удалено' if a.reason else 'удалено (реализовано, есть коммит)'}")
|
||
if kind_index == "sprint" and a.reason:
|
||
print(" задача закрыта прямо из спринта без реализации — назови это в докладе спринта")
|
||
return 0
|
||
|
||
|
||
# --- Спринт ---
|
||
|
||
def cmd_sprint_start(lay: Layout, a: argparse.Namespace) -> int:
|
||
if (err := bad_slug(a.goal)):
|
||
return fail(err)
|
||
entries, _ = parse_entries(read_lines(lay.index("sprint")))
|
||
if entries:
|
||
return fail(f"в {lay.name('sprint')} ещё есть набор ({len(entries)}) —"
|
||
f" закрой спринт: tasks.py sprint close")
|
||
gpath = lay.items / f"{a.goal}.md"
|
||
if not gpath.exists():
|
||
return fail(f"цели {a.goal}.md нет в {lay.cfg['items']}/")
|
||
goal = parse_task(gpath)
|
||
if goal["type"] != GOAL:
|
||
return fail(f"{a.goal} не цель (тип «{goal['type']}») — спринт набирается под [goal]")
|
||
date = a.date or datetime.date.today().isoformat()
|
||
if not DATE_RE.fullmatch(date):
|
||
return fail("дата в формате ГГГГ-ММ-ДД")
|
||
lines = sprint_header(lay, a.goal, goal["title"], date) + [f"## {lay.cfg['sprint_section']}", ""]
|
||
save_index(lay, "sprint", lines)
|
||
ready = [n[:-3] for n, t in tasks_of(lay).items()
|
||
if t["goal"] == a.goal and t["type"] in TAKEABLE
|
||
and QUESTION_TAG not in t["tags"]]
|
||
print(f"спринт начат: цель «{a.goal}», {date}")
|
||
print(f" кандидатов под цель без открытых вопросов: {len(ready)}"
|
||
+ (f" ({', '.join(sorted(ready))})" if ready else ""))
|
||
print(" набери: tasks.py sprint take <слаг> …; набор показывается человеку до старта работ")
|
||
return 0
|
||
|
||
|
||
def cmd_sprint_take(lay: Layout, a: argparse.Namespace) -> int:
|
||
goal_slug, _ = sprint_goal(lay)
|
||
if not goal_slug:
|
||
return fail("спринт не начат: tasks.py sprint start --goal <слаг>")
|
||
sprint_lines = read_lines(lay.index("sprint"))
|
||
backlog_lines = read_lines(lay.index("backlog"))
|
||
hi, section = find_section(sprint_lines, lay.cfg["sprint_section"])
|
||
if hi is None:
|
||
return fail(f"в {lay.name('sprint')} нет секции «{lay.cfg['sprint_section']}»")
|
||
taken = []
|
||
for slug in a.slugs:
|
||
if (err := bad_slug(slug)):
|
||
return fail(err)
|
||
path = lay.items / f"{slug}.md"
|
||
if not path.exists():
|
||
return fail(f"{slug}.md не найден в {lay.cfg['items']}/")
|
||
t = parse_task(path)
|
||
if t["type"] not in TAKEABLE:
|
||
return fail(f"{slug}: тип «{t['type']}» в спринт не берётся —"
|
||
f" идея идёт на штурм, эпик на декомпозицию, цель не берут вовсе")
|
||
if t["goal"] != goal_slug:
|
||
return fail(f"{slug}: цель «{t['goal'] or '—'}» не цель спринта «{goal_slug}» —"
|
||
f" набор служит одной цели, даже если взять удобно")
|
||
if QUESTION_TAG in t["tags"]:
|
||
return fail(f"{slug}: открыт вопрос — разбирается до взятия,"
|
||
f" вне очереди порции (`tasks.py edit {slug} --rm-tag {QUESTION_TAG}`)")
|
||
crit = lay.cfg["criteria_heading"].lower()
|
||
if not t["body"].get(crit):
|
||
return fail(f"{slug}: нет раздела «{lay.cfg['criteria_heading']}» —"
|
||
f" принимать будет не по чему")
|
||
ei = find_entry_index(backlog_lines, slug)
|
||
if ei is None:
|
||
return fail(f"{slug}: строки в {lay.name('backlog')} нет"
|
||
f" (уже в спринте? прогони check)")
|
||
entry = backlog_lines.pop(ei)
|
||
insert_entry(sprint_lines, section, entry)
|
||
taken.append(slug)
|
||
save_index(lay, "backlog", backlog_lines)
|
||
save_index(lay, "sprint", sprint_lines)
|
||
total = len(parse_entries(sprint_lines)[0])
|
||
print(f"взято в спринт: {', '.join(taken)}; в наборе {total}")
|
||
return 0
|
||
|
||
|
||
def cmd_sprint_drop(lay: Layout, a: argparse.Namespace) -> int:
|
||
if (err := bad_reason(a.reason)):
|
||
return fail(err)
|
||
sprint_lines = read_lines(lay.index("sprint"))
|
||
backlog_lines = read_lines(lay.index("backlog"))
|
||
dropped = []
|
||
for slug in a.slugs:
|
||
if (err := bad_slug(slug)):
|
||
return fail(err)
|
||
ei = find_entry_index(sprint_lines, slug)
|
||
if ei is None:
|
||
return fail(f"{slug}: в наборе спринта такой строки нет")
|
||
path = lay.items / f"{slug}.md"
|
||
t = parse_task(path)
|
||
hi, section = find_section(backlog_lines, t["section"])
|
||
if hi is None:
|
||
return fail(f"{slug}: секция «{t['section']}» не найдена в {lay.name('backlog')}")
|
||
if not update_meta(path, section, a.reason, None):
|
||
return fail(f"{slug}.md без мета-строки **Секция:** — прогони check и почини")
|
||
insert_entry(backlog_lines, section, sprint_lines.pop(ei))
|
||
dropped.append(slug)
|
||
save_index(lay, "backlog", backlog_lines)
|
||
save_index(lay, "sprint", sprint_lines)
|
||
print(f"вышло из спринта: {', '.join(dropped)} — причина записана в мета-строку")
|
||
print(f" живого предложения оставаться не должно; наработки, которые жалко,"
|
||
f" переносятся в тело задачи текстом")
|
||
return 0
|
||
|
||
|
||
def cmd_sprint_close(lay: Layout, a: argparse.Namespace) -> int:
|
||
if (err := bad_reason(a.reason)):
|
||
return fail(err)
|
||
entries, _ = parse_entries(read_lines(lay.index("sprint")))
|
||
goal_slug, _ = sprint_goal(lay)
|
||
if entries and not a.dissolve:
|
||
return fail(f"в наборе осталось задач: {len(entries)}"
|
||
f" ({', '.join(sorted(n[:-3] for n in entries))}). Спринт кончается, когда"
|
||
f" каждая либо сделана (close --implemented), либо вышла (sprint drop"
|
||
f" --reason …). Роспуск при блокере — sprint close --dissolve --reason …")
|
||
if a.dissolve:
|
||
if not a.reason:
|
||
return fail("--dissolve без --reason: роспуск объясняется блокером")
|
||
drop = argparse.Namespace(slugs=sorted(n[:-3] for n in entries), reason=a.reason)
|
||
if entries and (rc := cmd_sprint_drop(lay, drop)) != 0:
|
||
return rc
|
||
write_atomic(lay.index("sprint"), empty_sprint(lay))
|
||
print(f"спринт закрыт (цель «{goal_slug or '—'}»)"
|
||
+ (", набор распущен" if a.dissolve and entries else ""))
|
||
print(f" история наборов остаётся в git: `git log -p {lay.index('sprint')}`")
|
||
print(" дальше — сессия: разбор вопросов → разбор спринта → переоценка → новый набор")
|
||
return 0
|
||
|
||
|
||
# --- check --fix ---
|
||
|
||
def apply_fixes(lay: Layout) -> list[str]:
|
||
"""Детерминированная починка дрейфа. Чинит только безопасное, где истина
|
||
однозначно в файле: дубли строк, рассинхрон заголовка, задача не в своей
|
||
секции, отсутствующая строка. Неоднозначное (ссылка на исчезнувший файл,
|
||
задача сразу в двух индексах, битые строки) не трогает — это на суд человека."""
|
||
fixed: list[str] = []
|
||
tasks = tasks_of(lay)
|
||
idx = {k: read_lines(lay.index(k)) for k in lay.indexes}
|
||
dirty: set[str] = set()
|
||
|
||
# 1. Дубли строк на один файл — оставляем первую.
|
||
for kind, lines in idx.items():
|
||
seen: set[str] = set()
|
||
out: list[str] = []
|
||
for l in lines:
|
||
m = INDEX_ENTRY.match(l)
|
||
if m and Path(m.group(2)).name in seen:
|
||
fixed.append(f"{lay.name(kind)}: убран дубль строки {Path(m.group(2)).name}")
|
||
dirty.add(kind)
|
||
continue
|
||
if m:
|
||
seen.add(Path(m.group(2)).name)
|
||
out.append(l)
|
||
idx[kind] = out
|
||
|
||
# 2. Заголовок в индексе разошёлся с H1 — истина в файле, хук сохраняем.
|
||
for kind, lines in idx.items():
|
||
for i, l in enumerate(lines):
|
||
m = INDEX_ENTRY.match(l)
|
||
if not m:
|
||
continue
|
||
name = Path(m.group(2)).name
|
||
task = tasks.get(name)
|
||
if task and m.group(1) != task["title"]:
|
||
lines[i] = entry_line(lay, task["title"], name[:-3], (m.group(3) or "").strip())
|
||
fixed.append(f"{lay.name(kind)}: заголовок синхронизирован с файлом: {name}")
|
||
dirty.add(kind)
|
||
|
||
# 3. Нет строки вовсе / задача не в своей секции. Задачу из спринта не
|
||
# трогаем: «в спринте» — решение набора, а не свойство файла.
|
||
for name, task in tasks.items():
|
||
where = [k for k in lay.indexes if find_entry_index(idx[k], name[:-3]) is not None]
|
||
if len(where) > 1:
|
||
continue # неоднозначно — человеку
|
||
home = home_index(task)
|
||
if not where:
|
||
if not task["section"]:
|
||
continue
|
||
hi, section = find_section(idx[home], task["section"])
|
||
if hi is None:
|
||
continue
|
||
insert_entry(idx[home], section, entry_line(lay, task["title"], name[:-3], ""))
|
||
fixed.append(f"{lay.name(home)}: добавлена строка без хука: {name}")
|
||
dirty.add(home)
|
||
continue
|
||
kind = where[0]
|
||
if kind == "sprint" or kind != home or not task["section"]:
|
||
continue
|
||
hi, section = find_section(idx[kind], task["section"])
|
||
if hi is None:
|
||
continue
|
||
ei = find_entry_index(idx[kind], name[:-3])
|
||
cur = next((SECTION.match(idx[kind][j]).group(1)
|
||
for j in range(ei, -1, -1) if SECTION.match(idx[kind][j])), None)
|
||
if cur and cur.lower() != section.lower():
|
||
insert_entry(idx[kind], section, idx[kind].pop(ei))
|
||
fixed.append(f"{lay.name(kind)}: перенесена в секцию «{section}»: {name}")
|
||
dirty.add(kind)
|
||
|
||
for kind in dirty:
|
||
save_index(lay, kind, idx[kind])
|
||
return fixed
|
||
|
||
|
||
# --- init ---
|
||
|
||
def cmd_init(root: Path, a: argparse.Namespace) -> int:
|
||
if not dir_within_cwd(root):
|
||
return fail(f"--dir вне рабочего каталога: {root}")
|
||
cfg = {k: v for k, v in (("items", a.items), ("backlog", a.backlog), ("plan", a.plan),
|
||
("sprint", a.sprint), ("rejected", a.rejected)) if v}
|
||
lay = Layout(root, cfg)
|
||
if lay.index("backlog").exists():
|
||
return fail(f"{lay.index('backlog')} уже есть — каталог задач заведён")
|
||
|
||
def uniq(raw: str) -> list[str]:
|
||
out, seen = [], set()
|
||
for s in (s.strip() for s in raw.split(",")):
|
||
if s and s.lower() not in seen:
|
||
out.append(s)
|
||
seen.add(s.lower())
|
||
return out
|
||
|
||
sections, plan_sections = uniq(a.sections), uniq(a.plan_sections)
|
||
if not sections or not plan_sections:
|
||
return fail("пустой список секций")
|
||
|
||
root.mkdir(parents=True, exist_ok=True)
|
||
lay.items.mkdir(parents=True, exist_ok=True)
|
||
if cfg:
|
||
write_atomic(root / CONFIG_NAME, json.dumps(cfg, ensure_ascii=False, indent=2) + "\n")
|
||
|
||
write_atomic(lay.index("backlog"),
|
||
"# Беклог\n\n"
|
||
f"Что **можно взять**. Одна задача = один файл `{lay.cfg['items']}/<slug>.md`\n"
|
||
"+ строка здесь. Целей тут нет — они в "
|
||
f"[{lay.name('plan')}]({lay.name('plan')}): беклог — то, что берут,\n"
|
||
"план — то, подо что берут. Порядка внутри секции нет: «что делать\n"
|
||
f"дальше» отвечает набор спринта. Ведётся скиллом `tasks`.\n\n"
|
||
+ "".join(f"## {s}\n\n" for s in sections))
|
||
write_atomic(lay.index("plan"),
|
||
"# План\n\n"
|
||
f"Оглавление целей. Цель — файл `[goal]` в `{lay.cfg['items']}/`; её задачи\n"
|
||
"здесь **не перечисляются** — перечень даёт `tasks.py list --goal <слаг>`.\n"
|
||
f"Первая секция («{plan_sections[0]}») упорядочена, и порядок обосновывается\n"
|
||
"прозой; остальные — тематические кусты без порядка.\n\n"
|
||
+ "".join(f"## {s}\n\n" for s in plan_sections))
|
||
write_atomic(lay.index("sprint"), empty_sprint(lay))
|
||
write_atomic(lay.index("rejected"),
|
||
"# Ушедшее без реализации\n\n"
|
||
"Задачи, покинувшие беклог **без реализации**, с причиной и датой.\n"
|
||
f"Пишется `tasks.py close --reason`. Реализованные сюда не идут — у них\n"
|
||
"есть коммит. Это первое место, куда смотрит дедупликация при заведении.\n\n"
|
||
"<!-- - ГГГГ-ММ-ДД `slug` — Заголовок. Причина: … Была секция: … -->\n")
|
||
print(f"каталог задач заведён: {root}")
|
||
print(f" секции беклога: {', '.join(sections)}; части плана: {', '.join(plan_sections)}")
|
||
if cfg:
|
||
print(f" имена частей записаны в {root / CONFIG_NAME}")
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(prog="tasks.py")
|
||
sub = ap.add_subparsers(dest="command", required=True)
|
||
|
||
p = sub.add_parser("check", help="согласованность файлов и индексов")
|
||
p.add_argument("--dir")
|
||
p.add_argument("--fix", action="store_true",
|
||
help="починить безопасный дрейф (секция, заголовок, дубли)")
|
||
|
||
p = sub.add_parser("list", help="список задач и целей")
|
||
p.add_argument("--dir")
|
||
p.add_argument("--stale", action="store_true", help="от самой залежавшейся")
|
||
p.add_argument("--section")
|
||
p.add_argument("--type", choices=(*TYPES, PLAIN_TYPE))
|
||
p.add_argument("--tag", help="в том числе goal:<слаг>, question, sprint:<слаг>")
|
||
p.add_argument("--goal", help="задачи одной цели (перечень выводится, а не хранится)")
|
||
p.add_argument("--index", choices=("backlog", "sprint", "plan", "all"))
|
||
p.add_argument("--questions", action="store_true", help="только с открытым вопросом")
|
||
|
||
p = sub.add_parser("add", help="завести задачу, идею или цель")
|
||
p.add_argument("--dir")
|
||
p.add_argument("--slug", required=True)
|
||
p.add_argument("--title", required=True)
|
||
p.add_argument("--type", choices=TYPES)
|
||
p.add_argument("--section")
|
||
p.add_argument("--goal", help="слаг цели → тег goal:<слаг>")
|
||
p.add_argument("--hook")
|
||
p.add_argument("--reason")
|
||
p.add_argument("--tag")
|
||
|
||
p = sub.add_parser("edit", help="сменить заголовок/хук/тип/цель/теги")
|
||
p.add_argument("slug")
|
||
p.add_argument("--title")
|
||
p.add_argument("--hook")
|
||
p.add_argument("--type", choices=(*TYPES, PLAIN_TYPE))
|
||
p.add_argument("--goal", help="заменить тег goal:<слаг>")
|
||
p.add_argument("--add-tag", dest="add_tag")
|
||
p.add_argument("--rm-tag", dest="rm_tag")
|
||
p.add_argument("--dir")
|
||
|
||
p = sub.add_parser("move", help="перенести в другую секцию беклога или часть плана")
|
||
p.add_argument("slug")
|
||
p.add_argument("--section", required=True)
|
||
p.add_argument("--reason")
|
||
g = p.add_mutually_exclusive_group()
|
||
g.add_argument("--after", help="встать следом за этим слагом (упорядоченная часть плана)")
|
||
g.add_argument("--first", action="store_true")
|
||
p.add_argument("--dir")
|
||
|
||
p = sub.add_parser("close", help="закрыть задачу или цель")
|
||
p.add_argument("slug")
|
||
g = p.add_mutually_exclusive_group(required=True)
|
||
g.add_argument("--reason", help="ушла без реализации → строка в REJECTED")
|
||
g.add_argument("--implemented", action="store_true", help="реализована → просто удалить")
|
||
p.add_argument("--dir")
|
||
|
||
p = sub.add_parser("sprint", help="операции спринта")
|
||
ssub = p.add_subparsers(dest="sprint_command", required=True)
|
||
s = ssub.add_parser("start", help="начать спринт под названную цель")
|
||
s.add_argument("--goal", required=True)
|
||
s.add_argument("--date")
|
||
s.add_argument("--dir")
|
||
s = ssub.add_parser("take", help="взять задачи в набор")
|
||
s.add_argument("slugs", nargs="+")
|
||
s.add_argument("--dir")
|
||
s = ssub.add_parser("drop", help="вернуть задачи в беклог")
|
||
s.add_argument("slugs", nargs="+")
|
||
s.add_argument("--reason", required=True)
|
||
s.add_argument("--dir")
|
||
s = ssub.add_parser("close", help="закрыть спринт")
|
||
s.add_argument("--dissolve", action="store_true", help="распустить набор (блокер)")
|
||
s.add_argument("--reason")
|
||
s.add_argument("--dir")
|
||
|
||
p = sub.add_parser("init", help="завести каталог задач в новом проекте")
|
||
p.add_argument("--dir")
|
||
p.add_argument("--sections", default=DEFAULT_SECTIONS)
|
||
p.add_argument("--plan-sections", dest="plan_sections", default=DEFAULT_PLAN_SECTIONS)
|
||
p.add_argument("--items")
|
||
p.add_argument("--backlog")
|
||
p.add_argument("--plan")
|
||
p.add_argument("--sprint")
|
||
p.add_argument("--rejected")
|
||
|
||
a = ap.parse_args()
|
||
if a.command == "init":
|
||
return cmd_init(Path(a.dir or "docs/tasks"), a)
|
||
lay = resolve_layout(a.dir)
|
||
if a.command == "sprint":
|
||
return {"start": cmd_sprint_start, "take": cmd_sprint_take,
|
||
"drop": cmd_sprint_drop, "close": cmd_sprint_close}[a.sprint_command](lay, a)
|
||
return {
|
||
"check": lambda: check(lay, a.fix),
|
||
"list": lambda: list_tasks(lay, a),
|
||
"add": lambda: cmd_add(lay, a),
|
||
"edit": lambda: cmd_edit(lay, a),
|
||
"move": lambda: cmd_move(lay, a),
|
||
"close": lambda: cmd_close(lay, a),
|
||
}[a.command]()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|