128 lines
3.6 KiB
Python
128 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
from context.conversation_manager import get_current_context, get_current_sketch_pad
|
|
|
|
|
|
def _content_item_type(item: Any) -> Optional[str]:
|
|
if isinstance(item, dict):
|
|
value = item.get("type")
|
|
return value if isinstance(value, str) else None
|
|
value = getattr(item, "type", None)
|
|
return value if isinstance(value, str) else None
|
|
|
|
|
|
def _content_item_image_payload(item: Any) -> Any:
|
|
if isinstance(item, dict):
|
|
return item.get("image_url")
|
|
return getattr(item, "image_url", None)
|
|
|
|
|
|
def _image_payload_local_path(image_payload: Any) -> Optional[str]:
|
|
if isinstance(image_payload, dict):
|
|
value = image_payload.get("local_path")
|
|
return value if isinstance(value, str) else None
|
|
value = getattr(image_payload, "local_path", None)
|
|
return value if isinstance(value, str) else None
|
|
|
|
|
|
def _normalize_existing_file_path(path_value: Any) -> Optional[str]:
|
|
if not isinstance(path_value, (str, Path)):
|
|
return None
|
|
|
|
raw_value = str(path_value).strip()
|
|
if not raw_value:
|
|
return None
|
|
|
|
candidate = Path(raw_value).expanduser()
|
|
if not candidate.is_absolute():
|
|
candidate = (Path.cwd() / candidate).resolve()
|
|
|
|
try:
|
|
if candidate.exists() and candidate.is_file():
|
|
return str(candidate)
|
|
except OSError:
|
|
return None
|
|
|
|
return None
|
|
|
|
|
|
def _resolve_sketch_pad_image_path(reference: str) -> Optional[str]:
|
|
if not reference.startswith("key:"):
|
|
return None
|
|
|
|
sketch_pad = get_current_sketch_pad()
|
|
if sketch_pad is None:
|
|
return None
|
|
|
|
try:
|
|
value = sketch_pad.get_value(reference[4:])
|
|
except Exception:
|
|
return None
|
|
|
|
direct_path = _normalize_existing_file_path(value)
|
|
if direct_path is not None:
|
|
return direct_path
|
|
|
|
if not isinstance(value, dict):
|
|
return None
|
|
|
|
for key in ("local_path", "path", "image_path", "query_image_path"):
|
|
resolved_path = _normalize_existing_file_path(value.get(key))
|
|
if resolved_path is not None:
|
|
return resolved_path
|
|
|
|
return None
|
|
|
|
|
|
def get_latest_uploaded_reference_image_path() -> Optional[str]:
|
|
context = get_current_context()
|
|
if context is None:
|
|
return None
|
|
|
|
try:
|
|
messages = context.retrieve_full_messages()
|
|
except Exception:
|
|
try:
|
|
messages = context.retrieve_messages()
|
|
except Exception:
|
|
return None
|
|
|
|
for message in reversed(messages):
|
|
if getattr(message, "role", None) != "user":
|
|
continue
|
|
|
|
content = getattr(message, "content", None)
|
|
if not isinstance(content, list):
|
|
continue
|
|
|
|
for item in reversed(content):
|
|
if _content_item_type(item) != "image_url":
|
|
continue
|
|
|
|
image_payload = _content_item_image_payload(item)
|
|
resolved_path = _normalize_existing_file_path(
|
|
_image_payload_local_path(image_payload)
|
|
)
|
|
if resolved_path is not None:
|
|
return resolved_path
|
|
|
|
return None
|
|
|
|
|
|
def resolve_reference_image_path(query_image_path: Optional[str]) -> Optional[str]:
|
|
if isinstance(query_image_path, str) and query_image_path.strip():
|
|
stripped_path = query_image_path.strip()
|
|
|
|
sketch_pad_path = _resolve_sketch_pad_image_path(stripped_path)
|
|
if sketch_pad_path is not None:
|
|
return sketch_pad_path
|
|
|
|
direct_path = _normalize_existing_file_path(stripped_path)
|
|
if direct_path is not None:
|
|
return direct_path
|
|
|
|
return get_latest_uploaded_reference_image_path()
|