576a14c82f
Adds ivona (invoice-checker, timesheet-checker) and timesheet (check-my-timesheet, invoice-prep) as new plugins, plus invoice-prep-test — a one-off experiment that requests the Tempo API token via plugin userConfig at install time instead of a .env file, to check whether that value actually reaches the skill's script. Also adds the root LICENSE (MIT) referenced from the README.
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()))
|