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>
250 lines
10 KiB
Python
250 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""Check the current user's timesheet for gaps against the 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
|
||
|
||
|
||
# ── Config loading ────────────────────────────────────────────────────────────
|
||
|
||
def load_config() -> dict[str, Any]:
|
||
"""Read timesheet config from env vars MY_TIMESHEET_COUNTRY and MY_TIMESHEET_HOURS_PER_DAY."""
|
||
country = os.environ.get("MY_TIMESHEET_COUNTRY", "PL").strip().upper()
|
||
raw_hpd = os.environ.get("MY_TIMESHEET_HOURS_PER_DAY", "8").strip()
|
||
try:
|
||
hours_per_day = float(raw_hpd)
|
||
except ValueError:
|
||
print(f"Warning: invalid MY_TIMESHEET_HOURS_PER_DAY '{raw_hpd}' — using 8.", file=sys.stderr)
|
||
hours_per_day = 8.0
|
||
return {"country": country, "hours_per_day": hours_per_day}
|
||
|
||
|
||
# ── Period parsing ────────────────────────────────────────────────────────────
|
||
|
||
def parse_period(period_str: str) -> tuple[date, date]:
|
||
s = period_str.strip()
|
||
if ":" in s:
|
||
a, b = s.split(":", 1)
|
||
return date.fromisoformat(a.strip()), date.fromisoformat(b.strip())
|
||
raise ValueError(f"Invalid period '{period_str}'. Use YYYY-MM-DD:YYYY-MM-DD.")
|
||
|
||
|
||
# ── Public holidays ───────────────────────────────────────────────────────────
|
||
|
||
def get_holidays(country: str, years: set[int]) -> set[date]:
|
||
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": "check-my-timesheet/1.0"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as r:
|
||
holidays |= {date.fromisoformat(i["date"]) for i in json.loads(r.read())}
|
||
except Exception as exc:
|
||
print(f"Warning: could not fetch holidays for {country} {year}: {exc}", file=sys.stderr)
|
||
return holidays
|
||
|
||
|
||
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 r:
|
||
users = json.loads(r.read())
|
||
except urllib.error.HTTPError as exc:
|
||
if exc.code == 401:
|
||
print("Error: Jira 401. Check JIRA_EMAIL and JIRA_API_TOKEN.", file=sys.stderr)
|
||
sys.exit(1)
|
||
raise
|
||
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 ────────────────────────────────────────────────────────────
|
||
|
||
def fetch_worklogs(account_id: str, start: date, end: date,
|
||
token: str, base_url: str = "https://api.tempo.io/4") -> list[dict]:
|
||
results: list[dict] = []
|
||
headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
|
||
next_url: str | None = (
|
||
f"{base_url}/worklogs/user/{account_id}?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 r:
|
||
data = json.loads(r.read())
|
||
except urllib.error.HTTPError as exc:
|
||
body = exc.read().decode("utf-8", errors="replace")
|
||
if exc.code == 401:
|
||
print("Error: Tempo 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
|
||
|
||
|
||
# ── 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
|
||
|
||
cfg = load_config()
|
||
print(f"Config: country={cfg['country']}, hours_per_day={cfg['hours_per_day']}")
|
||
|
||
try:
|
||
start, end = parse_period(args.period)
|
||
except ValueError as exc:
|
||
print(f"Error: {exc}", file=sys.stderr)
|
||
return 1
|
||
|
||
print(f"Period: {start} – {end}")
|
||
print(f"User: {jira_email}")
|
||
|
||
print("Resolving Jira account...")
|
||
account_id = resolve_account_id(jira_email, jira_base_url, jira_email, jira_token)
|
||
print(f" {jira_email} → {account_id}")
|
||
|
||
print("Fetching public holidays...")
|
||
holidays = get_holidays(cfg["country"], {start.year, end.year})
|
||
wdays = working_days(start, end, holidays)
|
||
expected_hours = len(wdays) * cfg["hours_per_day"]
|
||
print(f" Working days: {len(wdays)} Expected: {expected_hours:.0f}h")
|
||
|
||
print("Fetching Tempo worklogs...")
|
||
worklogs = fetch_worklogs(account_id, start, end, tempo_token)
|
||
print(f" Worklogs: {len(worklogs)}")
|
||
|
||
# Aggregate hours per date
|
||
hours_by_date: dict[str, float] = {}
|
||
for wl in worklogs:
|
||
d = wl.get("startDate", "")
|
||
if d:
|
||
hours_by_date[d] = hours_by_date.get(d, 0.0) + wl.get("timeSpentSeconds", 0) / 3600
|
||
|
||
# Classify each working day
|
||
hpd = cfg["hours_per_day"]
|
||
days_out = []
|
||
for d in wdays:
|
||
logged = round(hours_by_date.get(d.isoformat(), 0.0), 2)
|
||
if logged >= hpd:
|
||
status = "ok"
|
||
entry = {"date": d.isoformat(), "logged_hours": logged, "status": "ok"}
|
||
elif logged > 0:
|
||
missing_h = round(hpd - logged, 2)
|
||
status = "partial"
|
||
entry = {"date": d.isoformat(), "logged_hours": logged,
|
||
"status": "partial", "missing": missing_h}
|
||
else:
|
||
status = "missing"
|
||
entry = {"date": d.isoformat(), "logged_hours": 0.0,
|
||
"status": "missing", "missing": hpd}
|
||
days_out.append(entry)
|
||
|
||
total_logged = round(sum(hours_by_date.get(d.isoformat(), 0.0) for d in wdays), 2)
|
||
missing_hours = round(max(0.0, expected_hours - total_logged), 2)
|
||
|
||
output = {
|
||
"period": {"start": str(start), "end": str(end)},
|
||
"config": cfg,
|
||
"working_days": len(wdays),
|
||
"expected_hours": expected_hours,
|
||
"total_logged_hours": total_logged,
|
||
"missing_hours": missing_hours,
|
||
"days": days_out,
|
||
}
|
||
|
||
output_dir = Path(args.output_dir)
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
out_path = output_dir / f"my_check_{start}_{end}.json"
|
||
out_path.write_text(json.dumps(output, indent=2, ensure_ascii=False), encoding="utf-8")
|
||
print(f"Wrote {out_path}")
|
||
|
||
# Quick summary
|
||
problem_days = [d for d in days_out if d["status"] != "ok"]
|
||
print(f"\nLogged: {total_logged:.1f}h / {expected_hours:.0f}h expected")
|
||
if not problem_days:
|
||
print("✓ Timesheet complete.")
|
||
else:
|
||
missing_days = [d for d in problem_days if d["status"] == "missing"]
|
||
partial_days = [d for d in problem_days if d["status"] == "partial"]
|
||
if missing_days:
|
||
print(f"✗ Missing ({len(missing_days)} day(s)): "
|
||
+ ", ".join(d["date"] for d in missing_days))
|
||
if partial_days:
|
||
print(f"⚠ Partial ({len(partial_days)} day(s)): "
|
||
+ ", ".join(f"{d['date']} ({d['logged_hours']}h)" for d in partial_days))
|
||
return 0
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
p = argparse.ArgumentParser(description="Check timesheet gaps for the current user.")
|
||
p.add_argument("--period", required=True, help="YYYY-MM-DD:YYYY-MM-DD")
|
||
p.add_argument("--output-dir", required=True)
|
||
return p
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(run(build_parser().parse_args()))
|