AirCoding/AirPlan/docs/spec/AirPlan-ParaV2/lib/air_runtime/todo_parser.py

from __future__ import annotations

import re
from pathlib import Path
from typing import Dict, List, Optional

from .contracts import TaskRecord

TASK_ID_RE = re.compile(r"T-\d+")
HEADER_RE = re.compile(r"^\|\s*ID\s*\|\s*Status\s*\|", re.IGNORECASE)
SEPARATOR_RE = re.compile(r"^\|\s*:?-+:?\s*(\|\s*:?-+:?\s*)+\|?$")
CODE_SPAN_RE = re.compile(r"`([^`]+)`")
DEPENDENCY_HINT_RE = re.compile(r"\[(?:deps?|depends)\s*:\s*([^\]]+)\]", re.IGNORECASE)


def _split_row(line: str) -> List[str]:
    return [cell.strip() for cell in line.strip().strip("|").split("|")]


def _normalize_paths(items: List[str]) -> List[str]:
    seen = set()
    ordered: List[str] = []
    for item in items:
        cleaned = item.strip().strip("`")
        if not cleaned or cleaned.lower() == "planned":
            continue
        if cleaned in seen:
            continue
        seen.add(cleaned)
        ordered.append(cleaned)
    return ordered


def extract_paths(cell_text: str) -> List[str]:
    code_paths = CODE_SPAN_RE.findall(cell_text)
    if code_paths:
        return _normalize_paths(code_paths)

    candidates = re.split(r"[,;+]", cell_text)
    paths = [
        token.strip()
        for token in candidates
        if "/" in token or "\\" in token or token.endswith((".md", ".py", ".json"))
    ]
    return _normalize_paths(paths)


def extract_dependencies(*cells: str) -> List[str]:
    found: List[str] = []
    for cell in cells:
        for match in DEPENDENCY_HINT_RE.finditer(cell):
            found.extend(TASK_ID_RE.findall(match.group(1)))
    return _normalize_paths(found)


def extract_global_doc_paths(*cells: str) -> List[str]:
    text = " ".join(cells)
    candidates: List[str] = []
    lowered = text.lower()

    if "agents.md" in lowered or "agents" in text:
        candidates.append("AirPlan/AGENTS.md")
    if "airplan/docs/architecture/adr" in lowered or "docs/architecture/adr" in lowered or "adr" in text:
        candidates.append("AirPlan/docs/architecture/adr/")
    if "airplan/docs/architecture/c4/module.md" in lowered or "docs/architecture/c4/module.md" in lowered or "c4" in text:
        candidates.append("AirPlan/docs/architecture/c4/module.md")
    if "plan.md" in lowered:
        candidates.append("AirPlan/plan.md")
    if "todo.md" in lowered or "todo" in text:
        candidates.append("AirPlan/todo.md")
    if "staticanalysis.md" in lowered:
        candidates.append("AirPlan/docs/staticanalysis.md")

    for path in extract_paths(text):
        normalized = path.replace("\\", "/")
        if normalized in {"AGENTS.md", "AirPlan/AGENTS.md"}:
            candidates.append("AirPlan/AGENTS.md")
        elif "AirPlan/docs/architecture/adr" in normalized or "docs/architecture/adr" in normalized:
            candidates.append("AirPlan/docs/architecture/adr/")
        elif "AirPlan/docs/architecture/c4/module.md" in normalized or "docs/architecture/c4/module.md" in normalized:
            candidates.append("AirPlan/docs/architecture/c4/module.md")
        elif normalized in {"plan.md", "AirPlan/plan.md"}:
            candidates.append("AirPlan/plan.md")
        elif normalized in {"todo.md", "AirPlan/todo.md"}:
            candidates.append("AirPlan/todo.md")
        elif normalized in {"staticanalysis.md", "AirPlan/docs/staticanalysis.md"}:
            candidates.append("AirPlan/docs/staticanalysis.md")

    return _normalize_paths(candidates)


def parse_tasks(todo_path: Path) -> List[TaskRecord]:
    lines = todo_path.read_text(encoding="utf-8").splitlines()
    header_index: Optional[int] = None

    for idx, line in enumerate(lines):
        if HEADER_RE.search(line):
            header_index = idx
            break

    if header_index is None:
        raise ValueError(f"no TODO task table found in {todo_path}")

    header_cells = _split_row(lines[header_index])
    tasks: List[TaskRecord] = []

    for line_number in range(header_index + 1, len(lines)):
        raw_line = lines[line_number]
        if not raw_line.strip():
            if tasks:
                break
            continue
        if not raw_line.strip().startswith("|"):
            if tasks:
                break
            continue
        if SEPARATOR_RE.match(raw_line):
            continue

        row_cells = _split_row(raw_line)
        if len(row_cells) < len(header_cells):
            row_cells += [""] * (len(header_cells) - len(row_cells))

        row: Dict[str, str] = dict(zip(header_cells, row_cells))
        task_id = row.get("ID", "").strip()
        if not task_id:
            continue

        files_dirs = row.get("Files/Dirs", "").strip()
        adr_c4_update = row.get("ADR/C4 Update", "").strip()
        task_text = row.get("Task", "").strip()
        validation = row.get("Validation", "").strip()

        tasks.append(
            TaskRecord(
                task_id=task_id,
                status=row.get("Status", "").strip().upper(),
                module=row.get("Module", "").strip(),
                task=task_text,
                files_dirs=files_dirs,
                done_when=row.get("Done When", "").strip(),
                validation=validation,
                adr_c4_update=adr_c4_update,
                line_number=line_number + 1,
                dependencies=extract_dependencies(task_text, files_dirs, validation, adr_c4_update),
                write_paths=extract_paths(files_dirs),
                global_doc_paths=extract_global_doc_paths(
                    files_dirs, adr_c4_update, task_text
                ),
            )
        )

    return tasks


def find_task(tasks: List[TaskRecord], task_id: str) -> Optional[TaskRecord]:
    for task in tasks:
        if task.task_id == task_id:
            return task
    return None