Jef Roosens
cd8ca789aa
- Add `write_csv_weekly()` to output.py: writes entries from multiple days as a single CSV with one header row, correct date per row - Add `-w`/`--weekly` flag to csv subparser - _cmd_csv branches on args.weekly: fetches week sections, formats per-day date strings, calls write_csv_weekly; --raw is honoured - Add TestWriteCsvWeekly with 6 tests - Update README with weekly csv usage examples
400 lines
14 KiB
Python
400 lines
14 KiB
Python
import csv
|
|
import sys
|
|
from collections import OrderedDict
|
|
from datetime import date
|
|
from typing import IO, TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from .status import DayStatus, WeekStatus
|
|
|
|
from .projects import resolve_project_task
|
|
from .utils import decimal_to_hhmm
|
|
|
|
|
|
def to_csv_entries(rows: list[dict]) -> list[dict]:
|
|
"""Convert raw parsed rows to write_csv-compatible entries without aggregating.
|
|
|
|
Each row becomes its own entry. Open entries (duration_hours is None) are
|
|
skipped. Rows are not combined, even if they share the same project and
|
|
description.
|
|
"""
|
|
from .parser import build_description
|
|
|
|
return [
|
|
{
|
|
"project": row["project"].strip(),
|
|
"description": build_description(row["story"], row["note"]),
|
|
"quantity": row["duration_hours"],
|
|
}
|
|
for row in rows
|
|
if row["duration_hours"] is not None
|
|
]
|
|
|
|
|
|
def write_csv_weekly(
|
|
day_sections: list[tuple[str, list[dict]]],
|
|
output: IO[str],
|
|
project_map: dict,
|
|
) -> None:
|
|
"""Write entries from multiple days as a single CSV with one header row.
|
|
|
|
day_sections is a list of (date_str, entries) pairs where entries are
|
|
already write_csv-compatible (project, description, quantity).
|
|
"""
|
|
writer = csv.writer(output)
|
|
writer.writerow(["Date*", "Project*", "Task", "Description", "Quantity"])
|
|
for date_str, entries in day_sections:
|
|
for entry in entries:
|
|
project, task = resolve_project_task(entry["project"], project_map)
|
|
writer.writerow(
|
|
[
|
|
date_str,
|
|
project,
|
|
task,
|
|
entry["description"],
|
|
f"{entry['quantity']:.2f}",
|
|
]
|
|
)
|
|
|
|
|
|
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
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Status output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def print_status(day: "DayStatus", week: "WeekStatus") -> None:
|
|
"""Print the status dashboard to stdout."""
|
|
hh = decimal_to_hhmm
|
|
W = 42 # label column width
|
|
|
|
def row(label: str, value: str, note: str = "") -> None:
|
|
suffix = f" ({note})" if note else ""
|
|
print(f" {label:<{W - 2}} {value}{suffix}")
|
|
|
|
def sep(char: str = "-", width: int = W + 12) -> None:
|
|
print(" " + char * width)
|
|
|
|
# ---- Today ----
|
|
print(f"\n Today {day.target_date.strftime('%A, %Y-%m-%d')}")
|
|
sep()
|
|
|
|
start_str = day.day_start.strftime("%H:%M") if day.day_start else "?"
|
|
row("Started", start_str)
|
|
|
|
logged_note = ""
|
|
if day.has_open_entry:
|
|
open_str = (
|
|
day.open_entry_start.strftime("%H:%M") if day.open_entry_start else "?"
|
|
)
|
|
logged_note = f"open entry since {open_str}"
|
|
if day.future_hours > 0:
|
|
logged_note += f", {hh(day.future_hours)} pre-logged ahead"
|
|
elif day.future_hours > 0:
|
|
logged_note = f"{hh(day.future_hours)} pre-logged ahead"
|
|
is_flex = round(day.daily_target * 60) != 480
|
|
if logged_note and is_flex:
|
|
logged_note += " (flex target)"
|
|
elif is_flex:
|
|
logged_note = "flex target"
|
|
row("Logged", f"{hh(day.logged_hours)} of {hh(day.daily_target)}", logged_note)
|
|
row("Remaining today", hh(day.remaining_hours))
|
|
|
|
if day.expected_end is not None:
|
|
if day.remaining_hours == 0:
|
|
note = "done"
|
|
elif day.latest_end_exceeds_expected:
|
|
note = "pre-logged entry ends later than needed"
|
|
else:
|
|
note = ""
|
|
row("Expected end", day.expected_end.strftime("%H:%M"), note)
|
|
else:
|
|
row("Expected end", "unknown")
|
|
|
|
# ---- Week ----
|
|
print(
|
|
f"\n Week W{day.target_date.isocalendar()[1]:02d} "
|
|
f"({week.week_start.strftime('%d/%m')} \u2013 {week.week_end.strftime('%d/%m')})"
|
|
)
|
|
sep()
|
|
|
|
row(
|
|
"Logged this week",
|
|
f"{hh(week.total_logged)} of {hh(week.weekly_target)}",
|
|
)
|
|
row("Remaining this week", hh(week.remaining_hours))
|
|
|
|
if week.daily_average is not None:
|
|
row(
|
|
"Daily average",
|
|
hh(week.daily_average),
|
|
f"{week.days_worked} day{'s' if week.days_worked != 1 else ''} worked",
|
|
)
|
|
|
|
if week.projected_total is not None:
|
|
deficit = week.weekly_target - week.projected_total
|
|
if deficit > 0.05:
|
|
note = f"{hh(deficit)} short"
|
|
elif deficit < -0.05:
|
|
note = f"{hh(abs(deficit))} over"
|
|
else:
|
|
note = "exact"
|
|
row("Projected week total", hh(week.projected_total), note)
|
|
|
|
if week.on_track is not None:
|
|
row("On track", "yes \u2713" if week.on_track else "no \u2717")
|
|
|
|
print()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stories output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def print_stories(rows: list[dict], project_map: dict | None = None) -> None:
|
|
"""
|
|
Print a Markdown list of stories worked on, grouped by project.
|
|
|
|
Only rows with a non-empty story field are included. Each story is
|
|
deduplicated across the rows; the raw story text (preserving any
|
|
markdown link) is used for display. Total time per story is appended.
|
|
Project names are resolved through project_map if provided.
|
|
"""
|
|
# Group by project, collecting (story_raw, total_hours) per story.
|
|
# Keyed by (project, stripped_story) to deduplicate; story_raw from
|
|
# the first row that has a link is preferred over a plain-text version.
|
|
from collections import defaultdict
|
|
|
|
_map = project_map or {}
|
|
project_order: list[str] = []
|
|
# resolved_label -> {stripped_story -> [story_raw, total_hours]}
|
|
stories: dict[str, dict[str, list]] = {}
|
|
|
|
for row in rows:
|
|
story = row.get("story", "").strip()
|
|
if not story:
|
|
continue
|
|
raw_project = row["project"].strip()
|
|
project_label, task = resolve_project_task(raw_project, _map)
|
|
label = f"{project_label} / {task}" if task else project_label
|
|
story_raw = row.get("story_raw", story).strip()
|
|
duration = row.get("duration_hours") or 0.0
|
|
|
|
if label not in stories:
|
|
project_order.append(label)
|
|
stories[label] = {}
|
|
|
|
if story not in stories[label]:
|
|
stories[label][story] = [story_raw, 0.0]
|
|
else:
|
|
# Prefer the version that contains a markdown link
|
|
existing_raw = stories[label][story][0]
|
|
if "](" in story_raw and "](" not in existing_raw:
|
|
stories[label][story][0] = story_raw
|
|
|
|
stories[label][story][1] += duration
|
|
|
|
for label in project_order:
|
|
print(f"- {label}")
|
|
for story_raw, total_hours in stories[label].values():
|
|
print(f" - {story_raw} ({decimal_to_hhmm(total_hours)})")
|