Hass/ha-text-ai
1
0
Fork 0
mirror of https://github.com/smkrv/ha-text-ai.git synced 2026-07-28 10:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

Release v2.0.0

Browse Source
This commit is contained in:
SMKRV
2024-11-23 01:09:38 +03:00
parent 7f8d8be5fb
commit 3b38a6dd29
1 changed files with 65 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/ha_text_ai/config_flow.py
+65 -71
View File
@@ -64,9 +64,7 @@ class HATextAIConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def _show_provider_config(self, provider): def _show_provider_config(self, provider):
"""Show configuration screen for selected provider.""" """Show configuration screen for selected provider."""
default_endpoint = DEFAULT_OPENAI_ENDPOINT if provider == API_PROVIDER_OPENAI else DEFAULT_ANTHROPIC_ENDPOINT default_endpoint = DEFAULT_OPENAI_ENDPOINT if provider == API_PROVIDER_OPENAI else DEFAULT_ANTHROPIC_ENDPOINT
default_name = f"HA Text AI {len(self._async_current_entries()) + 1}" default_name = f"HA Text AI {len(self._async_current_entries()) + 1}"
return self.async_show_form( return self.async_show_form(
step_id="user", step_id="user",
data_schema=vol.Schema({ data_schema=vol.Schema({
@@ -91,84 +89,80 @@ class HATextAIConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
description_placeholders={"provider": provider} description_placeholders={"provider": provider}
) )
async def _process_configuration(self, user_input): async def _process_configuration(self, user_input):
"""Validate and process user configuration.""" """Validate and process user configuration."""
errors = {} errors = {}
try: try:
# Генерируем уникальный идентификатор один раз unique_id = f"{user_input[CONF_API_KEY]}_{user_input[CONF_MODEL]}"
unique_id = f"{user_input[CONF_API_KEY]}_{user_input[CONF_MODEL]}" await self.async_set_unique_id(unique_id)
await self.async_set_unique_id(unique_id) self._abort_if_unique_id_configured()
self._abort_if_unique_id_configured()
session = async_get_clientsession(self.hass) session = async_get_clientsession(self.hass)
# Minimal API key validation
headers = {
"Authorization": f"Bearer {user_input[CONF_API_KEY]}",
"Content-Type": "application/json"
}
# Adjust headers and endpoint based on provider
if user_input[CONF_API_PROVIDER] == API_PROVIDER_ANTHROPIC:
headers = { headers = {
"x-api-key": user_input[CONF_API_KEY], "Authorization": f"Bearer {user_input[CONF_API_KEY]}",
"anthropic-version": "2023-06-01" "Content-Type": "application/json"
} }
# Basic connection test if user_input[CONF_API_PROVIDER] == API_PROVIDER_ANTHROPIC:
try: headers = {
async with session.get( "x-api-key": user_input[CONF_API_KEY],
f"{user_input[CONF_API_ENDPOINT]}/models", "anthropic-version": "2023-06-01"
headers=headers }
) as response:
if response.status not in [200, 404]: try:
errors["base"] = "cannot_connect" async with session.get(
f"{user_input[CONF_API_ENDPOINT]}/models",
headers=headers
) as response:
if response.status not in [200, 404]:
errors["base"] = "cannot_connect"
except Exception as e:
_LOGGER.error(f"Connection test failed: {e}")
errors["base"] = "cannot_connect"
if not errors:
return self.async_create_entry(
title=user_input.get("name", f"HA Text AI ({user_input[CONF_API_PROVIDER]})"),
data=user_input
)
except Exception as e: except Exception as e:
_LOGGER.error(f"Connection test failed: {e}") _LOGGER.error(f"Unexpected error: {e}")
errors["base"] = "cannot_connect" errors["base"] = "unknown"
if not errors: return self.async_show_form(
return self.async_create_entry( step_id="user",
title=user_input.get("name", f"HA Text AI ({user_input[CONF_API_PROVIDER]})"), data_schema=vol.Schema({
data=user_input vol.Required("name", default=user_input.get("name", f"HA Text AI {len(self._async_current_entries()) + 1}")): str,
) vol.Required(CONF_API_PROVIDER): vol.In([user_input[CONF_API_PROVIDER]]),
vol.Required(CONF_API_KEY): str,
vol.Required(CONF_MODEL, default=user_input.get(CONF_MODEL, DEFAULT_MODEL)): str,
vol.Optional(CONF_API_ENDPOINT, default=user_input.get(CONF_API_ENDPOINT,
DEFAULT_OPENAI_ENDPOINT if user_input[CONF_API_PROVIDER] == API_PROVIDER_OPENAI
else DEFAULT_ANTHROPIC_ENDPOINT)): str,
vol.Optional(CONF_TEMPERATURE, default=user_input.get(CONF_TEMPERATURE, DEFAULT_TEMPERATURE)): vol.All(
vol.Coerce(float),
vol.Range(min=MIN_TEMPERATURE, max=MAX_TEMPERATURE)
),
vol.Optional(CONF_MAX_TOKENS, default=user_input.get(CONF_MAX_TOKENS, DEFAULT_MAX_TOKENS)): vol.All(
vol.Coerce(int),
vol.Range(min=MIN_MAX_TOKENS, max=MAX_MAX_TOKENS)
),
vol.Optional(CONF_REQUEST_INTERVAL, default=user_input.get(CONF_REQUEST_INTERVAL, DEFAULT_REQUEST_INTERVAL)): vol.All(
vol.Coerce(float),
vol.Range(min=MIN_REQUEST_INTERVAL)
),
}),
description_placeholders={"provider": user_input[CONF_API_PROVIDER]},
errors=errors
)
except Exception as e: @staticmethod
_LOGGER.error(f"Unexpected error: {e}") async def async_get_options_flow(config_entry: config_entries.ConfigEntry) -> config_entries.OptionsFlow:
errors["base"] = "unknown" """Create the options flow."""
return OptionsFlowHandler(config_entry)
# Return to provider config, preserving previous input
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({
vol.Required("name", default=user_input.get("name", f"HA Text AI {len(self._async_current_entries()) + 1}")): str,
vol.Required(CONF_API_PROVIDER): vol.In([user_input[CONF_API_PROVIDER]]),
vol.Required(CONF_API_KEY): str,
vol.Required(CONF_MODEL, default=user_input.get(CONF_MODEL, DEFAULT_MODEL)): str,
vol.Optional(CONF_API_ENDPOINT, default=user_input.get(CONF_API_ENDPOINT,
DEFAULT_OPENAI_ENDPOINT if user_input[CONF_API_PROVIDER] == API_PROVIDER_OPENAI
else DEFAULT_ANTHROPIC_ENDPOINT)): str,
vol.Optional(CONF_TEMPERATURE, default=user_input.get(CONF_TEMPERATURE, DEFAULT_TEMPERATURE)): vol.All(
vol.Coerce(float),
vol.Range(min=MIN_TEMPERATURE, max=MAX_TEMPERATURE)
),
vol.Optional(CONF_MAX_TOKENS, default=user_input.get(CONF_MAX_TOKENS, DEFAULT_MAX_TOKENS)): vol.All(
vol.Coerce(int),
vol.Range(min=MIN_MAX_TOKENS, max=MAX_MAX_TOKENS)
),
vol.Optional(CONF_REQUEST_INTERVAL, default=user_input.get(CONF_REQUEST_INTERVAL, DEFAULT_REQUEST_INTERVAL)): vol.All(
vol.Coerce(float),
vol.Range(min=MIN_REQUEST_INTERVAL)
),
}),
description_placeholders={"provider": user_input[CONF_API_PROVIDER]},
errors=errors
)
def async_get_options_flow(config_entry: config_entries.ConfigEntry) -> config_entries.OptionsFlow:
"""Create the options flow."""
return OptionsFlowHandler(config_entry)
class OptionsFlowHandler(config_entries.OptionsFlow): class OptionsFlowHandler(config_entries.OptionsFlow):
"""Options flow handler.""" """Options flow handler."""