Create app.py

app.py

@@ -0,0 +1,310 @@
+import sys
+import os
+import tempfile
+from typing import List, Tuple, Optional, Dict
+from urllib.parse import urlparse
+import importlib
+import traceback
+
+# Base required imports
+import gradio as gr
+
+# Dictionary of required packages for each file type
+REQUIRED_PACKAGES = {
+    'url': ['langchain_community', 'requests', 'bs4'],
+    'pdf': ['langchain_community', 'pypdf'],
+    'docx': ['langchain_community', 'unstructured']
+}
+
+def check_and_import_packages(file_type: str) -> Tuple[bool, str, Optional[Exception]]:
+    """
+    Check and import required packages for a specific file type.
+    Returns (success, error_message, exception)
+    """
+    if file_type not in REQUIRED_PACKAGES:
+        return True, "", None
+        
+    missing_packages = []
+    for package in REQUIRED_PACKAGES[file_type]:
+        if not importlib.util.find_spec(package):
+            missing_packages.append(package)
+    
+    if missing_packages:
+        error_msg = (f"ERROR: Missing required packages for {file_type} processing:\n"
+                    f"  - Missing: {', '.join(missing_packages)}\n"
+                    f"  - Install with: pip install {' '.join(missing_packages)}")
+        return False, error_msg, None
+        
+    try:
+        if file_type == 'url':
+            from langchain_community.document_loaders import WebBaseLoader
+        elif file_type == 'pdf':
+            from langchain_community.document_loaders import PyPDFLoader
+        elif file_type == 'docx':
+            from langchain_community.document_loaders import UnstructuredWordDocumentLoader
+        return True, "", None
+    except Exception as e:
+        return False, f"ERROR: Failed to import required modules for {file_type}:\n  {str(e)}", e
+
+def count_characters(text: str) -> Dict[str, int]:
+    """Count characters in text."""
+    if not text:
+        return {
+            'total': 0,
+            'excluding_spaces': 0,
+            'japanese': 0
+        }
+    return {
+        'total': len(text),
+        'excluding_spaces': len(text.replace(' ', '').replace('\n', '').replace('\t', '')),
+        'japanese': len([c for c in text if '\u4e00' <= c <= '\u9fff' or '\u3040' <= c <= '\u309f' or '\u30a0' <= c <= '\u30ff'])
+    }
+
+def format_char_count(counts: Dict[str, int]) -> str:
+    """Format character count information."""
+    return (f"文字数（スペース・改行含む）: {counts['total']}\n"
+            f"文字数（スペース・改行除く）: {counts['excluding_spaces']}\n"
+            f"日本語文字数: {counts['japanese']}")
+
+def process_raw_text(text: str) -> Tuple[str, List[str]]:
+    """Process raw text input."""
+    errors = []
+    if not text or not text.strip():
+        return "", errors
+    
+    try:
+        return f"\n=== Raw Text Input ===\n{text.strip()}\n", errors
+    except Exception as e:
+        errors.append(f"ERROR: Failed to process raw text input:\n  {str(e)}")
+        return "", errors
+
+def is_valid_url(url: str) -> bool:
+    """Validate if the given string is a valid URL."""
+    try:
+        result = urlparse(url)
+        return all([result.scheme, result.netloc])
+    except:
+        return False
+
+def process_urls(urls: str) -> Tuple[str, List[str]]:
+    """Extract text from URLs using WebBaseLoader."""
+    errors = []
+    if not urls.strip():
+        return "", errors
+
+    success, error_msg, _ = check_and_import_packages('url')
+    if not success:
+        errors.append(error_msg)
+        return "", errors
+
+    from langchain_community.document_loaders import WebBaseLoader
+    
+    combined_text = ""
+    url_list = [url.strip() for url in urls.split('\n') if url.strip()]
+
+    for url in url_list:
+        if not is_valid_url(url):
+            errors.append(f"ERROR: Invalid URL format: {url}")
+            continue
+
+        try:
+            loader = WebBaseLoader(url)
+            documents = loader.load()
+            combined_text += f"\n=== Content from URL: {url} ===\n"
+            for doc in documents:
+                combined_text += doc.page_content + "\n"
+        except Exception as e:
+            errors.append(f"ERROR: Failed to process URL {url}:\n  {str(e)}\n  {traceback.format_exc()}")
+
+    return combined_text, errors
+
+def process_txt(txt_file: tempfile._TemporaryFileWrapper) -> Tuple[str, List[str]]:
+    """Process text files directly."""
+    errors = []
+    try:
+        with open(txt_file.name, 'r', encoding='utf-8') as f:
+            content = f.read()
+        return f"\n=== Content from TXT: {os.path.basename(txt_file.name)} ===\n{content}\n", errors
+    except UnicodeDecodeError:
+        try:
+            # Try with different encoding if UTF-8 fails
+            with open(txt_file.name, 'r', encoding='shift-jis') as f:
+                content = f.read()
+            return f"\n=== Content from TXT: {os.path.basename(txt_file.name)} ===\n{content}\n", errors
+        except Exception as e:
+            errors.append(f"ERROR: Failed to process text file {os.path.basename(txt_file.name)} (encoding error):\n  {str(e)}\n  {traceback.format_exc()}")
+            return "", errors
+    except Exception as e:
+        errors.append(f"ERROR: Failed to process text file {os.path.basename(txt_file.name)}:\n  {str(e)}\n  {traceback.format_exc()}")
+        return "", errors
+
+def process_pdf(pdf_file: tempfile._TemporaryFileWrapper) -> Tuple[str, List[str]]:
+    """Extract text from a PDF file using PyPDFLoader."""
+    errors = []
+    success, error_msg, _ = check_and_import_packages('pdf')
+    if not success:
+        errors.append(error_msg)
+        return "", errors
+
+    from langchain_community.document_loaders import PyPDFLoader
+
+    try:
+        loader = PyPDFLoader(pdf_file.name)
+        documents = loader.load()
+        
+        content = ""
+        for doc in documents:
+            content += doc.page_content + "\n"
+            
+        return f"\n=== Content from PDF: {os.path.basename(pdf_file.name)} ===\n{content}", errors
+    except Exception as e:
+        errors.append(f"ERROR: Failed to process PDF {os.path.basename(pdf_file.name)}:\n  {str(e)}\n  {traceback.format_exc()}")
+        return "", errors
+
+def process_docx(docx_file: tempfile._TemporaryFileWrapper) -> Tuple[str, List[str]]:
+    """Extract text from a DOCX file using UnstructuredWordDocumentLoader."""
+    errors = []
+    success, error_msg, _ = check_and_import_packages('docx')
+    if not success:
+        errors.append(error_msg)
+        return "", errors
+
+    from langchain_community.document_loaders import UnstructuredWordDocumentLoader
+
+    try:
+        loader = UnstructuredWordDocumentLoader(docx_file.name)
+        documents = loader.load()
+        
+        content = ""
+        for doc in documents:
+            content += doc.page_content + "\n"
+            
+        return f"\n=== Content from DOCX: {os.path.basename(docx_file.name)} ===\n{content}", errors
+    except Exception as e:
+        errors.append(f"ERROR: Failed to process DOCX {os.path.basename(docx_file.name)}:\n  {str(e)}\n  {traceback.format_exc()}")
+        return "", errors
+
+def process_file(file: tempfile._TemporaryFileWrapper) -> Tuple[str, List[str]]:
+    """Process a file based on its extension."""
+    errors = []
+    if not file:
+        return "", errors
+        
+    file_ext = os.path.splitext(file.name)[1].lower()
+    
+    # Process based on file extension
+    if file_ext == '.txt':
+        return process_txt(file)
+    elif file_ext == '.pdf':
+        return process_pdf(file)
+    elif file_ext in ['.doc', '.docx']:
+        return process_docx(file)
+    else:
+        return "", [f"ERROR: Unsupported file type: {file_ext}"]
+
+def combine_content(raw_text: str, url_input: str, files: List[tempfile._TemporaryFileWrapper]) -> Tuple[str, str, str, str]:
+    """Combine content from all sources into a single text file."""
+    combined_text = ""
+    all_errors = []
+    
+    # Process raw text if provided
+    if raw_text:
+        text_content, text_errors = process_raw_text(raw_text)
+        combined_text += text_content
+        all_errors.extend(text_errors)
+    
+    # Process URLs if provided
+    if url_input:
+        url_text, url_errors = process_urls(url_input)
+        combined_text += url_text
+        all_errors.extend(url_errors)
+    
+    # Process each uploaded file
+    if files:
+        for file in files:
+            file_text, file_errors = process_file(file)
+            combined_text += file_text
+            all_errors.extend(file_errors)
+
+    # Calculate character count
+    char_counts = count_characters(combined_text)
+    char_count_text = format_char_count(char_counts)
+
+    if not combined_text.strip():
+        if all_errors:
+            combined_text = "No content was extracted due to errors. Please check the error messages below."
+        else:
+            combined_text = "No content was extracted. Please provide some input (text, URLs, or files)."
+
+    # Format error messages
+    error_text = "\n".join(all_errors) if all_errors else "処理は正常に完了しました。"
+
+    # Save to temporary file for download
+    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as tmp_file:
+        tmp_file.write(combined_text)
+        output_path = tmp_file.name
+
+    return combined_text, output_path, char_count_text, error_text
+
+# Create Gradio interface
+with gr.Blocks(title="Document Content Extractor") as demo:
+    gr.Markdown("# Document Content Extractor")
+    gr.Markdown("""テキスト、URL、各種ドキュメントからテキストを抽出・結合するツールです。
+                
+対応ファイル形式:
+- テキストファイル (.txt)
+- PDFファイル (.pdf) - pypdfが必要
+- Wordドキュメント (.doc, .docx) - unstructuredが必要
+                
+必要なパッケージ:
+- URL処理用: langchain-community, requests, beautifulsoup4
+- PDF処理用: langchain-community, pypdf
+- DOCX処理用: langchain-community, unstructured""")
+
+    with gr.Row():
+        with gr.Column():
+            raw_text = gr.Textbox(
+                label="テキスト入力",
+                placeholder="直接テキストを入力できます...",
+                lines=5
+            )
+            url_input = gr.Textbox(
+                label="URL入力（1行に1つ）",
+                placeholder="URLを入力してください...",
+                lines=5
+            )
+            files = gr.File(
+                label="ファイルアップロード",
+                file_count="multiple",
+                file_types=[".txt", ".pdf", ".doc", ".docx"]
+            )
+            combine_btn = gr.Button("抽出・結合")
+
+        with gr.Column():
+            error_output = gr.Textbox(
+                label="エラー・警告",
+                lines=3,
+                interactive=False,
+                show_copy_button=True
+            )
+            char_count_output = gr.Textbox(
+                label="文字数",
+                lines=3,
+                interactive=False
+            )
+            text_output = gr.Textbox(
+                label="抽出されたテキスト",
+                lines=20,
+                interactive=False,
+                show_copy_button=True
+            )
+            file_output = gr.File(label="結合テキストをダウンロード")
+
+    combine_btn.click(
+        fn=combine_content,
+        inputs=[raw_text, url_input, files],
+        outputs=[text_output, file_output, char_count_output, error_output]
+    )
+
+if __name__ == "__main__":
+    demo.launch()