gigiachat_requests.py

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, model):
    url_completion = os.getenv('GIGA_TOKENS_URL')
        
    data_copm = json.dumps({
        "model": 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, model):
    url_completion = os.getenv('GIGA_COMPLETION_URL')
        
    data_copm = json.dumps({
        "model": 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, model):
    url_completion = os.getenv('GIGA_COMPLETION_URL')

    data_copm = json.dumps({
        "model": model,
        "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