task_map.py - master - dpavlov/datagrid_reporter

task_map.py
270 строк9 KB
Dmitriy Pavlov
Add notes-backed goals and user-linked efforts
17 май 2026, 13:54
17 май 2026, 13:54ab9be42
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, Optional

from datagrid_reporter.storage import LocalStore


@dataclass(frozen=True)
class TaskMapItem:
    kind: str
    key: str
    summary: str
    status: str = ""
    missing: tuple[str, ...] = ()
    evidence: tuple[str, ...] = ()
    url: str = ""

    def to_json(self) -> Dict[str, Any]:
        return {
            "kind": self.kind,
            "key": self.key,
            "summary": self.summary,
            "status": self.status,
            "missing": list(self.missing),
            "evidence": list(self.evidence),
            "url": self.url,
        }


def build_task_map(store: LocalStore, project_key: str) -> list[TaskMapItem]:
    items: list[TaskMapItem] = []
    items.extend(_jira_without_pr(store, project_key))
    items.extend(_pr_without_jira(store, project_key))
    items.extend(_pr_without_commit(store, project_key))
    items.extend(_commit_without_pr(store, project_key))
    items.extend(_commit_without_jira(store, project_key))
    items.extend(_goal_task_gaps(store, project_key))
    return sorted(items, key=lambda item: (item.kind, item.key))


def export_task_map(exports_dir: Path, project_key: str, items: Iterable[TaskMapItem]) -> tuple[Path, Path]:
    project_dir = exports_dir / project_key
    project_dir.mkdir(parents=True, exist_ok=True)
    rows = list(items)
    jsonl_path = project_dir / "task_map.jsonl"
    md_path = project_dir / "task_map.md"

    with jsonl_path.open("w", encoding="utf-8") as fh:
        for item in rows:
            fh.write(json.dumps(item.to_json(), ensure_ascii=False) + "\n")

    with md_path.open("w", encoding="utf-8") as fh:
        fh.write("# Task Map Gaps\n\n")
        for kind, grouped in _group_by_kind(rows).items():
            fh.write(f"## {kind}\n\n")
            for item in grouped:
                missing = ", ".join(item.missing) or "-"
                evidence = "; ".join(item.evidence) or "-"
                link = f" ({item.url})" if item.url else ""
                fh.write(f"- {item.key}: {item.summary}{link}; status {item.status or '-'}; missing {missing}; evidence {evidence}\n")
            fh.write("\n")

    return jsonl_path, md_path


def summarize_task_map(items: Iterable[TaskMapItem]) -> Dict[str, int]:
    summary: Dict[str, int] = {}
    for item in items:
        summary[item.kind] = summary.get(item.kind, 0) + 1
    return dict(sorted(summary.items()))


def _jira_without_pr(store: LocalStore, project_key: str) -> Iterable[TaskMapItem]:
    rows = store.conn.execute(
        """
        SELECT i.issue_key, i.summary, i.status, i.url
        FROM jira_issues i
        LEFT JOIN jira_pr_matches m
            ON m.project_key = i.project_key AND m.issue_key = i.issue_key
        WHERE i.project_key = ? AND m.pr_number IS NULL
          AND COALESCE(i.labels, '') NOT LIKE '%"ignite-3"%'
        ORDER BY i.updated_at DESC
        """,
        (project_key,),
    )
    for row in rows:
        yield TaskMapItem(
            kind="jira_without_pr",
            key=row["issue_key"],
            summary=row["summary"] or "",
            status=row["status"] or "",
            missing=("pr",),
            evidence=("no jira_pr_matches row",),
            url=row["url"] or "",
        )


def _pr_without_jira(store: LocalStore, project_key: str) -> Iterable[TaskMapItem]:
    rows = store.conn.execute(
        """
        SELECT p.number, p.title, p.state, p.html_url
        FROM github_prs p
        LEFT JOIN jira_pr_matches m
            ON m.project_key = p.project_key
           AND m.repo_owner = p.repo_owner
           AND m.repo_name = p.repo_name
           AND m.pr_number = p.number
        WHERE p.project_key = ? AND m.issue_key IS NULL
        ORDER BY p.updated_at DESC
        """,
        (project_key,),
    )
    for row in rows:
        yield TaskMapItem(
            kind="pr_without_jira",
            key=f"PR #{row['number']}",
            summary=row["title"] or "",
            status=row["state"] or "",
            missing=("jira",),
            evidence=("no jira_pr_matches row",),
            url=row["html_url"] or "",
        )


