Bulk-upload time entries to YouTrack from a CSV or XLSX file.

youtrack_issue_trimmer.js

// define the json gotten from youtrack
const data = ;//

function getField(fields, name) {
  const f = fields.find(f => f.projectCustomField?.field?.name === name);
  if (!f) return null;
  const val = f.value;
  if (val === null || val === undefined) return null;
  if (Array.isArray(val)) return val.map(v => v.name || v.fullName || v).filter(Boolean);
  if (typeof val === 'object') return val.name || val.fullName || null;
  return val;
}

const result = data.flat().map(issue => ({
  id: issue.id,
  idReadable: issue.idReadable,
  summary: issue.summary,
  created: issue.created,
  updated: issue.updated,
  resolved: issue.resolved,
  project: issue.project?.name || null,
  reporter: issue.reporter?.fullName || null,
  updater: issue.updater?.fullName || null,
  votes: issue.votes,
  type: getField(issue.fields, 'Type'),
  state: getField(issue.fields, 'State'),
  resolution: getField(issue.fields, 'Resolution'),
  assignee: getField(issue.fields, 'Assignee'),
  priority: getField(issue.fields, 'Priority'),
  storyPoints: getField(issue.fields, 'Story Points'),
  subsystems: getField(issue.fields, 'Subsystems'),
  fixVersions: getField(issue.fields, 'Fix versions'),
  dueDate: getField(issue.fields, 'Due Date'),
  devResources: getField(issue.fields, 'Dev Resources'),
  airtableRecordId: getField(issue.fields, 'Airtable Record ID'),
}));

copy(JSON.stringify(result, null, 2));
console.log('Done! Result copied to clipboard ✅', result);

youtrack_timesheet.py

#!/usr/bin/env python3
"""
youtrack_timesheet.py
Bulk-upload time entries to YouTrack from a CSV or XLSX file.

Resumable & dedup-safe: every successful post is appended to a JSONL ledger.
A re-run loads that ledger, skips anything already logged, and continues from
where it left off. Identical rows within a single file are skipped too.

AUTH (browser-token workflow)
-----------------------------
This instance only issues short-lived session tokens (~1 hour), so grab a fresh
one from the browser per batch:
  1. YouTrack -> F12 -> Network tab; trigger any request to the server.
  2. Right-click it -> Copy -> Copy as cURL; paste into a file, e.g. auth.curl
  3. Run with --auth-from-curl auth.curl  (pulls the Bearer token, the
     cf_clearance cookie, and the base URL out of the pasted curl).
Alternatively set env vars: YT_TOKEN, YT_CF_CLEARANCE, YT_BASE_URL, YT_AUTHOR_ID.
The script refuses to start on an expired token.

INPUT FILE
----------
Headers are case-insensitive (aliases accepted). Columns:
    issue     (required)  e.g. WBMDMOB-59226
    date      (required)  e.g. 2026-05-08   (ISO is safest)
    duration  (required)  e.g. 1h, 30m, 1h 30m, 90m
    text      (optional)  description / comment
    type      (optional)  work item type NAME, e.g. Development

USAGE
-----
    python3 youtrack_timesheet.py timesheet.csv --auth-from-curl auth.curl --dry-run
    python3 youtrack_timesheet.py timesheet.csv --auth-from-curl auth.curl
    python3 youtrack_timesheet.py timesheet.xlsx --auth-from-curl auth.curl --sheet Sheet1
    python3 youtrack_timesheet.py --auth-from-curl auth.curl --list-types
    python3 youtrack_timesheet.py timesheet.csv --ledger run1.jsonl   # custom ledger
"""

from __future__ import annotations

import argparse
import base64
import csv
import hashlib
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional

DEFAULT_BASE_URL = "https://youtrack.<organization>.com"

HEADER_ALIASES = {
    "issue": "issue", "issue_id": "issue", "id": "issue", "ticket": "issue", "key": "issue",
    "date": "date", "day": "date",
    "duration": "duration", "time": "duration", "spent": "duration", "hours": "duration",
    "text": "text", "comment": "text", "description": "text", "notes": "text", "note": "text",
    "type": "type", "worktype": "type", "work_type": "type", "work type": "type",
}

DATE_FORMATS = ("%Y-%m-%d", "%Y/%m/%d", "%m/%d/%Y", "%m-%d-%Y", "%d.%m.%Y", "%d/%m/%Y")


