jason/open-project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ver 1.0.0 initial commit

Browse Source
This commit is contained in:
Jason Stedwell
2026-06-27 17:25:55 -05:00
parent 550111295a
commit 2a7efe9c5d
20 changed files with 1850 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/openproject/scripts/op.py
+775
View File
@@ -0,0 +1,775 @@
#!/usr/bin/env python3
"""
op — command-line interface to an OpenProject instance.
A thin, friendly CLI over the OpenProject API v3, designed for Claude (and you)
to read and manage projects, work packages, time entries, and more.
Run `op.py --help` or `op.py <command> --help` for usage.
Output: human-readable tables by default; add `--json` to any command for the
raw HAL+JSON (useful for chaining / precise field access).
Examples:
op.py ping
op.py projects --search infra
op.py wp list --project 3 --assignee me --status open
op.py wp show 1234
op.py wp create --project 3 --type Task --subject "Fix the thing" \
--description "details..." --assignee me --priority High
op.py wp update 1234 --status "In progress" --done-ratio 50
op.py wp comment 1234 --text "Working on it"
op.py wp close 1234
op.py wp delete 1234 --yes
op.py time log --wp 1234 --hours 1.5 --comment "investigation"
op.py raw GET /api/v3/statuses
"""
from __future__ import annotations
import argparse
import datetime
import json
import sys
from opclient import OpenProjectClient, OpenProjectError
def _today():
return datetime.date.today().isoformat()
# --------------------------------------------------------------------------- #
# Output helpers
# --------------------------------------------------------------------------- #
def emit_json(obj):
print(json.dumps(obj, indent=2, ensure_ascii=False))
def table(rows, headers):
"""Render a simple aligned text table."""
if not rows:
print("(no results)")
return
cols = list(zip(*([headers] + rows)))
widths = [max(len(str(c)) for c in col) for col in cols]
line = " ".join(str(h).ljust(w) for h, w in zip(headers, widths))
print(line)
print(" ".join("-" * w for w in widths))
for row in rows:
print(" ".join(str(c).ljust(w) for c, w in zip(row, widths)))
def wp_row(wp, client):
return [
wp.get("id"),
(wp.get("subject") or "")[:50],
client.link_title(wp, "type") or "",
client.link_title(wp, "status") or "",
client.link_title(wp, "assignee") or "—",
client.link_title(wp, "project") or "",
]
# --------------------------------------------------------------------------- #
# Filter building
# --------------------------------------------------------------------------- #
def build_wp_filters(client, args):
filters = []
if getattr(args, "project", None):
proj = client.find_project(args.project)
filters.append({"project": {"operator": "=", "values": [str(proj["id"])]}})
if getattr(args, "assignee", None):
if args.assignee.lower() == "me":
filters.append({"assignee": {"operator": "=", "values": ["me"]}})
else:
user = client.find_user(args.assignee)
filters.append({"assignee": {"operator": "=", "values": [str(user["id"])]}})
if getattr(args, "status", None):
s = args.status.lower()
if s == "open":
filters.append({"status": {"operator": "o", "values": None}})
elif s in ("closed", "close"):
filters.append({"status": {"operator": "c", "values": None}})
else:
st = client.find_status(args.status)
filters.append({"status": {"operator": "=", "values": [str(st["id"])]}})
if getattr(args, "type", None):
t = client.find_type(args.type)
filters.append({"type": {"operator": "=", "values": [str(t["id"])]}})
if getattr(args, "query", None):
filters.append({"search": {"operator": "**", "values": [args.query]}})
return filters
# --------------------------------------------------------------------------- #
# Commands
# --------------------------------------------------------------------------- #
def cmd_ping(client, args):
me = client.me()
name = me.get("name") or me.get("login")
print(f"OK — authenticated as {name} (id {me.get('id')}) at {client.base_url}")
def cmd_me(client, args):
me = client.me()
if args.json:
emit_json(me)
return
for k in ("id", "name", "login", "email", "admin", "status"):
if k in me:
print(f"{k:>8}: {me[k]}")
def cmd_projects(client, args):
filters = None
if args.search:
filters = [{"name_and_identifier": {"operator": "~", "values": [args.search]}}]
items = client.get_all("/api/v3/projects", params={"filters": filters} if filters else None,
max_items=args.limit)
if args.json:
emit_json(items)
return
rows = [[p.get("id"), p.get("identifier"), (p.get("name") or "")[:50],
"archived" if not p.get("active", True) else "active"] for p in items]
table(rows, ["ID", "IDENTIFIER", "NAME", "STATE"])
print(f"\n{len(items)} project(s)")
def cmd_wp_list(client, args):
filters = build_wp_filters(client, args)
params = {"pageSize": args.limit}
if filters:
params["filters"] = filters
if args.sort:
params["sortBy"] = [args.sort.split(":")] if ":" in args.sort else [[args.sort, "asc"]]
items = client.get_all("/api/v3/work_packages", params=params, max_items=args.limit)
if args.json:
emit_json(items)
return
rows = [wp_row(wp, client) for wp in items]
table(rows, ["ID", "SUBJECT", "TYPE", "STATUS", "ASSIGNEE", "PROJECT"])
print(f"\n{len(items)} work package(s)")
def cmd_wp_show(client, args):
wp = client.get(f"/api/v3/work_packages/{args.id}")
if args.json:
emit_json(wp)
return
L = client.link_title
print(f"#{wp.get('id')} {wp.get('subject')}")
print(f" project : {L(wp,'project')}")
print(f" type : {L(wp,'type')}")
print(f" status : {L(wp,'status')} ({wp.get('percentageDone', 0)}% done)")
print(f" priority: {L(wp,'priority')}")
print(f" assignee: {L(wp,'assignee') or '—'}")
print(f" author : {L(wp,'author')}")
if wp.get("startDate") or wp.get("dueDate"):
print(f" dates : {wp.get('startDate') or '?'} → {wp.get('dueDate') or '?'}")
print(f" version : lockVersion={wp.get('lockVersion')}")
desc = (wp.get("description") or {}).get("raw") or ""
if desc.strip():
print(" description:")
for line in desc.splitlines():
print(f" {line}")
def _wp_link_body(client, args):
"""Build the _links + scalar body for create/update from common args."""
body = {}
links = {}
if getattr(args, "subject", None) is not None:
body["subject"] = args.subject
if getattr(args, "description", None) is not None:
body["description"] = {"format": "markdown", "raw": args.description}
if getattr(args, "start", None) is not None:
body["startDate"] = args.start
if getattr(args, "due", None) is not None:
body["dueDate"] = args.due
if getattr(args, "done_ratio", None) is not None:
body["percentageDone"] = int(args.done_ratio)
if getattr(args, "project", None):
links["project"] = {"href": f"/api/v3/projects/{client.find_project(args.project)['id']}"}
if getattr(args, "type", None):
links["type"] = {"href": f"/api/v3/types/{client.find_type(args.type)['id']}"}
if getattr(args, "status", None):
links["status"] = {"href": f"/api/v3/statuses/{client.find_status(args.status)['id']}"}
if getattr(args, "priority", None):
links["priority"] = {"href": f"/api/v3/priorities/{client.find_priority(args.priority)['id']}"}
if getattr(args, "assignee", None):
if args.assignee.lower() in ("none", "unassign"):
links["assignee"] = {"href": None}
else:
links["assignee"] = {"href": f"/api/v3/users/{client.find_user(args.assignee)['id']}"}
if getattr(args, "parent", None):
links["parent"] = {"href": f"/api/v3/work_packages/{args.parent}"}
if links:
body["_links"] = links
return body
def cmd_wp_create(client, args):
body = _wp_link_body(client, args)
if "subject" not in body:
raise OpenProjectError("--subject is required to create a work package.")
if "_links" not in body or "project" not in body["_links"]:
raise OpenProjectError("--project is required to create a work package.")
if "type" not in body.get("_links", {}):
# Default to the project's first available type.
proj_id = client.id_from_href(body["_links"]["project"]["href"])
types = client.get_all(f"/api/v3/projects/{proj_id}/types", max_items=1)
if types:
body["_links"]["type"] = {"href": types[0]["_links"]["self"]["href"]}
created = client.create_work_package(body)
if args.json:
emit_json(created)
return
print(f"Created #{created.get('id')}: {created.get('subject')}")
print(f" {client.base_url}/work_packages/{created.get('id')}")
def cmd_wp_update(client, args):
current = client.get(f"/api/v3/work_packages/{args.id}")
body = _wp_link_body(client, args)
if not body:
raise OpenProjectError("Nothing to update — pass at least one field.")
body["lockVersion"] = current.get("lockVersion")
updated = client.patch(f"/api/v3/work_packages/{args.id}", body=body)
if args.json:
emit_json(updated)
return
print(f"Updated #{updated.get('id')}: {updated.get('subject')} "
f"[{client.link_title(updated,'status')}]")
def cmd_wp_comment(client, args):
body = {"comment": {"format": "markdown", "raw": args.text}}
res = client.post(f"/api/v3/work_packages/{args.id}/activities", body=body)
if args.json:
emit_json(res)
return
print(f"Comment added to #{args.id}.")
def cmd_wp_close(client, args):
current = client.get(f"/api/v3/work_packages/{args.id}")
status = client.find_status(args.status_name)
body = {
"lockVersion": current.get("lockVersion"),
"_links": {"status": {"href": f"/api/v3/statuses/{status['id']}"}},
}
updated = client.patch(f"/api/v3/work_packages/{args.id}", body=body)
print(f"#{args.id} → {client.link_title(updated,'status')}")
def cmd_wp_delete(client, args):
if not args.yes:
raise OpenProjectError(
f"Refusing to delete #{args.id} without --yes. "
f"This permanently removes the work package."
)
client.delete(f"/api/v3/work_packages/{args.id}")
print(f"Deleted work package #{args.id}.")
def cmd_time_log(client, args):
wp = client.get(f"/api/v3/work_packages/{args.wp}")
proj_href = wp["_links"]["project"]["href"]
body = {
"hours": f"PT{args.hours}H" if not str(args.hours).startswith("PT") else args.hours,
"_links": {
"workPackage": {"href": f"/api/v3/work_packages/{args.wp}"},
"project": {"href": proj_href},
},
}
if args.comment:
body["comment"] = {"format": "markdown", "raw": args.comment}
# spentOn is required by the API; default to today when not given.
body["spentOn"] = args.date or _today()
if args.activity:
act = client._resolve_by_name("/api/v3/time_entries/activities", args.activity)
body["_links"]["activity"] = {"href": act["_links"]["self"]["href"]}
res = client.post("/api/v3/time_entries", body=body)
if args.json:
emit_json(res)
return
print(f"Logged {args.hours}h on #{args.wp}.")
def cmd_users(client, args):
filters = None
if args.search:
filters = [{"name": {"operator": "~", "values": [args.search]}}]
items = client.get_all("/api/v3/users", params={"filters": filters} if filters else None,
max_items=args.limit)
if args.json:
emit_json(items)
return
rows = [[u.get("id"), u.get("login"), u.get("name"), u.get("email", "")] for u in items]
table(rows, ["ID", "LOGIN", "NAME", "EMAIL"])
def _simple_list(client, path, args, name_field="name"):
items = client.get_all(path, max_items=200)
if args.json:
emit_json(items)
return
rows = [[i.get("id"), i.get(name_field)] for i in items]
table(rows, ["ID", "NAME"])
def cmd_types(client, args):
if args.project:
proj = client.find_project(args.project)
_simple_list(client, f"/api/v3/projects/{proj['id']}/types", args)
else:
_simple_list(client, "/api/v3/types", args)
def cmd_statuses(client, args):
_simple_list(client, "/api/v3/statuses", args)
def cmd_priorities(client, args):
_simple_list(client, "/api/v3/priorities", args)
def cmd_roles(client, args):
_simple_list(client, "/api/v3/roles", args)
def cmd_project_create(client, args):
created = client.create_project(
args.name, identifier=args.identifier,
description=args.description, parent=args.parent)
if args.json:
emit_json(created)
return
print(f"Created project '{created.get('name')}' "
f"(id {created.get('id')}, identifier {created.get('identifier')})")
print(f" {client.base_url}/projects/{created.get('identifier')}")
def cmd_members_add(client, args):
proj = client.find_project(args.project)
user = client.find_user(args.user)
client.add_membership(proj["id"], user["id"], args.role)
print(f"Added {user.get('name')} to '{proj.get('name')}' as {args.role}.")
def cmd_wp_relate(client, args):
res = client.create_relation(args.id, args.to, rel_type=args.type,
description=args.description)
print(f"#{args.id} {args.type} #{args.to} (relation {res.get('id')})")
# --------------------------------------------------------------------------- #
# Import: spreadsheet -> project + work packages
# --------------------------------------------------------------------------- #
# Default value mappings (tuned for the APTA planning register; override-able
# would be a future flag, but these are sensible defaults for H/M/L + statuses).
PRIORITY_MAP = {"H": "High", "M": "Normal", "L": "Low",
"HIGH": "High", "MEDIUM": "Normal", "LOW": "Low"}
STATUS_MAP = {
"OPEN": "New", "NEW": "New",
"IN PROGRESS": "In progress", "INPROGRESS": "In progress",
"COMPLETE": "Closed", "COMPLETED": "Closed", "DONE": "Closed",
"CLOSED": "Closed",
}
DEFAULT_PRIORITY = "Normal"
DEFAULT_STATUS = "New"
# Column header -> internal field. Header lookup is case-insensitive and matches
# on a prefix so "Priority (H/M/L)" maps to priority.
COLUMN_ALIASES = {
"task": "subject", "title": "subject", "name": "subject",
"category": "category", "group": "category",
"owner": "owner", "assignee": "owner", "assigned": "owner",
"priority": "priority",
"status": "status",
"due": "due", "due date": "due",
"notes": "notes", "notes / dependencies": "notes", "dependencies": "notes",
"#": "ref", "id": "ref",
}
def _map_columns(headers):
"""Build {internal_field: header_string} from the sheet headers."""
mapping = {}
for h in headers:
if not h:
continue
key = h.strip().lower()
field = COLUMN_ALIASES.get(key)
if not field:
for alias, fld in COLUMN_ALIASES.items():
if key.startswith(alias):
field = fld
break
if field and field not in mapping:
mapping[field] = h
return mapping
def _build_description(rec, cols, owner_noted):
notes = (rec.get(cols.get("notes", "")) or "").strip()
meta = []
if cols.get("ref") and rec.get(cols["ref"]):
ref = str(rec[cols["ref"]]).rstrip("0").rstrip(".")
meta.append(f"- Source row #: {ref}")
if cols.get("category") and rec.get(cols["category"]):
meta.append(f"- Category: {rec[cols['category']].strip()}")
if owner_noted:
meta.append(f"- Owner: {owner_noted}")
if cols.get("due") and (rec.get(cols["due"]) or "").strip():
meta.append(f"- Due (from sheet): {rec[cols['due']].strip()}")
parts = []
if notes:
parts.append(notes)
if meta:
parts.append("---\n" + "\n".join(meta))
return "\n\n".join(parts) if parts else None
def cmd_import_xlsx(client, args):
import xlsx_read # local import so the CLI still loads without the file present
headers, records = xlsx_read.read_table(args.path, args.sheet)
cols = _map_columns(headers)
if "subject" not in cols:
raise OpenProjectError(
f"Could not find a task/title column in sheet '{args.sheet}'. "
f"Headers seen: {[h for h in headers if h]}")
# Keep only rows with a real subject.
rows = [r for r in records if (r.get(cols["subject"]) or "").strip()]
if args.limit:
rows = rows[: args.limit]
# Category normalization (merge near-duplicates), e.g.
# --merge-category "Display & Technology=Display & Tech"
merge_map = {}
for pair in getattr(args, "merge_category", None) or []:
if "=" in pair:
old, new = pair.split("=", 1)
merge_map[old.strip().lower()] = new.strip()
cat_col = cols.get("category")
def norm_cat(r):
c = (r.get(cat_col) or "Uncategorized").strip() if cat_col else "Tasks"
return merge_map.get(c.lower(), c)
# Preserve first-seen category order.
categories = []
for r in rows:
c = norm_cat(r)
if c not in categories:
categories.append(c)
# Resolve owners once (match existing users, else note in description).
owner_col = cols.get("owner")
owner_cache = {}
if owner_col:
for r in rows:
o = (r.get(owner_col) or "").strip()
if not o or o.upper() == "TBD" or o in owner_cache:
continue
try:
owner_cache[o] = client.find_user(o)
except OpenProjectError:
owner_cache[o] = None
def map_priority(r):
raw = (r.get(cols["priority"]) or "").strip().upper() if "priority" in cols else ""
return PRIORITY_MAP.get(raw, DEFAULT_PRIORITY)
def map_status(r):
raw = (r.get(cols["status"]) or "").strip().upper() if "status" in cols else ""
return STATUS_MAP.get(raw, DEFAULT_STATUS)
# ----- Dry run: report the plan, write nothing ----- #
if not args.execute:
print(f"DRY RUN — would import {len(rows)} task(s) from sheet "
f"'{args.sheet}' into a new project '{args.project_name}'.\n")
print(f"Detected columns: {cols}\n")
print(f"Hierarchy: {len(categories)} parent Summary task(s), each with its rows as children:")
for c in categories:
n = sum(1 for r in rows if norm_cat(r) == c)
print(f" • {c} ({n})")
if owner_col:
matched = {o: u for o, u in owner_cache.items() if u}
unmatched = sorted(o for o, u in owner_cache.items() if not u)
print(f"\nOwners: {len(matched)} matched to users "
f"{[u.get('name') for u in matched.values()]}; "
f"{len(unmatched)} recorded in description: {unmatched}")
print("\nSample of first 5 mapped tasks:")
for r in rows[:5]:
print(f" [{map_priority(r)}/{map_status(r)}] "
f"{(r.get(cols['subject']) or '').strip()[:60]}")
print("\nNothing was written. Re-run with --execute to create everything.")
return
# ----- Execute: create project -> parents -> children ----- #
print(f"Creating project '{args.project_name}'...")
project = client.create_project(args.project_name, identifier=args.identifier,
description=args.project_description)
pid = project["id"]
proj_href = f"/api/v3/projects/{pid}"
summary_type = client.find_type("Summary task")
task_type = client.find_type(args.task_type)
print(f" project id {pid} — {client.base_url}/projects/{project['identifier']}")
# Parent per category.
parent_id = {}
for c in categories:
body = {"subject": c, "_links": {
"project": {"href": proj_href},
"type": {"href": f"/api/v3/types/{summary_type['id']}"}}}
wp = client.create_work_package(body)
parent_id[c] = wp["id"]
print(f" parent: {c} -> #{wp['id']}")
created, errors = 0, []
for r in rows:
subject = (r.get(cols["subject"]) or "").strip()
cat = norm_cat(r)
owner_raw = (r.get(owner_col) or "").strip() if owner_col else ""
owner_user = owner_cache.get(owner_raw)
owner_noted = owner_raw if (owner_raw and owner_raw.upper() != "TBD" and not owner_user) else None
links = {
"project": {"href": proj_href},
"type": {"href": f"/api/v3/types/{task_type['id']}"},
"parent": {"href": f"/api/v3/work_packages/{parent_id[cat]}"},
"priority": {"href": f"/api/v3/priorities/{client.find_priority(map_priority(r))['id']}"},
"status": {"href": f"/api/v3/statuses/{client.find_status(map_status(r))['id']}"},
}
if owner_user:
links["assignee"] = {"href": f"/api/v3/users/{owner_user['id']}"}
body = {"subject": subject[:255], "_links": links}
desc = _build_description(r, cols, owner_noted)
if desc:
body["description"] = {"format": "markdown", "raw": desc}
try:
wp = client.create_work_package(body)
created += 1
if created % 20 == 0:
print(f" ... {created}/{len(rows)} tasks created")
except OpenProjectError as exc:
errors.append((subject[:50], str(exc)))
print(f"\nDone. Project '{project['name']}': "
f"{len(categories)} parents + {created} tasks created.")
print(f" {client.base_url}/projects/{project['identifier']}/work_packages")
if errors:
print(f"\n{len(errors)} row(s) failed:")
for subj, err in errors[:15]:
print(f" ✗ {subj}: {err}")
def cmd_raw(client, args):
body = json.loads(args.data) if args.data else None
method = args.method.upper()
if method == "GET":
res = client.get(args.path)
elif method == "POST":
res = client.post(args.path, body=body)
elif method == "PATCH":
res = client.patch(args.path, body=body)
elif method == "DELETE":
res = client.delete(args.path)
else:
raise OpenProjectError(f"Unsupported method {method}")
emit_json(res)
# --------------------------------------------------------------------------- #
# Argument parser
# --------------------------------------------------------------------------- #
def build_parser():
p = argparse.ArgumentParser(prog="op", description="OpenProject CLI")
p.add_argument("--json", action="store_true", help="raw JSON output")
sub = p.add_subparsers(dest="command", required=True)
sub.add_parser("ping", help="verify connectivity + auth")
sub.add_parser("me", help="show the authenticated user")
sp = sub.add_parser("projects", help="list/search projects")
sp.add_argument("--search", help="name/identifier fragment")
sp.add_argument("--limit", type=int, default=100)
us = sub.add_parser("users", help="list/search users")
us.add_argument("--search")
us.add_argument("--limit", type=int, default=100)
ty = sub.add_parser("types", help="list work package types")
ty.add_argument("--project", help="restrict to types enabled in a project")
sub.add_parser("statuses", help="list statuses")
sub.add_parser("priorities", help="list priorities")
sub.add_parser("roles", help="list membership roles")
pj = sub.add_parser("project", help="project admin")
pjsub = pj.add_subparsers(dest="project_command", required=True)
pc = pjsub.add_parser("create", help="create a project")
pc.add_argument("--name", required=True)
pc.add_argument("--identifier", help="url slug (default: derived from name)")
pc.add_argument("--description")
pc.add_argument("--parent", help="parent project ref (for a subproject)")
mb = sub.add_parser("members", help="project memberships")
mbsub = mb.add_subparsers(dest="members_command", required=True)
ma = mbsub.add_parser("add", help="add a user to a project with a role")
ma.add_argument("--project", required=True)
ma.add_argument("--user", required=True)
ma.add_argument("--role", required=True, help="role name or id (e.g. Member)")
wp = sub.add_parser("wp", help="work packages")
wpsub = wp.add_subparsers(dest="wp_command", required=True)
wl = wpsub.add_parser("list", help="list/search work packages")
wl.add_argument("--project")
wl.add_argument("--assignee", help="user ref or 'me'")
wl.add_argument("--status", help="'open', 'closed', or a status name/id")
wl.add_argument("--type")
wl.add_argument("--query", help="free-text search")
wl.add_argument("--sort", help="field or field:asc|desc")
wl.add_argument("--limit", type=int, default=50)
ws = wpsub.add_parser("show", help="show one work package")
ws.add_argument("id")
wc = wpsub.add_parser("create", help="create a work package")
wc.add_argument("--project", required=True)
wc.add_argument("--subject", required=True)
wc.add_argument("--type")
wc.add_argument("--description")
wc.add_argument("--assignee")
wc.add_argument("--status")
wc.add_argument("--priority")
wc.add_argument("--parent")
wc.add_argument("--start")
wc.add_argument("--due")
wc.add_argument("--done-ratio", dest="done_ratio")
wu = wpsub.add_parser("update", help="update a work package (auto lockVersion)")
wu.add_argument("id")
wu.add_argument("--subject")
wu.add_argument("--description")
wu.add_argument("--assignee", help="user ref, 'me', or 'none' to unassign")
wu.add_argument("--status")
wu.add_argument("--priority")
wu.add_argument("--type")
wu.add_argument("--parent")
wu.add_argument("--start")
wu.add_argument("--due")
wu.add_argument("--done-ratio", dest="done_ratio")
wcm = wpsub.add_parser("comment", help="add a comment")
wcm.add_argument("id")
wcm.add_argument("--text", required=True)
wcl = wpsub.add_parser("close", help="set status (default 'Closed')")
wcl.add_argument("id")
wcl.add_argument("--status-name", dest="status_name", default="Closed")
wd = wpsub.add_parser("delete", help="delete a work package (needs --yes)")
wd.add_argument("id")
wd.add_argument("--yes", action="store_true")
wr = wpsub.add_parser("relate", help="link two work packages")
wr.add_argument("id")
wr.add_argument("--to", required=True, help="target work package id")
wr.add_argument("--type", default="relates",
help="relates|blocks|blocked|precedes|follows|requires|...")
wr.add_argument("--description")
tm = sub.add_parser("time", help="time entries")
tmsub = tm.add_subparsers(dest="time_command", required=True)
tl = tmsub.add_parser("log", help="log time on a work package")
tl.add_argument("--wp", required=True)
tl.add_argument("--hours", required=True)
tl.add_argument("--comment")
tl.add_argument("--date", help="YYYY-MM-DD (default today)")
tl.add_argument("--activity")
im = sub.add_parser("import", help="import a spreadsheet into a project")
imsub = im.add_subparsers(dest="import_command", required=True)
ix = imsub.add_parser("xlsx", help="import an .xlsx sheet as a project of tasks")
ix.add_argument("path", help="path to the .xlsx file")
ix.add_argument("--sheet", required=True, help="worksheet name to import")
ix.add_argument("--project-name", dest="project_name", required=True)
ix.add_argument("--identifier", help="project url slug")
ix.add_argument("--project-description", dest="project_description")
ix.add_argument("--task-type", dest="task_type", default="Task")
ix.add_argument("--merge-category", dest="merge_category", action="append",
default=[], help="merge a category: 'OLD=NEW' (repeatable)")
ix.add_argument("--limit", type=int, help="only import the first N rows")
ix.add_argument("--execute", action="store_true",
help="actually write (default is a dry run)")
rw = sub.add_parser("raw", help="raw API call (escape hatch)")
rw.add_argument("method")
rw.add_argument("path")
rw.add_argument("--data", help="JSON request body")
return p
DISPATCH = {
("ping",): cmd_ping,
("me",): cmd_me,
("projects",): cmd_projects,
("users",): cmd_users,
("types",): cmd_types,
("statuses",): cmd_statuses,
("priorities",): cmd_priorities,
("roles",): cmd_roles,
("project", "create"): cmd_project_create,
("members", "add"): cmd_members_add,
("wp", "list"): cmd_wp_list,
("wp", "show"): cmd_wp_show,
("wp", "create"): cmd_wp_create,
("wp", "update"): cmd_wp_update,
("wp", "comment"): cmd_wp_comment,
("wp", "close"): cmd_wp_close,
("wp", "delete"): cmd_wp_delete,
("wp", "relate"): cmd_wp_relate,
("time", "log"): cmd_time_log,
("import", "xlsx"): cmd_import_xlsx,
("raw",): cmd_raw,
}
def main(argv=None):
args = build_parser().parse_args(argv)
key = (args.command,)
if args.command == "wp":
key = ("wp", args.wp_command)
elif args.command == "time":
key = ("time", args.time_command)
elif args.command == "project":
key = ("project", args.project_command)
elif args.command == "members":
key = ("members", args.members_command)
elif args.command == "import":
key = ("import", args.import_command)
handler = DISPATCH.get(key)
if not handler:
print(f"Unknown command: {key}", file=sys.stderr)
return 2
try:
client = OpenProjectClient.from_config()
handler(client, args)
return 0
except OpenProjectError as exc:
print(f"error: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
skills/openproject/scripts/opclient.py
+355
View File
@@ -0,0 +1,355 @@
#!/usr/bin/env python3
"""
OpenProject API v3 client — standard library only (no pip installs).
Talks to an OpenProject instance over its HAL+JSON REST API.
Auth: API token via HTTP Basic auth (username "apikey", password = token).
Config resolution order (first hit wins for each key):
1. Environment variables OPENPROJECT_URL / OPENPROJECT_API_KEY
2. config.json next to the plugin root (../config.json relative to this file)
3. ~/.config/openproject/config.json
This module is imported by op.py (the CLI) and can also be used directly:
from opclient import OpenProjectClient
op = OpenProjectClient.from_config()
me = op.get("/api/v3/users/me")
"""
from __future__ import annotations
import base64
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
__all__ = ["OpenProjectClient", "OpenProjectError", "load_config"]
DEFAULT_TIMEOUT = 30
class OpenProjectError(Exception):
"""Raised for API and configuration errors. Carries status + parsed body."""
def __init__(self, message, status=None, body=None):
super().__init__(message)
self.status = status
self.body = body
# --------------------------------------------------------------------------- #
# Configuration
# --------------------------------------------------------------------------- #
def _config_search_paths():
"""Look for config.json at the plugin root and common locations.
Scripts live at <root>/skills/openproject/scripts/, so we walk up a few
parent directories to find the plugin root regardless of install depth.
"""
paths = []
if os.environ.get("CLAUDE_PLUGIN_ROOT"):
paths.append(os.path.join(os.environ["CLAUDE_PLUGIN_ROOT"], "config.json"))
cur = os.path.dirname(os.path.abspath(__file__))
for _ in range(5): # scripts -> openproject -> skills -> <root> -> ...
paths.append(os.path.join(cur, "config.json"))
cur = os.path.dirname(cur)
paths.append(os.path.expanduser("~/.config/openproject/config.json"))
seen, ordered = set(), []
for p in paths:
if p not in seen:
seen.add(p)
ordered.append(p)
return ordered
def load_config():
"""Merge config sources. Env vars override file values."""
cfg = {"url": None, "api_key": None, "default_project": None}
for path in _config_search_paths():
if os.path.isfile(path):
try:
with open(path, "r", encoding="utf-8") as fh:
data = json.load(fh)
for key in cfg:
if data.get(key) not in (None, "", "REPLACE_WITH_YOUR_API_TOKEN"):
cfg[key] = data[key]
except (OSError, ValueError) as exc:
print(f"warning: could not read {path}: {exc}", file=sys.stderr)
break # first existing config file wins
# Environment overrides
if os.environ.get("OPENPROJECT_URL"):
cfg["url"] = os.environ["OPENPROJECT_URL"]
if os.environ.get("OPENPROJECT_API_KEY"):
cfg["api_key"] = os.environ["OPENPROJECT_API_KEY"]
if os.environ.get("OPENPROJECT_DEFAULT_PROJECT"):
cfg["default_project"] = os.environ["OPENPROJECT_DEFAULT_PROJECT"]
return cfg
# --------------------------------------------------------------------------- #
# Client
# --------------------------------------------------------------------------- #
class OpenProjectClient:
def __init__(self, url, api_key, default_project=None, timeout=DEFAULT_TIMEOUT):
if not url:
raise OpenProjectError(
"No OpenProject URL configured. Set OPENPROJECT_URL or config.json 'url'."
)
if not api_key:
raise OpenProjectError(
"No API key configured. Set OPENPROJECT_API_KEY or config.json 'api_key'."
)
self.base_url = url.rstrip("/")
self.api_key = api_key
self.default_project = default_project
self.timeout = timeout
token = base64.b64encode(f"apikey:{api_key}".encode()).decode()
self._auth_header = f"Basic {token}"
@classmethod
def from_config(cls, timeout=DEFAULT_TIMEOUT):
cfg = load_config()
return cls(cfg["url"], cfg["api_key"], cfg.get("default_project"), timeout)
# -- low-level HTTP ----------------------------------------------------- #
def _request(self, method, path, params=None, body=None):
# Allow callers to pass either "/api/v3/..." or a full URL.
if path.startswith("http://") or path.startswith("https://"):
url = path
else:
url = self.base_url + path
if params:
# Drop None values; JSON-encode any non-string (filters/sortBy).
clean = {}
for k, v in params.items():
if v is None:
continue
clean[k] = v if isinstance(v, str) else json.dumps(v)
url += "?" + urllib.parse.urlencode(clean)
data = None
headers = {
"Authorization": self._auth_header,
"Accept": "application/json",
}
if body is not None:
data = json.dumps(body).encode("utf-8")
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
raw = resp.read()
if not raw:
return {}
return json.loads(raw.decode("utf-8"))
except urllib.error.HTTPError as exc:
raw = exc.read().decode("utf-8", "replace")
parsed = None
try:
parsed = json.loads(raw)
except ValueError:
pass
msg = self._format_api_error(exc.code, parsed, raw)
raise OpenProjectError(msg, status=exc.code, body=parsed or raw)
except urllib.error.URLError as exc:
raise OpenProjectError(
f"Could not reach {self.base_url} ({exc.reason}). "
f"Check the URL and network."
)
@staticmethod
def _format_api_error(code, parsed, raw):
if isinstance(parsed, dict):
message = parsed.get("message", "")
details = []
embedded = parsed.get("_embedded", {})
for err in embedded.get("errors", []) or []:
if isinstance(err, dict) and err.get("message"):
details.append(err["message"])
if details:
message = message + " | " + "; ".join(details)
if message:
return f"HTTP {code}: {message}"
return f"HTTP {code}: {raw[:400]}"
def get(self, path, params=None):
return self._request("GET", path, params=params)
def post(self, path, body=None, params=None):
return self._request("POST", path, params=params, body=body or {})
def patch(self, path, body=None, params=None):
return self._request("PATCH", path, params=params, body=body or {})
def delete(self, path, params=None):
return self._request("DELETE", path, params=params)
# -- pagination --------------------------------------------------------- #
def get_all(self, path, params=None, max_items=1000):
"""Follow nextByOffset links, accumulating collection elements."""
params = dict(params or {})
params.setdefault("pageSize", 100)
params.setdefault("offset", 1)
items = []
while True:
page = self.get(path, params=params)
embedded = page.get("_embedded", {})
elements = embedded.get("elements", [])
items.extend(elements)
if len(items) >= max_items:
return items[:max_items]
total = page.get("total", 0)
if len(items) >= total or not elements:
return items
params["offset"] = int(params["offset"]) + 1
# -- helpers / resolvers ------------------------------------------------ #
@staticmethod
def id_from_href(href):
if not href:
return None
return href.rstrip("/").split("/")[-1]
@staticmethod
def link_title(resource, link_name):
link = (resource.get("_links") or {}).get(link_name) or {}
return link.get("title")
def me(self):
return self.get("/api/v3/users/me")
def find_project(self, ref):
"""Resolve a project by numeric id, identifier, or (partial) name."""
if ref is None:
return None
ref = str(ref)
if ref.isdigit():
return self.get(f"/api/v3/projects/{ref}")
# Try identifier lookup directly.
try:
return self.get(f"/api/v3/projects/{urllib.parse.quote(ref)}")
except OpenProjectError:
pass
# Fall back to name search.
flt = [{"name_and_identifier": {"operator": "~", "values": [ref]}}]
results = self.get_all("/api/v3/projects", params={"filters": flt}, max_items=50)
if not results:
raise OpenProjectError(f"No project matching '{ref}'.")
return results[0]
def find_user(self, ref):
"""Resolve a user by 'me', numeric id, login, or name fragment."""
if ref is None:
return None
ref = str(ref)
if ref.lower() == "me":
return self.me()
if ref.isdigit():
return self.get(f"/api/v3/users/{ref}")
flt = [{"name": {"operator": "~", "values": [ref]}}]
results = self.get_all("/api/v3/users", params={"filters": flt}, max_items=25)
if not results:
raise OpenProjectError(f"No user matching '{ref}'.")
return results[0]
def _resolve_by_name(self, collection_path, ref, fields=("name",)):
"""Generic resolver for small collections (types, statuses, priorities)."""
ref = str(ref)
if ref.isdigit():
return self.get(f"{collection_path}/{ref}")
for item in self.get_all(collection_path, max_items=200):
for field in fields:
val = item.get(field)
if val and val.lower() == ref.lower():
return item
# second pass: substring match
for item in self.get_all(collection_path, max_items=200):
for field in fields:
val = item.get(field)
if val and ref.lower() in val.lower():
return item
raise OpenProjectError(f"No item in {collection_path} matching '{ref}'.")
def find_type(self, ref):
return self._resolve_by_name("/api/v3/types", ref)
def find_status(self, ref):
return self._resolve_by_name("/api/v3/statuses", ref)
def find_priority(self, ref):
return self._resolve_by_name("/api/v3/priorities", ref)
def find_role(self, ref):
return self._resolve_by_name("/api/v3/roles", ref)
# -- create / relate helpers ------------------------------------------- #
@staticmethod
def slugify(name):
"""Make a valid OpenProject project identifier from a name."""
slug = re.sub(r"[^a-z0-9]+", "-", str(name).lower()).strip("-")
return slug[:100] or "project"
def create_project(self, name, identifier=None, description=None, parent=None):
body = {"name": name, "identifier": identifier or self.slugify(name)}
if description:
body["description"] = {"format": "markdown", "raw": description}
if parent:
pid = self.find_project(parent)["id"]
body["_links"] = {"parent": {"href": f"/api/v3/projects/{pid}"}}
return self.post("/api/v3/projects", body=body)
def add_membership(self, project_id, user_id, role_ref):
role = self.find_role(role_ref)
body = {
"_links": {
"project": {"href": f"/api/v3/projects/{project_id}"},
"principal": {"href": f"/api/v3/users/{user_id}"},
"roles": [{"href": f"/api/v3/roles/{role['id']}"}],
}
}
return self.post("/api/v3/memberships", body=body)
# OpenProject relation types: relates, duplicates, blocks, blocked, precedes,
# follows, includes, partof, requires, required.
def create_relation(self, from_id, to_id, rel_type="relates", description=None):
body = {
"type": rel_type,
"_links": {"to": {"href": f"/api/v3/work_packages/{to_id}"}},
}
if description:
body["description"] = description
return self.post(f"/api/v3/work_packages/{from_id}/relations", body=body)
def create_work_package(self, body, notify=False):
"""Create a WP; falls back to create-without-status + PATCH when the
workflow rejects setting a non-default status on creation."""
params = {"notify": "false"} if not notify else None
try:
return self.post("/api/v3/work_packages", body=body, params=params)
except OpenProjectError as exc:
links = body.get("_links", {})
if exc.status in (422,) and "status" in links:
status_link = links.pop("status")
created = self.post("/api/v3/work_packages", body=body, params=params)
patch = {"lockVersion": created.get("lockVersion"),
"_links": {"status": status_link}}
try:
return self.patch(
f"/api/v3/work_packages/{created['id']}", body=patch)
except OpenProjectError:
return created # created at default status; status change rejected
raise
skills/openproject/scripts/xlsx_read.py
+135
View File
@@ -0,0 +1,135 @@
#!/usr/bin/env python3
"""
Minimal .xlsx reader — standard library only (no openpyxl/pandas).
An .xlsx file is a zip of XML parts. This reads worksheet cell values, resolving
shared strings and inline strings, and returns plain Python rows. Enough for
tabular imports; it does NOT evaluate formulas (it returns the cached value if
present) and ignores styling/number formats.
API:
from xlsx_read import read_sheet, sheet_names
names = sheet_names("file.xlsx")
rows = read_sheet("file.xlsx", "To-Do List") # -> list[list[str|None]]
table = read_table("file.xlsx", "To-Do List") # -> (headers, list[dict])
"""
from __future__ import annotations
import re
import zipfile
from xml.etree import ElementTree as ET
_NS = "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}"
_RID = "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"
_RNS = "{http://schemas.openxmlformats.org/package/2006/relationships}"
def _col_to_num(ref):
"""'B7' -> 2 (1-based column index)."""
letters = re.match(r"[A-Z]+", ref).group()
n = 0
for ch in letters:
n = n * 26 + (ord(ch) - 64)
return n
def _shared_strings(zf):
if "xl/sharedStrings.xml" not in zf.namelist():
return []
root = ET.fromstring(zf.read("xl/sharedStrings.xml"))
out = []
for si in root.findall(f"{_NS}si"):
out.append("".join(t.text or "" for t in si.iter(f"{_NS}t")))
return out
def _sheet_map(zf):
"""Return ordered list of (sheet_name, worksheet_part_path)."""
wb = ET.fromstring(zf.read("xl/workbook.xml"))
rels = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels"))
rid2target = {r.get("Id"): r.get("Target")
for r in rels.iter(f"{_RNS}Relationship")}
sheets = []
for s in wb.iter(f"{_NS}sheet"):
target = rid2target.get(s.get(_RID))
if not target:
continue
if not target.startswith("xl/"):
target = "xl/" + target.lstrip("/")
sheets.append((s.get("name"), target))
return sheets
def sheet_names(path):
with zipfile.ZipFile(path) as zf:
return [name for name, _ in _sheet_map(zf)]
def read_sheet(path, sheet_name):
"""Return rows as lists of cell values (str or None), padded to max width."""
with zipfile.ZipFile(path) as zf:
shared = _shared_strings(zf)
target = None
for name, part in _sheet_map(zf):
if name == sheet_name:
target = part
break
if target is None:
raise KeyError(f"Sheet '{sheet_name}' not found. "
f"Available: {[n for n, _ in _sheet_map(zf)]}")
root = ET.fromstring(zf.read(target))
rows = []
max_col = 0
for row in root.iter(f"{_NS}row"):
cells = {}
for c in row.findall(f"{_NS}c"):
ref = c.get("r")
if not ref:
continue
ctype = c.get("t")
v = c.find(f"{_NS}v")
inline = c.find(f"{_NS}is")
val = None
if ctype == "s" and v is not None:
val = shared[int(v.text)]
elif ctype == "inlineStr" and inline is not None:
val = "".join(t.text or "" for t in inline.iter(f"{_NS}t"))
elif v is not None:
val = v.text
idx = _col_to_num(ref)
cells[idx] = val
max_col = max(max_col, idx)
rows.append(cells)
return [[r.get(i) for i in range(1, max_col + 1)] for r in rows]
def read_table(path, sheet_name, header_row=0):
"""Return (headers, list[dict]) using the given 0-based header row index.
Rows above the header are skipped; the first row at/after header_row whose
first cell is non-empty is treated as the header.
"""
rows = read_sheet(path, sheet_name)
if not rows:
return [], []
headers = [(h or "").strip() for h in rows[header_row]]
records = []
for r in rows[header_row + 1:]:
rec = {}
for i, h in enumerate(headers):
if not h:
continue
rec[h] = r[i] if i < len(r) else None
records.append(rec)
return headers, records
if __name__ == "__main__":
import sys
p = sys.argv[1]
print("Sheets:", sheet_names(p))
if len(sys.argv) > 2:
hdrs, recs = read_table(p, sys.argv[2])
print("Headers:", hdrs)
print(f"{len(recs)} data rows")