def _pr_without_commit(store: LocalStore, project_key: str) -> Iterable[TaskMapItem]:
    rows = store.conn.execute(
        """
        SELECT p.number, p.title, p.state, p.html_url
        FROM github_prs p
        LEFT JOIN github_pr_commits pc
            ON pc.project_key = p.project_key AND pc.pr_number = p.number
        WHERE p.project_key = ? AND pc.commit_sha IS NULL
        ORDER BY p.updated_at DESC
        """,
        (project_key,),
    )
    for row in rows:
        yield TaskMapItem(
            kind="pr_without_commit",
            key=f"PR #{row['number']}",
            summary=row["title"] or "",
            status=row["state"] or "",
            missing=("commit",),
            evidence=("no github_pr_commits row",),
            url=row["html_url"] or "",
        )


def _commit_without_pr(store: LocalStore, project_key: str) -> Iterable[TaskMapItem]:
    rows = store.conn.execute(
        """
        SELECT c.sha, c.message, c.html_url, c.committed_at
        FROM github_commits c
        LEFT JOIN github_pr_commits pc
            ON pc.project_key = c.project_key AND pc.commit_sha = c.sha
        WHERE c.project_key = ? AND pc.pr_number IS NULL
        ORDER BY c.committed_at DESC
        """,
        (project_key,),
    )
    for row in rows:
        sha = row["sha"] or ""
        yield TaskMapItem(
            kind="commit_without_pr",
            key=sha[:7],
            summary=_first_line(row["message"]),
            status=row["committed_at"] or "",
            missing=("pr",),
            evidence=("no github_pr_commits row",),
            url=row["html_url"] or "",
        )


def _commit_without_jira(store: LocalStore, project_key: str) -> Iterable[TaskMapItem]:
    rows = store.conn.execute(
        """
        SELECT c.sha, c.message, c.html_url, c.committed_at,
               GROUP_CONCAT(pc.pr_number) AS pr_numbers,
               GROUP_CONCAT(m.issue_key) AS issue_keys
        FROM github_commits c
        LEFT JOIN github_pr_commits pc
            ON pc.project_key = c.project_key AND pc.commit_sha = c.sha
        LEFT JOIN github_prs p
            ON p.project_key = pc.project_key AND p.number = pc.pr_number
        LEFT JOIN jira_pr_matches m
            ON m.project_key = pc.project_key
           AND m.repo_owner = p.repo_owner
           AND m.repo_name = p.repo_name
           AND m.pr_number = pc.pr_number
        WHERE c.project_key = ?
        GROUP BY c.project_key, c.sha
        HAVING issue_keys IS NULL
        ORDER BY c.committed_at DESC
        """,
        (project_key,),
    )
    for row in rows:
        sha = row["sha"] or ""
        prs = row["pr_numbers"] or "-"
        yield TaskMapItem(
            kind="commit_without_jira",
            key=sha[:7],
            summary=_first_line(row["message"]),
            status=row["committed_at"] or "",
            missing=("jira",),
            evidence=(f"prs={prs}",),
            url=row["html_url"] or "",
        )


def _goal_task_gaps(store: LocalStore, project_key: str) -> Iterable[TaskMapItem]:
    rows = store.conn.execute(
        """
        SELECT t.goal_id, t.issue_key, t.pr_number, t.commit_sha, t.summary, t.status, t.sources
        FROM goal_tasks t
        LEFT JOIN jira_issues i
            ON i.project_key = t.project_key AND i.issue_key = t.issue_key
        WHERE t.project_key = ?
          AND (t.issue_key IS NULL OR COALESCE(i.labels, '') NOT LIKE '%"ignite-3"%')
          AND (t.issue_key IS NULL OR t.pr_number IS NULL OR t.commit_sha IS NULL)
        ORDER BY t.goal_id, t.issue_key, t.pr_number, t.commit_sha
        """,
        (project_key,),
    )
    for row in rows:
        missing = []
        if row["issue_key"] is None:
            missing.append("jira")
        if row["pr_number"] is None:
            missing.append("pr")
        if row["commit_sha"] is None:
            missing.append("commit")

        parts = [row["goal_id"]]
        if row["issue_key"]:
            parts.append(row["issue_key"])
        if row["pr_number"]:
            parts.append(f"PR #{row['pr_number']}")
        if row["commit_sha"]:
            parts.append(str(row["commit_sha"])[:7])

        yield TaskMapItem(
            kind="goal_task_gap",
            key=" / ".join(parts),
            summary=row["summary"] or "",
            status=row["status"] or "",
            missing=tuple(missing),
            evidence=tuple(_json_loads(row["sources"], [])),
        )


def _group_by_kind(items: Iterable[TaskMapItem]) -> Dict[str, list[TaskMapItem]]:
    result: Dict[str, list[TaskMapItem]] = {}
    for item in items:
        result.setdefault(item.kind, []).append(item)
    return result


def _first_line(value: Optional[str]) -> str:
    return (value or "").splitlines()[0]


def _json_loads(value: Any, default: Any) -> Any:
    if value is None:
        return default
    try:
        return json.loads(value)
    except (TypeError, json.JSONDecodeError):
        return default