redmine/redmine_ticket_attachments.py

#!/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