feat: attach photos from Search preview replies (Closes #955)

src/gitea_proxy.py

@@ -1364,6 +1364,32 @@ async def upload_assigned_issue_attachment(
     return {"name": name, "url": url, "size": size}
 
 
+async def upload_preview_attachment(
+    repository: str,
+    number: int,
+    filename: str,
+    content_type: str,
+    content: bytes,
+) -> dict:
+    """Upload evidence after the caller verifies an exact Search preview target."""
+    response = await _get_client().post(
+        f"/api/v1/repos/{repository}/issues/{number}/assets",
+        headers=_auth(),
+        params={"name": filename},
+        files={"attachment": (filename, content, content_type)},
+    )
+    response.raise_for_status()
+    attachment = response.json()
+    if not isinstance(attachment, dict):
+        raise ValueError("Gitea attachment response was not an object")
+    name = attachment.get("name")
+    url = _safe_web_url(attachment.get("browser_download_url"))
+    size = attachment.get("size")
+    if not isinstance(name, str) or not name or not url or not isinstance(size, int):
+        raise ValueError("Gitea did not confirm the attachment")
+    return {"name": name, "url": url, "size": size}
+
+
 async def upload_assigned_pull_attachment(
     repository: str,
     number: int,

src/main.py

@@ -3136,6 +3136,71 @@ async def comment_on_global_search_preview(
     return JSONResponse(result, status_code=201)
 
 
+@app.post(
+    "/api/v1/repos/{owner}/{repo}/issues/{number}/preview/attachments",
+    status_code=201,
+)
+async def attach_to_global_search_preview(
+    request: Request,
+    owner: str,
+    repo: str,
+    number: int = PathParam(gt=0),
+    kind: Literal["issue", "pull"] = Query(),
+    idempotency_key: str | None = Header(default=None, max_length=128),
+) -> JSONResponse:
+    repository = f"{owner}/{repo}"
+    try:
+        form = await request.form()
+        uploaded = form.get("file")
+        if not isinstance(uploaded, UploadFile):
+            raise ValueError("screenshot file is required")
+        filename = str(uploaded.filename or "")
+        content_type = str(uploaded.content_type or "")
+        content = _validate_binary_attachment(filename, content_type, await uploaded.read())
+    except (ValueError, ValidationError) as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+    async def upload_attachment():
+        preview = await gitea_proxy.work_preview(repository, kind, number)
+        if (
+            preview.get("repository") != repository
+            or preview.get("kind") != kind
+            or preview.get("number") != number
+            or preview.get("commentable") is not True
+        ):
+            raise HTTPException(status_code=404, detail="Search result not found")
+        result = await gitea_proxy.upload_preview_attachment(
+            repository, number, filename, content_type, content
+        )
+        safe_name = (
+            result["name"].replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]")
+            .replace("\r", " ").replace("\n", " ")
+        )
+        safe_url = result["url"].replace("<", "%3C").replace(">", "%3E")
+        result["markdown"] = f"![{safe_name}](<{safe_url}>)"
+        return result
+
+    try:
+        result = await _run_idempotent_authored_action(
+            upload_attachment(),
+            idempotency_key=idempotency_key,
+            fingerprint=(
+                "search-preview-attachment", repository, kind, number, filename,
+                content_type, hashlib.sha256(content).hexdigest(),
+            ),
+            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
+        )
+    except HTTPException:
+        raise
+    except Exception:
+        return JSONResponse(
+            {"error": "The photo could not be uploaded. Your reply is safe; please retry."},
+            status_code=503,
+            headers={"Retry-After": "1"},
+        )
+    return JSONResponse(result, status_code=201)
+
+
 @app.get("/api/v1/work-route")
 async def resolve_work_route(
     kind: Literal["issue", "filed", "pull", "review", "update"] = Query(),