import csv import sys from collections import OrderedDict from datetime import date from typing import IO from .projects import resolve_project_task from .utils import decimal_to_hhmm def write_csv( aggregated: list[dict], output: IO[str], date_str: str, project_map: dict, ) -> None: """Write the aggregated timesheet data as CSV.""" writer = csv.writer(output) writer.writerow(["Date*", "Project*", "Task", "Description", "Quantity"]) for entry in aggregated: project, task = resolve_project_task(entry["project"], project_map) writer.writerow( [ date_str, project, task, entry["description"], f"{entry['quantity']:.2f}", ] ) # --------------------------------------------------------------------------- # Summary helpers # --------------------------------------------------------------------------- def _group_by_project_label( aggregated: list[dict], project_map: dict ) -> "OrderedDict[str, list[dict]]": """Group aggregated entries by resolved project/task label.""" grouped: OrderedDict[str, list[dict]] = OrderedDict() for entry in aggregated: project, task = resolve_project_task(entry["project"], project_map) label = f"{project} / {task}" if task else project grouped.setdefault(label, []).append(entry) return grouped def _sum_by_project_label( aggregated: list[dict], project_map: dict ) -> "OrderedDict[str, float]": """Return total hours per project label.""" totals: OrderedDict[str, float] = OrderedDict() for entry in aggregated: project, task = resolve_project_task(entry["project"], project_map) label = f"{project} / {task}" if task else project totals[label] = totals.get(label, 0.0) + entry["quantity"] return totals # --------------------------------------------------------------------------- # Full detail summary (default) # --------------------------------------------------------------------------- def print_summary(aggregated: list[dict], project_map: dict) -> None: """Print a full human-readable summary with per-entry descriptions.""" grouped = _group_by_project_label(aggregated, project_map) total_all = sum(e["quantity"] for e in aggregated) all_descs = [e["description"] for e in aggregated] desc_width = max(max((len(d) for d in all_descs), default=40), 40) separator = "-" * (desc_width + 16) for label, entries in grouped.items(): project_total = sum(e["quantity"] for e in entries) print(f"\n {label} ({decimal_to_hhmm(project_total)})") print(" " + separator) for entry in entries: print( f" {entry['description']:<{desc_width}} {decimal_to_hhmm(entry['quantity'])}" ) print(" " + separator) print(f"\n {'TOTAL':<{desc_width + 2}} {decimal_to_hhmm(total_all)}\n") # --------------------------------------------------------------------------- # Short summary (-s): one line per project label # --------------------------------------------------------------------------- def print_summary_short(aggregated: list[dict], project_map: dict) -> None: """Print one line per project label with total hours.""" totals = _sum_by_project_label(aggregated, project_map) total_all = sum(totals.values()) label_width = max((len(l) for l in totals), default=20) label_width = max(label_width, 20) for label, hours in totals.items(): print(f" {label:<{label_width}} {decimal_to_hhmm(hours)}") print(f" {'TOTAL':<{label_width}} {decimal_to_hhmm(total_all)}") # --------------------------------------------------------------------------- # Weekly summary (--weekly) # --------------------------------------------------------------------------- def print_summary_weekly( day_sections: list[tuple[date, list[dict]]], project_map: dict ) -> None: """Print a full summary for each day of the week.""" week_total = sum(e["duration_hours"] for _, rows in day_sections for e in rows) for day, rows in day_sections: aggregated = _aggregate(rows) day_total = sum(e["quantity"] for e in aggregated) print(f"\n{'═' * 60}") print(f" {day.strftime('%A, %Y-%m-%d')} ({decimal_to_hhmm(day_total)})") print(f"{'═' * 60}") print_summary(aggregated, project_map) print(f"\n {'WEEK TOTAL':<42} {decimal_to_hhmm(week_total)}\n") # --------------------------------------------------------------------------- # Weekly short (-s --weekly): per-day project totals # --------------------------------------------------------------------------- def print_summary_weekly_short( day_sections: list[tuple[date, list[dict]]], project_map: dict ) -> None: """Print one line per project per day, plus day and week totals.""" week_total = 0.0 all_labels: list[str] = [] day_data: list[tuple[date, OrderedDict[str, float]]] = [] for day, rows in day_sections: aggregated = _aggregate(rows) totals = _sum_by_project_label(aggregated, project_map) day_data.append((day, totals)) all_labels.extend(totals.keys()) label_width = max((len(l) for l in all_labels), default=20) label_width = max(label_width, 20) for day, totals in day_data: day_total = sum(totals.values()) week_total += day_total print(f"\n {day.strftime('%A, %Y-%m-%d')}") for label, hours in totals.items(): print(f" {label:<{label_width}} {decimal_to_hhmm(hours)}") print(f" {'day total':<{label_width}} {decimal_to_hhmm(day_total)}") print(f"\n {'WEEK TOTAL':<{label_width + 2}} {decimal_to_hhmm(week_total)}\n") # --------------------------------------------------------------------------- # Weekly day-total-only (-ss --weekly): one line per day # --------------------------------------------------------------------------- def print_summary_weekly_totals( day_sections: list[tuple[date, list[dict]]], project_map: dict ) -> None: """Print only the total hours per day plus a week total.""" week_total = 0.0 rows_out = [] for day, rows in day_sections: day_total = sum(e["duration_hours"] for e in rows) week_total += day_total rows_out.append((day.strftime("%A"), day.strftime("%Y-%m-%d"), day_total)) day_width = max((len(r[0]) for r in rows_out), default=9) for day_name, date_str, hours in rows_out: print(f" {day_name:<{day_width}}, {date_str} {decimal_to_hhmm(hours)}") print(f" {'WEEK TOTAL':<{day_width + 13}} {decimal_to_hhmm(week_total)}") # --------------------------------------------------------------------------- # Internal helper # --------------------------------------------------------------------------- def _aggregate(rows: list[dict]) -> list[dict]: """Aggregate raw rows — thin wrapper to avoid importing parser in output.""" from collections import defaultdict from .parser import build_description key_order: list[tuple] = [] totals: dict[tuple, float] = defaultdict(float) for row in rows: description = build_description(row["story"], row["note"]) key = (row["project"].strip(), description) if key not in totals: key_order.append(key) totals[key] += row["duration_hours"] return [ {"project": p, "description": d, "quantity": totals[(p, d)]} for p, d in key_order ]