import os import base64 import json import requests import streamlit as st def get_access_token(): username = st.secrets["GIGA_USERNAME"] password = st.secrets["GIGA_SECRET"] # Получаем строку с базовой авторизацией в формате Base64 auth_str = f'{username}:{password}' auth_bytes = auth_str.encode('utf-8') auth_base64 = base64.b64encode(auth_bytes).decode('utf-8') url = os.getenv('GIGA_AUTH_URL') headers = { 'Authorization': f'Basic {auth_base64}', # вставляем базовую авторизацию 'RqUID': os.getenv('GIGA_rquid'), 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } data = { 'scope': os.getenv('GIGA_SCOPE') } response = requests.post(url, headers=headers, data=data, verify=False) access_token = response.json()['access_token'] print('Got access token') return access_token def get_number_of_tokens(prompt, access_token): url_completion = os.getenv('GIGA_TOKENS_URL') data_copm = json.dumps({ "model": os.getenv('GIGA_MODEL'), "input": [ prompt ], }) headers_comp = { 'Content-Type': 'application/json', 'Accept': 'application/json', 'Authorization': 'Bearer ' + access_token } response = requests.post(url_completion, headers=headers_comp, data=data_copm, verify=False) response_data = response.json() return response_data[0]['tokens'] def get_completion_from_gigachat(prompt, max_tokens, access_token): url_completion = os.getenv('GIGA_COMPLETION_URL') data_copm = json.dumps({ "model": os.getenv('GIGA_MODEL'), "messages": [ { "role": "user", "content": prompt } ], "stream": False, "max_tokens": max_tokens, }) headers_comp = { 'Content-Type': 'application/json', 'Accept': 'application/json', 'Authorization': 'Bearer ' + access_token } response = requests.post(url_completion, headers=headers_comp, data=data_copm, verify=False) response_data = response.json() answer_from_llm = response_data['choices'][0]['message']['content'] return answer_from_llm def process_transcribation_with_gigachat(prompt, transcript, max_tokens, access_token): url_completion = os.getenv('GIGA_COMPLETION_URL') data_copm = json.dumps({ "model": 'GigaChat-Plus', "max_tokens": max_tokens, "messages": [ { "role": "user", "content": prompt + transcript } ], "stream": True, }) headers_comp = { 'Content-Type': 'application/json', 'Accept': 'application/json', 'Authorization': 'Bearer ' + access_token } output_text = '' st.write("Результат обработки:") text_container = st.empty() response = requests.post(url_completion, headers=headers_comp, data=data_copm, verify=False, stream=True) for line in response.iter_lines(): if line: decoded_line = line.decode('utf-8') if decoded_line.startswith('data:'): decoded_line = decoded_line[len('data:'):].strip() try: data = json.loads(decoded_line) if "choices" in data: for choice in data["choices"]: if "delta" in choice and "content" in choice["delta"]: output_text += choice["delta"]["content"] text_container.text(output_text) except json.JSONDecodeError: continue return output_text