77 lines
2.8 KiB
Python
77 lines
2.8 KiB
Python
"""Structured user-message attachments and model-context compatibility rendering."""
|
|
from __future__ import annotations
|
|
|
|
import mimetypes
|
|
from collections.abc import Iterable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .artifacts import ArtifactPathError, resolve_artifact_path
|
|
|
|
ATTACHMENT_REF_VERSION = 1
|
|
MAX_ATTACHMENTS_PER_MESSAGE = 10
|
|
_IMAGE_EXTENSIONS = {
|
|
".avif", ".bmp", ".gif", ".heic", ".heif", ".jpeg", ".jpg",
|
|
".png", ".svg", ".tif", ".tiff", ".webp",
|
|
}
|
|
|
|
|
|
def _attachment_kind(requested: str, path: Path, media_type: str) -> str:
|
|
detected_image = media_type.startswith("image/") or path.suffix.lower() in _IMAGE_EXTENSIONS
|
|
if requested == "image" and detected_image:
|
|
return "image"
|
|
return "file"
|
|
|
|
|
|
def normalize_attachment_refs(
|
|
refs: Iterable[Any], *, working_dir: Path, user_root: Path,
|
|
) -> list[dict]:
|
|
"""Validate client refs and return task-relative, display-ready metadata."""
|
|
output: list[dict] = []
|
|
seen: set[str] = set()
|
|
for raw in refs:
|
|
if len(output) >= MAX_ATTACHMENTS_PER_MESSAGE:
|
|
raise ArtifactPathError(
|
|
f"at most {MAX_ATTACHMENTS_PER_MESSAGE} attachments are allowed"
|
|
)
|
|
data = raw.model_dump() if hasattr(raw, "model_dump") else dict(raw or {})
|
|
absolute, rel = resolve_artifact_path(
|
|
str(data.get("path") or ""),
|
|
working_dir=working_dir,
|
|
user_root=user_root,
|
|
require_file=True,
|
|
)
|
|
if rel in seen:
|
|
continue
|
|
seen.add(rel)
|
|
media_type = mimetypes.guess_type(absolute.name)[0] or "application/octet-stream"
|
|
label = str(data.get("label") or absolute.name).strip() or absolute.name
|
|
output.append({
|
|
"version": ATTACHMENT_REF_VERSION,
|
|
"scope": "working_dir",
|
|
"path": rel,
|
|
"label": label,
|
|
"kind": _attachment_kind(str(data.get("kind") or "file"), absolute, media_type),
|
|
"media_type": media_type,
|
|
"size_bytes": absolute.stat().st_size,
|
|
})
|
|
return output
|
|
|
|
|
|
def content_for_model(content: str, refs: Iterable[dict], working_dir_name: str) -> str:
|
|
"""Append the legacy-readable file hints only to the provider-bound message."""
|
|
lines: list[str] = []
|
|
wd = str(working_dir_name or "").strip().strip("/\\")
|
|
for ref in refs or ():
|
|
path = str((ref or {}).get("path") or "").strip().replace("\\", "/")
|
|
if not path:
|
|
continue
|
|
full_path = f"{wd}/{path}" if wd else path
|
|
marker = "[用户上传的参考图]" if (ref or {}).get("kind") == "image" else "[用户上传的文件]"
|
|
lines.append(f"{marker} {full_path}")
|
|
text = str(content or "").strip()
|
|
if not lines:
|
|
return text
|
|
suffix = "\n".join(lines)
|
|
return f"{text}\n\n{suffix}" if text else suffix
|