Avatar download and validation for user profiles.
Simplified: URL-only format, saved to media/images/ like regular media. Avatars go through regular media enrichment pipeline (no special moderation).
Bases: EgregoraError
Error during avatar processing.
dataclass
AvatarContext(
docs_dir: Path,
media_dir: Path,
profiles_dir: Path,
vision_model: str,
avatar_namespace: UUID,
cache: EnrichmentCache | None = None,
)
Context for avatar processing operations.
download_avatar_from_url(
url: str,
media_dir: Path,
namespace: UUID,
timeout: float = DEFAULT_DOWNLOAD_TIMEOUT,
client: Client | None = None,
) -> tuple[uuid.UUID, Path]
Download avatar from URL and save to avatars directory.
Parameters:
| Name | Type | Description | Default |
url | str | | required |
media_dir | Path | Root media directory (e.g., site_root/media) | required |
namespace | UUID | UUID namespace for avatar generation | required |
timeout | float | | DEFAULT_DOWNLOAD_TIMEOUT |
client | Client | None | Optional httpx.Client to reuse | None |
Returns:
| Type | Description |
tuple[UUID, Path] | Tuple of (avatar_uuid, avatar_path) |
Raises:
Source code in src/egregora/agents/avatar.py
| @sleep_and_retry
@limits(calls=10, period=60)
def download_avatar_from_url(
url: str,
media_dir: Path,
namespace: uuid.UUID,
timeout: float = DEFAULT_DOWNLOAD_TIMEOUT,
client: httpx.Client | None = None,
) -> tuple[uuid.UUID, Path]:
"""Download avatar from URL and save to avatars directory.
Args:
url: URL of the avatar image
media_dir: Root media directory (e.g., site_root/media)
namespace: UUID namespace for avatar generation
timeout: HTTP timeout in seconds
client: Optional httpx.Client to reuse
Returns:
Tuple of (avatar_uuid, avatar_path)
Raises:
AvatarProcessingError: If download fails or image is invalid
"""
try:
with safe_dns_validation(url):
if client:
return _download_avatar_with_client(client, url, media_dir, namespace)
with _create_secure_client(timeout) as new_client:
return _download_avatar_with_client(new_client, url, media_dir, namespace)
except SSRFValidationError as exc:
raise AvatarProcessingError(str(exc)) from exc
|
process_avatar_commands(
messages_table: Table, context: AvatarContext
) -> dict[str, str]
Process all avatar commands from messages table.
Source code in src/egregora/agents/avatar.py
| def process_avatar_commands(
messages_table: Table,
context: AvatarContext,
) -> dict[str, str]:
"""Process all avatar commands from messages table."""
logger.info("Processing avatar commands from messages")
commands = extract_commands(messages_table)
avatar_commands = [cmd for cmd in commands if cmd.get("command", {}).get("target") == "avatar"]
if not avatar_commands:
logger.info("No avatar commands found")
return {}
logger.info("Found %s avatar command(s)", len(avatar_commands))
results: dict[str, str] = {}
with _create_secure_client() as client:
for cmd_entry in avatar_commands:
author_uuid = cmd_entry["author"]
timestamp_raw = cmd_entry["timestamp"]
command = cmd_entry["command"]
cmd_type = command["command"]
target = command["target"]
if cmd_type in ("set", "unset") and target == "avatar":
if cmd_type == "set":
timestamp_dt = ensure_datetime(timestamp_raw)
result = _process_set_avatar_command(
author_uuid=author_uuid,
timestamp=timestamp_dt,
context=context,
value=command.get("value"),
client=client,
)
results[author_uuid] = result
elif cmd_type == "unset":
result = _process_unset_avatar_command(
author_uuid=author_uuid,
timestamp=str(timestamp_raw),
profiles_dir=context.profiles_dir,
)
results[author_uuid] = result
return results
|