def die(msg: str, code: int = 1):
    print(f"ERROR: {msg}", file=sys.stderr)
    sys.exit(code)


def to_epoch_ms(value) -> int:
    """Parse a date string/cell into epoch milliseconds at UTC midnight."""
    if isinstance(value, datetime):
        d = value
    else:
        s = str(value).strip()
        d = None
        for fmt in DATE_FORMATS:
            try:
                d = datetime.strptime(s, fmt)
                break
            except ValueError:
                continue
        if d is None:
            raise ValueError(f"unrecognised date {value!r} (use YYYY-MM-DD)")
    return int(datetime(d.year, d.month, d.day, tzinfo=timezone.utc).timestamp() * 1000)


# --------------------------------------------------------------------------- #
# WorkItem — the identity + wire-format of a single time entry
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class WorkItem:
    issue: str
    date_ms: int
    duration: str
    text: str = ""
    type_name: Optional[str] = None

    @property
    def date(self) -> str:
        return datetime.fromtimestamp(self.date_ms / 1000, tz=timezone.utc).date().isoformat()

    @property
    def key(self) -> str:
        """Stable content hash used for dedup / resume."""
        norm = "|".join([
            self.issue.upper(),
            str(self.date_ms),
            " ".join(self.duration.lower().split()),
            self.text.strip(),
            (self.type_name or "").lower().strip(),
        ])
        return hashlib.sha1(norm.encode()).hexdigest()[:16]

    def payload(self, author_id: Optional[str], type_id: Optional[str]) -> dict:
        body = {
            "date": self.date_ms,
            "duration": {"presentation": self.duration},
            "text": self.text,
            "type": {"id": type_id} if type_id else None,
        }
        if author_id:
            body["author"] = {"id": author_id}
        return body


def parse_row(row: dict) -> WorkItem:
    """Validate one input row and turn it into a WorkItem (raises ValueError)."""
    def field(name: str) -> str:
        return str(row.get(name, "") or "").strip()

    issue, duration, date_v = field("issue"), field("duration"), row.get("date", "")
    missing = [n for n, v in (("issue", issue), ("date", str(date_v).strip()),
                              ("duration", duration)) if not v]
    if missing:
        raise ValueError(f"missing required field(s): {', '.join(missing)}")
    return WorkItem(
        issue=issue,
        date_ms=to_epoch_ms(date_v),
        duration=duration,
        text=field("text"),
        type_name=field("type") or None,
    )


# --------------------------------------------------------------------------- #
# Input reading — CSV / XLSX into normalized row dicts
# --------------------------------------------------------------------------- #
def _normalize_header(h) -> str:
    key = str(h or "").strip().lower()
    return HEADER_ALIASES.get(key, key)


def _rows_from_table(header_cells, data_rows) -> list[dict]:
    headers = [_normalize_header(h) for h in header_cells]
    rows = []
    for raw in data_rows:
        cells = list(raw)
        if not any(str(c).strip() for c in cells if c is not None):
            continue  # skip blank lines
        rows.append({headers[i]: (cells[i] if i < len(cells) and cells[i] is not None else "")
                     for i in range(len(headers))})
    return rows


def _read_csv(path: str) -> list[dict]:
    with open(path, newline="", encoding="utf-8-sig") as f:
        table = list(csv.reader(f))
    return _rows_from_table(table[0], table[1:]) if table else []


def _read_xlsx(path: str, sheet: Optional[str]) -> list[dict]:
    try:
        import openpyxl
    except ImportError:
        die("Reading .xlsx needs openpyxl:  pip install openpyxl  (or export as .csv)")
    wb = openpyxl.load_workbook(path, data_only=True, read_only=True)
    ws = wb[sheet] if sheet else wb[wb.sheetnames[0]]
    table = list(ws.iter_rows(values_only=True))
    return _rows_from_table(table[0], table[1:]) if table else []


def read_rows(path: str, sheet: Optional[str] = None) -> list[dict]:
    ext = os.path.splitext(path)[1].lower()
    return _read_xlsx(path, sheet) if ext in (".xlsx", ".xlsm") else _read_csv(path)


