Unclear reason for the backup

2026-10-18 10:45:14 -04:00
parent 4e10a21142
commit 7ad3723525
6 changed files with 5499 additions and 0 deletions
@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""Attachment helpers for the power-user Redmine ticket client.
+
+List (cache-first), download to local dir, upload via Redmine REST:
+  POST /uploads.json?filename=…  (raw body)
+  PUT  /issues/:id.json with issue.uploads token
+"""
+
+from __future__ import annotations
+
+import json
+import mimetypes
+import re
+from pathlib import Path
+from typing import Any
+from urllib import error as urllib_error
+from urllib import parse as urllib_parse
+from urllib import request as urllib_request
+
+
+def safe_filename(name: str) -> str:
+    name = (name or "attachment.bin").strip().replace("\\", "_").replace("/", "_")
+    name = re.sub(r"[^\w.\- ()\[\]+@]+", "_", name)
+    return name or "attachment.bin"
+
+
+def attachments_from_issue(issue: dict[str, Any]) -> list[dict[str, Any]]:
+    atts = issue.get("attachments")
+    if isinstance(atts, list):
+        return [a for a in atts if isinstance(a, dict)]
+    return []
+
+
+def attachment_dir(cache_dir: Path, project: str, issue_id: int) -> Path:
+    path = Path(cache_dir) / project / "attachments" / str(int(issue_id))
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def pick_attachment(attachments: list[dict[str, Any]], attachment_id: int | None) -> dict[str, Any]:
+    if not attachments:
+        raise LookupError("Issue has no attachments.")
+    if attachment_id is None:
+        if len(attachments) == 1:
+            return attachments[0]
+        raise LookupError(
+            "Multiple attachments; pass --attachment-id. ids="
+            + ",".join(str(a.get("id")) for a in attachments)
+        )
+    for att in attachments:
+        if int(att.get("id") or 0) == int(attachment_id):
+            return att
+    raise LookupError(f"Attachment id {attachment_id} not found on issue.")
+
+
+def download_url_for_attachment(att: dict[str, Any], base_url: str) -> str:
+    urls = download_url_candidates(att, base_url)
+    return urls[0]
+
+
+def download_url_candidates(att: dict[str, Any], base_url: str) -> list[str]:
+    """Ordered download URL candidates (host rewrite + path variants)."""
+    base = base_url.rstrip("/")
+    aid = att.get("id")
+    filename = str(att.get("filename") or "").strip()
+    candidates: list[str] = []
+
+    def add(url: str) -> None:
+        if url and url not in candidates:
+            candidates.append(url)
+
+    for key in ("content_url", "contentUrl"):
+        raw = att.get(key)
+        if not raw:
+            continue
+        raw = str(raw)
+        add(raw)
+        # rewrite host to active base_url (prod content_url on LAN client, etc.)
+        try:
+            from urllib.parse import urlparse, urlunparse
+
+            parsed = urlparse(raw)
+            base_p = urlparse(base if "://" in base else f"http://{base}")
+            rewritten = urlunparse(
+                (
+                    base_p.scheme or parsed.scheme,
+                    base_p.netloc or parsed.netloc,
+                    parsed.path,
+                    "",
+                    parsed.query,
+                    "",
+                )
+            )
+            add(rewritten)
+        except Exception:
+            pass
+
+    if aid is not None:
+        add(f"{base}/attachments/download/{int(aid)}")
+        if filename:
+            from urllib.parse import quote
+
+            add(f"{base}/attachments/download/{int(aid)}/{quote(filename)}")
+    if not candidates:
+        raise LookupError("Attachment missing id and content_url.")
+    return candidates
+
+
+def download_attachment_bytes(
+    *,
+    url: str,
+    api_key: str,
+    impersonate_user: str | None = None,
+    timeout: int = 120,
+    fallback_urls: list[str] | None = None,
+) -> bytes:
+    headers = {
+        "X-Redmine-API-Key": api_key,
+        "Accept": "*/*",
+    }
+    if impersonate_user:
+        headers["X-Redmine-Switch-User"] = impersonate_user
+    urls = [url] + [u for u in (fallback_urls or []) if u != url]
+    last_err: Exception | None = None
+    for u in urls:
+        req = urllib_request.Request(u, headers=headers, method="GET")
+        try:
+            with urllib_request.urlopen(req, timeout=timeout) as resp:
+                return resp.read()
+        except urllib_error.HTTPError as exc:
+            detail = exc.read().decode("utf-8", errors="replace")[:200]
+            last_err = RuntimeError(f"HTTP {exc.code} downloading attachment ({u}): {detail}")
+            continue
+        except urllib_error.URLError as exc:
+            last_err = RuntimeError(f"Could not download attachment ({u}): {exc.reason}")
+            continue
+    raise last_err or RuntimeError("Attachment download failed")
+
+
+def upload_file_token(
+    *,
+    base_url: str,
+    api_key: str,
+    path: Path,
+    filename: str | None = None,
+    impersonate_user: str | None = None,
+    timeout: int = 120,
+) -> dict[str, Any]:
+    """POST raw file to /uploads.json?filename=… Returns upload object with token."""
+    path = Path(path)
+    if not path.is_file():
+        raise FileNotFoundError(f"File not found: {path}")
+    name = filename or path.name
+    data = path.read_bytes()
+    q = urllib_parse.urlencode({"filename": name})
+    url = f"{base_url.rstrip('/')}/uploads.json?{q}"
+    headers = {
+        "X-Redmine-API-Key": api_key,
+        "Content-Type": "application/octet-stream",
+        "Accept": "application/json",
+    }
+    if impersonate_user:
+        headers["X-Redmine-Switch-User"] = impersonate_user
+    req = urllib_request.Request(url, data=data, headers=headers, method="POST")
+    try:
+        with urllib_request.urlopen(req, timeout=timeout) as resp:
+            raw = resp.read()
+    except urllib_error.HTTPError as exc:
+        detail = exc.read().decode("utf-8", errors="replace")[:500]
+        raise RuntimeError(f"HTTP {exc.code} uploading file: {detail}") from exc
+    except urllib_error.URLError as exc:
+        raise RuntimeError(f"Could not upload file: {exc.reason}") from exc
+    if not raw:
+        raise RuntimeError("Empty response from /uploads.json")
+    payload = json.loads(raw.decode("utf-8"))
+    upload = payload.get("upload") if isinstance(payload, dict) else None
+    if not isinstance(upload, dict) or not upload.get("token"):
+        raise RuntimeError(f"Unexpected upload response: {payload!r}")
+    content_type = mimetypes.guess_type(name)[0] or "application/octet-stream"
+    upload.setdefault("filename", name)
+    upload.setdefault("content_type", content_type)
+    return upload