150 lines
6.3 KiB
Python
150 lines
6.3 KiB
Python
import json
|
|
import logging
|
|
|
|
import formencode
|
|
|
|
from pyramid.httpexceptions import HTTPFound
|
|
from rhodecode.apps._base import BaseAppView
|
|
from rhodecode.apps._base.navigation import navigation_list
|
|
from rhodecode.apps.ai_agents.ai_service import get_ai_service
|
|
from rhodecode.apps.ai_agents.ai_settings import AIModelName
|
|
from rhodecode.apps.ai_agents.models.base import AIService
|
|
from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator, CSRFRequired
|
|
from rhodecode.lib import helpers as h
|
|
from rhodecode.model.db import User
|
|
from rhodecode.model.forms import AiSettingsForm, AiSettingsModelVersionUpdateForm
|
|
from rhodecode.model.settings import SettingsModel
|
|
from rhodecode.model.meta import Session
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class AdminAiView(BaseAppView):
|
|
def load_default_context(self):
|
|
c = self._get_local_tmpl_context()
|
|
return c
|
|
|
|
@LoginRequired()
|
|
@HasPermissionAllDecorator("hg.admin")
|
|
def admin_settings_ai(self):
|
|
c = self.load_default_context()
|
|
c.active = "ai"
|
|
c.navlist = navigation_list(self.request)
|
|
|
|
app_settings = c.rc_config
|
|
c.selected_ai_model = app_settings.get("rhodecode_ai_model", AIModelName.GPT.name)
|
|
c.selected_ai_model_version = app_settings.get("rhodecode_ai_model_version")
|
|
c.api_key = app_settings.get("rhodecode_ai_api_key")
|
|
c.ai_features_enabled = app_settings.get("rhodecode_ai_features_enabled", False)
|
|
c.ai_default_code_review = app_settings.get("rhodecode_ai_default_code_review", False)
|
|
|
|
c.ai_instructions = app_settings.get("rhodecode_ai_code_review_instructions")
|
|
if not c.ai_instructions:
|
|
# to not run formatting each time
|
|
c.ai_instructions = "\n".join(AIService.DEFAULT_BASIC_REVIEW_POINTS)
|
|
|
|
gpt_versions = app_settings.get(f"rhodecode_ai_model_{AIModelName.GPT.name}_versions", "[]")
|
|
claude_versions = app_settings.get(f"rhodecode_ai_model_{AIModelName.Claude.name}_versions", "[]")
|
|
gemini_versions = app_settings.get(f"rhodecode_ai_model_{AIModelName.Gemini.name}_versions", "[]")
|
|
|
|
c.model_map = {
|
|
AIModelName.GPT.name: json.loads(gpt_versions),
|
|
AIModelName.Claude.name: json.loads(claude_versions),
|
|
AIModelName.Gemini.name: json.loads(gemini_versions),
|
|
}
|
|
return self._get_template_context(c)
|
|
|
|
@LoginRequired()
|
|
@HasPermissionAllDecorator("hg.admin")
|
|
def admin_settings_ai_update_models(self):
|
|
_ = self.request.translate
|
|
versions = []
|
|
try:
|
|
data = self._parse_form(_, form_class=AiSettingsModelVersionUpdateForm)
|
|
model = data["rhodecode_ai_model"]
|
|
api_key = data["rhodecode_ai_api_key"]
|
|
|
|
ai_service = get_ai_service(api_key, model_name=model)
|
|
versions = ai_service.list_model_api_and_display_names()
|
|
if versions:
|
|
message, category = self._save_model_versions(_, model=model, versions=versions)
|
|
else:
|
|
message = _("AI model API returned empty version list")
|
|
category = "warning"
|
|
except Exception as e:
|
|
log.exception("Exception updating AI model versions: %s", e)
|
|
category = "error"
|
|
message = _(f"Error occurred during updating AI model versions, error: {str(e)}")
|
|
|
|
return {
|
|
"success": True,
|
|
"message": message,
|
|
"message_category": category,
|
|
"versions": versions,
|
|
}
|
|
|
|
@CSRFRequired()
|
|
@LoginRequired()
|
|
@HasPermissionAllDecorator("hg.admin")
|
|
def admin_settings_ai_update(self):
|
|
_ = self.request.translate
|
|
c = self.load_default_context()
|
|
c.active = "ai"
|
|
|
|
data = self._parse_form(_)
|
|
self._save_settings(_, data)
|
|
|
|
return HTTPFound(h.route_path("admin_settings_ai"))
|
|
|
|
def _save_model_versions(self, _, model, versions):
|
|
try:
|
|
setting = f"ai_model_{model}_versions"
|
|
str_list = json.dumps(versions)
|
|
sett = SettingsModel().create_or_update_setting(setting, str_list)
|
|
Session().add(sett)
|
|
Session().commit()
|
|
SettingsModel().invalidate_settings_cache()
|
|
return _("AI model versions updated"), "success"
|
|
except Exception as e:
|
|
log.exception("Exception saving AI model versions: %s", e)
|
|
return _(f"Error occurred during saving AI model versions, error: {str(e)}"), "error"
|
|
|
|
def _save_settings(self, _, data):
|
|
try:
|
|
settings = [
|
|
("ai_features_enabled", "rhodecode_ai_features_enabled", "bool"),
|
|
("ai_default_code_review", "rhodecode_ai_default_code_review", "bool"),
|
|
("ai_model", "rhodecode_ai_model", "unicode"),
|
|
("ai_model_version", "rhodecode_ai_model_version", "unicode"),
|
|
("ai_api_key", "rhodecode_ai_api_key", "unicode"),
|
|
("ai_code_review_instructions", "rhodecode_ai_code_review_instructions", "unicode"),
|
|
]
|
|
for setting, form_key, type_ in settings:
|
|
sett = SettingsModel().create_or_update_setting(setting, data[form_key], type_)
|
|
if setting == "ai_features_enabled" and isinstance(data[form_key], bool):
|
|
self._activate_deactivate_ai_user(data, form_key)
|
|
|
|
Session().add(sett)
|
|
|
|
Session().commit()
|
|
SettingsModel().invalidate_settings_cache()
|
|
h.flash(_("AI settings saved"), category="success")
|
|
except Exception as e:
|
|
log.exception("Exception saving AI settings: %s", e)
|
|
h.flash(_(f"Error occurred during saving AI settings, error: {str(e)}"), category="error")
|
|
|
|
def _activate_deactivate_ai_user(self, data, form_key):
|
|
log.debug("%s AI user" % "Activating" if data[form_key] else "Deactivating")
|
|
ai_user = User.get_ai_user()
|
|
ai_user.active = data[form_key]
|
|
Session().add(ai_user)
|
|
|
|
def _parse_form(self, _, form_class=AiSettingsForm):
|
|
try:
|
|
form = form_class()()
|
|
data = form.to_python(self.request.POST)
|
|
except formencode.Invalid as errors:
|
|
error = errors
|
|
h.flash(_(f"Invalid form error: {error}"), category="error")
|
|
raise HTTPFound(h.route_path("admin_settings_ai"))
|
|
return data
|