import math import inquirer import re import sys from collections import defaultdict from datetime import datetime, timedelta from typing import Any, DefaultDict, Dict, List, Sequence, Tuple, Optional from tqdm import tqdm import base32_crockford as base32 import click import tabulate import api.projects import api.stories from config import Config from util import require_login from color import Color from . import ( COLOR_HEADER, COLOR_PLANNED, COLOR_STARTED, COLOR_FINISHED, COLOR_DELIVERED, COLOR_ACCEPTED, COLOR_REJECTED, _format_state, _get_persons, ) from ._stories_info import stories_info from .cli import cli STATES = ( "unstarted", "planned", "started", "finished", "delivered", "accepted", "rejected", ) Persons = Dict[int, Dict[str, Any]] Totals = DefaultDict[int, Dict[str, int]] def __burndown(color: Color, letter: str, points: float) -> str: return color.format(letter * int(math.ceil(points))) def __get_row( item: Tuple[int, Dict[str, int]], persons: Persons, show_accepted: bool ) -> Sequence: owner_id, points = item name = persons[owner_id]["name"] estimates = [round(points[state], 1) for state in STATES[1:]] progress = "[" if show_accepted: progress += __burndown(COLOR_ACCEPTED, "A", points["accepted"]) progress += ( __burndown(COLOR_DELIVERED, "D", points["delivered"]) + __burndown(COLOR_FINISHED, "F", points["finished"]) + __burndown(COLOR_STARTED, "S", points["started"]) + __burndown(COLOR_REJECTED, "R", points["rejected"]) + __burndown(COLOR_PLANNED, "P", points["planned"]) + "]" ) return name, (*estimates), sum(estimates), progress def __format_story( story: Dict[str, Any], persons: Persons, totals: Totals ) -> Tuple[str, str, str, str, str, int]: code = base32.encode(story["id"]) is_owner = False initials = [] owner_ids = story.get("owner_ids", []) for owner_id in owner_ids: email = persons[owner_id]["email"] value = persons[owner_id]["initials"] if email == Config["user"]["email"]: is_owner = True value = Color(Color.CYAN).format(value) initials.append(value) type_ = story["story_type"] if type_ == "release": code = f"\033[44m{code}" estimate = story.get("estimate") if estimate: for owner_id in owner_ids: state = story["current_state"] totals[owner_id][state] += estimate / len(owner_ids) if is_owner: code = f"\033[1;97m{code}" estimate = f"\033[97m{estimate}\033[0m" if estimate else None owners = ", ".join(initials) state = _format_state(story["current_state"]) return code, type_, story["name"], owners, state, estimate, "\033[0m" def __print_stories( stories_: List[Dict[str, Any]], persons: Persons, totals: Totals, show_accepted: bool, ) -> None: table = ( __format_story(story, persons, totals) for story in stories_ if show_accepted or story["current_state"] != "accepted" ) headers = "Code", "Type", "Story name", "Owners", "State", "Pts" print(tabulate.tabulate(table, headers=headers), end="\n\n") def __print_totals(totals: Totals, persons: Persons, show_accepted: bool) -> None: COLOR_HEADER.print("Point totals:", end="\n\n") state_headers = [_format_state(state) for state in STATES] headers = ("Owner", *state_headers[1:], "Total", "Progress") data = sorted( (__get_row(item, persons, show_accepted) for item in totals.items()), key=lambda row: row[-2], reverse=True, ) print(tabulate.tabulate(data, headers), end="\n\n") def __print_burndown(iteration: Dict[str, Any], show_accepted: bool) -> None: COLOR_HEADER.print("Burndown:", end="\n\n") start = datetime.strptime(iteration["start"], "%Y-%m-%dT%H:%M:%SZ") history = api.projects.get_history_days( iteration["project_id"], start - timedelta(days=1) ) accepted_points = history["data"][0][1] last_counts: List[int] = [] for date, *counts in history["data"][1:]: if len(counts) > 0: # Update the last_counts variable if an update is available. last_counts = counts elif len(last_counts) > 0: # If no counts are available, use those of the previous day. counts = last_counts else: # If there are no last_counts either, just skip the day continue progress = "" if show_accepted: progress += __burndown(COLOR_ACCEPTED, "A", counts[0] - accepted_points) progress += ( __burndown(COLOR_DELIVERED, "D", counts[1]) + __burndown(COLOR_FINISHED, "F", counts[2]) + __burndown(COLOR_STARTED, "S", counts[3]) + __burndown(COLOR_PLANNED, "P", counts[5]) + __burndown(COLOR_PLANNED, "U", counts[6]) ) print(f"{date}: {progress}") print() def _stories_current(project: str, scope: str, show_accepted: bool) -> None: try: project_id = int(Config["project_aliases"][project]) except KeyError: project_id = base32.decode(project) iterations = api.projects.get_iterations(project_id, scope=scope) if not iterations: print(f"No stories in {scope}") return iteration = iterations[0] persons = _get_persons(project_id=project_id) totals: DefaultDict[int, Dict[str, int]] = defaultdict( lambda: dict((state, 0) for state in STATES) ) __print_stories(iteration["stories"], persons, totals, show_accepted) __print_totals(totals, persons, show_accepted) if scope == "current": __print_burndown(iteration, show_accepted) def _set_story_state(story: str, state: str) -> None: story_id = base32.decode(story) api.stories.put_story(story_id, current_state=state) def _complete_projects( ctx: click.Context, args: List[str], incomplete: str ) -> List[str]: return [ alias for alias in Config["project_aliases"] if alias.startswith(incomplete) ] @cli.command("stories") @click.argument("project", type=click.STRING, autocompletion=_complete_projects) @click.argument("story", required=False) @click.argument("action", required=False) @click.option("--scope", default="current") @click.option("--show-accepted/--hide-accepted", default=True) @click.option("--set-state", type=click.Choice(STATES)) @require_login def stories( project: str, story: Optional[str], action: Optional[str], scope: str, show_accepted: bool, set_state: str, ) -> None: if story is not None: state_actions = "start", "finish", "deliver", "accept", "reject" if set_state is not None: _set_story_state(story, set_state) elif action is not None: if action in state_actions: _set_story_state(story, f"{action}ed") if action == "comment": # todo pass stories_info(story) else: _stories_current(project, scope, show_accepted)