112 lines
5.3 KiB
Python
112 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
||
import os
|
||
import sys
|
||
import time
|
||
import google.generativeai as genai
|
||
|
||
# Setup proxies for Google API routing through HTTP proxy (port 10809 is supported by gRPC)
|
||
os.environ["https_proxy"] = "http://127.0.0.1:10809"
|
||
os.environ["http_proxy"] = "http://127.0.0.1:10809"
|
||
os.environ["all_proxy"] = "http://127.0.0.1:10809"
|
||
|
||
# Configure Gemini with the active AI Studio key
|
||
API_KEY = "AIzaSyAJMfQ_mRZ1picRXVFNaeFJ7fpAkZxUKyk"
|
||
genai.configure(api_key=API_KEY)
|
||
|
||
BOT_TOKEN = "8725618164:AAH1tGalq-pw1l0t4P5c0sdLkCVJdh7IE7M"
|
||
CHAT_ID = "197957361"
|
||
|
||
def send_tg(msg):
|
||
import urllib.parse
|
||
import urllib.request
|
||
try:
|
||
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
|
||
# Split message if it's too long for TG limit (4096 chars)
|
||
for chunk in [msg[i:i+4000] for i in range(0, len(msg), 4000)]:
|
||
data = urllib.parse.urlencode({"chat_id": CHAT_ID, "text": chunk, "parse_mode": "markdown"}).encode("utf-8")
|
||
req = urllib.request.Request(url, data=data)
|
||
urllib.request.urlopen(req, timeout=15)
|
||
except Exception as e:
|
||
print(f"Failed to send Telegram message: {e}", flush=True)
|
||
|
||
def transcribe_audio(audio_path, output_prefix, model_name="gemini-2.5-flash"):
|
||
if not os.path.exists(audio_path):
|
||
print(f"Error: File not found: {audio_path}", flush=True)
|
||
sys.exit(1)
|
||
|
||
size_mb = os.path.getsize(audio_path) / (1024 * 1024)
|
||
print(f"File size: {size_mb:.2f} MB", flush=True)
|
||
|
||
send_tg(f"🎙 **Транскрибация аудио через Gemini:**\nФайл: `{os.path.basename(audio_path)}` ({size_mb:.1f} MB)\nМодель: `{model_name}`\n\n*Запуск выгрузки в Google API...*")
|
||
|
||
# 1. Upload file
|
||
print(f"Uploading {audio_path} to Google Gemini File API...", flush=True)
|
||
audio_file = genai.upload_file(path=audio_path)
|
||
print(f"File uploaded successfully! API Name: {audio_file.name}", flush=True)
|
||
|
||
# 2. Wait for processing
|
||
start_time = time.time()
|
||
while audio_file.state.name == "PROCESSING":
|
||
print("File is processing on Google servers... waiting 10s", flush=True)
|
||
time.sleep(10)
|
||
audio_file = genai.get_file(name=audio_file.name)
|
||
|
||
if audio_file.state.name == "FAILED":
|
||
send_tg("❌ **Ошибка**: Обработка аудиофайла на сервере Google завершилась неудачей.")
|
||
raise ValueError("Audio processing failed on Google servers.")
|
||
|
||
print("Audio file is active and ready!", flush=True)
|
||
send_tg("⚡️ **Аудиофайл готов.** Запускаю расшифровку речи в Gemini (это может занять до 2-3 минут)...")
|
||
|
||
# 3. Transcribe with model
|
||
model = genai.GenerativeModel(model_name)
|
||
prompt = (
|
||
"Сделай максимально точную расшифровку (транскрибацию) этой аудиозаписи на русском языке. "
|
||
"Пожалуйста, проанализируй голоса и разделяй речь по спикерам (например: 'Спикер 1:', 'Спикер 2:'), если говорят разные люди. "
|
||
"Обязательно расставляй таймкоды реплик или таймкоды каждые 1-2 минуты (например, [04:12]). "
|
||
"Оформи результат в красивом и удобном Markdown-формате с заголовками, абзацами и выделением ключевых моментов."
|
||
)
|
||
|
||
try:
|
||
response = model.generate_content([audio_file, prompt])
|
||
transcript_text = response.text
|
||
|
||
# Save MD file
|
||
md_file = f"{output_prefix}.md"
|
||
with open(md_file, "w", encoding="utf-8") as f:
|
||
f.write(transcript_text)
|
||
|
||
# Save plain text copy
|
||
txt_file = f"{output_prefix}.txt"
|
||
with open(txt_file, "w", encoding="utf-8") as f:
|
||
f.write(transcript_text)
|
||
|
||
elapsed = time.time() - start_time
|
||
print(f"Success! Unpacked in {elapsed:.1f}s", flush=True)
|
||
|
||
# Send transcript to Telegram
|
||
send_tg(f"✅ **Расшифровка завершена!** ({elapsed/60:.1f} мин)\nРезультаты сохранены в:\n- `{md_file}`\n- `{txt_file}`\n\n📝 **Текст расшифровки:**\n\n{transcript_text}")
|
||
|
||
except Exception as e:
|
||
print(f"Error during transcription generation: {e}", flush=True)
|
||
send_tg(f"❌ **Ошибка во время генерации текста:**\n`{e}`")
|
||
|
||
finally:
|
||
# Delete remote file from Google storage
|
||
print("Cleaning up file from Google storage...", flush=True)
|
||
try:
|
||
genai.delete_file(name=audio_file.name)
|
||
print("Cleanup done.", flush=True)
|
||
except Exception as e:
|
||
print(f"Cleanup error: {e}", flush=True)
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) < 3:
|
||
print("Usage: python3 gemini_transcribe.py <audio_path> <output_prefix_path> [model_name]")
|
||
sys.exit(1)
|
||
|
||
audio_path = sys.argv[1]
|
||
output_prefix = sys.argv[2]
|
||
model_name = sys.argv[3] if len(sys.argv) > 3 else "gemini-2.5-flash"
|
||
|
||
transcribe_audio(audio_path, output_prefix, model_name)
|