tahamajs/NS / slides /generate_2x_parallel.py
tahamajs's picture
download
raw
9.93 kB
import os
import re
import asyncio
from bs4 import BeautifulSoup
import edge_tts
# High concurrency semaphore for ultra-fast generation
SEMAPHORE = asyncio.Semaphore(12)
def normalize_text_for_speech(text):
replacements = {
r'\bIPsec\b': 'آی‌پی‌سک',
r'\bIPSec\b': 'آی‌پی‌سک',
r'\bSSL/TLS\b': 'اس‌اس‌ال / تی‌ال‌اس',
r'\bTLS\b': 'تی‌ال‌اس',
r'\bSSL\b': 'اس‌اس‌ال',
r'\bHMAC\b': 'اچ‌مَک',
r'\bMAC\b': 'مَک',
r'\bSHA-256\b': 'اس‌اچ‌ای ۲۵۶',
r'\bSHA-512\b': 'اس‌اچ‌ای ۵۱۲',
r'\bSHA-1\b': 'اس‌اچ‌ای ۱',
r'\bSHA-3\b': 'اس‌اچ‌ای ۳',
r'\bMD5\b': 'ام‌دی ۵',
r'\bAES\b': 'ای‌ای‌اس',
r'\bDES\b': 'دِس',
r'\b3DES\b': 'تری‌دِس',
r'\bRSA\b': 'آر‌اس‌ای',
r'\bNAT\b': 'نَت',
r'\bVPN\b': 'وی‌پی‌ان',
r'\bDMZ\b': 'دی‌ام‌زد',
r'\bGDPR\b': 'جی‌دی‌پی‌آر',
r'\bNIST\b': 'نیست',
r'\bRAM\b': 'رم',
r'\bDNS\b': 'دی‌ان‌اس',
r'\bTCP\b': 'تی‌سی‌پی',
r'\bUDP\b': 'یو‌دی‌پی',
r'\bICMP\b': 'آی‌سی‌ام‌پی',
r'\bPGP\b': 'پی‌جی‌پی',
r'\bWEP\b': 'وِپ',
r'\bWPA2\b': 'دبلیو‌پی‌ای ۲',
r'\bWPA\b': 'دبلیو‌پی‌ای',
r'\bIDS\b': 'آی‌دی‌اس',
r'\bIPS\b': 'آی‌پی‌اس',
r'\bOSI\b': 'او‌اس‌آی',
r'\bHTTP\b': 'اچ‌تی‌تی‌پی',
r'\bHTTPS\b': 'اچ‌تی‌تی‌پی‌اس',
r'\bSMTP\b': 'اس‌ام‌تی‌پی',
r'\bIMAP\b': 'آی‌مپ',
r'\bPOP3\b': 'پاپ ۳',
r'\bAPI\b': 'ای‌پی‌آی',
r'\bURL\b': 'یو‌آر‌ال',
r'\bIP\b': 'آی‌پی',
r'\bDoS\b': 'داس',
r'\bDDoS\b': 'دی‌داس',
r'\bSA\b': 'اس‌ای',
r'\bSAD\b': 'اس‌ای‌دی',
r'\bSPD\b': 'اس‌پی‌دی',
r'\bSPI\b': 'اس‌پی‌آی',
r'\bAH\b': 'ای‌ایچ',
r'\bESP\b': 'ای‌اس‌پی',
r'\bIKE\b': 'آی‌کی‌ای',
r'\bIV\b': 'آی‌وی',
r'\bCRC32\b': 'سی‌آرسی ۳۲',
r'\bTKIP\b': 'تی‌کیپ',
r'\bCCMP\b': 'سی‌سی‌ام‌پی',
r'\bPKI\b': 'پی‌کی‌آی',
r'\bCA\b': 'سی‌ای',
}
for k, v in replacements.items():
text = re.sub(k, v, text)
text = re.sub(r'\\\((.*?)\\\)', r' \1 ', text)
text = re.sub(r'\\\[(.*?)\\\]', r' \1 ', text)
text = text.replace('^', ' به توان ')
text = text.replace('\\oplus', ' XOR ')
text = text.replace('\\neq', ' مخالف ')
text = text.replace('\\approx', ' تقریباً برابر ')
text = text.replace('||', ' چسبیده به ')
text = text.replace('،', '، ')
text = text.replace('.', '. ')
return text
def extract_paragraphs(html_filepath):
with open(html_filepath, 'r', encoding='utf-8') as f:
soup = BeautifulSoup(f.read(), 'html.parser')
chunks = []
for elem in soup.find_all(['h2', 'h3', 'p']):
txt = elem.get_text().strip()
if not txt:
continue
txt = normalize_text_for_speech(txt)
chunks.append(txt)
return chunks
async def render_chunk(idx, text, temp_filename, voice="fa-IR-FaridNeural"):
async with SEMAPHORE:
for retry in range(3):
try:
# rate="+100%" for 2x speed
communicate = edge_tts.Communicate(text, voice, rate="+100%", pitch="+0Hz")
await communicate.save(temp_filename)
return temp_filename
except Exception as e:
await asyncio.sleep(0.5)
return None
async def generate_file_parallel(filepath, out_filename, voice="fa-IR-FaridNeural"):
chunks = extract_paragraphs(filepath)
print(f"🚀 Generating 2x speed audio for {os.path.basename(out_filename)} ({len(chunks)} chunks)...")
temp_dir = f"{out_filename}_temp"
os.makedirs(temp_dir, exist_ok=True)
tasks = []
temp_files = []
for idx, chunk in enumerate(chunks):
temp_path = os.path.join(temp_dir, f"chunk_{idx:03d}.mp3")
temp_files.append(temp_path)
tasks.append(render_chunk(idx, chunk, temp_path, voice))
await asyncio.gather(*tasks)
with open(out_filename, 'wb') as outfile:
for temp_path in temp_files:
if os.path.exists(temp_path):
with open(temp_path, 'rb') as infile:
outfile.write(infile.read())
os.remove(temp_path)
try:
os.rmdir(temp_dir)
except:
pass
print(f"✅ FINISHED 2x: {os.path.basename(out_filename)}")
async def main():
audio_dir = "/Users/tahamajs/Documents/uni/NS/slides/audio"
os.makedirs(audio_dir, exist_ok=True)
# Process all parts concurrently
file_tasks = [
generate_file_parallel("/Users/tahamajs/Documents/uni/NS/slides/part1_crisp.html", os.path.join(audio_dir, "part_1.mp3")),
generate_file_parallel("/Users/tahamajs/Documents/uni/NS/slides/part2_crisp.html", os.path.join(audio_dir, "part_2.mp3")),
generate_file_parallel("/Users/tahamajs/Documents/uni/NS/slides/part3_crisp.html", os.path.join(audio_dir, "part_3.mp3")),
]
await asyncio.gather(*file_tasks)
# Final exam section
final_exam_text = normalize_text_for_speech("""
بخش ضمیمه: بررسی و حل تشریحی سوالات امتحان پایان ترم امنیت شبکه.
سوال اول: سناریوی راه‌اندازی پروتکل آی‌پی‌سک برای اتصال کاربران راه دور به سرورهای داخلی یک سازمان.
در این سناریو به محرمانگی و احراز هویت نیاز داریم. دو انجمن امنیتی یا اس‌ای برای مسیر رفت و برگشت الزامی است. پروتکل ای‌اس‌پ انتخاب مناسبی برای ارائه همزمان رمزنگاری و احراز هویت است.
استفاده از حالت تونل الزامی است تا آدرس‌های آی‌پی شبکه داخلی لو نرود و کل بسته کپسوله شود.
در صورت استفاده همزمان از ای‌ایچ و ای‌اس‌پ در حالت ترنسپورت، مدل ای‌اس‌پ پس از ای‌ایچ برای اجرای اصل کدهای احراز هویت پیش از رمزگشایی جهت جلوگیری از حملات داس اعمال می‌شود.
سوال دوم: بررسی تفاوت نشست و اتصال در پروتکل اس‌اس‌ال و تی‌ال‌اس.
نشست یک رابطه منطقی پایدار حاوی پارامترهای سنگین مانند مستر سکرت است. اتصال صرفاً ارتباط فیزیکی موقتی است.
امکان تعریف چندین اتصال روی یک نشست برای جلوگیری از تکرار فرآیند دست‌تکانی وجود دارد. کلیدهای نشست هر اتصال به دلیل اعداد تصادفی کلاینت و سرور کاملاً متفاوت است.
سوال سوم: دیواره آتش و تداخل نت و آی‌پی‌سک.
فایروال باید پورت ۵۰ ای‌اس‌پ و پورت ۵۰۰ یو‌دی‌پی آی‌کی‌ای رو باز بگذارد. ترافیک ای‌اس‌پ فاقد پورت لایه ۴ است بنابراین نت توانایی ترجمه آن را ندارد و بسته افت می‌کند.
سوال چهارم: امضای دیجیتال و الگوی اول امضا بعد رمزنگاری.
فرستنده هش پیام را با کلید خصوصی امضا کرده، به پیام چسبانده و کل پکیج را با کلید عمومی گیرنده رمز می‌کند. این روش امکان ذخیره‌سازی امضای جداگانه را فراهم می‌سازد.
""")
exam_chunks = [c.strip() for c in final_exam_text.split('\n') if c.strip()]
temp_dir = os.path.join(audio_dir, "final_exam_temp")
os.makedirs(temp_dir, exist_ok=True)
exam_tasks = []
exam_temp_files = []
for idx, chunk in enumerate(exam_chunks):
tp = os.path.join(temp_dir, f"chunk_{idx:03d}.mp3")
exam_temp_files.append(tp)
exam_tasks.append(render_chunk(idx, chunk, tp))
await asyncio.gather(*exam_tasks)
out_exam_path = os.path.join(audio_dir, "final_exam.mp3")
with open(out_exam_path, 'wb') as outfile:
for tp in exam_temp_files:
if os.path.exists(tp):
with open(tp, 'rb') as infile:
outfile.write(infile.read())
os.remove(tp)
try:
os.rmdir(temp_dir)
except:
pass
print("✅ FINAL EXAM 2x DONE!")
# Merge ALL into one single complete master presentation audio file
master_file = os.path.join(audio_dir, "full_presentation_2x.mp3")
print(f"📦 Merging all 4 parts into a single file: {master_file}...")
all_parts = [
os.path.join(audio_dir, "part_1.mp3"),
os.path.join(audio_dir, "part_2.mp3"),
os.path.join(audio_dir, "part_3.mp3"),
out_exam_path
]
with open(master_file, 'wb') as outfile:
for pf in all_parts:
if os.path.exists(pf):
with open(pf, 'rb') as infile:
outfile.write(infile.read())
print(f"🎉 MASTER 2X FILE CREATED SUCCESSFULLY: {master_file}")
if __name__ == "__main__":
asyncio.run(main())

Xet Storage Details

Size:
9.93 kB
·
Xet hash:
66a936bc64deca3588524b3b0f736afc0aa8892b97d3f31b13395889615ef493

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.