# --------------------------------------------------------------------------- #
# Credentials — auth loading + expiry awareness
# --------------------------------------------------------------------------- #
@dataclass
class Credentials:
    token: str
    base_url: str = DEFAULT_BASE_URL
    cf_clearance: Optional[str] = None

    @classmethod
    def from_curl(cls, path: str) -> "Credentials":
        try:
            text = open(path, encoding="utf-8").read()
        except OSError as e:
            die(f"could not read curl file {path!r}: {e}")
        token = re.search(r"authorization:\s*Bearer\s+([A-Za-z0-9._\-]+)", text, re.I)
        if not token:
            die(f"no 'authorization: Bearer ...' header found in {path!r}")
        cf = re.search(r"cf_clearance=([^;'\"\s]+)", text)
        url = re.search(r"https?://[^/'\"\s]+", text)
        return cls(
            token=token.group(1),
            base_url=url.group(0).rstrip("/") if url else DEFAULT_BASE_URL,
            cf_clearance=cf.group(1) if cf else None,
        )

    @classmethod
    def from_env(cls) -> Optional["Credentials"]:
        token = os.environ.get("YT_TOKEN")
        if not token:
            return None
        return cls(
            token=token,
            base_url=os.environ.get("YT_BASE_URL", DEFAULT_BASE_URL).rstrip("/"),
            cf_clearance=os.environ.get("YT_CF_CLEARANCE"),
        )

    def minutes_left(self) -> Optional[float]:
        if self.token.count(".") != 2:
            return None  # not a JWT (e.g. perm: token) — can't tell
        try:
            payload = self.token.split(".")[1]
            payload += "=" * (-len(payload) % 4)
            exp = json.loads(base64.urlsafe_b64decode(payload)).get("exp")
            return (exp - time.time()) / 60.0 if exp else None
        except Exception:
            return None

    def ensure_valid(self):
        mins = self.minutes_left()
        if mins is not None and mins <= 0:
            die("Token has EXPIRED — grab a fresh one from the browser and retry.")
        if mins is not None and mins < 5:
            print(f"WARNING: token expires in ~{mins:.0f} min; a long batch may fail partway.\n")


# --------------------------------------------------------------------------- #
# YouTrackClient — all HTTP / API concerns
# --------------------------------------------------------------------------- #
class YouTrackClient:
    def __init__(self, creds: Credentials, author_id: Optional[str] = None):
        self.creds = creds
        self.author_id = author_id
        self._types: Optional[dict] = None

    def _request(self, method: str, path: str, body=None):
        data = json.dumps(body).encode() if body is not None else None
        req = urllib.request.Request(f"{self.creds.base_url}{path}", data=data, method=method)
        req.add_header("Authorization", f"Bearer {self.creds.token}")
        req.add_header("Accept", "application/json")
        if data is not None:
            req.add_header("Content-Type", "application/json")
        if self.creds.cf_clearance:
            req.add_header("Cookie", f"cf_clearance={self.creds.cf_clearance}")
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                raw = resp.read().decode()
                return json.loads(raw) if raw else {}
        except urllib.error.HTTPError as e:
            body = e.read().decode(errors="replace")[:300]
            raise RuntimeError(f"HTTP {e.code} {e.reason}: {body}") from None
        except urllib.error.URLError as e:
            raise RuntimeError(f"connection failed: {e.reason}") from None

    def work_types(self) -> dict:
        if self._types is None:
            items = self._request("GET", "/api/admin/timeTrackingSettings/workItemTypes?fields=id,name")
            self._types = {it["name"].lower(): it["id"] for it in items if it.get("name")}
        return self._types

    def resolve_type(self, name: Optional[str]) -> Optional[str]:
        if not name:
            return None
        tid = self.work_types().get(name.lower())
        if tid is None:
            die(f"unknown work type {name!r}. Valid: {', '.join(sorted(self.work_types())) or '(none)'}")
        return tid

    def create_work_item(self, item: WorkItem) -> str:
        type_id = self.resolve_type(item.type_name)
        path = (f"/api/issues/{item.issue}/timeTracking/workItems"
                "?fields=id,date,duration(presentation,minutes),text")
        res = self._request("POST", path, item.payload(self.author_id, type_id))
        return res.get("id", "?")


