Create app.py

app.py

@@ -0,0 +1,461 @@
+import gradio as gr
+import os
+import tempfile
+import requests
+import subprocess
+import zipfile
+import shutil
+import time
+import asyncio
+import re
+import hashlib
+from bs4 import BeautifulSoup
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from urllib.parse import urljoin
+import PyPDF2
+
+# ==========================================
+# ⚙️ CONFIGURATION & CONSTANTS
+# ==========================================
+THREADS = 10
+download_door = asyncio.Semaphore(1)  
+kitchen_door = asyncio.Semaphore(1)  
+
+CACHE_DIR = os.path.join(os.getcwd(), "AudioNovel_Cache")
+os.makedirs(CACHE_DIR, exist_ok=True)
+
+# ==========================================
+# 🛠️ HELPER FUNCTIONS
+# ==========================================
+def sanitize_filename(title):
+    return re.sub(r'[\\/*?:"<>|]', "", title).strip()
+
+def get_chunk_hash(text):
+    return hashlib.md5(text.encode('utf-8')).hexdigest()
+
+def smart_sanitize(text):
+    text = re.sub(r'http[s]?://\S+', '', text)
+    patterns = [
+        r'(?i)(please\s+)?vote for.*?(power stones|tickets)',
+        r'(?i)support me on patreon',
+        r'(?i)join my discord',
+        r'(?i)ko-fi\.com',
+        r'(?i)buy me a coffee',
+        r'(?i)author\'s note[s]?[\s\S]*',
+        r'(?i)translator\'s note[s]?[\s\S]*'
+    ]
+    for p in patterns:
+        text = re.sub(p, '', text)
+    return re.sub(r'\n{3,}', '\n\n', text).strip()
+
+def extract_novel_info(main_url, headers):
+    try:
+        # Bypass cookies for 18+ Mature walls
+        bypass_cookies = {'isAdult': '1', 'adult': 'true', 'over18': '1', 'age_verified': '1'}
+        
+        response = requests.get(main_url, headers=headers, cookies=bypass_cookies, timeout=15)
+        response.raise_for_status()
+        soup = BeautifulSoup(response.text, 'html.parser')
+        
+        # 1. Get Title
+        title = "Unknown_Novel"
+        h1_tag = soup.find('h1')
+        if h1_tag: 
+            title = h1_tag.get_text(strip=True)
+        else: 
+            title = soup.title.get_text(strip=True).split('-')[0]
+        safe_title = sanitize_filename(title)
+        
+        # 2. Get Cover Art
+        cover_url = None
+        og_img = soup.find('meta', property='og:image')
+        if og_img and og_img.get('content'):
+            cover_url = og_img['content']
+        
+        # 3. Get Chapter Links
+        links = []
+        morelist_div = soup.find('div', id='morelist')
+        
+        if morelist_div:
+            # Standard Novelhall Layout
+            for a_tag in morelist_div.find_all('a', href=True):
+                links.append(urljoin(main_url, a_tag['href']))
+        else:
+            # SUPER RADAR: If 'morelist' is hidden (e.g. adult novels or short stories)
+            for a_tag in soup.find_all('a', href=True):
+                href = a_tag['href']
+                full_url = urljoin(main_url, href)
+                # Filter out obvious junk links, keep things that look like chapters
+                if full_url.startswith(main_url) and full_url != main_url and len(href) > 5:
+                    if full_url not in links:
+                        links.append(full_url)
+        
+        if not links:
+            return None, [], None, "No chapter links could be found on the page. The site layout might have changed."
+            
+        return safe_title, links, cover_url, None
+    except requests.exceptions.HTTPError as e:
+        return None, [], None, f"HTTP Error {e.response.status_code}: The website blocked the request or the page doesn't exist."
+    except Exception as e:
+        return None, [], None, f"Extraction failed: {str(e)}"
+
+def extract_local_text(path, temp_dir):
+    content = ""
+    ext = os.path.splitext(path)[1].lower()
+    try:
+        if ext == '.txt':
+            with open(path, 'r', encoding='utf-8', errors='ignore') as f: 
+                content = ' '.join(f.read().split())
+        elif ext == '.epub':
+            with zipfile.ZipFile(path, 'r') as z:
+                for filename in z.namelist():
+                    if filename.endswith(('.html', '.xhtml')):
+                        with z.open(filename) as f:
+                            soup = BeautifulSoup(f.read(), 'html.parser')
+                            clean_part = ' '.join(soup.get_text(separator=' ').split())
+                            if clean_part: content += clean_part + " "
+        elif ext == '.pdf':
+            reader = PyPDF2.PdfReader(path)
+            for page in reader.pages: 
+                txt = page.extract_text()
+                if txt: content += ' '.join(txt.split()) + " "
+    except Exception as e:
+        return None
+    
+    if not content: return None
+    out = os.path.join(temp_dir, f"extracted_{int(time.time()*1000)}.txt")
+    with open(out, 'w', encoding='utf-8') as f: f.write(content)
+    return out
+
+# ==========================================
+# 🗣️ TEXT-TO-SPEECH ENGINE
+# ==========================================
+class GoogleTTSEngine:
+    def split_text(self, text, max_len=200):
+        chunks = []
+        while text:
+            if len(text) <= max_len:
+                chunks.append(text.strip())
+                break
+            best_i = -1
+            for punct in [". ", "! ", "? ", "; "]:
+                i = text[:max_len].rfind(punct)
+                if i > best_i: best_i = i
+            if best_i == -1: best_i = text[:max_len].rfind(" ")
+            if best_i == -1: best_i = max_len
+            else: best_i += 1 
+            chunks.append(text[:best_i].strip())
+            text = text[best_i:].lstrip()
+        return chunks
+
+    def strip_id3(self, mp3_data):
+        if mp3_data[:3] == b"ID3":
+            size = ((mp3_data[6] & 0x7F) << 21 | (mp3_data[7] & 0x7F) << 14 | 
+                    (mp3_data[8] & 0x7F) << 7 | (mp3_data[9] & 0x7F))
+            mp3_data = mp3_data[10 + size:]
+        if len(mp3_data) > 128 and mp3_data[-128:-125] == b"TAG": 
+            mp3_data = mp3_data[:-128]
+        return mp3_data
+
+    def download_chunk(self, text):
+        chunk_hash = get_chunk_hash(text)
+        cache_path = os.path.join(CACHE_DIR, f"{chunk_hash}.mp3")
+        
+        if os.path.exists(cache_path):
+            with open(cache_path, "rb") as f:
+                return f.read()
+
+        url = "https://translate.google.com/translate_tts"
+        params = {"ie": "UTF-8", "q": text, "tl": "en-IN", "client": "tw-ob"}
+        headers = {"User-Agent": "Mozilla/5.0"}
+        for _ in range(5):
+            try:
+                r = requests.get(url, params=params, headers=headers, timeout=5)
+                if r.status_code == 200:
+                    clean_audio = self.strip_id3(r.content)
+                    with open(cache_path, "wb") as f: f.write(clean_audio)
+                    return clean_audio
+            except: time.sleep(0.5)
+        return None
+
+def sync_download_worker(text_file, temp_dir, progress_obj, book_idx, total_books, base_name):
+    book_label = f"[{book_idx}/{total_books}]"
+    engine = GoogleTTSEngine()
+    
+    with open(text_file, 'r', encoding='utf-8') as f: 
+        raw_text = f.read()
+    
+    full_text = smart_sanitize(raw_text)
+    chunks = engine.split_text(full_text)
+    total_chunks = len(chunks)
+    
+    if total_chunks == 0:
+        return None
+        
+    chunk_dir = os.path.join(temp_dir, "chunks")
+    os.makedirs(chunk_dir, exist_ok=True)
+
+    start_time = time.time()
+    completed = 0
+
+    with ThreadPoolExecutor(max_workers=THREADS) as pool:
+        futures = {pool.submit(engine.download_chunk, chunk): i for i, chunk in enumerate(chunks)}
+        for f in as_completed(futures):
+            idx = futures[f]
+            data = f.result()
+            if data: 
+                with open(os.path.join(chunk_dir, f"{idx:05d}.mp3"), "wb") as af: af.write(data)
+            completed += 1
+            elapsed = time.time() - start_time
+            avg = elapsed / completed
+            eta = avg * (total_chunks - completed)
+            progress_obj((completed/total_chunks), desc=f"📥 {book_label} TTS Audio: {completed}/{total_chunks} | ⏳ ETA: {int(eta//60):02d}:{int(eta%60):02d}")
+
+    merged_path = os.path.join(temp_dir, "raw.mp3")
+    with open(merged_path, 'wb') as outfile:
+        for i in range(len(chunks)):
+            p = os.path.join(chunk_dir, f"{i:05d}.mp3")
+            if os.path.exists(p):
+                with open(p, 'rb') as infile: shutil.copyfileobj(infile, outfile)
+    return merged_path
+
+def sync_ffmpeg_worker(merged_path, final_output_m4b, temp_dir, speed, title, cover_url):
+    cover_path = os.path.join(temp_dir, "cover.jpg")
+    has_cover = False
+    
+    if cover_url:
+        try:
+            r = requests.get(cover_url, timeout=5)
+            if r.status_code == 200:
+                with open(cover_path, 'wb') as f: f.write(r.content)
+                has_cover = True
+        except: pass
+
+    cmd = ["ffmpeg", "-y", "-i", merged_path]
+    
+    if has_cover:
+        cmd.extend(["-i", cover_path, "-map", "0:a", "-map", "1", "-c:v", "copy", "-disposition:v", "attached_pic"])
+    else:
+        cmd.extend(["-map", "0:a"])
+
+    cmd.extend([
+        "-af", f"silenceremove=1:0:-50dB,atempo={speed}",
+        "-c:a", "aac", "-b:a", "64k", 
+        "-metadata", f"title={title}",
+        "-metadata", "artist=Ultimate AudioNovel Studio",
+        "-metadata", "album=Web Novels",
+        "-threads", "0",
+        final_output_m4b
+    ])
+    
+    subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    return final_output_m4b
+
+async def convert_text_to_audio(text_file, title, cover_url, speed, progress_obj, book_idx, total_books):
+    temp_dir = tempfile.mkdtemp()
+    
+    try:
+        async with download_door:
+            merged_path = await asyncio.to_thread(sync_download_worker, text_file, temp_dir, progress_obj, book_idx, total_books, title)
+        
+        if not merged_path: 
+            return None
+        
+        final_output_m4b = os.path.join(tempfile.gettempdir(), f"{title}_Audiobook.m4b")
+        
+        progress_obj(1.0, desc=f"🚀 Packaging {title} into .m4b format...")
+        async with kitchen_door:
+            output_file = await asyncio.to_thread(sync_ffmpeg_worker, merged_path, final_output_m4b, temp_dir, speed, title, cover_url)
+        return output_file
+    finally:
+        # Cleanup temp files safely
+        try: shutil.rmtree(temp_dir)
+        except: pass
+
+async def master_tts_pipeline(text_info_list, speed, progress_obj):
+    total_books = len(text_info_list)
+    finished_files = []
+    
+    for idx, (txt_path, title, cover_url) in enumerate(text_info_list, 1):
+        m4b_path = await convert_text_to_audio(txt_path, title, cover_url, speed, progress_obj, idx, total_books)
+        if m4b_path:
+            finished_files.append(m4b_path)
+            yield finished_files
+            
+    if len(finished_files) > 1:
+        zip_path = os.path.join(tempfile.gettempdir(), "AudioBook_MasterBatch.zip")
+        with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
+            for f in finished_files: zipf.write(f, os.path.basename(f))
+        finished_files.insert(0, zip_path)
+        yield finished_files
+
+# ==========================================
+# 🚀 CORE ENTRY POINTS
+# ==========================================
+async def process_links_to_audio(input_mode, input_text, speed, progress=gr.Progress()):
+    if not input_text.strip():
+        yield [], "❌ Error: No input provided."
+        return
+
+    headers = {'User-Agent': 'Mozilla/5.0'}
+    bypass_cookies = {'isAdult': '1', 'adult': 'true', 'over18': '1', 'age_verified': '1'}
+    
+    input_urls = [line.strip() for line in input_text.split('\n') if line.strip()]
+    text_info_list = []
+    overall_status = ""
+
+    if input_mode == "Main Novel Page URLs (Batch Supported)":
+        for novel_idx, main_url in enumerate(input_urls, 1):
+            progress(0, desc=f"Extracting Info: Novel {novel_idx}/{len(input_urls)}")
+            novel_title, links, cover_url, err = extract_novel_info(main_url, headers)
+            
+            if err or not links:
+                overall_status += f"❌ Failed {main_url} - {err}\n"
+                continue
+            
+            compiled_text = ""
+            for idx, link in progress.tqdm(enumerate(links, 1), total=len(links), desc=f"Scraping '{novel_title}'"):
+                try:
+                    response = requests.get(link, headers=headers, cookies=bypass_cookies, timeout=15)
+                    soup = BeautifulSoup(response.text, 'html.parser')
+                    lines = []
+                    
+                    if 'novelhall.com' in link:
+                        header_div = soup.find('div', class_='single-header')
+                        if header_div and header_div.find('h1'): lines.append(header_div.find('h1').get_text(strip=True))
+                        content_div = soup.find('div', id='htmlContent')
+                        if content_div:
+                            for br in content_div.find_all('br'): br.replace_with('\n')
+                            for p in content_div.get_text(separator='\n', strip=True).split('\n'):
+                                if p.strip(): lines.append(p.strip())
+                    elif 'novelbin.com' in link:
+                        content_div = soup.find('div', id='chr-content')
+                        if content_div:
+                            h4 = content_div.find('h4')
+                            if h4: lines.append(h4.get_text(strip=True))
+                            for p in content_div.find_all('p'):
+                                if p.get_text(strip=True): lines.append(p.get_text(strip=True))
+                                
+                    compiled_text += '\n\n'.join(lines) + '\n\n\n'
+                except Exception as e: pass
+                time.sleep(0.2)
+            
+            if not compiled_text.strip():
+                overall_status += f"❌ Failed {novel_title} - Could not extract chapter text.\n"
+                continue
+                
+            temp_file = tempfile.NamedTemporaryFile(delete=False, mode='w', encoding='utf-8', suffix='.txt')
+            temp_file.write(compiled_text)
+            temp_file.close()
+            text_info_list.append((temp_file.name, novel_title, cover_url))
+            overall_status += f"✅ Scraped {novel_title} ({len(links)} chapters)\n"
+            
+    else: # Custom Chapters
+        novel_title = "Custom_Scraped_Chapters"
+        compiled_text = ""
+        for idx, link in progress.tqdm(enumerate(input_urls, 1), total=len(input_urls), desc="Scraping Custom Chapters"):
+            try:
+                response = requests.get(link, headers=headers, cookies=bypass_cookies, timeout=15)
+                soup = BeautifulSoup(response.text, 'html.parser')
+                lines = []
+                content_div = soup.find('div', id='htmlContent') or soup.find('div', id='chr-content')
+                if content_div:
+                    for br in content_div.find_all('br'): br.replace_with('\n')
+                    for p in content_div.get_text(separator='\n', strip=True).split('\n'):
+                        if p.strip(): lines.append(p.strip())
+                compiled_text += '\n\n'.join(lines) + '\n\n\n'
+            except Exception as e: pass
+            time.sleep(0.2)
+            
+        if compiled_text.strip():
+            temp_file = tempfile.NamedTemporaryFile(delete=False, mode='w', encoding='utf-8', suffix='.txt')
+            temp_file.write(compiled_text)
+            temp_file.close()
+            text_info_list.append((temp_file.name, novel_title, None))
+            overall_status += f"✅ Scraped {len(input_urls)} Custom Chapters\n"
+        else:
+            overall_status += "❌ Failed - Could not extract any text from the provided links.\n"
+
+    if not text_info_list:
+        yield [], overall_status
+        return
+
+    overall_status += "⚙️ Initiating TTS Audio Conversion...\n"
+    yield [], overall_status
+    
+    async for audio_files in master_tts_pipeline(text_info_list, speed, progress):
+        yield audio_files, overall_status + f"✅ Audiobook Process Complete!"
+
+async def process_files_to_audio(file_paths, speed, progress=gr.Progress()):
+    if not file_paths:
+        yield [], "❌ Error: No files uploaded."
+        return
+        
+    if not isinstance(file_paths, list): file_paths = [file_paths]
+    text_info_list = []
+    
+    for i, path in enumerate(file_paths):
+        progress(i/len(file_paths), desc="Extracting text from local files...")
+        temp_dir = tempfile.mkdtemp()
+        txt_path = extract_local_text(path, temp_dir)
+        if txt_path:
+            base_name = sanitize_filename(os.path.splitext(os.path.basename(path))[0])
+            text_info_list.append((txt_path, base_name, None))
+            
+    if not text_info_list:
+        yield [], "❌ Error: Could not extract readable text from the uploaded files."
+        return
+        
+    overall_status = "✅ Local files parsed. Initiating TTS Audio Conversion...\n"
+    yield [], overall_status
+    
+    async for audio_files in master_tts_pipeline(text_info_list, speed, progress):
+        yield audio_files, overall_status + f"✅ Audiobook Process Complete!"
+
+# ==========================================
+# 🎨 GRADIO UI
+# ==========================================
+with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo")) as app:
+    gr.Markdown("# 🎧 The Ultimate AudioNovel Studio")
+    gr.Markdown("Transform Web Novels into True **.m4b Audiobooks** featuring Cover Art Injection, Save-State Caching, and Smart Text Sanitization!")
+    
+    with gr.Tabs():
+        with gr.Tab("🌐 Link to Audio (Web Scraper)"):
+            with gr.Row():
+                with gr.Column(scale=1):
+                    input_mode = gr.Radio(
+                        choices=["Main Novel Page URLs (Batch Supported)", "List of Chapter Links"],
+                        value="Main Novel Page URLs (Batch Supported)",
+                        label="Source Type"
+                    )
+                    links_input = gr.Textbox(lines=8, label="Paste Links Here (One per line)")
+                    speed_slider_1 = gr.Slider(minimum=1.0, maximum=3.0, value=1.6, step=0.1, label="Voice Speedup")
+                    with gr.Row():
+                        start_btn_1 = gr.Button("🚀 Scrape & Convert", variant="primary")
+                        stop_btn_1 = gr.Button("🛑 Stop", variant="stop")
+                
+                with gr.Column(scale=1):
+                    status_output_1 = gr.Textbox(label="Status Log", interactive=False, lines=4)
+                    file_output_1 = gr.File(label="📥 Download Ready Audiobook (.m4b)", file_count="multiple", interactive=False)
+                    
+            process_event_1 = start_btn_1.click(fn=process_links_to_audio, inputs=[input_mode, links_input, speed_slider_1], outputs=[file_output_1, status_output_1])
+            stop_btn_1.click(fn=None, inputs=None, outputs=None, cancels=[process_event_1])
+
+        with gr.Tab("📂 File to Audio (Local Converter)"):
+            with gr.Row():
+                with gr.Column(scale=1):
+                    file_in = gr.File(label="Upload Books (.txt, .epub, .pdf)", file_count="multiple")
+                    speed_slider_2 = gr.Slider(minimum=1.0, maximum=3.0, value=1.6, step=0.1, label="Voice Speedup")
+                    with gr.Row():
+                        start_btn_2 = gr.Button("🚀 Convert Files to Audio", variant="primary")
+                        stop_btn_2 = gr.Button("🛑 Stop", variant="stop")
+                
+                with gr.Column(scale=1):
+                    status_output_2 = gr.Textbox(label="Status Log", interactive=False, lines=4)
+                    file_output_2 = gr.File(label="📥 Download Ready Audiobook (.m4b)", file_count="multiple", interactive=False)
+                    
+            process_event_2 = start_btn_2.click(fn=process_files_to_audio, inputs=[file_in, speed_slider_2], outputs=[file_output_2, status_output_2])
+            stop_btn_2.click(fn=None, inputs=None, outputs=None, cancels=[process_event_2])
+
+if __name__ == "__main__":
+    app.queue().launch()