פתיחת שלוחות מרובות עם קבצי שמע / טקסט
-
יצירת שלוחות מרובות
שלום לכולם!
ב"ה יצא לי לעשות סקריפט פשוט, שעוזר ליצור ולהעלות שלוחות למערכת ימות המשיח בקלות, כולל יצירת קבצי שמע (TTS) וקובצי הגדרות (ext.ini) אוטומטית.שלב אחד:
מה הסקריפט עושה?
הסקריפט קורא קובץ אקסל פשוט, ועל פיו:- יוצר שלוחות ותתי שלוחות במערכת ימות המשיח.
- מעלה קבצי הגדרה (ext.ini) אוטומטית.
- ממיר טקסט לקבצי שמע (TTS) בעברית ומעלה אותם לשלוחה לפי הנתיב
- מעלה קובצי טקסט/TTS רגילים.
- הוא דואג להתקנה אוטומטית של תוכנת FFmpeg.
שלב שתיים:
מכינים את הקבצים...
קובץ הסקריפט: שמרו את קוד הפייתון שפה (בספוילר) בקובץ בשם make_audio.py (או כל שם אחר עם סיומת .py).import pandas as pd import requests from requests_toolbelt.multipart.encoder import MultipartEncoder import os import asyncio import edge_tts import subprocess import json import urllib.request import zipfile import shutil import warnings warnings.filterwarnings("ignore") DEFAULT_TTS_VOICE = "he-IL-AvriNeural" # נגדיר את FFMPEG_EXECUTABLE כ-None בהתחלה, ורק נאתחל אותו כשנצטרך FFMPEG_EXECUTABLE = None def read_excel_data(excel_file_path): print(f"📖 קורא קובץ אקסל: {excel_file_path}") try: df = pd.read_excel(excel_file_path, header=None) username = str(df.iloc[0, 1]).strip() password = str(df.iloc[1, 1]).strip() if not username: print("❌ שגיאה: אנא הכנס מספר מערכת (תא B1 באקסל).") return None, None, None if not password: print("❌ שגיאה: אנא הכנס סיסמא (תא B2 באקסל).") return None, None, None extensions = df.iloc[5:, [0, 1, 2, 3, 4]].dropna(how='all') extensions.columns = ['נתיב', 'הגדרה', 'שם השלוחה', 'קובץ שמע להעלאה', 'קובץ טקסט להעלאה'] extensions = extensions.apply(lambda x: x.astype(str).str.strip().replace('nan', '') if x.dtype == 'object' else x) print(f"✔️ נקראו בהצלחה: משתמש={username}, סיסמה={'*' * len(password)}") print(f"📋 נתוני השלוחות שנקראו:\n{extensions.to_string()}") return username, password, extensions.to_dict('records') except FileNotFoundError: print(f"❌ שגיאה: קובץ האקסל '{excel_file_path}' לא נמצא. ודא שהוא באותה תיקייה כמו הסקריפט.") return None, None, None except Exception as e: print(f"❌ שגיאה בקריאת קובץ האקסל: {e}") return None, None, None def create_ext_ini_file(extension_type, extension_name, output_file_name, is_subpath=False, direct_ini_content=None): content = "" if direct_ini_content: content = direct_ini_content.strip() if not content.endswith('\n'): content += '\n' print(f"⚙️ יוצר קובץ ext.ini מתוכן INI ישיר.") elif is_subpath: content = "type=menu\n" else: normalized_extension_type = str(extension_type).strip().lower() if extension_type else "" if pd.isna(extension_type): normalized_extension_type = "" if normalized_extension_type == 'השמעת קבצים': content = "type=playfile\n" elif normalized_extension_type == 'תפריט': content = "type=menu\n" elif normalized_extension_type in ['הקלטה', 'הקלטות']: content = "type=record\n" else: print(f"⚠️ סוג שלוחה לא מוכר '{extension_type}'. מוגדר כ-'תפריט' כברירת מחדל.") content = "type=menu\n" if extension_name and extension_name.strip(): content += f"title={extension_name.strip()}\n" try: with open(output_file_name, 'w', encoding='utf-8') as f: f.write(content) print(f"✅ נוצר קובץ {output_file_name} עם תוכן: {content.strip() or '[ריק]'}") return True except Exception as e: print(f"❌ שגיאה ביצירת קובץ {output_file_name}: {e}") return False async def create_and_convert_audio_file(text, output_wav_name): output_mp3_temp = f"temp_{os.path.splitext(output_wav_name)[0]}.mp3" print(f"🔊 יוצר קובץ שמע {output_wav_name} מטקסט: '{text}'") # הבדיקה והאיתחול של FFMPEG_EXECUTABLE מתרחשים כאן בלבד global FFMPEG_EXECUTABLE if not FFMPEG_EXECUTABLE or not os.path.exists(FFMPEG_EXECUTABLE): print(f"⏳ מנסה להוריד או לאתר את FFmpeg כיוון שנדרש לטיפול בקבצי שמע...") ensure_ffmpeg() # קוראים ל-ensure_ffmpeg רק כאן if not FFMPEG_EXECUTABLE or not os.path.exists(FFMPEG_EXECUTABLE): print(f"❌ לא ניתן ליצור קובץ שמע {output_wav_name} כי FFmpeg לא זמין.") return False try: comm = edge_tts.Communicate(text, voice=DEFAULT_TTS_VOICE) await comm.save(output_mp3_temp) print(f"✅ קובץ אודיו זמני נוצר בהצלחה: {output_mp3_temp}") result = subprocess.run( [FFMPEG_EXECUTABLE, "-loglevel", "error", "-y", "-i", output_mp3_temp, "-ar", "8000", "-ac", "1", "-acodec", "pcm_s16le", output_wav_name], check=True ) print(f"✅ קובץ WAV סופי נוצר בהצלחה: {output_wav_name}") return True except edge_tts.exceptions.NoAudioReceived as e: print(f"❌ שגיאה ביצירת קובץ שמע {output_wav_name}: {e}") print("❗ ודא/י חיבור אינטרנט תקין, אין חסימות חומת אש/אנטי-וירוס, או נסה/י קול אחר.") except subprocess.CalledProcessError as e: print(f"❌ שגיאה בהמרת אודיו (FFmpeg) עבור {output_wav_name}: {e}. ודא/י ש-FFmpeg מותקן ונגיש.") except FileNotFoundError: print(f"❌ שגיאה: FFmpeg לא נמצא בנתיב המוגדר. ודא/י שהורד והוגדר כראוי.") except Exception as e: print(f"❌ שגיאה כללית ביצירת/המרת אודיו עבור {output_wav_name}: {e}") finally: if os.path.exists(output_mp3_temp): os.remove(output_mp3_temp) return False def create_text_file(text, output_file_name): print(f"📝 יוצר קובץ טקסט {output_file_name} עם תוכן: {text}") try: with open(output_file_name, 'w', encoding='utf-8') as f: f.write(text) print(f"✅ קובץ טקסט {output_file_name} נוצר בהצלחה") return True except Exception as e: print(f"❌ שגיאה ביצירת קובץ טקסט {output_file_name}: {e}") return False def upload_file_to_yemot(file_path, yemot_full_path, username, password): token = f"{username}:{password}" file_ext = file_path.lower() if file_ext.endswith('.wav'): content_type = 'audio/wav' elif file_ext.endswith('.txt') or file_ext.endswith('.tts'): content_type = 'text/plain' elif file_ext.endswith('.ini'): content_type = 'text/plain' else: content_type = 'application/octet-stream' try: with open(file_path, 'rb') as f_read: m = MultipartEncoder(fields={ "token": token, "path": yemot_full_path, "upload": (os.path.basename(file_path), f_read, content_type) }) print(f"⬆️ מעלה קובץ {os.path.basename(file_path)} לנתיב: {yemot_full_path}") r = requests.post("https://www.call2all.co.il/ym/api/UploadFile", data=m, headers={'Content-Type': m.content_type}) r.raise_for_status() print(f"✔️ הקובץ '{os.path.basename(file_path)}' הועלה בהצלחה") return True except FileNotFoundError: print(f"❌ שגיאה: קובץ מקור להעלאה לא נמצא: {file_path}") return False except requests.exceptions.RequestException as e: print(f"❌ שגיאה בהעלאת קובץ {os.path.basename(file_path)}: {e}") print(f"📤 תגובת שרת: {r.text if 'r' in locals() else 'אין תגובה מהשרת'}") return False except Exception as e: print(f"❌ שגיאה בלתי צפויה בהעלאת קובץ {os.path.basename(file_path)}: {e}") return False def generate_subpaths(full_path): parts = [p for p in full_path.split('/') if p] subpaths = [] current_path_parts = [] for part in parts[:-1]: current_path_parts.append(part) subpaths.append('/'.join(current_path_parts)) return subpaths def test_yemot_credentials(username, password): token = f"{username}:{password}" dummy_file_name = "temp_cred_test.txt" dummy_file_path_local = os.path.join(os.getcwd(), dummy_file_name) yemot_target_path = "ivr2:/temp_credential_test_folder/test_file.txt" try: with open(dummy_file_path_local, "w") as f: f.write("test_content") print("⏳ בודק שם משתמש וסיסמא מול מערכת ימות המשיח...") with open(dummy_file_path_local, 'rb') as f_read_dummy: m = MultipartEncoder(fields={ "token": token, "path": yemot_target_path, "upload": (dummy_file_name, f_read_dummy, 'text/plain') }) r = requests.post("https://www.call2all.co.il/ym/api/UploadFile", data=m, headers={'Content-Type': m.content_type}) response_text = r.text.strip() response_text_upper = response_text.upper() if r.status_code == 401 or r.status_code == 403: print("❌ שם משתמש ו/או סיסמא לא תקין. שגיאת אימות (Unauthorized/Forbidden).") return False try: response_json = json.loads(response_text) if response_json.get("responseStatus") == "OK" and response_json.get("success") == True: print("✅ שם משתמש וסיסמא תקינים.") return True elif "TOKEN INVALID" in response_text_upper or \ "המספר אינו מורשה" in response_text_upper or \ "THE NUMBER IS NOT VALID" in response_text_upper or \ "מספר המערכת אינו תקין" in response_text_upper: print("❌ שם משתמש ו/או סיסמא לא תקין.") return False else: print(f"❌ שם משתמש ו/או סיסמא לא תקין. תגובת שרת לא צפויה: {response_text}") return False except json.JSONDecodeError: if "ERROR:" in response_text_upper: if "TOKEN INVALID" in response_text_upper or \ "המספר אינו מורשה" in response_text_upper or \ "THE NUMBER IS NOT VALID" in response_text_upper or \ "מספר המערכת אינו תקין" in response_text_upper: print("❌ שם משתמש ו/או סיסמא לא תקין.") return False else: print(f"❌ שם משתמש ו/או סיסמא לא תקין. שגיאת שרת לא ספציפית: {response_text}") return False else: print(f"❌ שם משתמש ו/או סיסמא לא תקין. תגובת שרת בפורמט לא צפוי: {response_text}") return False except requests.exceptions.RequestException as e: print(f"❌ שגיאת חיבור או תקשורת בעת בדיקת התחברות: {e}") print("❗ ודא/י חיבור אינטרנט תקין.") return False except Exception as e: print(f"❌ שגיאה בלתי צפויה בעת בדיקת התחברות: {e}") return False finally: if os.path.exists(dummy_file_path_local): try: os.remove(dummy_file_path_local) except OSError as e: print(f"⚠️ אזהרה: לא ניתן למחוק את קובץ הדמה '{dummy_file_path_local}': {e}") def ensure_ffmpeg(): global FFMPEG_EXECUTABLE local_ffmpeg_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bin", "ffmpeg.exe") if os.path.exists(local_ffmpeg_path): print("⏩ ffmpeg כבר קיים בנתיב המקומי.") FFMPEG_EXECUTABLE = local_ffmpeg_path return print("⬇️ מוריד ומגדיר ffmpeg...") os.makedirs(os.path.dirname(local_ffmpeg_path), exist_ok=True) url = "https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-essentials.zip" archive_path = os.path.join(os.path.dirname(local_ffmpeg_path), "ffmpeg.zip") try: print(f"📥 מוריד קובץ FFmpeg מ: {url}") urllib.request.urlretrieve(url, archive_path) print(f"📦 חולץ את קובץ FFmpeg מ: {archive_path}") with zipfile.ZipFile(archive_path, 'r') as zip_ref: found_ffmpeg_in_zip = False for member in zip_ref.namelist(): if member.endswith('ffmpeg.exe'): source = zip_ref.open(member) target = open(local_ffmpeg_path, "wb") with source, target: shutil.copyfileobj(source, target) print(f"✅ FFmpeg הוכן בהצלחה ב: {local_ffmpeg_path}") FFMPEG_EXECUTABLE = local_ffmpeg_path found_ffmpeg_in_zip = True break if not found_ffmpeg_in_zip: print("❌ שגיאה: לא נמצא ffmpeg.exe בתוך קובץ ה-ZIP שהורד.") FFMPEG_EXECUTABLE = None return except Exception as e: print(f"❌ שגיאה בהורדה או בחילוץ של FFmpeg: {e}") print("❗ ודא/י חיבור אינטרנט תקין ונסה/י שוב.") FFMPEG_EXECUTABLE = None finally: if os.path.exists(archive_path): try: os.remove(archive_path) except OSError as e: print(f"⚠️ אזהרה: לא ניתן למחוק את קובץ ה-ZIP הזמני '{archive_path}': {e}") if not os.path.exists(local_ffmpeg_path): print("❌ FFmpeg לא הותקן אוטומטית. אנא הורד/י והתקן/י אותו ידנית.") print("ניתן להוריד מכאן: https://www.gyan.dev/ffmpeg/builds/ (בחר/י את גרסת Essentials או Full עבור Windows).") FFMPEG_EXECUTABLE = None async def main(): print("🚀 מתחיל תהליך יצירת והעלאת שלוחות...") # ensure_ffmpeg() הוסר מכאן - יופעל רק במידת הצורך # if not FFMPEG_EXECUTABLE or not os.path.exists(FFMPEG_EXECUTABLE): # print("⚠️ התהליך נעצר עקב חוסר ב-FFmpeg. אנא התקן/י אותו ידנית כדי להמשיך.") # return script_dir = os.path.dirname(os.path.abspath(__file__)) excel_file = os.path.join(script_dir, "משני.xlsx") print(f"📂 תיקיית הסקריפט: {script_dir}") print(f"🔍 מחפש קובץ אקסל: {excel_file}") if not os.path.exists(excel_file): print(f"❌ קובץ האקסל '{excel_file}' לא נמצא בתיקיית הסקריפט!") print(f"📋 קבצים בתיקייה: {os.listdir(script_dir)}") return username, password, extensions_data = read_excel_data(excel_file) if not username or not password or not extensions_data: print("⚠️ התהליך נעצר עקב חוסר בנתוני התחברות או נתוני שלוחות.") return if not test_yemot_credentials(username, password): print("⚠️ התהליך נעצר עקב פרטי התחברות לא תקינים.") return processed_subfolders = set() for ext in extensions_data: path = ext.get('נתיב') extension_type = ext.get('הגדרה') extension_name = ext.get('שם השלוחה') audio_content = ext.get('קובץ שמע להעלאה', '') text_content = ext.get('קובץ טקסט להעלאה', '') if isinstance(path, str): path = path.strip() if isinstance(extension_type, str): extension_type = extension_type.strip() if isinstance(extension_name, str): extension_name = extension_name.strip() if isinstance(audio_content, str): audio_content = audio_content.strip().replace('...', '') if isinstance(text_content, str): text_content = text_content.strip().replace('...', '') if not path: print(f"⚠️ דילוג על שלוחה: נתיב לא תקין או חסר '{path}'") continue clean_path = path print(f"\n📋 מעבד שלוחה: {extension_name} ({path})") subpaths_to_create = generate_subpaths(clean_path) for subpath_item in subpaths_to_create: full_sub_yemot_ini_path = f"ivr2:/{subpath_item}/ext.ini" if full_sub_yemot_ini_path not in processed_subfolders: print(f"⚙️ מעבד תת-שלוחה (ברירת מחדל menu): {full_sub_yemot_ini_path}") ini_file_name = "ext.ini" if create_ext_ini_file(None, None, ini_file_name, is_subpath=True): if not upload_file_to_yemot(ini_file_name, full_sub_yemot_ini_path, username, password): print(f"❌ שגיאה בהעלאת INI לתת-שלוחה {full_sub_yemot_ini_path}. המשך עיבוד...") else: print(f"❌ שגיאה ביצירת INI לתת-שלוחה {full_sub_yemot_ini_path}. מדלג.") if os.path.exists(ini_file_name): os.remove(ini_file_name) processed_subfolders.add(full_sub_yemot_ini_path) main_ini_file = "ext.ini" target_yemot_ini_path = f"ivr2:/{clean_path}/ext.ini" should_create_main_ext_ini = True is_direct_ini_input = False ini_content_for_direct_upload = "" if isinstance(extension_type, str) and extension_type.strip(): if ("type=" in extension_type.strip().lower() and "=" in extension_type.strip()) or "\n" in extension_type.strip(): is_direct_ini_input = True ini_content_for_direct_upload = extension_type.strip() if is_direct_ini_input: if not create_ext_ini_file(None, None, main_ini_file, direct_ini_content=ini_content_for_direct_upload): print(f"❌ שגיאה ביצירת INI ראשי מתוכן ישיר עבור {extension_name}. מדלג על שלוחה זו.") continue else: normalized_ext_type_for_check = str(extension_type).strip().lower() if extension_type else "" if pd.isna(extension_type): normalized_ext_type_for_check = "" if normalized_ext_type_for_check == 'תיקייה' or not normalized_ext_type_for_check: print(f"🗂️ שלוחה מסוג '{normalized_ext_type_for_check or 'ריק'}'. לא נוצר קובץ ext.ini ראשי.") should_create_main_ext_ini = False else: if not create_ext_ini_file(extension_type, extension_name, main_ini_file): print(f"❌ שגיאה ביצירת INI ראשי עבור {extension_name}. מדלג על שלוחה זו.") continue if should_create_main_ext_ini: if not upload_file_to_yemot(main_ini_file, target_yemot_ini_path, username, password): print(f"❌ שגיאה בהעלאת INI ראשי עבור {extension_name}. המשך עיבוד...") if os.path.exists(main_ini_file): os.remove(main_ini_file) continue if os.path.exists(main_ini_file): os.remove(main_ini_file) # השינוי העיקרי: בדיקת audio_content לפני ניסיון יצירת קובץ שמע if audio_content and audio_content.strip(): audio_texts = [t.strip() for t in audio_content.split(' + ') if t.strip()] for i, text_to_tts in enumerate(audio_texts): audio_wav_final = f"{str(i).zfill(3)}.wav" # קריאה ל-create_and_convert_audio_file תטפל באיתחול FFmpeg אם נדרש if await create_and_convert_audio_file(text_to_tts, audio_wav_final): audio_yemot_path = f"ivr2:/{clean_path}/{audio_wav_final}" upload_file_to_yemot(audio_wav_final, audio_yemot_path, username, password) else: print(f"⚠️ דילוג על העלאת קובץ שמע {audio_wav_final} עקב שגיאה ביצירה/המרה.") if os.path.exists(audio_wav_final): os.remove(audio_wav_final) # קבצי טקסט אינם דורשים את FFmpeg, אז הם ממשיכים לפעול כרגיל if text_content and text_content.strip() and (extension_type and pd.notna(extension_type)): text_files_list = [t.strip() for t in text_content.split(' + ') if t.strip()] for i, item_text_content in enumerate(text_files_list): if os.path.exists(item_text_content): file_name_to_upload = os.path.basename(item_text_content) print(f"📄 מעלה קובץ טקסט קיים: {file_name_to_upload}") text_yemot_path = f"ivr2:/{clean_path}/{file_name_to_upload}" upload_file_to_yemot(item_text_content, text_yemot_path, username, password) else: text_file_name_final = f"{str(i).zfill(3)}.tts" if create_text_file(item_text_content, text_file_name_final): text_yemot_path = f"ivr2:/{clean_path}/{text_file_name_final}" upload_file_to_yemot(text_file_name_final, text_yemot_path, username, password) else: print(f"⚠️ דילוג על העלאת קובץ טקסט {text_file_name_final} עקב שגיאה ביצירה.") if os.path.exists(text_file_name_final): os.remove(text_file_name_final) print("\n✅ סיום עיבוד כל השלוחות בהצלחה!") if __name__ == "__main__": asyncio.run(main())
קובץ האקסל: ניתן להוריד מפה משני.xlsx
חשוב מאוד: שימרו על מבנה האקסל כמו בדוגמה המקורית שקיבלתם.
חשוב מאוד ששתי הקבצים (הקוד להרצה והקובץ אקסל) יהיו באותה תיקייה על המחשב!!שלב שלוש:
פשוט תמלאו את הקובץ אקסל לפי השלוחות שתרצו לייצר לדוגמא:כמה כללים בנוגע למילוי השדות:
-
השדות שהם חובה זה - למעלה בשם משתמש והסיסמה, חוץ מזה הכל יכול להישאר ריק - אם "הגדרות שלוחה" נשאר ריק - זה מייצר תיקייה בלבד בלי שום הגדרות.
-
כשממלאים הגדרות שלוחה בעיקרון צריך להכניס בשדה את ההגדרות של השלוחה אבל אם זה שלוחה - או תפריט או השמעת קבצים או הקלטה - אפשר להכניס את המילים "הקלטה" או "השמעת קבצים" או "תפריט" וזה ייצר לבד את הגדרות השלוחה.
-
אם רוצים למשל לעשות מאה תיקיות / שלוחות - לא צריך למלאות מאה שדות אפשר פשוט לעשות שתי שדות למשל:
1/1/1
1/1/2
ואז לגרור כלפי מטה. -
במידה ומגדירים ב"נתיב" למשל שלוחה 1/1/1 ומגדירים ב"הגדרות השלוחה" שיהיה השמעת קבצים או כל הגדרה אחרת - נוצר שלוחה בתוך שלוחה בתוך שלוחה - ומכיון שרק השלוחה האחרונה הוגדרה - ברירת מחדל שהשלוחות מוגדרות ל"תפריט".
-
ב"העלאת קובץ שמע / טקסט - ברירת מחדל נוצר קובץ בשם 000 אפשר לייצר כמה קבצים בכל שלוחה על ידי הפרדה של + לדוגמה "שלום וברוכים הבאים לקו המידע + לתכנים נוספים הקש כוכבית"
כך נוצרים שתי קבצים 000, 001.
שלב ארבע:
שימו לב לפני שמריצים את הקוד צריך לוודא שפייתון מותקן על המחשב
אפשר להתקין עם הפקודה הזו בשורת הפקודהwinget install Python.Python.3.12 --source winget
לאחר מכן תריצו את הפקודה הזו
pip install pandas requests requests-toolbelt edge-tts
ואז לגרור את הקובץ פייתון לתוך השורת פקודה (או להעתיק את הנתיב שלו) וכמובן - אנטר!
השימוש באחריות המשתמש בלבד!
אם מישהו מסתבך עם ההרצה של הקוד אשמח לעזור
אשמח לשמוע שזה עזר למישהו... ואפילו למישהו אחד... דיינו! -
-