c9336941ac
Add four skills for Tempo/KSeF timesheet and invoice workflows: - check-my-timesheet: show/log the current user's Tempo time entries - timesheet-checker: audit timesheet completion across all reporters - invoice-checker: pull KSeF invoices, check contractors + MF white list - invoice-prep: summarize Tempo hours per Jira project for invoicing Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
649 lines
26 KiB
Python
649 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
"""Fetch Tempo worklogs and compute timesheet completion for a given period."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.error
|
||
import urllib.request
|
||
from datetime import date, timedelta
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
# ── .env loading ─────────────────────────────────────────────────────────────
|
||
|
||
def load_dotenv(*paths: Path) -> Path | None:
|
||
"""Load the first .env file found from the given paths.
|
||
|
||
Sets missing keys into os.environ (existing env vars take priority).
|
||
Returns the path that was loaded, or None if none were found.
|
||
"""
|
||
for path in paths:
|
||
if not path.exists():
|
||
continue
|
||
for raw in path.read_text(encoding="utf-8").splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, _, value = line.partition("=")
|
||
key = key.strip()
|
||
value = value.strip().strip('"').strip("'")
|
||
if key and key not in os.environ:
|
||
os.environ[key] = value
|
||
return path
|
||
return None
|
||
|
||
|
||
# ── Utilities ─────────────────────────────────────────────────────────────────
|
||
|
||
def clean_text(value: Any) -> str:
|
||
return re.sub(r"\s+", " ", str(value or "")).strip()
|
||
|
||
|
||
def normalize_key(value: Any) -> str:
|
||
return re.sub(r"[^a-z0-9]+", "_", clean_text(value).lower()).strip("_")
|
||
|
||
|
||
# ── Period parsing ────────────────────────────────────────────────────────────
|
||
|
||
def parse_period(period_str: str, today: date | None = None) -> tuple[date, date, str]:
|
||
"""Return (start, end, human_label) for the given period string.
|
||
|
||
Supported keywords:
|
||
last-week Previous Monday–Friday
|
||
last-month Previous calendar month
|
||
last-year Previous calendar year
|
||
yesterday Yesterday only
|
||
current-week Monday of this week to yesterday
|
||
current-month 1st of this month to yesterday
|
||
current-year 1st January of this year to yesterday
|
||
YYYY-MM-DD:YYYY-MM-DD Explicit date range
|
||
|
||
For current-* periods the end date is always yesterday so the current
|
||
in-progress day is excluded. If yesterday falls before the period start
|
||
(e.g. calling current-week on a Monday), end < start and the period
|
||
will contain zero working days — a warning is printed by run().
|
||
"""
|
||
if today is None:
|
||
today = date.today()
|
||
yesterday = today - timedelta(days=1)
|
||
s = period_str.strip().lower()
|
||
|
||
if s == "last-week":
|
||
last_monday = today - timedelta(days=today.weekday() + 7)
|
||
return last_monday, last_monday + timedelta(days=4), "last week"
|
||
if s == "last-month":
|
||
end = today.replace(day=1) - timedelta(days=1)
|
||
return end.replace(day=1), end, "last month"
|
||
if s == "last-year":
|
||
return date(today.year - 1, 1, 1), date(today.year - 1, 12, 31), "last year"
|
||
if s == "yesterday":
|
||
return yesterday, yesterday, "yesterday"
|
||
if s == "current-week":
|
||
start = today - timedelta(days=today.weekday())
|
||
return start, yesterday, "current week to date"
|
||
if s == "current-month":
|
||
return today.replace(day=1), yesterday, "current month to date"
|
||
if s == "current-year":
|
||
return date(today.year, 1, 1), yesterday, "current year to date"
|
||
if ":" in s:
|
||
parts = s.split(":", 1)
|
||
start = date.fromisoformat(parts[0].strip())
|
||
end = date.fromisoformat(parts[1].strip())
|
||
return start, end, f"{start} to {end}"
|
||
raise ValueError(
|
||
f"Unknown period '{period_str}'. "
|
||
f"Use last-week, last-month, last-year, yesterday, "
|
||
f"current-week, current-month, current-year, or YYYY-MM-DD:YYYY-MM-DD."
|
||
)
|
||
|
||
|
||
# ── Public holidays ───────────────────────────────────────────────────────────
|
||
|
||
def get_public_holidays(country: str, years: set[int]) -> set[date]:
|
||
"""Fetch public holidays from date.nager.at for the given country and years.
|
||
|
||
country is an ISO 3166-1 alpha-2 code (e.g. PL, GB, DE, FR).
|
||
Returns an empty set with a warning if the country is not supported.
|
||
"""
|
||
holidays: set[date] = set()
|
||
for year in sorted(years):
|
||
url = f"https://date.nager.at/api/v3/PublicHolidays/{year}/{country.upper()}"
|
||
req = urllib.request.Request(url, headers={"User-Agent": "timesheet-checker/1.0"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
items = json.loads(resp.read())
|
||
year_holidays = {date.fromisoformat(item["date"]) for item in items}
|
||
holidays |= year_holidays
|
||
print(f" {country.upper()} {year}: {len(year_holidays)} public holidays")
|
||
except urllib.error.HTTPError as exc:
|
||
if exc.code == 404:
|
||
print(
|
||
f"Warning: country '{country.upper()}' not supported by date.nager.at "
|
||
f"— no holidays will be excluded for {year}.",
|
||
file=sys.stderr,
|
||
)
|
||
else:
|
||
print(f"Warning: could not fetch holidays for {country.upper()} {year}: {exc}", file=sys.stderr)
|
||
except urllib.error.URLError as exc:
|
||
print(f"Warning: could not fetch holidays for {country.upper()} {year}: {exc}", file=sys.stderr)
|
||
return holidays
|
||
|
||
|
||
def get_working_days(start: date, end: date, holidays: set[date]) -> list[date]:
|
||
"""Return weekdays (Mon–Fri) in [start, end] that are not public holidays."""
|
||
result = []
|
||
current = start
|
||
while current <= end:
|
||
if current.weekday() < 5 and current not in holidays:
|
||
result.append(current)
|
||
current += timedelta(days=1)
|
||
return result
|
||
|
||
|
||
# ── Employees config ──────────────────────────────────────────────────────────
|
||
|
||
def is_table_row(line: str) -> bool:
|
||
s = line.strip()
|
||
return s.startswith("|") and s.endswith("|")
|
||
|
||
|
||
def split_table_row(line: str) -> list[str]:
|
||
s = line.strip().lstrip("|").rstrip("|")
|
||
return [c.strip() for c in s.split("|")]
|
||
|
||
|
||
def is_separator_row(cells: list[str]) -> bool:
|
||
return all(re.fullmatch(r":?-{3,}:?", c.strip()) for c in cells)
|
||
|
||
|
||
def parse_markdown_tables(path: Path) -> list[dict[str, Any]]:
|
||
lines = path.read_text(encoding="utf-8").splitlines()
|
||
tables: list[dict[str, Any]] = []
|
||
i = 0
|
||
while i < len(lines):
|
||
if not is_table_row(lines[i]):
|
||
i += 1
|
||
continue
|
||
block: list[str] = []
|
||
while i < len(lines) and is_table_row(lines[i]):
|
||
block.append(lines[i])
|
||
i += 1
|
||
if len(block) < 3:
|
||
continue
|
||
headers = [normalize_key(c) for c in split_table_row(block[0])]
|
||
if not is_separator_row(split_table_row(block[1])):
|
||
continue
|
||
rows = []
|
||
for line in block[2:]:
|
||
vals = split_table_row(line)
|
||
if len(vals) != len(headers):
|
||
raise ValueError(f"Malformed table row in {path}: {line!r}")
|
||
rows.append(dict(zip(headers, vals)))
|
||
tables.append({"headers": headers, "rows": rows})
|
||
return tables
|
||
|
||
|
||
def find_table(tables: list[dict[str, Any]], required_headers: set[str]) -> list[dict[str, str]]:
|
||
for t in tables:
|
||
if required_headers.issubset(set(t["headers"])):
|
||
return t["rows"]
|
||
raise ValueError(f"No table with headers {sorted(required_headers)} found in employees file.")
|
||
|
||
|
||
def load_employees(path: Path) -> list[dict[str, Any]]:
|
||
tables = parse_markdown_tables(path)
|
||
|
||
# Load defaults (optional table with setting/value columns).
|
||
defaults: dict[str, Any] = {"country": "PL", "expected_hours_per_day": 8.0}
|
||
try:
|
||
for row in find_table(tables, {"setting", "value"}):
|
||
key = normalize_key(row["setting"])
|
||
val = clean_text(row["value"])
|
||
if not val:
|
||
continue
|
||
if key == "country":
|
||
defaults["country"] = val.upper()
|
||
elif key == "expected_hours_per_day":
|
||
try:
|
||
defaults["expected_hours_per_day"] = float(val)
|
||
except ValueError:
|
||
print(f"Warning: invalid default expected_hours_per_day '{val}' — using 8.", file=sys.stderr)
|
||
except ValueError:
|
||
pass # no defaults table; built-in defaults apply
|
||
|
||
rows = find_table(tables, {"name", "email"})
|
||
employees = []
|
||
for row in rows:
|
||
name = clean_text(row["name"])
|
||
email = clean_text(row["email"]).lower()
|
||
if not name or not email:
|
||
continue
|
||
|
||
raw_h = clean_text(row.get("expected_hours_per_day", ""))
|
||
try:
|
||
expected_h = float(raw_h) if raw_h else defaults["expected_hours_per_day"]
|
||
except ValueError:
|
||
expected_h = defaults["expected_hours_per_day"]
|
||
|
||
raw_country = clean_text(row.get("country", "")).upper()
|
||
country = raw_country if raw_country else defaults["country"]
|
||
|
||
raw_start = clean_text(row.get("start_date", ""))
|
||
start_date: date | None = None
|
||
if raw_start:
|
||
try:
|
||
start_date = date.fromisoformat(raw_start)
|
||
except ValueError:
|
||
print(
|
||
f"Warning: could not parse start_date '{raw_start}' for {name} — ignoring.",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
raw_term = clean_text(row.get("termination_date", ""))
|
||
termination_date: date | None = None
|
||
if raw_term:
|
||
try:
|
||
termination_date = date.fromisoformat(raw_term)
|
||
except ValueError:
|
||
print(
|
||
f"Warning: could not parse termination_date '{raw_term}' for {name} — ignoring.",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
employees.append({
|
||
"name": name,
|
||
"email": email,
|
||
"expected_hours_per_day": expected_h,
|
||
"country": country,
|
||
"start_date": start_date,
|
||
"termination_date": termination_date,
|
||
})
|
||
|
||
if not employees:
|
||
raise ValueError(f"No employees found in {path}. Check the table format.")
|
||
return employees
|
||
|
||
|
||
# ── Jira user lookup ──────────────────────────────────────────────────────────
|
||
|
||
def resolve_account_ids(
|
||
emails: list[str],
|
||
jira_base_url: str,
|
||
jira_email: str,
|
||
jira_token: str,
|
||
) -> dict[str, str]:
|
||
"""Return {email: account_id} by searching Jira's user API for each address."""
|
||
credentials = base64.b64encode(f"{jira_email}:{jira_token}".encode()).decode()
|
||
headers = {"Authorization": f"Basic {credentials}", "Accept": "application/json"}
|
||
resolved: dict[str, str] = {}
|
||
not_found: list[str] = []
|
||
|
||
for email in emails:
|
||
url = f"{jira_base_url.rstrip('/')}/rest/api/3/user/search?query={email}&maxResults=10"
|
||
req = urllib.request.Request(url, headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
users = json.loads(resp.read())
|
||
except urllib.error.HTTPError as exc:
|
||
if exc.code == 401:
|
||
print(
|
||
"Error: Jira API returned 401 Unauthorized.\n"
|
||
"Check JIRA_EMAIL and JIRA_API_TOKEN (generate at id.atlassian.com).",
|
||
file=sys.stderr,
|
||
)
|
||
sys.exit(1)
|
||
body = exc.read().decode("utf-8", errors="replace")
|
||
print(f"Error: Jira API returned {exc.code} for {email}: {body}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
match = next(
|
||
(u for u in users if u.get("emailAddress", "").lower() == email.lower()),
|
||
None,
|
||
)
|
||
if match:
|
||
resolved[email] = match["accountId"]
|
||
print(f" {email} → {match['accountId']}")
|
||
else:
|
||
not_found.append(email)
|
||
|
||
if not_found:
|
||
print(
|
||
f"Error: Could not find Jira account for: {', '.join(not_found)}\n"
|
||
"Check the email addresses in employees.md or your JIRA_BASE_URL.",
|
||
file=sys.stderr,
|
||
)
|
||
sys.exit(1)
|
||
|
||
return resolved
|
||
|
||
|
||
# ── Tempo API ─────────────────────────────────────────────────────────────────
|
||
|
||
def fetch_tempo_worklogs(
|
||
start: date,
|
||
end: date,
|
||
token: str,
|
||
base_url: str,
|
||
) -> dict[str, dict[str, float]]:
|
||
"""Return {account_id: {date_str: hours}} for ALL reporters in the period.
|
||
|
||
Paginates automatically. Date strings are ISO format (YYYY-MM-DD).
|
||
"""
|
||
totals: dict[str, dict[str, float]] = {}
|
||
headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
|
||
next_url: str | None = f"{base_url}/worklogs?from={start}&to={end}&limit=1000"
|
||
page = 0
|
||
while next_url:
|
||
page += 1
|
||
req = urllib.request.Request(next_url, headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
data = json.loads(resp.read())
|
||
except urllib.error.HTTPError as exc:
|
||
body = exc.read().decode("utf-8", errors="replace")
|
||
if exc.code == 401:
|
||
print(
|
||
"Error: Tempo API returned 401 Unauthorized. Check your TEMPO_API_TOKEN.",
|
||
file=sys.stderr,
|
||
)
|
||
else:
|
||
print(f"Error: Tempo API returned {exc.code}: {body}", file=sys.stderr)
|
||
sys.exit(1)
|
||
for worklog in data.get("results", []):
|
||
aid = worklog.get("author", {}).get("accountId", "")
|
||
date_str = worklog.get("startDate", "")
|
||
if aid and date_str:
|
||
if aid not in totals:
|
||
totals[aid] = {}
|
||
totals[aid][date_str] = (
|
||
totals[aid].get(date_str, 0.0) + worklog.get("timeSpentSeconds", 0) / 3600
|
||
)
|
||
next_url = data.get("metadata", {}).get("next")
|
||
print(f"Fetched worklogs (pages: {page})")
|
||
return totals
|
||
|
||
|
||
def resolve_account_info(
|
||
account_ids: list[str],
|
||
jira_base_url: str,
|
||
jira_email: str,
|
||
jira_token: str,
|
||
) -> dict[str, dict]:
|
||
"""Return {account_id: {display_name, email}} for the given Jira account IDs."""
|
||
credentials = base64.b64encode(f"{jira_email}:{jira_token}".encode()).decode()
|
||
headers = {"Authorization": f"Basic {credentials}", "Accept": "application/json"}
|
||
result: dict[str, dict] = {}
|
||
for aid in account_ids:
|
||
url = f"{jira_base_url.rstrip('/')}/rest/api/3/user?accountId={aid}"
|
||
req = urllib.request.Request(url, headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
user = json.loads(resp.read())
|
||
result[aid] = {
|
||
"display_name": user.get("displayName", ""),
|
||
"email": user.get("emailAddress", ""),
|
||
}
|
||
except (urllib.error.HTTPError, urllib.error.URLError):
|
||
result[aid] = {"display_name": f"Unknown ({aid})", "email": ""}
|
||
return result
|
||
|
||
|
||
# ── Stats computation ─────────────────────────────────────────────────────────
|
||
|
||
def compute_stats(
|
||
employees: list[dict[str, Any]],
|
||
period_start: date,
|
||
period_end: date,
|
||
holidays_by_country: dict[str, set[date]],
|
||
worklogs: dict[str, dict[str, float]],
|
||
) -> list[dict[str, Any]]:
|
||
results = []
|
||
for emp in employees:
|
||
holidays = holidays_by_country.get(emp["country"], set())
|
||
all_working_days = get_working_days(period_start, period_end, holidays)
|
||
emp_start: date | None = emp.get("start_date")
|
||
emp_term: date | None = emp.get("termination_date")
|
||
emp_working_days = all_working_days
|
||
if emp_start:
|
||
emp_working_days = [d for d in emp_working_days if d >= emp_start]
|
||
if emp_term:
|
||
emp_working_days = [d for d in emp_working_days if d <= emp_term]
|
||
expected_h = emp["expected_hours_per_day"] * len(emp_working_days)
|
||
|
||
# Partition logged hours into in-range vs out-of-range
|
||
date_logs = worklogs.get(emp["account_id"], {})
|
||
hours_in_range = 0.0
|
||
hours_before_start = 0.0
|
||
hours_after_termination = 0.0
|
||
for date_str, h in date_logs.items():
|
||
d = date.fromisoformat(date_str)
|
||
if emp_start and d < emp_start:
|
||
hours_before_start += h
|
||
elif emp_term and d > emp_term:
|
||
hours_after_termination += h
|
||
else:
|
||
hours_in_range += h
|
||
|
||
reported_h = round(hours_in_range, 2)
|
||
hours_before_start = round(hours_before_start, 2)
|
||
hours_after_termination = round(hours_after_termination, 2)
|
||
missing_h = round(max(0.0, expected_h - reported_h), 2)
|
||
pct = round((reported_h / expected_h * 100) if expected_h > 0 else 100.0, 1)
|
||
results.append({
|
||
"name": emp["name"],
|
||
"account_id": emp["account_id"],
|
||
"country": emp["country"],
|
||
"start_date": emp_start.isoformat() if emp_start else None,
|
||
"termination_date": emp_term.isoformat() if emp_term else None,
|
||
"effective_working_days": len(emp_working_days),
|
||
"expected_hours": expected_h,
|
||
"reported_hours": reported_h,
|
||
"hours_before_start": hours_before_start,
|
||
"hours_after_termination": hours_after_termination,
|
||
"missing_hours": missing_h,
|
||
"completion_pct": pct,
|
||
})
|
||
return sorted(results, key=lambda r: r["missing_hours"], reverse=True)
|
||
|
||
|
||
# ── Entry point ───────────────────────────────────────────────────────────────
|
||
|
||
def _resolve_employees_path(cli_value: str | None) -> Path | None:
|
||
"""Return the employees file to use, or None if not found.
|
||
|
||
Search order:
|
||
1. Explicit --employees argument (if given)
|
||
2. employees.md in the current working directory
|
||
3. references/employees.md in the skill folder (next to this script)
|
||
"""
|
||
if cli_value:
|
||
return Path(cli_value)
|
||
skill_dir = Path(__file__).parent.parent
|
||
for candidate in (
|
||
Path.cwd() / "employees.md",
|
||
skill_dir / "references" / "employees.md",
|
||
):
|
||
if candidate.exists():
|
||
return candidate
|
||
return None
|
||
|
||
|
||
def run(args: argparse.Namespace) -> int:
|
||
# Load .env from working directory first, then skill directory as fallback.
|
||
# Existing environment variables always take priority over .env values.
|
||
skill_dir = Path(__file__).parent.parent
|
||
loaded = load_dotenv(Path.cwd() / ".env", skill_dir / ".env")
|
||
if loaded:
|
||
print(f"Loaded credentials from {loaded}")
|
||
|
||
tempo_token = args.tempo_token or os.environ.get("TEMPO_API_TOKEN", "")
|
||
jira_base_url = args.jira_base_url or os.environ.get("JIRA_BASE_URL", "")
|
||
jira_email = args.jira_email or os.environ.get("JIRA_EMAIL", "")
|
||
jira_token = args.jira_token or os.environ.get("JIRA_API_TOKEN", "")
|
||
|
||
missing = [
|
||
name for name, val in [
|
||
("TEMPO_API_TOKEN", tempo_token),
|
||
("JIRA_BASE_URL", jira_base_url),
|
||
("JIRA_EMAIL", jira_email),
|
||
("JIRA_API_TOKEN", jira_token),
|
||
] if not val
|
||
]
|
||
if missing:
|
||
print(
|
||
f"Error: missing required credentials: {', '.join(missing)}\n"
|
||
"Set them as environment variables or pass via CLI flags.\n"
|
||
"See README.md for how to obtain each credential.",
|
||
file=sys.stderr,
|
||
)
|
||
return 1
|
||
|
||
start, end, label = parse_period(args.period)
|
||
print(f"Period: {label} ({start} to {end})")
|
||
|
||
employees_path = _resolve_employees_path(args.employees)
|
||
if employees_path is None:
|
||
skill_dir = Path(__file__).parent.parent
|
||
print(
|
||
"Error: No employees.md found.\n"
|
||
f"Looked in:\n"
|
||
f" {Path.cwd() / 'employees.md'}\n"
|
||
f" {skill_dir / 'references' / 'employees.md'}\n"
|
||
"Create one in your project directory or pass --employees PATH.",
|
||
file=sys.stderr,
|
||
)
|
||
return 1
|
||
employees = load_employees(employees_path)
|
||
print(f"Employees loaded: {len(employees)} (from {employees_path})")
|
||
|
||
countries = {e["country"] for e in employees}
|
||
years = {start.year, end.year}
|
||
print(f"Fetching public holidays from date.nager.at for: {', '.join(sorted(countries))} ...")
|
||
holidays_by_country = {c: get_public_holidays(c, years) for c in countries}
|
||
|
||
print("Resolving Jira account IDs from email addresses...")
|
||
email_to_id = resolve_account_ids(
|
||
[e["email"] for e in employees], jira_base_url, jira_email, jira_token
|
||
)
|
||
for emp in employees:
|
||
emp["account_id"] = email_to_id[emp["email"]]
|
||
|
||
known_ids = {e["account_id"] for e in employees}
|
||
print(f"Fetching Tempo worklogs from {args.tempo_base_url}...")
|
||
all_worklogs = fetch_tempo_worklogs(start, end, tempo_token, args.tempo_base_url)
|
||
|
||
# Detect reporters not in employees.md
|
||
unknown_ids = [aid for aid in all_worklogs if aid not in known_ids]
|
||
unknown_reporters: list[dict] = []
|
||
if unknown_ids:
|
||
print(f"Resolving {len(unknown_ids)} unknown reporter(s) from Jira...")
|
||
account_info = resolve_account_info(unknown_ids, jira_base_url, jira_email, jira_token)
|
||
for aid in unknown_ids:
|
||
info = account_info[aid]
|
||
total_h = round(sum(all_worklogs[aid].values()), 2)
|
||
unknown_reporters.append({
|
||
"account_id": aid,
|
||
"display_name": info["display_name"],
|
||
"email": info["email"],
|
||
"reported_hours": total_h,
|
||
})
|
||
print(f" Not in employees.md: {info['display_name']} ({info['email']}) — {total_h:.1f}h")
|
||
unknown_reporters.sort(key=lambda r: r["reported_hours"], reverse=True)
|
||
|
||
stats = compute_stats(employees, start, end, holidays_by_country, all_worklogs)
|
||
total_expected = sum(r["expected_hours"] for r in stats)
|
||
total_reported = round(sum(r["reported_hours"] for r in stats), 2)
|
||
overall_pct = round((total_reported / total_expected * 100) if total_expected > 0 else 100.0, 1)
|
||
|
||
# Annotate with email for readability
|
||
for stat in stats:
|
||
emp = next(e for e in employees if e["account_id"] == stat["account_id"])
|
||
stat["email"] = emp["email"]
|
||
|
||
working_days_by_country = {
|
||
c: len(get_working_days(start, end, holidays_by_country[c]))
|
||
for c in sorted(countries)
|
||
}
|
||
|
||
output = {
|
||
"period": {
|
||
"start": start.isoformat(),
|
||
"end": end.isoformat(),
|
||
"label": label,
|
||
"working_days_by_country": working_days_by_country,
|
||
},
|
||
"summary": {
|
||
"total_employees": len(stats),
|
||
"total_expected_hours": total_expected,
|
||
"total_reported_hours": total_reported,
|
||
"overall_completion_pct": overall_pct,
|
||
"employees_fully_complete": sum(1 for r in stats if r["completion_pct"] >= 100),
|
||
"employees_with_gaps": sum(1 for r in stats if r["completion_pct"] < 100),
|
||
},
|
||
"employees": stats,
|
||
"unknown_reporters": unknown_reporters,
|
||
}
|
||
|
||
output_dir = Path(args.output_dir)
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
out_path = output_dir / "timesheet_data.json"
|
||
out_path.write_text(json.dumps(output, indent=2), encoding="utf-8")
|
||
print(f"Wrote {out_path}")
|
||
return 0
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(
|
||
description="Fetch Tempo worklogs and compute timesheet completion."
|
||
)
|
||
parser.add_argument(
|
||
"--period",
|
||
required=True,
|
||
help="Period to check: 'last-week', 'last-month', or 'YYYY-MM-DD:YYYY-MM-DD'.",
|
||
)
|
||
parser.add_argument(
|
||
"--employees",
|
||
help=(
|
||
"Path to the employees markdown config file. "
|
||
"If omitted, looks for employees.md in the current directory, "
|
||
"then references/employees.md in the skill folder."
|
||
),
|
||
)
|
||
parser.add_argument(
|
||
"--output-dir",
|
||
required=True,
|
||
help="Directory for timesheet_data.json.",
|
||
)
|
||
parser.add_argument(
|
||
"--tempo-token",
|
||
help="Tempo API token. Defaults to TEMPO_API_TOKEN env var.",
|
||
)
|
||
parser.add_argument(
|
||
"--tempo-base-url",
|
||
default="https://api.tempo.io/4",
|
||
help="Tempo API base URL. Default: https://api.tempo.io/4 (Tempo Cloud).",
|
||
)
|
||
parser.add_argument(
|
||
"--jira-base-url",
|
||
help="Jira base URL, e.g. https://company.atlassian.net. Defaults to JIRA_BASE_URL env var.",
|
||
)
|
||
parser.add_argument(
|
||
"--jira-email",
|
||
help="Email address used to authenticate with Jira. Defaults to JIRA_EMAIL env var.",
|
||
)
|
||
parser.add_argument(
|
||
"--jira-token",
|
||
help="Jira personal API token. Defaults to JIRA_API_TOKEN env var.",
|
||
)
|
||
return parser
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(run(build_parser().parse_args()))
|