1 Commits

Author SHA1 Message Date
Voskik
bb9ac27d85 Update .gitignore 2025-09-13 01:50:17 +03:00
21 changed files with 1176 additions and 1587 deletions

1
.gitignore vendored
View File

@@ -2,4 +2,3 @@
__pycache__/
outputs/*
venv/

187
app.py
View File

@@ -1,128 +1,195 @@
import gradio as gr
from config import *
from services.llm_factory import get_llm_provider
from pathlib import Path
# Подгрузка сервисов
from services.llm import Llm
from services.fasterWhisper import FasterWhisper
from handlers.gradioHandler import GradioHandlers
from handlers.fileHandlers import FileHandlers
from handlers.convertMdToPdf import ConvertMdToPdf
from handlers.glueAudio import GlueAudio
from services.convertMdToPdf import ConvertMdToPdf
gh = GradioHandlers(get_llm_provider, ConvertMdToPdf, FileHandlers, FasterWhisper, GlueAudio)
# Загрузка параметров конфигурации
from config import *
# Функция транскрибации
def generateByCondition(api_key, llm_model, system_prompt, recognized_text, llm_temperature, is_pipeline_enabled, trigger, isSaveFile, filename, filenamePdf):
llm = Llm(api_key)
# если чекбокс включен и событие было change → обрабатываем
if is_pipeline_enabled and trigger == "change":
result, md = llm.generate(llm_model, system_prompt, recognized_text, llm_temperature)
# Конвертируем текст с латексом в юникод
pdf, unicodeText = ConvertMdToPdf().convertLatexToText(md)
if isSaveFile:
savePdf(filenamePdf, pdf)
saveFile(filename, result)
return result, unicodeText
# если чекбокс выключен и событие было click → обрабатываем
if not is_pipeline_enabled and trigger == "click":
result, md = llm.generate(llm_model, system_prompt, recognized_text, llm_temperature)
# Конвертируем текст с латексом в юникод
pdf, unicodeText = ConvertMdToPdf().convertLatexToText(md)
if isSaveFile:
savePdf(filenamePdf, pdf)
saveFile(filename, result)
return result, unicodeText
# если нет чекбокса и было событие change
return gr.skip(), gr.skip()
def savePdf(filename, pdf):
directory = Path(OUTPUT_PATH)
filePath = directory / filename
filePath.parent.mkdir(parents=True, exist_ok=True)
pdf.save(filePath)
# Функция сохранеhния файла
def saveFile(filename, text):
directory = Path(OUTPUT_PATH)
filePath = directory / filename
filePath.parent.mkdir(parents=True, exist_ok=True)
filePath.write_text(text, encoding='utf-8')
# ConvertMdToPdf().convert(text)
# Функция для динамического обновления кнопки в зависимости от состояния checkbox
def updateButton(isChecked):
if not isChecked:
variant = 'primary'
else:
variant = 'secondary'
return gr.update(interactive=not isChecked, variant=variant)
def updateTextbox(isChecked):
return gr.update(visible=isChecked)
###################################################
# ____ ___.___ ___. .__ #
#| | \ | \_ |__ ____ | | ______ _ __#
#| | / | | __ \_/ __ \| | / _ \ \/ \/ /#
#| | /| | | \_\ \ ___/| |_( <_> ) / #
#|______/ |___| |___ /\___ >____/\____/ \/\_/ #
# \/ \/ #
###################################################
def main():
with gr.Blocks() as demo:
gr.HTML('<div align=center><h1>Faster Whisper WebUI</h1></div>')
gr.HTML('''
<div align=center>
<h1>
Faster Whisper WebUI
</h1>
</div>
''')
with gr.Row():
# Вкладка с основным взаимодействием
with gr.Tab('Actions'):
isPipelineEnabledCheckbox = gr.Checkbox(label='is pipeline enabled', value=True, interactive=True)
with gr.Row():
with gr.Accordion(label='Recognization and integration'):
with gr.Column():
audioFiles = gr.Files(label='Load audio for transcribe', type="filepath")
audioFile = gr.Audio(label='Load audio for transcribe', type="filepath")
images = gr.Files(label='Upload images', file_types=['image'])
recognizeBtn = gr.Button('recognize and integrate', variant='primary')
with gr.Accordion(label='Recognized text'):
recognizedText = gr.TextArea(label='')
with gr.Accordion(label='LLM'):
with gr.Column():
refineTextBtn = gr.Button('refine text', variant='secondary', interactive=False)
with gr.Accordion(label='Refined text raw'):
refinedText = gr.Textbox(label='', show_copy_button=True)
with gr.Accordion(label='Refined text md formated'):
refinedTextMD = gr.Markdown(label='')
# Вкладка с настройками
with gr.Tab('Settings'):
with gr.Column():
# Первое поле на всю ширину в акордионе настроек
with gr.Accordion('File settings'):
saveFileCheckbox = gr.Checkbox(label='save file', value=True, interactive=True)
filename = gr.Textbox(label='Output filename', value='output.md', interactive=True)
filename = gr.Textbox(label='Output filename', value='output.txt', interactive=True)
filenamePdf = gr.Textbox(label='Output filename for pdf', value='output.pdf', interactive=True)
# Акордион настроек faster whisper
with gr.Accordion(label='Faster whisper settings'):
with gr.Row():
# Левая колонка в акордионе
with gr.Column():
device = gr.Dropdown(label='Device', choices=DEVICES, value=DEVICES[1], interactive=True)
compute_type = gr.Dropdown(label='compute_type', choices=COMPUTE_TYPE, value=COMPUTE_TYPE[0], interactive=True)
device = gr.Dropdown(label='Device', choices=["cpu", "cuda"], value="cuda", interactive=True)
compute_type = gr.Dropdown(label='compute_type', choices=["auto", "int8", "float16", "float32"], value="auto", interactive=True)
fastWhisperModel = gr.Dropdown(label='Model', choices=FAST_WHISPER_MODELS, value=FAST_WHISPER_MODELS[11], interactive=True)
beamSize = gr.Number(label='beam_size', value=8, interactive=True)
noSpeechThreshold = gr.Number(label='no_speech_threshold', value=0.5, interactive=True)
vadFilter = gr.Checkbox(label='vad_filter', value=True, interactive=True)
wordTimestamps = gr.Checkbox(label='word_timestamps', value=True, interactive=True)
conditionOnPreviousText = gr.Checkbox(label='condition_on_previous_text', value=False, interactive=True)
# Правая колонка в акордионе
with gr.Column():
with gr.Accordion(label='Vad parameters'):
minSilenceDurationMs = gr.Number(label='min_silence_duration_ms', value=300, interactive=True)
speechPadMs = gr.Number(label='speech_pad_ms', value=200, interactive=True)
with gr.Accordion(label='Temperature'):
temp0 = gr.Number(label='temp_0', value=0.0, interactive=True)
temp1 = gr.Number(label='temp_1', value=0.2, interactive=True)
temp2 = gr.Number(label='temp_2', value=0.4, interactive=True)
with gr.Accordion(label='LLM settings'):
apiKey = gr.Textbox(label='API key (required for io.net, Gemini)', value=DEFAULT_API_KEY, interactive=True)
# Нижний акордион настроек для api ключа llm
with gr.Accordion(label='ai.io.net api settings'):
apiKey = gr.Textbox(label='API key', value=DEFAULT_API_KEY, interactive=True)
with gr.Accordion(label='System prompt'):
systemPrompt = gr.Textbox(label='', value=DEFAULT_SYSTEM_PROMPT, interactive=True)
with gr.Row():
llmProvider = gr.Dropdown(label='LLM Provider', choices=LLM_PROVIDERS, value=LLM_PROVIDERS[0], interactive=True)
llmModel = gr.Dropdown(label='Models', choices=LLM_MODELS[LLM_PROVIDERS[0]], value=LLM_MODELS[LLM_PROVIDERS[0]][1], interactive=True)
llmModel = gr.Dropdown(label='models', choices=LLM_MODELS, value=LLM_MODELS[1], interactive=True)
llmTemperature = gr.Number(label='Temperature', value=0.8, interactive=True )
# Настройки Custom провайдера
with gr.Accordion(label='Custom Provider Settings', open=True):
customBaseUrl = gr.Textbox(
label='Base URL',
value='http://127.0.0.1:1234/v1/',
interactive=True,
visible=False # Скрыто по умолчанию
)
######################################################################
#.____ .__ ___. .__ #
#| | ____ ____ |__| ____ \_ |__ ____ | | ______ _ __#
#| | / _ \ / ___\| |/ ___\ | __ \_/ __ \| | / _ \ \/ \/ /#
#| |__( <_> ) /_/ > \ \___ | \_\ \ ___/| |_( <_> ) / #
#|_______ \____/\___ /|__|\___ > |___ /\___ >____/\____/ \/\_/ #
# \/ /_____/ \/ \/ \/ #
######################################################################
# Обработчики событий
isPipelineEnabledCheckbox.change(gh.updateButton, inputs=[isPipelineEnabledCheckbox], outputs=refineTextBtn)
saveFileCheckbox.change(gh.updateTextbox, inputs=saveFileCheckbox, outputs=filename)
saveFileCheckbox.change(gh.updateTextbox, inputs=saveFileCheckbox, outputs=filenamePdf)
isPipelineEnabledCheckbox.change(updateButton, inputs=[isPipelineEnabledCheckbox], outputs=refineTextBtn)
saveFileCheckbox.change(updateTextbox, inputs=saveFileCheckbox, outputs=filename)
saveFileCheckbox.change(updateTextbox, inputs=saveFileCheckbox, outputs=filenamePdf)
recognizeBtn.click(
gh.handleRecognizeBtn,
inputs=[audioFiles, fastWhisperModel, device, compute_type, beamSize,
vadFilter, minSilenceDurationMs, speechPadMs, temp0, temp1, temp2,
wordTimestamps, noSpeechThreshold, conditionOnPreviousText, gr.State(GLUED_AUDIO_FILENAME), gr.State(OUTPUT_PATH)],
outputs=[recognizedText],
)
recognizeBtn.click(FasterWhisper().recognize, outputs=[recognizedText], inputs=[fastWhisperModel, device, compute_type, audioFile, beamSize, vadFilter, minSilenceDurationMs, speechPadMs, temp0, temp1, temp2, wordTimestamps, noSpeechThreshold, conditionOnPreviousText])
# --- ИСПРАВЛЕНИЕ: ДОБАВЛЕН customBaseUrl В INPUTS ---
# Если пайплайн включен то тогда делаем автоматически
# автоматический пайплайн
recognizedText.change(
gh.generateByCondition,
inputs=[apiKey, llmProvider, llmModel, systemPrompt, recognizedText, llmTemperature,
isPipelineEnabledCheckbox, gr.State("change"), saveFileCheckbox, filename, filenamePdf, gr.State(OUTPUT_PATH), customBaseUrl],
generateByCondition,
inputs=[apiKey, llmModel, systemPrompt, recognizedText, llmTemperature, isPipelineEnabledCheckbox, gr.State("change"), saveFileCheckbox, filename, filenamePdf],
outputs=[refinedText, refinedTextMD]
)
# Обновление выпадающего списка моделей и поля API ключа
llmProvider.change(
gh.update_model_dropdown,
inputs=llmProvider,
outputs=[llmModel, apiKey]
)
# Переключение видимости URL для Custom провайдера
llmProvider.change(
fn=gh.toggle_custom_url,
inputs=llmProvider,
outputs=[customBaseUrl]
)
# ручной запуск по кнопке
refineTextBtn.click(
gh.generateByCondition,
inputs=[apiKey, llmProvider, llmModel, systemPrompt, recognizedText, llmTemperature,
isPipelineEnabledCheckbox, gr.State("click"), saveFileCheckbox, filename, filenamePdf, gr.State(OUTPUT_PATH), customBaseUrl],
generateByCondition,
inputs=[apiKey, llmModel, systemPrompt, recognizedText, llmTemperature, isPipelineEnabledCheckbox, gr.State("click"), saveFileCheckbox, filename, filenamePdf],
outputs=[refinedText, refinedTextMD]
)
demo.launch()
if __name__ == '__main__':
main()

View File

@@ -4,58 +4,14 @@ from dotenv import load_dotenv
load_dotenv()
FAST_WHISPER_MODELS = ['tiny', 'base', 'small', 'medium', 'large-v1', 'large-v2', 'large-v3', 'large', 'distil-large-v2', 'distil-large-v3', 'distil-large-v3.5', 'large-v3-turbo', 'turbo']
DEVICES = ['cpu', 'cuda']
COMPUTE_TYPE = ['auto', 'int8', 'float16', 'float32']
LLM_MODELS = ['openai/gpt-oss-120b', 'Qwen/Qwen3-235B-A22B-Thinking-2507', 'deepseek-ai/DeepSeek-R1-0528', 'meta-llama/Llama-4-Maverick-17B-128E-Instruct-FP8', 'openai/gpt-oss-20b', 'Intel/Qwen3-Coder-480B-A35B-Instruct-int4-mixed-ar', 'meta-llama/Llama-3.2-90B-Vision-Instruct', 'mistralai/Mistral-Nemo-Instruct-2407', 'Qwen/Qwen2.5-VL-32B-Instruct', 'meta-llama/Llama-3.3-70B-Instruct', 'mistralai/Devstral-Small-2505', 'mistralai/Magistral-Small-2506', 'mistralai/Mistral-Large-Instruct-2411', 'CohereForAI/aya-expanse-32b']
# Стандартный API ключ
IO_API_KEY=os.getenv('IO_API_KEY')
GEMINI_API_KEY=os.getenv('GEMINI_API_KEY')
DEFAULT_API_KEY=IO_API_KEY
# Словарь провайдеров и их моделей
LLM_PROVIDERS = ['io.net', 'Gemini', 'gpt4free', 'Custom']
LLM_MODELS = {
'io.net': [
'openai/gpt-oss-120b', 'Qwen/Qwen3-235B-A22B-Thinking-2507',
'deepseek-ai/DeepSeek-R1-0528', 'meta-llama/Llama-4-Maverick-17B-128E-Instruct-FP8',
'openai/gpt-oss-20b', 'Intel/Qwen3-Coder-480B-A35B-Instruct-int4-mixed-ar',
'meta-llama/Llama-3.2-90B-Vision-Instruct', 'mistralai/Mistral-Nemo-Instruct-2407',
'Qwen/Qwen2.5-VL-32B-Instruct', 'meta-llama/Llama-3.3-70B-Instruct',
'mistralai/Devstral-Small-2505', 'mistralai/Magistral-Small-2506',
'mistralai/Mistral-Large-Instruct-2411', 'CohereForAI/aya-expanse-32b'
],
'Gemini': [
'gemini-2.5-pro',
'gemini-2.5-flash',
'gemini-2.5-flash-lite'
],
'gpt4free': [
'default',
'gpt-4',
'sonar-reasoning',
'command-r-plus',
'llama-3.3-70b',
'hermes-3-llama-3.1-405b'
'qwen-3-235b',
'gpt-4o-mini',
'deepseek-r1',
'PollinationsAI:gpt-5-nano'
],
'Custom': [
'qwen/qwen3-vl-30b',
'qwen/qwen3-coder-30b',
'openai/gpt-oss-20b',
'qwen3-vl-8b-thinking',
'qwen/qwen3-vl-8b',
],
}
DEFAULT_API_KEY=os.getenv('API_KEY')
# Задаем выходную директорию
OUTPUT_PATH='outputs'
GLUED_AUDIO_FILENAME='glued.mp3'
DEFAULT_SYSTEM_PROMPT='''You are a diligent university student who has recorded a lecture as an audio file and later transcribed it into raw text.
Your task is to rewrite this unstructured transcript into a clear, logically organized, and detailed lecture summary (lecture notes).
@@ -88,6 +44,5 @@ Final Output: A cohesive, detailed, and well-structured lecture summary, suitabl
Use only russian language!
USE LATEX IN DOLLAR SIGN ($)!
EXTRA BIG LENTH OF CONSPECT!
MAKE AS LONG AS POSIBLE AND AS BE GOOD!
'''

View File

View File

@@ -1,31 +0,0 @@
from pathlib import Path
# Для аннотации типов
from markdown_pdf import MarkdownPdf
from pydub import AudioSegment
class FileHandlers:
# Функция сохранения файла
def saveFile(self, filename, content, output_path, format='mp3'):
'''
Сохраняет текст, pdf из markdown_pdf или склеенный аудиофайл в файл с указанным названием и директорией.
Args:
:param filename: название файла;
:param content: содержание файла;
:param output_path: выходная диретория файла.
'''
# Создание объекта директории
directory = Path(output_path)
filePath = directory / filename # Добавление пути директории
filePath.parent.mkdir(parents=True, exist_ok=True) # Создание директории если не существует
# Сохранение для разных типов
if type(content) == MarkdownPdf:
return content.save(filePath)
elif type(content) == str:
return filePath.write_text(content, encoding='utf-8')
elif type(content) == AudioSegment:
return content.export(filePath, format=format)

View File

@@ -1,61 +0,0 @@
import subprocess
from pathlib import Path
class GlueAudio():
def glue(self, audio_files: list, output_path: str, output_filename: str) -> Path:
"""
Склеивает аудиофайлы РАЗНЫХ форматов с помощью FFmpeg и filter_complex.
Это универсальный и эффективный по памяти метод.
Args:
audio_files (list): Список путей к исходным аудиофайлам.
output_path (str): Директория для сохранения итогового файла.
output_filename (str): Имя итогового склеенного файла.
Returns:
Path: Путь к созданному склеенному файлу.
"""
output_dir = Path(output_path)
output_dir.mkdir(parents=True, exist_ok=True)
final_audio_path = output_dir / output_filename
if not audio_files:
raise ValueError("Список аудиофайлов для склейки пуст.")
# 1. Формируем часть команды с входными файлами (-i file1 -i file2 ...)
input_args = []
for file_path in audio_files:
input_args.extend(['-i', str(Path(file_path).resolve())])
# 2. Формируем строку для filter_complex
num_files = len(audio_files)
stream_specifiers = "".join([f"[{i}:a]" for i in range(num_files)])
filter_complex_str = f"{stream_specifiers}concat=n={num_files}:v=0:a=1[outa]"
# 3. Собираем полную команду
command = [
'ffmpeg',
*input_args, # Распаковываем список входных файлов
'-filter_complex', filter_complex_str,
'-map', '[outa]',
'-c:a', 'libmp3lame',
'-q:a', '2',
str(final_audio_path),
'-y'
]
try:
# 4. Выполняем команду
print(f"Выполнение команды FFmpeg: {' '.join(command)}")
subprocess.run(command, check=True, capture_output=True, text=True)
print("FFmpeg успешно завершил склейку.")
except FileNotFoundError:
raise FileNotFoundError("FFmpeg не найден. Убедитесь, что он установлен и доступен в системной переменной PATH.")
except subprocess.CalledProcessError as e:
print("Ошибка при выполнении FFmpeg!")
print("Stderr:", e.stderr)
raise RuntimeError(f"Ошибка FFmpeg при склейке файлов: {e.stderr}")
return final_audio_path

View File

@@ -1,87 +0,0 @@
from config import LLM_MODELS, GEMINI_API_KEY, IO_API_KEY
import gradio as gr
class GradioHandlers:
def __init__(self, llm_factory, ConvertMdToPdf, FileHandlers, FasterWhisper, GlueAudio):
self.fh = FileHandlers()
self.ga = GlueAudio()
self.ConvertMdToPdf = ConvertMdToPdf()
self.FasterWhisper = FasterWhisper()
self.llm_factory = llm_factory
def handleRecognizeBtn(self, audioFiles, model, device, compute_type, beamSize, vadFilter,
minSilenceDurationMs, speechPadMs, temp0, temp1, temp2,
wordTimestamps, noSpeechThreshold, conditionOnPreviousText, filename, outPath):
try:
glued_audio_path = self.ga.glue(
audio_files=[f.name for f in audioFiles],
output_path=outPath,
output_filename=filename
)
except (FileNotFoundError, RuntimeError) as e:
gr.Warning(str(e))
return ""
return self.FasterWhisper.recognize(model, device, compute_type, str(glued_audio_path), beamSize, vadFilter, minSilenceDurationMs, speechPadMs, temp0, temp1, temp2, wordTimestamps, noSpeechThreshold, conditionOnPreviousText)
# Добавил аргумент custom_base_url в конец
def generateByCondition(self, api_key, llm_provider,
llm_model, system_prompt, recognized_text,
llm_temperature, is_pipeline_enabled, trigger,
isSaveFile, filename, filenamePdf, output_path, custom_base_url):
try:
if llm_provider == "Custom":
# Передаем base_url только для Custom
provider = self.llm_factory(llm_provider, api_key, base_url=custom_base_url)
else:
provider = self.llm_factory(llm_provider, api_key)
except ValueError as e:
gr.Warning(str(e))
return gr.skip(), gr.skip()
def process():
# Добавлена обработка ошибок генерации
try:
result, md = provider.generate(llm_model, system_prompt, recognized_text, llm_temperature)
except Exception as e:
raise gr.Error(f"Ошибка генерации LLM: {e}")
pdf, unicodeText = self.ConvertMdToPdf.convertLatexToText(md)
if isSaveFile:
self.fh.saveFile(filenamePdf, pdf, output_path)
self.fh.saveFile(filename, result, output_path)
return result, unicodeText
if (is_pipeline_enabled and trigger == "change") or (not is_pipeline_enabled and trigger == "click"):
return process()
return gr.skip(), gr.skip()
def update_model_dropdown(self, provider):
models = LLM_MODELS.get(provider, [])
default_value = models[0] if models else None
# Обновляем список моделей и настройки поля API Key
if provider == 'io.net':
return gr.update(choices=models, value=default_value), gr.update(label='API key', value=IO_API_KEY, interactive=True, visible=True)
if provider == 'Gemini':
return gr.update(choices=models, value=default_value), gr.update(label='API key', value=GEMINI_API_KEY, interactive=True, visible=True)
if provider == 'gpt4free':
return gr.update(choices=models, value=default_value), gr.update(label='API key (not required)', value="", interactive=False, visible=True)
if provider == "Custom":
return gr.update(choices=models, value=default_value), gr.update(label='API key (optional)', value="", interactive=True, visible=True) # Для Custom ключ может понадобиться
def updateButton(self, isChecked):
variant = 'secondary' if isChecked else 'primary'
return gr.update(interactive=not isChecked, variant=variant)
def toggle_custom_url(self, provider):
"""Показывает поле Base URL только если выбран Custom"""
return gr.update(visible=(provider == 'Custom'))
def update_custom_url(self, base_url):
return None
def updateTextbox(self, isChecked):
return gr.update(visible=isChecked)

View File

@@ -1,21 +0,0 @@
import os
from PIL import Image, ExifTags
import subprocess
import json
from datetime import datetime
class MetadataHandler:
def get_image_timestamp(self, image_path: str) -> datetime | None:
"""Извлекает метку времени из метаданных изображения, если она доступна."""
try:
image = Image.open(image_path)
exif_data = image._getexif()
if exif_data:
for tag, value in exif_data.items():
decoded_tag = ExifTags.TAGS.get(tag, tag)
if decoded_tag == 'DateTimeOriginal':
return value
return None
except Exception as e:
print(f"Error extracting metadata from image: {e}")
return None

View File

@@ -1,7 +0,0 @@
pandoc "out.md" -o output1310.pdf \
--pdf-engine=xelatex \
-V geometry:margin=2.5cm \
-V fontsize=12pt \
-V mainfont="Times New Roman" \
-V colorlinks=true \
-V linkcolor=blue\

View File

@@ -6,13 +6,6 @@ from markdown_pdf import Section
class ConvertMdToPdf:
# Конвертирует md в pdf
def convertLatexToText(self, text:str):
'''
Функция для конвертации LaTeX в текст;
Args:
:param text: текст содержащий LaTeX.
'''
# Обрабатываем только математические выражения
text = re.sub(
r'\$\$(.*?)\$\$|\$(.*?)\$',

View File

@@ -1,11 +1,7 @@
from faster_whisper import WhisperModel
class FasterWhisper:
def recognize(self, model, device, compute_type,
audioFile, beamSize, vadFilter,
minSilenceDurationMs, speechPadMs,
temp0, temp1, temp2, wordTimestamps,
noSpeechThreshold, conditionOnPreviousText):
def recognize(self, model, device, compute_type, audioFile, beamSize, vadFilter, minSilenceDurationMs, speechPadMs, temp0, temp1, temp2, wordTimestamps, noSpeechThreshold, conditionOnPreviousText):
model = WhisperModel(model, device=device, compute_type=compute_type) # Задаем модель
segments, _ = model.transcribe( # Распознаем текст
@@ -26,9 +22,8 @@ class FasterWhisper:
for seg in segments:
text += f"[{self.format_timestamp(seg.start)} -> {self.format_timestamp(seg.end)}] {seg.text}" + '\n'
print(f"[{self.format_timestamp(seg.start)} -> {self.format_timestamp(seg.end)}] {seg.text}")
return text
return(text)
def format_timestamp(self, seconds: float) -> str:
millis = int(seconds * 1000)

31
services/llm.py Normal file
View File

@@ -0,0 +1,31 @@
import openai
class Llm:
def __init__(self, apiKey:str):
self.client = openai.OpenAI(
api_key=apiKey,
base_url='https://api.intelligence.io.solutions/api/v1/'
)
def generate(self, model:str, systemPrompt:str, userPrompt:str, temp:float):
'''
prompt[user_promtp, system_ptompt]
Function generate text by prompt
'''
# Получаем ответ от нейросети
response = self.client.chat.completions.create(
model=model,
messages=[
{'role': 'system', 'content': systemPrompt},
{'role': 'user', 'content': userPrompt},
],
temperature=temp,
stream=False
)
# Достаем текст
text = str(response.choices[0].message.content)
return text, text

View File

@@ -1,27 +0,0 @@
# services/llm_factory.py
from services.llm_providers.ionet_provider import IoNetProvider
from services.llm_providers.gemini_provider import GeminiProvider
from services.llm_providers.gpt4free_provider import Gpt4FreeProvider
from services.llm_providers.base_provider import BaseLLMProvider
from services.llm_providers.custom_provider import CustomProvider
def get_llm_provider(provider_name: str, api_key: str | None = None, base_url: str | None = None) -> BaseLLMProvider:
"""
Фабричная функция для получения экземпляра провайдера LLM.
"""
if provider_name == 'io.net':
if not api_key:
raise ValueError("API ключ обязателен для io.net")
return IoNetProvider(api_key)
elif provider_name == 'Gemini':
if not api_key:
raise ValueError("API ключ обязателен для Gemini")
return GeminiProvider(api_key)
elif provider_name == 'gpt4free':
return Gpt4FreeProvider()
elif provider_name == 'Custom':
if not base_url:
raise ValueError("Base URL обязателен для Custom провайдера")
return CustomProvider(api_key, base_url) # base_url будет установлен позже
else:
raise ValueError(f"Неизвестный провайдер: {provider_name}")

View File

@@ -1,18 +0,0 @@
from abc import ABC, abstractmethod
class BaseLLMProvider(ABC):
"""
Абстрактный базовый класс для всех провайдеров LLM.
Каждый провайдер должен реализовать метод generate.
"""
def __init__(self, api_key: str | None = None):
self.api_key = api_key
@abstractmethod
def generate(self, model: str, system_prompt: str, user_prompt: str, temp: float):
"""
Основной метод для генерации текста.
Должен возвращать кортеж из двух строк: (чистый_текст, markdown_текст)
"""
pass

View File

@@ -1,44 +0,0 @@
import requests
from .base_provider import BaseLLMProvider
class CustomProvider(BaseLLMProvider):
def __init__(self, api_key: str | None = None, base_url: str | None = None):
super().__init__(api_key)
self.base_url = base_url or 'http://127.0.0.1:1234/v1/'
# ГАРАНТИРУЕМ наличие слеша в конце URL
if not self.base_url.endswith('/'):
self.base_url += '/'
def generate(self, model: str, system_prompt: str, user_prompt: str, temp: float):
url = f"{self.base_url}chat/completions"
# Некоторые Custom провайдеры (как vLLM или Ollama) могут требовать API Key, даже если он фиктивный
headers = {"Content-Type": "application/json"}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
data = {
"model": model,
"messages": [
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': user_prompt},
],
"temperature": temp,
"stream": False
}
try:
response = requests.post(url, headers=headers, json=data, timeout=300)
response.raise_for_status()
result = response.json()
# Обработка разных форматов ответа (на всякий случай)
if 'choices' in result and len(result['choices']) > 0:
text = str(result['choices'][0]['message']['content'])
return text, text
else:
return f"Неожиданный ответ от сервера: {result}", str(result)
except requests.exceptions.RequestException as e:
return f"Ошибка при запросе к CustomProvider API: {e}", f"Ошибка: {e}"

View File

@@ -1,74 +0,0 @@
# services/llm_providers/gemini_provider.py
import requests
from .base_provider import BaseLLMProvider
class GeminiProvider(BaseLLMProvider):
"""
Провайдер для Google Gemini, использующий прямые REST API вызовы
через библиотеку requests для надежной работы с SOCKS-прокси.
"""
def __init__(self, api_key: str):
super().__init__(api_key)
self.base_url = "https://generativelanguage.googleapis.com/v1beta/models/"
def generate(self, model: str, system_prompt: str, user_prompt: str, temp: float):
"""
Генерирует текст с помощью модели Gemini, отправляя запрос через прокси.
"""
# 1. Формируем URL для запроса
api_url = f"{self.base_url}{model}:generateContent?key={self.api_key}"
# 2. Задаем настройки прокси из вашего примера
# socks5h:// означает, что DNS-запросы также будут идти через прокси
proxies = {
'http': 'socks5://192.168.1.6:2080',
'https': 'socks5h://192.168.1.6:2080'
}
# 3. Собираем тело запроса (payload) в формате, который ожидает Gemini API
data = {
"system_instruction": {
"parts": {"text": system_prompt}
},
"contents": [{
"parts": [{"text": user_prompt}]
}],
"generationConfig": {
"temperature": temp
}
}
try:
# 4. Отправляем POST-запрос с данными и настройками прокси
response = requests.post(api_url, json=data, proxies=proxies, timeout=400)
# Проверяем, не вернул ли сервер ошибку (например, 4xx или 5xx)
response.raise_for_status()
# 5. Парсим JSON-ответ и извлекаем сгенерированный текст
response_json = response.json()
# Добавим проверку на случай, если контент был заблокирован
if "candidates" not in response_json or not response_json["candidates"]:
block_reason = response_json.get("promptFeedback", {}).get("blockReason", "неизвестная причина")
error_message = f"Контент заблокирован. Причина: {block_reason}"
return error_message, error_message
text = response_json["candidates"][0]["content"]["parts"][0]["text"]
return text, text
except requests.exceptions.ProxyError as e:
error_message = f"Ошибка подключения к прокси. Убедитесь, что Nekobox запущен и слушает порт 2080. Ошибка: {e}"
print(error_message)
return error_message, error_message
except requests.exceptions.RequestException as e:
# Ловим все остальные ошибки requests (таймаут, проблемы с сетью и т.д.)
error_message = f"Произошла ошибка при обращении к API Gemini: {e}"
print(error_message)
return error_message, error_message
except (KeyError, IndexError) as e:
# Ловим ошибки, если структура JSON-ответа неожиданная
error_message = f"Не удалось разобрать ответ от API Gemini. Структура ответа изменилась. Ошибка: {e}"
print(error_message)
return error_message, error_message

View File

@@ -1,47 +0,0 @@
# services/llm_providers/gpt4free_provider.py
from g4f.client import Client
from .base_provider import BaseLLMProvider
class Gpt4FreeProvider(BaseLLMProvider):
"""
Провайдер для работы с моделью GPT через библиотеку gpt4free.
Этот класс реализует интерфейс BaseLLMProvider и предоставляет возможность
взаимодействия с различными LLM через сервис gpt4free, который не требует
API ключа для работы.
"""
# gpt4free не требует API ключа
def __init__(self, api_key: str | None = None):
super().__init__(api_key)
self.client = Client()
def generate(self, model: str, system_prompt: str, user_prompt: str, temp: float):
"""
Генерирует ответ от модели GPT с использованием gpt4free.
Args:
model (str): Название модели для генерации ответа
system_prompt (str): Системное сообщение для контекста
user_prompt (str): Пользовательский запрос
temp (float): Температура генерации ( controls randomness of responses)
Returns:
tuple: Кортеж из двух одинаковых строк - сгенерированного ответа и его копии
"""
# temp в g4f может работать не для всех внутренних провайдеров
try:
response = self.client.chat.completions.create(
model=model, # Пример модели, может варьироваться в зависимости от доступности провайдеров
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=temp
)
text = response.choices[0].message.content
return text, text
except Exception as e:
error_message = f"Ошибка при работе с gpt4free: {e}"
print(error_message)
return error_message, error_message

View File

@@ -1,34 +0,0 @@
import requests
from .base_provider import BaseLLMProvider
class IoNetProvider(BaseLLMProvider):
def __init__(self, api_key: str):
super().__init__(api_key)
self.base_url = 'https://api.intelligence.io.solutions/api/v1'
def generate(self, model: str, system_prompt: str, user_prompt: str, temp: float):
url = f"{self.base_url}/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
data = {
"model": model,
"messages": [
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': user_prompt},
],
"temperature": temp
}
try:
response = requests.post(url, headers=headers, json=data)
response.raise_for_status()
result = response.json()
text = str(result['choices'][0]['message']['content'])
return text, text # Возвращаем как чистый текст, так и Markdown
except requests.exceptions.RequestException as e:
raise Exception(f"Ошибка при запросе к IO.net API: {e}")