# --------------------------------------------------------------------------- #
# Ledger — append-only, crash-safe record of what's already posted
# --------------------------------------------------------------------------- #
class Ledger:
    def __init__(self, path: str):
        self.path = path
        self.keys: set[str] = set()
        if os.path.exists(path):
            with open(path, encoding="utf-8") as f:
                for line in f:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        self.keys.add(json.loads(line)["key"])
                    except (ValueError, KeyError):
                        continue  # tolerate a partially-written final line

    def __contains__(self, key: str) -> bool:
        return key in self.keys

    def record(self, item: WorkItem, work_item_id: str):
        entry = {
            "key": item.key,
            "issue": item.issue,
            "date": item.date,
            "duration": item.duration,
            "text": item.text,
            "type": item.type_name,
            "work_item_id": work_item_id,
            "posted_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        }
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry) + "\n")
        self.keys.add(item.key)


# --------------------------------------------------------------------------- #
# Orchestration
# --------------------------------------------------------------------------- #
def default_ledger_path(input_path: str) -> str:
    return f"{os.path.splitext(input_path)[0]}.ledger.jsonl"


def process(input_path: str, client: Optional[YouTrackClient], ledger: Ledger,
            sheet: Optional[str] = None, dry_run: bool = False, delay: float = 0.3) -> int:
    rows = read_rows(input_path, sheet)
    if not rows:
        die("no data rows found in the file.")

    print(f"{len(rows)} row(s) in {input_path}")
    print(f"ledger: {ledger.path} ({len(ledger.keys)} already logged)\n")

    seen_this_run: set[str] = set()
    posted = already = dup = failed = 0

    for i, row in enumerate(rows, 1):
        try:
            item = parse_row(row)
        except ValueError as e:
            print(f"[{i:>3}] SKIP  invalid — {e}")
            failed += 1
            continue

        label = f"[{i:>3}] {item.issue}  {item.duration:<8}  {item.date}  {item.text[:38]}"

        if item.key in ledger:
            print(f"{label}   · already logged")
            already += 1
            continue
        if item.key in seen_this_run:
            print(f"{label}   · duplicate in file")
            dup += 1
            continue
        seen_this_run.add(item.key)

        if dry_run:
            print(f"{label}   → would post")
            continue

        try:
            wid = client.create_work_item(item)
            ledger.record(item, wid)
            print(f"{label}   → posted id={wid}")
            posted += 1
        except RuntimeError as e:
            print(f"{label}   → FAILED: {e}")
            failed += 1
        time.sleep(delay)

    new_count = len(seen_this_run) if dry_run else posted
    verb = "would post" if dry_run else "posted"
    print(f"\nDone. {new_count} {verb}, {already} already logged, "
          f"{dup} dup-in-file, {failed} failed/invalid.")
    return 0 if failed == 0 else 2


# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Bulk-upload YouTrack time entries from CSV/XLSX (resumable, dedup-safe).")
    p.add_argument("file", nargs="?", help="path to .csv or .xlsx")
    p.add_argument("--auth-from-curl", metavar="FILE",
                   help="read Bearer token + cf_clearance from a pasted 'Copy as cURL' file")
    p.add_argument("--sheet", help="worksheet name for .xlsx (default: first sheet)")
    p.add_argument("--ledger", metavar="FILE", help="ledger path (default: <input>.ledger.jsonl)")
    p.add_argument("--dry-run", action="store_true", help="preview without posting")
    p.add_argument("--delay", type=float, default=0.3, help="seconds between posts (default 0.3)")
    p.add_argument("--list-types", action="store_true", help="list valid work item type names and exit")
    return p


def load_credentials(args) -> Optional[Credentials]:
    return Credentials.from_curl(args.auth_from_curl) if args.auth_from_curl else Credentials.from_env()


def main():
    parser = build_parser()
    args = parser.parse_args()
    creds = load_credentials(args)

    if args.list_types:
        if not creds:
            die("no token. Use --auth-from-curl FILE or set YT_TOKEN.")
        creds.ensure_valid()
        for name in sorted(YouTrackClient(creds).work_types()):
            print(name)
        return

    if not args.file:
        parser.error("provide a CSV/XLSX file (or use --list-types)")
    if not os.path.exists(args.file):
        die(f"file not found: {args.file}")

    client = None
    if not args.dry_run:
        if not creds:
            die("no token. Use --auth-from-curl FILE or set YT_TOKEN. See the docstring.")
        creds.ensure_valid()
        client = YouTrackClient(creds, author_id=os.environ.get("YT_AUTHOR_ID"))

    ledger = Ledger(args.ledger or default_ledger_path(args.file))
    sys.exit(process(args.file, client, ledger,
                     sheet=args.sheet, dry_run=args.dry_run, delay=args.delay))


if __name__ == "__main__":
    main()