เอกสาร Parsing ด้วย AI: การเชื่อมต่อ PDF Word Excel เข้า Claude API สำหรับ Production

Document Processor: รองรับ PDF, Word, Excel

# services/document_processor.py
from abc import ABC, abstractmethod
from typing import Optional, Dict, Any
import docx
import pdfplumber
import openpyxl
from pathlib import Path

class DocumentProcessor(ABC):
    @abstractmethod
    def read(self, file_path: str) -> str:
        pass
    
    @abstractmethod
    def get_metadata(self) -> Dict[str, Any]:
        pass

class PDFProcessor(DocumentProcessor):
    def __init__(self):
        self._metadata = {}
    
    def read(self, file_path: str) -> str:
        text_content = []
        with pdfplumber.open(file_path) as pdf:
            self._metadata = {
                "page_count": len(pdf.pages),
                "metadata": pdf.metadata
            }
            for page in pdf.pages:
                text = page.extract_text()
                if text:
                    text_content.append(text)
        
        # รวม Tables ด้วย
        tables = []
        with pdfplumber.open(file_path) as pdf:
            for page in pdf.pages:
                page_tables = page.extract_tables()
                if page_tables:
                    for table in page_tables:
                        tables.append(self._format_table(table))
        
        full_content = "\n\n".join(text_content)
        if tables:
            full_content += "\n\n[TABLES]\n" + "\n".join(tables)
        
        return full_content
    
    def _format_table(self, table: list) -> str:
        """Format Table เป็น Text ที่ AI อ่านง่าย"""
        rows = []
        for row in table:
            rows.append(" | ".join(str(cell) if cell else "" for cell in row))
        return "\n".join(rows)
    
    def get_metadata(self) -> Dict[str, Any]:
        return self._metadata

class WordProcessor(DocumentProcessor):
    def read(self, file_path: str) -> str:
        doc = docx.Document(file_path)
        paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
        
        # ดึงข้อมูลจาก Tables
        tables_content = []
        for table in doc.tables:
            table_text = []
            for row in table.rows:
                row_text = [cell.text for cell in row.cells]
                table_text.append(" | ".join(row_text))
            tables_content.append("\n".join(table_text))
        
        full_content = "\n\n".join(paragraphs)
        if tables_content:
            full_content += "\n\n[TABLES]\n" + "\n".join(tables_content)
        
        return full_content
    
    def get_metadata(self) -> Dict[str, Any]:
        return {"type": "docx"}

class ExcelProcessor(DocumentProcessor):
    def read(self, file_path: str) -> str:
        wb = openpyxl.load_workbook(file_path, data_only=True)
        content_parts = []
        
        for sheet_name in wb.sheetnames:
            sheet = wb[sheet_name]
            content_parts.append(f"[Sheet: {sheet_name}]")
            
            for row in sheet.iter_rows(values_only=True):
                row_values = [str(cell) if cell is not None else "" for cell in row]
                if any(row_values):
                    content_parts.append(" | ".join(row_values))
        
        return "\n".join(content_parts)
    
    def get_metadata(self) -> Dict[str, Any]:
        return {"type": "xlsx"}

class DocumentProcessorFactory:
    @staticmethod
    def get_processor(file_path: str) -> DocumentProcessor:
        extension = Path(file_path).suffix.lower()
        
        processors = {
            ".pdf": PDFProcessor,
            ".docx": WordProcessor,
            ".doc": WordProcessor,
            ".xlsx": ExcelProcessor,
            ".xls": ExcelProcessor,
        }
        
        processor_class = processors.get(extension)
        if not processor_class:
            raise ValueError(f"ไม่รองรับไฟล์ประเภท: {extension}")
        
        return processor_class()

Concurrency Control: Async Queue สำหรับ Batch Processing

# services/concurrency_controller.py
import asyncio
from typing import List, Callable, Any
from dataclasses import dataclass, field
from datetime import datetime
import time

@dataclass
class ProcessingTask:
    task_id: str
    document_path: str
    schema: str
    status: str = "pending"
    result: Any = None
    error: str = None
    created_at: datetime = field(default_factory=datetime.now)
    completed_at: datetime = None

class AsyncQueueProcessor:
    """
    ระบบจัดการคิวแบบ Asynchronous สำหรับประมวลผลเอกสารจำนวนมาก
    ควบคุม Concurrency เพื่อไม่ให้เกิน Rate Limit
    """
    
    def __init__(self, max_concurrent: int = 10):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.tasks: List[ProcessingTask] = []
        self._lock = asyncio.Lock()
    
    async def process_batch(
        self, 
        tasks: List[ProcessingTask],
        processor_func: Callable
    ) -> List[ProcessingTask]:
        """ประมวลผล Batch พร้อมกันด้วย Semaphore Control"""
        
        async def process_with_semaphore(task: ProcessingTask):
            async with self.semaphore:
                task.status = "processing"
                try:
                    start_time = time.time()
                    result = await processor_func(task)
                    task.result = result
                    task.status = "completed"
                    task.completed_at = datetime.now()
                    print(f"✓ Task {task.task_id} เสร็จใน {time.time() - start_time:.2f}s")
                except Exception as e:
                    task.status = "failed"
                    task.error = str(e)
                    print(f"✗ Task {task.task_id} ล้มเหลว: {e}")
        
        # สร้าง Coroutines ทั้งหมด
        coroutines = [process_with_semaphore(task) for task in tasks]
        
        # รันพร้อมกันด้วย gather
        await asyncio.gather(*coroutines, return_exceptions=True)
        
        return tasks
    
    async def get_progress(self) -> dict:
        """ดึงสถานะความคืบหน้า"""
        total = len(self.tasks)
        completed = sum(1 for t in self.tasks if t.status == "completed")
        failed = sum(1 for t in self.tasks if t.status == "failed")
        
        return {
            "total": total,
            "completed": completed,
            "failed": failed,
            "pending": total - completed - failed,
            "progress_percent": (completed / total * 100) if total > 0 else 0
        }

ตัวอย่างการใช้งาน
async def main():
    processor = AsyncQueueProcessor(max_concurrent=5)
    
    tasks = [
        ProcessingTask(f"task_{i}", f"document_{i}.pdf", invoice_schema)
        for i in range(100)
    ]
    
    # Process 100 เอกสารพร้อมก
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
th southeast asia developers low latency ai api setup 2026 0
th deepseek r1 tuilimoxing api diaoyongyucanshupeizhi 2026 0