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>
353 lines
14 KiB
Python
353 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Generate invoice line items from the user's own Tempo worklogs for a given month.
|
||
|
||
Groups hours by Jira project code and reports missing time vs the Polish
|
||
working calendar.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import calendar
|
||
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:
|
||
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
|
||
|
||
|
||
# ── Period helpers ────────────────────────────────────────────────────────────
|
||
|
||
def parse_month(month_str: str) -> tuple[date, date]:
|
||
"""Parse 'YYYY-MM' into (first_day, last_day) of that month."""
|
||
s = month_str.strip()
|
||
if not re.fullmatch(r"\d{4}-\d{2}", s):
|
||
raise ValueError(f"Invalid month '{month_str}'. Use YYYY-MM.")
|
||
year, month = int(s[:4]), int(s[5:7])
|
||
first = date(year, month, 1)
|
||
last = date(year, month, calendar.monthrange(year, month)[1])
|
||
return first, last
|
||
|
||
|
||
# ── Public holidays ───────────────────────────────────────────────────────────
|
||
|
||
def get_public_holidays(country: str, year: int) -> set[date]:
|
||
url = f"https://date.nager.at/api/v3/PublicHolidays/{year}/{country.upper()}"
|
||
req = urllib.request.Request(url, headers={"User-Agent": "invoice-prep/1.0"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
return {date.fromisoformat(item["date"]) for item in json.loads(resp.read())}
|
||
except Exception as exc:
|
||
print(f"Warning: could not fetch holidays for {country} {year}: {exc}", file=sys.stderr)
|
||
return set()
|
||
|
||
|
||
def working_days(start: date, end: date, holidays: set[date]) -> list[date]:
|
||
result, cur = [], start
|
||
while cur <= end:
|
||
if cur.weekday() < 5 and cur not in holidays:
|
||
result.append(cur)
|
||
cur += timedelta(days=1)
|
||
return result
|
||
|
||
|
||
# ── Jira account lookup ───────────────────────────────────────────────────────
|
||
|
||
def resolve_account_id(email: str, jira_base_url: str,
|
||
jira_email: str, jira_token: str) -> str:
|
||
creds = base64.b64encode(f"{jira_email}:{jira_token}".encode()).decode()
|
||
headers = {"Authorization": f"Basic {creds}", "Accept": "application/json"}
|
||
url = f"{jira_base_url.rstrip('/')}/rest/api/3/user/search?query={email}&maxResults=5"
|
||
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. Check JIRA_EMAIL and JIRA_API_TOKEN.",
|
||
file=sys.stderr)
|
||
sys.exit(1)
|
||
raise RuntimeError(f"Jira API {exc.code}") from exc
|
||
match = next(
|
||
(u for u in users if u.get("emailAddress", "").lower() == email.lower()),
|
||
None,
|
||
)
|
||
if not match:
|
||
print(f"Error: Jira account not found for {email}.", file=sys.stderr)
|
||
sys.exit(1)
|
||
return match["accountId"]
|
||
|
||
|
||
# ── Tempo worklogs ────────────────────────────────────────────────────────────
|
||
|
||
_PROJECT_CODE_RE = re.compile(r"\b([A-Z][A-Z0-9]{1,9})-\d+\b")
|
||
|
||
|
||
def load_jira_ignore(path: Path) -> set[str]:
|
||
"""Load issue keys to ignore from a markdown table in jira_ignore.md."""
|
||
if not path.exists():
|
||
return set()
|
||
ignored: set[str] = set()
|
||
in_table = False
|
||
header_seen = False
|
||
for line in path.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not (line.startswith("|") and line.endswith("|")):
|
||
in_table = False
|
||
header_seen = False
|
||
continue
|
||
cells = [c.strip() for c in line.strip("|").split("|")]
|
||
if not header_seen:
|
||
# First row is header or separator
|
||
if all(re.fullmatch(r":?-{3,}:?", c) for c in cells if c):
|
||
continue # separator row
|
||
header_seen = True
|
||
in_table = True
|
||
continue
|
||
if all(re.fullmatch(r":?-{3,}:?", c) for c in cells if c):
|
||
continue # separator row
|
||
if in_table and cells and cells[0]:
|
||
ignored.add(cells[0].strip().upper())
|
||
return ignored
|
||
|
||
|
||
def extract_project_code(issue_key: str) -> str | None:
|
||
"""Extract project code from a Jira issue key, e.g. 'IAA' from 'IAA-42'."""
|
||
m = _PROJECT_CODE_RE.match(issue_key.strip())
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def fetch_worklogs(account_id: str, start: date, end: date,
|
||
token: str, base_url: str = "https://api.tempo.io/4") -> list[dict]:
|
||
"""Return all Tempo worklog entries for the account in [start, end]."""
|
||
results: list[dict] = []
|
||
headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
|
||
next_url: str | None = (
|
||
f"{base_url}/worklogs/user/{account_id}"
|
||
f"?from={start}&to={end}&limit=1000"
|
||
)
|
||
while next_url:
|
||
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. Check TEMPO_API_TOKEN.", file=sys.stderr)
|
||
sys.exit(1)
|
||
raise RuntimeError(f"Tempo API {exc.code}: {body[:200]}") from exc
|
||
results.extend(data.get("results", []))
|
||
next_url = data.get("metadata", {}).get("next")
|
||
return results
|
||
|
||
|
||
def resolve_issue_keys(worklogs: list[dict], jira_email: str,
|
||
jira_token: str) -> dict[int, str]:
|
||
"""
|
||
Resolve Jira issue IDs → keys by calling each issue's self URL directly.
|
||
Uses the self URL already present in each Tempo worklog entry.
|
||
"""
|
||
creds = base64.b64encode(f"{jira_email}:{jira_token}".encode()).decode()
|
||
headers = {"Authorization": f"Basic {creds}", "Accept": "application/json"}
|
||
seen: dict[int, str] = {}
|
||
for wl in worklogs:
|
||
issue = wl.get("issue", {})
|
||
issue_id = issue.get("id")
|
||
if not issue_id or int(issue_id) in seen:
|
||
continue
|
||
self_url = issue.get("self", "")
|
||
if not self_url:
|
||
continue
|
||
# Swap api/2 → api/3 if needed; append fields filter
|
||
url = re.sub(r"/rest/api/\d+/", "/rest/api/3/", self_url) + "?fields=key"
|
||
req = urllib.request.Request(url, headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
data = json.loads(resp.read())
|
||
seen[int(issue_id)] = data.get("key", "")
|
||
except Exception as exc:
|
||
print(f"Warning: could not resolve issue {issue_id}: {exc}", file=sys.stderr)
|
||
return seen
|
||
|
||
|
||
# ── Main ──────────────────────────────────────────────────────────────────────
|
||
|
||
def run(args: argparse.Namespace) -> int:
|
||
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 = os.environ.get("TEMPO_API_TOKEN", "")
|
||
jira_base_url = os.environ.get("JIRA_BASE_URL", "")
|
||
jira_email = os.environ.get("JIRA_EMAIL", "")
|
||
jira_token = os.environ.get("JIRA_API_TOKEN", "")
|
||
|
||
missing = [n for n, v in [
|
||
("TEMPO_API_TOKEN", tempo_token), ("JIRA_BASE_URL", jira_base_url),
|
||
("JIRA_EMAIL", jira_email), ("JIRA_API_TOKEN", jira_token),
|
||
] if not v]
|
||
if missing:
|
||
print(f"Error: missing credentials: {', '.join(missing)}", file=sys.stderr)
|
||
return 1
|
||
|
||
try:
|
||
start, end = parse_month(args.month)
|
||
except ValueError as exc:
|
||
print(f"Error: {exc}", file=sys.stderr)
|
||
return 1
|
||
|
||
# Load ignore list
|
||
ignore_path = Path.cwd() / "jira_ignore.md"
|
||
ignored_keys = load_jira_ignore(ignore_path)
|
||
if ignored_keys:
|
||
print(f"Ignoring issues: {', '.join(sorted(ignored_keys))}")
|
||
|
||
print(f"Month: {args.month} ({start} – {end})")
|
||
print(f"User: {jira_email}")
|
||
|
||
# Resolve account ID
|
||
print("Resolving Jira account...")
|
||
account_id = resolve_account_id(jira_email, jira_base_url, jira_email, jira_token)
|
||
print(f" {jira_email} → {account_id}")
|
||
|
||
# Working days (Polish calendar)
|
||
print("Fetching public holidays...")
|
||
holidays = get_public_holidays("PL", start.year)
|
||
wdays = working_days(start, end, holidays)
|
||
expected_hours = len(wdays) * 8.0
|
||
print(f" Working days: {len(wdays)} Expected hours: {expected_hours:.0f}h")
|
||
|
||
# Fetch Tempo worklogs
|
||
print("Fetching Tempo worklogs...")
|
||
worklogs = fetch_worklogs(account_id, start, end, tempo_token)
|
||
print(f" Worklogs fetched: {len(worklogs)}")
|
||
|
||
# Resolve issue IDs → keys via each issue's self URL (Tempo v4 omits the key)
|
||
unique_ids = {int(wl["issue"]["id"]) for wl in worklogs if wl.get("issue", {}).get("id")}
|
||
issue_key_map: dict[int, str] = {}
|
||
if unique_ids:
|
||
print(f" Resolving {len(unique_ids)} unique issue key(s) from Jira...")
|
||
issue_key_map = resolve_issue_keys(worklogs, jira_email, jira_token)
|
||
print(f" Resolved: {len(issue_key_map)}")
|
||
|
||
# Group hours by project code and by date, skipping ignored issues
|
||
project_hours: dict[str, float] = {}
|
||
hours_by_date: dict[str, float] = {}
|
||
unmatched_hours = 0.0
|
||
ignored_hours = 0.0
|
||
|
||
for wl in worklogs:
|
||
issue_id = int(wl.get("issue", {}).get("id", 0))
|
||
issue_key = issue_key_map.get(issue_id, "")
|
||
hours = wl.get("timeSpentSeconds", 0) / 3600
|
||
wl_date = wl.get("startDate", "")
|
||
|
||
if issue_key.upper() in ignored_keys:
|
||
ignored_hours += hours
|
||
# Still count toward date totals so missing-day detection stays accurate
|
||
if wl_date:
|
||
hours_by_date[wl_date] = hours_by_date.get(wl_date, 0.0) + hours
|
||
continue
|
||
|
||
code = extract_project_code(issue_key)
|
||
if code:
|
||
project_hours[code] = project_hours.get(code, 0.0) + hours
|
||
else:
|
||
unmatched_hours += hours
|
||
if wl_date:
|
||
hours_by_date[wl_date] = hours_by_date.get(wl_date, 0.0) + hours
|
||
|
||
# Sort projects by hours descending
|
||
projects = sorted(
|
||
[{"code": k, "hours": round(v, 2)} for k, v in project_hours.items()],
|
||
key=lambda x: x["hours"],
|
||
reverse=True,
|
||
)
|
||
|
||
total_logged = round(sum(project_hours.values()) + unmatched_hours + ignored_hours, 2)
|
||
total_billable = round(sum(project_hours.values()) + unmatched_hours, 2)
|
||
missing_hours = round(max(0.0, expected_hours - total_logged), 2)
|
||
|
||
# Find working days with no logged time
|
||
days_missing = [
|
||
d.isoformat() for d in wdays
|
||
if hours_by_date.get(d.isoformat(), 0.0) == 0.0
|
||
]
|
||
|
||
output = {
|
||
"month": args.month,
|
||
"user_email": jira_email,
|
||
"account_id": account_id,
|
||
"working_days": len(wdays),
|
||
"expected_hours": expected_hours,
|
||
"total_logged_hours": total_logged,
|
||
"total_billable_hours": total_billable,
|
||
"ignored_hours": round(ignored_hours, 2),
|
||
"ignored_keys": sorted(ignored_keys),
|
||
"missing_hours": missing_hours,
|
||
"unmatched_hours": round(unmatched_hours, 2),
|
||
"days_with_no_hours": days_missing,
|
||
"projects": projects,
|
||
}
|
||
|
||
output_dir = Path(args.output_dir)
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
out_path = output_dir / f"invoice_prep_{args.month}.json"
|
||
out_path.write_text(json.dumps(output, indent=2, ensure_ascii=False), encoding="utf-8")
|
||
print(f"Wrote {out_path}")
|
||
|
||
# Print quick summary
|
||
print(f"\nProjects:")
|
||
for p in projects:
|
||
print(f" {p['code']:12s} {p['hours']:.2f}h")
|
||
if unmatched_hours:
|
||
print(f" (no project) {unmatched_hours:.2f}h")
|
||
if ignored_hours:
|
||
print(f" (ignored) {ignored_hours:.2f}h ({', '.join(sorted(ignored_keys))})")
|
||
print(f"\nBillable total: {total_billable:.2f}h")
|
||
print(f"Total logged: {total_logged:.2f}h / {expected_hours:.0f}h expected")
|
||
if missing_hours > 0:
|
||
print(f"⚠ Missing: {missing_hours:.2f}h ({len(days_missing)} day(s) with no time logged)")
|
||
|
||
return 0
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(
|
||
description="Generate invoice line items from Tempo worklogs."
|
||
)
|
||
parser.add_argument("--month", required=True, help="Month to check: YYYY-MM")
|
||
parser.add_argument("--output-dir", required=True,
|
||
help="Directory for invoice_prep_YYYY-MM.json")
|
||
return parser
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(run(build_parser().parse_args()))
|