@@ -1,11 +1,14 @@
import hashlib
import json
import logging
import time
from google import genai
from google . genai import types
import httpx
from sqlalchemy import select
from sqlalchemy . ext . asyncio import AsyncSession
from app . config import settings
from app . models . app_settings import AppSettings
logger = logging . getLogger ( " app.services.ai_advisor " )
@@ -34,71 +37,133 @@ SYSTEM_PROMPT = """
"""
async def _get_setting ( db : AsyncSession , key : str , default : str = " " ) - > str :
result = await db . execute ( select ( AppSettings . value ) . where ( AppSettings . key == key ) )
row = result . scalar_one_or_none ( )
return row if row is not None else default
def _make_proxy_signature ( salt : str ) - > tuple [ str , int ] :
timestamp = int ( time . time ( ) )
hash_input = f " { timestamp } { salt } " . encode ( )
signature = hashlib . sha256 ( hash_input ) . hexdigest ( )
return signature , timestamp
async def _call_via_proxy ( proxy_url : str , proxy_salt : str , messages : list [ dict ] ) - > str :
""" Send request through AI proxy. """
signature , timestamp = _make_proxy_signature ( proxy_salt )
headers = {
" x-signature " : signature ,
" x-timestamp " : str ( timestamp ) ,
}
async with httpx . AsyncClient ( timeout = 60.0 ) as client :
resp = await client . post (
f " { proxy_url . rstrip ( ' / ' ) } /generate_text " ,
json = { " messages " : messages } ,
headers = headers ,
)
if resp . status_code != 200 :
logger . error ( " AI proxy error %d : %s " , resp . status_code , resp . text [ : 500 ] )
raise ValueError ( f " AI proxy вернул ошибку: { resp . status_code } " )
data = resp . json ( )
text = data . get ( " response " , " " )
if not text :
raise ValueError ( f " AI proxy вернул пустой ответ (finish_reason= { data . get ( ' finish_reason ' ) } ) " )
return text
async def _call_direct ( api_key : str , system : str , user_message : str ) - > str :
""" Call Google GenAI directly via REST (no SDK dependency). """
url = f " https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent?key= { api_key } "
payload = {
" system_instruction " : { " parts " : [ { " text " : system } ] } ,
" contents " : [ { " role " : " user " , " parts " : [ { " text " : user_message } ] } ] ,
" generationConfig " : { " temperature " : 0.3 , " maxOutputTokens " : 1024 } ,
}
async with httpx . AsyncClient ( timeout = 60.0 ) as client :
resp = await client . post ( url , json = payload )
if resp . status_code != 200 :
logger . error ( " Google API error %d : %s " , resp . status_code , resp . text [ : 500 ] )
raise ValueError ( f " Google API ошибка: { resp . status_code } " )
data = resp . json ( )
candidates = data . get ( " candidates " , [ ] )
if not candidates :
raise ValueError ( " Google API вернул пустой ответ " )
parts = candidates [ 0 ] . get ( " content " , { } ) . get ( " parts " , [ ] )
return parts [ 0 ] . get ( " text " , " " ) if parts else " "
def _parse_json_response ( text : str ) - > dict :
try :
return json . loads ( text )
except json . JSONDecodeError :
start = text . find ( " { " )
end = text . rfind ( " } " ) + 1
if start != - 1 and end > start :
return json . loads ( text [ start : end ] )
raise ValueError ( " AI вернул невалидный JSON " )
async def get_material_recommendation (
task_description : str ,
materials_data : list [ dict ] ,
budget_preference : str = " optimal " ,
file_info : dict | None = None ,
db : AsyncSession | None = None ,
) - > dict :
""" Get material recommendation from Google Gemini API . """
""" Get material recommendation — via proxy or direct, based on DB settings . """
logger . info ( " === AI Advisor request === " )
logger . info ( " Task: %s " , task_description )
logger . info ( " Budget preference: %s " , budget_preference )
logger . info ( " File info: %s " , file_info )
logger . info ( " Materials count: %d " , len ( materials_data ) )
if not settings . GOOGLE_API_KEY :
logger . error ( " GOOGLE_API_KEY is not configured " )
raise ValueError ( " GOOGLE_API_KEY not configured " )
# Read settings from DB, fallback to config
use_proxy = True
proxy_url = settings . AI_PROXY_URL
proxy_salt = settings . AI_PROXY_SALT
direct_api_key = settings . GOOGLE_API_KEY
if db :
use_proxy_str = await _get_setting ( db , " ai_use_proxy " , " true " )
use_proxy = use_proxy_str . lower ( ) in ( " true " , " 1 " , " yes " )
db_proxy_url = await _get_setting ( db , " ai_proxy_url " )
db_proxy_salt = await _get_setting ( db , " ai_proxy_salt " )
db_api_key = await _get_setting ( db , " ai_direct_api_key " )
if db_proxy_url :
proxy_url = db_proxy_url
if db_proxy_salt :
proxy_salt = db_proxy_salt
if db_api_key :
direct_api_key = db_api_key
materials_json = json . dumps ( materials_data , ensure_ascii = False , indent = 2 )
system = SYSTEM_PROMPT . format ( materials_json = materials_json )
logger . debug ( " System prompt length: %d chars " , len ( system ) )
user_message = f " Описание задачи: { task_description } \n Предпочтение по бюджету: { budget_preference } "
if file_info :
user_message + = f " \n Информация о { json . dumps ( file_info , ensure_ascii = False ) } "
logger . debug ( " User message: %s " , user_message )
logger . info ( " Sending request to Gemini API (model: gemini-2.0-flash)... " )
start_time = time . time ( )
client = genai . Client ( api_key = settings . GOOGLE_API_KEY )
response = await client . aio . models . generate_content (
model = " gemini-3-flash-preview " ,
contents = user_message ,
config = types . GenerateContentConfig (
system_instruction = system ,
max_output_tokens = 1024 ,
temperature = 0.3 ,
) ,
)
if use_proxy :
logger . info ( " Mode: PROXY ( %s ) " , proxy_url )
if not proxy_url :
raise ValueError ( " AI proxy URL не настроен " )
messages = [ { " role " : " user " , " content " : system + " \n \n " + user_message } ]
response_text = await _call_via_proxy ( proxy_url , proxy_salt , messages )
else :
logger . info ( " Mode: DIRECT (Google API) " )
if not direct_api_key :
raise ValueError ( " Google API Key не настроен " )
response_text = await _call_direct ( direct_api_key , system , user_message )
elapsed = time . time ( ) - start_time
logger . info ( " Gemini AP I responded in %.2f seconds " , elapsed )
response_text = response . text
logger . info ( " A I responded in %.2f seconds " , elapsed )
logger . debug ( " Raw response ( %d chars): %s " , len ( response_text ) , response_text [ : 500 ] )
try :
result = json . loads ( response_text )
logger . info ( " Response parsed as JSON successfully " )
logger . info ( " Recommended material: id= %s , name= %s " ,
result . get ( " recommended_material_id " ) , result . get ( " recommended_material_name " ) )
logger . info ( " Alternatives: %d , Questions: %d " ,
len ( result . get ( " alternatives " , [ ] ) ) , len ( result . get ( " questions " , [ ] ) ) )
logger . info ( " === AI Advisor complete === " )
return result
except json . JSONDecodeError :
logger . warning ( " Direct JSON parse failed, trying to extract JSON from response... " )
start = response_text . find ( " { " )
end = response_text . rfind ( " } " ) + 1
if start != - 1 and end > start :
extracted = response_text [ start : end ]
logger . debug ( " Extracted JSON substring [ %d : %d ]: %s " , start , end , extracted [ : 300 ] )
result = json . loads ( extracted )
logger . info ( " Extracted JSON parsed successfully " )
logger . info ( " === AI Advisor complete === " )
return result
logger . error ( " Failed to extract JSON from AI response: %s " , response_text [ : 200 ] )
raise ValueError ( " AI вернул невалидный JSON " )
result = _parse_json_response ( response_text )
logger . info ( " Recommended: id= %s , name= %s " ,
result . get ( " recommended_material_id " ) , result . get ( " recommended_material_name " ) )
logger . info ( " === AI Advisor complete === " )
return result
