init commit

Dockerfile

@@ -0,0 +1,35 @@
+FROM python:3.11-slim
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libsndfile1 \
+    espeak-ng \
+    ffmpeg \
+    git \
+    wget \
+    && rm -rf /var/lib/apt/lists/*
+RUN useradd -m -u 1000 user
+
+# Switch to the "user" user
+USER user
+
+# Set home to the user's home directory
+ENV HOME=/home/user \
+	PATH=/home/user/.local/bin:$PATH
+
+# Set the working directory to the user's home directory
+WORKDIR $HOME/app
+
+# Copy and install Python dependencies
+COPY requirements.txt $HOME/app/
+RUN pip install --no-cache-dir -r requirements.txt && pip install --upgrade pip
+
+# Copy the current directory contents into the container at $HOME/app setting the owner to the user
+COPY --chown=user . $HOME/app
+
+
+# Expose port
+EXPOSE 7860
+
+# Run the application
+CMD ["python", "app.py"]

README.md

@@ -1,6 +1,6 @@
 ---
-title: NetMonTTS
-emoji: 📚
+title: PrivateTest
+emoji: 🐨
 colorFrom: indigo
 colorTo: indigo
 sdk: docker

app.py

@@ -0,0 +1,302 @@
+from flask import Flask, request, jsonify, send_from_directory, abort
+from transformers import WhisperProcessor, WhisperForConditionalGeneration
+import librosa
+import torch
+import numpy as np
+from onnxruntime import InferenceSession
+import soundfile as sf
+import os
+import sys
+import uuid
+import logging
+from flask_cors import CORS
+import threading
+import tempfile
+from huggingface_hub import snapshot_download
+from huggingface_hub.utils import RepositoryNotFoundError, HfHubHTTPError
+import time
+from tts_processor import preprocess_all
+import hashlib
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = Flask(__name__)
+CORS(app, resources={r"/*": {"origins": "*"}})
+
+# Global lock to ensure one method runs at a time
+global_lock = threading.Lock()
+
+# Repository ID and paths
+kokoro_model_id = 'onnx-community/Kokoro-82M-v1.0-ONNX'
+model_path = 'kokoro_model'
+voice_name = 'am_adam'  # Example voice: af (adjust as needed)
+
+# Directory to serve files from
+SERVE_DIR = os.environ.get("SERVE_DIR", "./files")  # Default to './files' if not provided
+
+os.makedirs(SERVE_DIR, exist_ok=True)
+def validate_audio_file(file):
+    if file.content_type not in ["audio/wav", "audio/x-wav", "audio/mpeg", "audio/mp3"]:
+        raise ValueError("Unsupported file type")
+    file.seek(0, os.SEEK_END)
+    file_size = file.tell()
+    file.seek(0)  # Reset file pointer
+    if file_size > 10 * 1024 * 1024:  # 10 MB limit
+        raise ValueError("File is too large (max 10 MB)")
+
+def validate_text_input(text):
+    if not isinstance(text, str):
+        raise ValueError("Text input must be a string")
+    if len(text.strip()) == 0:
+        raise ValueError("Text input cannot be empty")
+    if len(text) > 1024:  # Limit to 1024 characters
+        raise ValueError("Text input is too long (max 1024 characters)")
+
+file_cache = {}
+
+def is_cached(cached_file_path):
+    """
+    Check if a file exists in the cache.
+    If the file is not in the cache, perform a disk check and update the cache.
+    """
+    if cached_file_path in file_cache:
+        return file_cache[cached_file_path]  # Return cached result
+    exists = os.path.exists(cached_file_path)  # Perform disk check
+    file_cache[cached_file_path] = exists  # Update the cache
+    return exists
+import time
+from huggingface_hub import snapshot_download
+from huggingface_hub.utils import RepositoryNotFoundError, HfHubHTTPError
+
+def initialize_models():
+    global sess, voice_style, processor, whisper_model
+
+    max_retries = 5  # Maximum number of retries
+    retry_delay = 2  # Initial delay in seconds (will double after each retry)
+
+    for attempt in range(max_retries):
+        try:
+            # Download the ONNX model if not already downloaded
+            if not os.path.exists(model_path):
+                logger.info(f"Attempt {attempt + 1} to download and load Kokoro model...")
+                kokoro_dir = snapshot_download(kokoro_model_id, cache_dir=model_path)
+                logger.info(f"Kokoro model directory: {kokoro_dir}")
+            else:
+                kokoro_dir = model_path
+                logger.info(f"Using cached Kokoro model directory: {kokoro_dir}")
+
+            # Validate ONNX file path
+            onnx_path = None
+            for root, _, files in os.walk(kokoro_dir):
+                if 'model.onnx' in files:
+                    onnx_path = os.path.join(root, 'model.onnx')
+                    break
+
+            if not onnx_path or not os.path.exists(onnx_path):
+                raise FileNotFoundError(f"ONNX file not found after redownload at {kokoro_dir}")
+
+            logger.info("Loading ONNX session...")
+            sess = InferenceSession(onnx_path)
+            logger.info(f"ONNX session loaded successfully from {onnx_path}")
+
+            # Load the voice style vector
+            voice_style_path = None
+            for root, _, files in os.walk(kokoro_dir):
+                if f'{voice_name}.bin' in files:
+                    voice_style_path = os.path.join(root, f'{voice_name}.bin')
+                    break
+
+            if not voice_style_path or not os.path.exists(voice_style_path):
+                raise FileNotFoundError(f"Voice style file not found at {voice_style_path}")
+
+            logger.info("Loading voice style vector...")
+            voice_style = np.fromfile(voice_style_path, dtype=np.float32).reshape(-1, 1, 256)
+            logger.info(f"Voice style vector loaded successfully from {voice_style_path}")
+
+            # Initialize Whisper model for S2T
+            logger.info("Downloading and loading Whisper model...")
+            processor = WhisperProcessor.from_pretrained("openai/whisper-base")
+            whisper_model = WhisperForConditionalGeneration.from_pretrained("openai/whisper-base")
+            whisper_model.config.forced_decoder_ids = None
+            logger.info("Whisper model loaded successfully")
+
+            # If everything succeeds, break out of the retry loop
+            break
+
+        except (RepositoryNotFoundError, HfHubHTTPError, FileNotFoundError) as e:
+            logger.error(f"Attempt {attempt + 1} failed: {str(e)}")
+            if attempt == max_retries - 1:
+                logger.error("Max retries reached. Failed to initialize models.")
+                raise  # Re-raise the exception if max retries are reached
+            time.sleep(retry_delay)
+            retry_delay *= 2  # Exponential backoff
+
+# Initialize models
+initialize_models()
+
+# Health check endpoint
+@app.route('/health', methods=['GET'])
+def health_check():
+    try:
+        return jsonify({"status": "healthy"}), 200
+    except Exception as e:
+        logger.error(f"Health check failed: {str(e)}")
+        return jsonify({"status": "unhealthy"}), 500
+
+# Text-to-Speech (T2S) Endpoint
+@app.route('/generate_audio', methods=['POST'])
+def generate_audio():
+    """Text-to-Speech (T2S) Endpoint"""
+    with global_lock:  # Acquire global lock to ensure only one instance runs
+        try:
+            logger.debug("Received request to /generate_audio")
+            data = request.json
+            text = data['text']
+            output_dir = data.get('output_dir')
+
+            validate_text_input(text)
+            logger.debug(f"Text: {text}")
+            if not output_dir:
+                raise ValueError("Output directory is required but not provided")
+
+            # Ensure output_dir is an absolute path and valid
+            if not os.path.isabs(output_dir):
+                raise ValueError("Output directory must be an absolute path")
+            if not os.path.exists(output_dir):
+                raise ValueError(f"Output directory does not exist: {output_dir}")
+
+            # Generate a unique hash for the text
+            text = preprocess_all(text)
+            logger.debug(f"Processed Text {text}")
+            text_hash = hashlib.sha256(text.encode('utf-8')).hexdigest()
+            hashed_file_name = f"{text_hash}.wav"
+            cached_file_path = os.path.join(output_dir, hashed_file_name)
+            logger.debug(f"Generated hash for processed text: {text_hash}")
+            logger.debug(f"Output directory: {output_dir}")
+            logger.debug(f"Cached file path: {cached_file_path}")
+
+            # Check if cached file exists
+            if is_cached(cached_file_path):
+                logger.info(f"Returning cached audio for text: {text}")
+                return jsonify({"status": "success", "output_path": cached_file_path})
+
+            # Tokenize text
+            logger.debug("Tokenizing text...")
+            from kokoro import phonemize, tokenize  # Import dynamically
+            tokens = tokenize(phonemize(text, 'a'))
+            logger.debug(f"Initial tokens: {tokens}")
+            if len(tokens) > 510:
+                logger.warning("Text too long; truncating to 510 tokens.")
+                tokens = tokens[:510]
+            tokens = [[0, *tokens, 0]]  # Add pad tokens
+            logger.debug(f"Final tokens: {tokens}")
+
+            # Get style vector based on token length
+            logger.debug("Fetching style vector...")
+            ref_s = voice_style[len(tokens[0]) - 2]  # Shape: (1, 256)
+            logger.debug(f"Style vector shape: {ref_s.shape}")
+
+            # Run ONNX inference
+            logger.debug("Running ONNX inference...")
+            audio = sess.run(None, dict(
+                input_ids=np.array(tokens, dtype=np.int64),
+                style=ref_s,
+                speed=np.ones(1, dtype=np.float32),
+            ))[0]
+            logger.debug(f"Audio generated with shape: {audio.shape}")
+
+            # Fix audio data for saving
+            audio = np.squeeze(audio)  # Remove extra dimension
+            audio = audio.astype(np.float32)  # Ensure correct data type
+
+            # Save audio
+            logger.debug(f"Saving audio to {cached_file_path}...")
+            sf.write(cached_file_path, audio, 24000)  # Save with 24 kHz sample rate
+            logger.info(f"Audio saved successfully to {cached_file_path}")
+            return jsonify({"status": "success", "output_path": cached_file_path})
+        except Exception as e:
+            logger.error(f"Error generating audio: {str(e)}")
+            return jsonify({"status": "error", "message": str(e)}), 500
+
+# Speech-to-Text (S2T) Endpoint
+@app.route('/transcribe_audio', methods=['POST'])
+def transcribe_audio():
+    """Speech-to-Text (S2T) Endpoint"""
+    with global_lock:  # Acquire global lock to ensure only one instance runs
+        audio_path = None
+        try:
+            logger.debug("Received request to /transcribe_audio")
+            file = request.files['file']
+            validate_audio_file(file)
+            # Generate a unique filename using uuid
+            unique_filename = f"{uuid.uuid4().hex}_{file.filename}"
+            audio_path = os.path.join("/tmp", unique_filename)
+            file.save(audio_path)
+            logger.debug(f"Audio file saved to {audio_path}")
+
+            # Load and preprocess audio
+            logger.debug("Processing audio for transcription...")
+            audio_array, sampling_rate = librosa.load(audio_path, sr=16000)
+
+            input_features = processor(
+                audio_array,
+                sampling_rate=sampling_rate,
+                return_tensors="pt"
+            ).input_features
+
+            # Generate transcription
+            logger.debug("Generating transcription...")
+            predicted_ids = whisper_model.generate(input_features)
+            transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True)[0]
+            logger.info(f"Transcription: {transcription}")
+
+            return jsonify({"status": "success", "transcription": transcription})
+        except Exception as e:
+            logger.error(f"Error transcribing audio: {str(e)}")
+            return jsonify({"status": "error", "message": str(e)}), 500
+        finally:
+            # Ensure temporary file is removed
+            if audio_path and os.path.exists(audio_path):
+                os.remove(audio_path)
+                logger.debug(f"Temporary file {audio_path} removed")
+
+@app.route('/files/<filename>', methods=['GET'])
+def serve_wav_file(filename):
+    """
+    Serve a .wav file from the configured directory.
+    Only serves files ending with '.wav'.
+    """
+    # Ensure only .wav files are allowed
+    if not filename.lower().endswith('.wav'):
+        abort(400, "Only .wav files are allowed.")
+    
+    # Check if the file exists in the directory
+    file_path = os.path.join(SERVE_DIR, filename)
+    logger.debug(f"Looking for file at: {file_path}")
+    if not os.path.isfile(file_path):
+        logger.error(f"File not found: {file_path}")
+        abort(404, "File not found.")
+    
+    # Serve the file
+    return send_from_directory(SERVE_DIR, filename)
+
+# Error handlers
+@app.errorhandler(400)
+def bad_request(error):
+    """Handle 400 errors."""
+    return {"error": "Bad Request", "message": str(error)}, 400
+
+@app.errorhandler(404)
+def not_found(error):
+    """Handle 404 errors."""
+    return {"error": "Not Found", "message": str(error)}, 404
+
+@app.errorhandler(500)
+def internal_error(error):
+    """Handle unexpected errors."""
+    return {"error": "Internal Server Error", "message": "An unexpected error occurred."}, 500
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=7860)

commit

@@ -0,0 +1,3 @@
+git add .
+git commit -m "$*"
+git push

kokoro.py

@@ -0,0 +1,165 @@
+import phonemizer
+import re
+import torch
+import numpy as np
+
+def split_num(num):
+    num = num.group()
+    if '.' in num:
+        return num
+    elif ':' in num:
+        h, m = [int(n) for n in num.split(':')]
+        if m == 0:
+            return f"{h} o'clock"
+        elif m < 10:
+            return f'{h} oh {m}'
+        return f'{h} {m}'
+    year = int(num[:4])
+    if year < 1100 or year % 1000 < 10:
+        return num
+    left, right = num[:2], int(num[2:4])
+    s = 's' if num.endswith('s') else ''
+    if 100 <= year % 1000 <= 999:
+        if right == 0:
+            return f'{left} hundred{s}'
+        elif right < 10:
+            return f'{left} oh {right}{s}'
+    return f'{left} {right}{s}'
+
+def flip_money(m):
+    m = m.group()
+    bill = 'dollar' if m[0] == '$' else 'pound'
+    if m[-1].isalpha():
+        return f'{m[1:]} {bill}s'
+    elif '.' not in m:
+        s = '' if m[1:] == '1' else 's'
+        return f'{m[1:]} {bill}{s}'
+    b, c = m[1:].split('.')
+    s = '' if b == '1' else 's'
+    c = int(c.ljust(2, '0'))
+    coins = f"cent{'' if c == 1 else 's'}" if m[0] == '$' else ('penny' if c == 1 else 'pence')
+    return f'{b} {bill}{s} and {c} {coins}'
+
+def point_num(num):
+    a, b = num.group().split('.')
+    return ' point '.join([a, ' '.join(b)])
+
+def normalize_text(text):
+    text = text.replace(chr(8216), "'").replace(chr(8217), "'")
+    text = text.replace('«', chr(8220)).replace('»', chr(8221))
+    text = text.replace(chr(8220), '"').replace(chr(8221), '"')
+    text = text.replace('(', '«').replace(')', '»')
+    for a, b in zip('、。！，：；？', ',.!,:;?'):
+        text = text.replace(a, b+' ')
+    text = re.sub(r'[^\S \n]', ' ', text)
+    text = re.sub(r'  +', ' ', text)
+    text = re.sub(r'(?<=\n) +(?=\n)', '', text)
+    text = re.sub(r'\bD[Rr]\.(?= [A-Z])', 'Doctor', text)
+    text = re.sub(r'\b(?:Mr\.|MR\.(?= [A-Z]))', 'Mister', text)
+    text = re.sub(r'\b(?:Ms\.|MS\.(?= [A-Z]))', 'Miss', text)
+    text = re.sub(r'\b(?:Mrs\.|MRS\.(?= [A-Z]))', 'Mrs', text)
+    text = re.sub(r'\betc\.(?! [A-Z])', 'etc', text)
+    text = re.sub(r'(?i)\b(y)eah?\b', r"\1e'a", text)
+    text = re.sub(r'\d*\.\d+|\b\d{4}s?\b|(?<!:)\b(?:[1-9]|1[0-2]):[0-5]\d\b(?!:)', split_num, text)
+    text = re.sub(r'(?<=\d),(?=\d)', '', text)
+    text = re.sub(r'(?i)[$£]\d+(?:\.\d+)?(?: hundred| thousand| (?:[bm]|tr)illion)*\b|[$£]\d+\.\d\d?\b', flip_money, text)
+    text = re.sub(r'\d*\.\d+', point_num, text)
+    text = re.sub(r'(?<=\d)-(?=\d)', ' to ', text)
+    text = re.sub(r'(?<=\d)S', ' S', text)
+    text = re.sub(r"(?<=[BCDFGHJ-NP-TV-Z])'?s\b", "'S", text)
+    text = re.sub(r"(?<=X')S\b", 's', text)
+    text = re.sub(r'(?:[A-Za-z]\.){2,} [a-z]', lambda m: m.group().replace('.', '-'), text)
+    text = re.sub(r'(?i)(?<=[A-Z])\.(?=[A-Z])', '-', text)
+    return text.strip()
+
+def get_vocab():
+    _pad = "$"
+    _punctuation = ';:,.!?¡¿—…"«»“” '
+    _letters = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'
+    _letters_ipa = "ɑɐɒæɓʙβɔɕçɗɖðʤəɘɚɛɜɝɞɟʄɡɠɢʛɦɧħɥʜɨɪʝɭɬɫɮʟɱɯɰŋɳɲɴøɵɸθœɶʘɹɺɾɻʀʁɽʂʃʈʧʉʊʋⱱʌɣɤʍχʎʏʑʐʒʔʡʕʢǀǁǂǃˈˌːˑʼʴʰʱʲʷˠˤ˞↓↑→↗↘'̩'ᵻ"
+    symbols = [_pad] + list(_punctuation) + list(_letters) + list(_letters_ipa)
+    dicts = {}
+    for i in range(len((symbols))):
+        dicts[symbols[i]] = i
+    return dicts
+
+VOCAB = get_vocab()
+def tokenize(ps):
+    return [i for i in map(VOCAB.get, ps) if i is not None]
+
+phonemizers = dict(
+    a=phonemizer.backend.EspeakBackend(language='en-us', preserve_punctuation=True, with_stress=True),
+    b=phonemizer.backend.EspeakBackend(language='en-gb', preserve_punctuation=True, with_stress=True),
+)
+def phonemize(text, lang, norm=True):
+    if norm:
+        text = normalize_text(text)
+    ps = phonemizers[lang].phonemize([text])
+    ps = ps[0] if ps else ''
+    # https://en.wiktionary.org/wiki/kokoro#English
+    ps = ps.replace('kəkˈoːɹoʊ', 'kˈoʊkəɹoʊ').replace('kəkˈɔːɹəʊ', 'kˈəʊkəɹəʊ')
+    ps = ps.replace('ʲ', 'j').replace('r', 'ɹ').replace('x', 'k').replace('ɬ', 'l')
+    ps = re.sub(r'(?<=[a-zɹː])(?=hˈʌndɹɪd)', ' ', ps)
+    ps = re.sub(r' z(?=[;:,.!?¡¿—…"«»“” ]|$)', 'z', ps)
+    if lang == 'a':
+        ps = re.sub(r'(?<=nˈaɪn)ti(?!ː)', 'di', ps)
+    ps = ''.join(filter(lambda p: p in VOCAB, ps))
+    return ps.strip()
+
+def length_to_mask(lengths):
+    mask = torch.arange(lengths.max()).unsqueeze(0).expand(lengths.shape[0], -1).type_as(lengths)
+    mask = torch.gt(mask+1, lengths.unsqueeze(1))
+    return mask
+
+@torch.no_grad()
+def forward(model, tokens, ref_s, speed):
+    device = ref_s.device
+    tokens = torch.LongTensor([[0, *tokens, 0]]).to(device)
+    input_lengths = torch.LongTensor([tokens.shape[-1]]).to(device)
+    text_mask = length_to_mask(input_lengths).to(device)
+    bert_dur = model.bert(tokens, attention_mask=(~text_mask).int())
+    d_en = model.bert_encoder(bert_dur).transpose(-1, -2)
+    s = ref_s[:, 128:]
+    d = model.predictor.text_encoder(d_en, s, input_lengths, text_mask)
+    x, _ = model.predictor.lstm(d)
+    duration = model.predictor.duration_proj(x)
+    duration = torch.sigmoid(duration).sum(axis=-1) / speed
+    pred_dur = torch.round(duration).clamp(min=1).long()
+    pred_aln_trg = torch.zeros(input_lengths, pred_dur.sum().item())
+    c_frame = 0
+    for i in range(pred_aln_trg.size(0)):
+        pred_aln_trg[i, c_frame:c_frame + pred_dur[0,i].item()] = 1
+        c_frame += pred_dur[0,i].item()
+    en = d.transpose(-1, -2) @ pred_aln_trg.unsqueeze(0).to(device)
+    F0_pred, N_pred = model.predictor.F0Ntrain(en, s)
+    t_en = model.text_encoder(tokens, input_lengths, text_mask)
+    asr = t_en @ pred_aln_trg.unsqueeze(0).to(device)
+    return model.decoder(asr, F0_pred, N_pred, ref_s[:, :128]).squeeze().cpu().numpy()
+
+def generate(model, text, voicepack, lang='a', speed=1, ps=None):
+    ps = ps or phonemize(text, lang)
+    tokens = tokenize(ps)
+    if not tokens:
+        return None
+    elif len(tokens) > 510:
+        tokens = tokens[:510]
+        print('Truncated to 510 tokens')
+    ref_s = voicepack[len(tokens)]
+    out = forward(model, tokens, ref_s, speed)
+    ps = ''.join(next(k for k, v in VOCAB.items() if i == v) for i in tokens)
+    return out, ps
+
+def generate_full(model, text, voicepack, lang='a', speed=1, ps=None):
+    ps = ps or phonemize(text, lang)
+    tokens = tokenize(ps)
+    if not tokens:
+        return None
+    outs = []
+    loop_count = len(tokens)//510 + (1 if len(tokens) % 510 != 0 else 0)
+    for i in range(loop_count):
+        ref_s = voicepack[len(tokens[i*510:(i+1)*510])]
+        out = forward(model, tokens[i*510:(i+1)*510], ref_s, speed)
+        outs.append(out)
+    outs = np.concatenate(outs)
+    ps = ''.join(next(k for k, v in VOCAB.items() if i == v) for i in tokens)
+    return outs, ps

requirements.txt

@@ -0,0 +1,17 @@
+flask
+flask-cors
+transformers
+librosa
+numpy
+soundfile
+huggingface_hub
+phonemizer
+munch
+werkzeug
+num2words
+dateparser
+inflect
+ftfy
+sentencepiece
+torch --index-url https://download.pytorch.org/whl/cpu
+onnxruntime

tts_processor.py

@@ -0,0 +1,142 @@
+import re
+from dateutil.parser import parse
+from num2words import num2words
+import inflect
+from ftfy import fix_text
+
+# Initialize the inflect engine
+inflect_engine = inflect.engine()
+
+# Define alphabet pronunciation mapping
+alphabet_map = {
+    "A": " Eh ", "B": " Bee ", "C": " See ", "D": " Dee ", "E": " Eee ",
+    "F": " Eff ", "G": " Jee ", "H": " Aitch ", "I": " Eye ", "J": " Jay ",
+    "K": " Kay ", "L": " El ", "M": " Emm ", "N": " Enn ", "O": " Ohh ",
+    "P": " Pee ", "Q": " Queue ", "R": " Are ", "S": " Ess ", "T": " Tee ",
+    "U": " You ", "V": " Vee ", "W": " Double You ", "X": " Ex ", "Y": " Why ", "Z": " Zed "
+}
+
+# Function to add ordinal suffix to a number
+def add_ordinal_suffix(day):
+    """Adds ordinal suffix to a day (e.g., 13 -> 13th)."""
+    if 11 <= day <= 13:  # Special case for 11th, 12th, 13th
+        return f"{day}th"
+    elif day % 10 == 1:
+        return f"{day}st"
+    elif day % 10 == 2:
+        return f"{day}nd"
+    elif day % 10 == 3:
+        return f"{day}rd"
+    else:
+        return f"{day}th"
+
+# Function to format dates in a human-readable form
+def format_date(parsed_date, include_time=True):
+    """Formats a parsed date into a human-readable string."""
+    if not parsed_date:
+        return None
+
+    # Convert the day into an ordinal (e.g., 13 -> 13th)
+    day = add_ordinal_suffix(parsed_date.day)
+
+    # Format the date in a TTS-friendly way
+    if include_time and parsed_date.hour != 0 and parsed_date.minute != 0:
+        return parsed_date.strftime(f"%B {day}, %Y at %-I:%M %p")  # Unix
+    return parsed_date.strftime(f"%B {day}, %Y")  # Only date
+
+# Normalize dates in the text
+def normalize_dates(text):
+    """
+    Finds and replaces date strings with a nicely formatted, TTS-friendly version.
+    """
+    def replace_date(match):
+        raw_date = match.group(0)
+        try:
+            parsed_date = parse(raw_date)
+            if parsed_date:
+                include_time = "T" in raw_date or " " in raw_date  # Include time only if explicitly provided
+                return format_date(parsed_date, include_time)
+        except ValueError:
+            pass
+        return raw_date
+
+    # Match common date formats
+    date_pattern = r"\b(\d{4}-\d{2}-\d{2}(?:[ T]\d{2}:\d{2}:\d{2})?|\d{2}/\d{2}/\d{4}|\d{1,2} \w+ \d{4})\b"
+    return re.sub(date_pattern, replace_date, text)
+
+# Replace invalid characters and clean text
+def replace_invalid_chars(string):
+    string = fix_text(string)
+    replacements = {
+        "**": "",
+        '&#x27;': "'",
+        'AI;': 'Artificial Intelligence!',
+        'iddqd;': 'Immortality cheat code',
+        '😉;': 'wink wink!',
+        ':D': '*laughs* Ahahaha!',
+        ';D': '*laughs* Ahahaha!'
+    }
+    for old, new in replacements.items():
+        string = string.replace(old, new)
+    return string
+
+# Replace numbers with their word equivalents
+def replace_numbers(string):
+    ipv4_pattern = r'(\b\d{1,3}(\.\d{1,3}){3}\b)'
+    ipv6_pattern = r'([0-9a-fA-F]{1,4}:){2,7}[0-9a-fA-F]{1,4}'
+    range_pattern = r'\b\d+-\d+\b'  # Detect ranges like 1-4
+    date_pattern = r'\b\d{4}-\d{2}-\d{2}(?:T\d{2}:\d{2}:\d{2})?\b'
+    alphanumeric_pattern = r'\b[A-Za-z]+\d+|\d+[A-Za-z]+\b'
+
+    # Do not process IP addresses, date patterns, or alphanumerics
+    if re.search(ipv4_pattern, string) or re.search(ipv6_pattern, string) or re.search(range_pattern, string) or re.search(date_pattern, string) or re.search(alphanumeric_pattern, string):
+        return string
+
+    # Convert standalone numbers and port numbers
+    def convert_number(match):
+        number = match.group()
+        return num2words(int(number)) if number.isdigit() else number
+
+    pattern = re.compile(r'\b\d+\b')
+    return re.sub(pattern, convert_number, string)
+
+# Replace abbreviations with expanded form
+def replace_abbreviations(string):
+    words = string.split()
+    for i, word in enumerate(words):
+        if word.isupper() and len(word) <= 4 and not any(char.isdigit() for char in word) and word not in ["ID", "AM", "PM"]:
+            words[i] = ''.join([alphabet_map.get(char, char) for char in word])
+    return ' '.join(words)
+
+# Clean up whitespace in the text
+def clean_whitespace(string):
+    string = re.sub(r'\s+([.,?!])', r'\1', string)
+    return ' '.join(string.split())
+
+# Main preprocessing pipeline
+def preprocess_all(string):
+    string = normalize_dates(string)
+    string = replace_invalid_chars(string)
+    string = replace_numbers(string)
+    string = replace_abbreviations(string)
+    string = clean_whitespace(string)
+    return string
+
+# Expose a testing function for external use
+def test_preprocessing(file_path):
+    with open(file_path, 'r') as file:
+        lines = file.readlines()
+    for line in lines:
+        original = line.strip()
+        processed = preprocess_all(original)
+        print(f"Original: {original}")
+        print(f"Processed: {processed}\n")
+
+if __name__ == "__main__":
+    import sys
+    if len(sys.argv) > 1:
+        test_file = sys.argv[1]
+        test_preprocessing(test_file)
+    else:
+        print("Please provide a file path as an argument.")
+