2025-11-04 13:30:59 -05:00
|
|
|
from urllib.parse import quote
|
|
|
|
|
|
2026-02-06 02:48:35 +04:00
|
|
|
from open_webui.env import (
|
2026-02-06 03:23:37 +04:00
|
|
|
FORWARD_USER_INFO_HEADER_USER_NAME,
|
|
|
|
|
FORWARD_USER_INFO_HEADER_USER_ID,
|
|
|
|
|
FORWARD_USER_INFO_HEADER_USER_EMAIL,
|
|
|
|
|
FORWARD_USER_INFO_HEADER_USER_ROLE,
|
2026-02-06 02:48:35 +04:00
|
|
|
)
|
|
|
|
|
|
2025-11-04 13:30:59 -05:00
|
|
|
|
|
|
|
|
def include_user_info_headers(headers, user):
|
|
|
|
|
return {
|
|
|
|
|
**headers,
|
2026-03-17 17:58:01 -05:00
|
|
|
FORWARD_USER_INFO_HEADER_USER_NAME: quote(user.name, safe=' '),
|
2026-02-06 03:23:37 +04:00
|
|
|
FORWARD_USER_INFO_HEADER_USER_ID: user.id,
|
|
|
|
|
FORWARD_USER_INFO_HEADER_USER_EMAIL: user.email,
|
|
|
|
|
FORWARD_USER_INFO_HEADER_USER_ROLE: user.role,
|
2025-11-04 13:30:59 -05:00
|
|
|
}
|
2026-05-09 06:01:02 +09:00
|
|
|
|
2026-05-09 15:25:27 +09:00
|
|
|
|
2026-05-09 06:01:02 +09:00
|
|
|
def get_custom_headers(custom_headers: dict, user=None, metadata: dict = None) -> dict:
|
|
|
|
|
if not custom_headers or not isinstance(custom_headers, dict):
|
|
|
|
|
return {}
|
|
|
|
|
|
|
|
|
|
metadata = metadata or {}
|
|
|
|
|
template_vars = {
|
|
|
|
|
'{{CHAT_ID}}': metadata.get('chat_id', '') or '',
|
|
|
|
|
'{{MESSAGE_ID}}': metadata.get('message_id', '') or '',
|
|
|
|
|
'{{USER_ID}}': (user.id if user else '') or '',
|
|
|
|
|
'{{USER_NAME}}': (user.name if user else '') or '',
|
|
|
|
|
}
|
2026-05-09 15:25:27 +09:00
|
|
|
|
2026-05-09 06:01:02 +09:00
|
|
|
parsed_headers = {}
|
|
|
|
|
for key, value in custom_headers.items():
|
|
|
|
|
if not isinstance(value, str):
|
|
|
|
|
value = str(value)
|
|
|
|
|
for token, val in template_vars.items():
|
|
|
|
|
value = value.replace(token, val)
|
|
|
|
|
parsed_headers[key] = value
|
2026-05-09 15:25:27 +09:00
|
|
|
|
2026-05-09 06:01:02 +09:00
|
|
|
return parsed_headers
|