Buckets:
| #!/usr/bin/env python3 | |
| """ | |
| پردازشگر پیشرفته اسناد با Gemini API | |
| Advanced Document Processor with Gemini API | |
| """ | |
| import os | |
| import json | |
| import logging | |
| from typing import List, Dict, Any, Optional, Tuple | |
| from pathlib import Path | |
| import hashlib | |
| from datetime import datetime | |
| # LangChain imports | |
| from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings | |
| from langchain_community.document_loaders import ( | |
| TextLoader, PyPDFLoader, DirectoryLoader, | |
| CSVLoader, JSONLoader, WebBaseLoader, | |
| UnstructuredFileLoader, UnstructuredMarkdownLoader | |
| ) | |
| from langchain_community.vectorstores import Chroma, FAISS | |
| from langchain.text_splitter import ( | |
| RecursiveCharacterTextSplitter, | |
| TokenTextSplitter, | |
| SpacyTextSplitter, | |
| NLTKTextSplitter | |
| ) | |
| from langchain_core.documents import Document | |
| from langchain_core.prompts import PromptTemplate | |
| from langchain.chains import RetrievalQA, MapReduceDocumentsChain, ReduceDocumentsChain | |
| from langchain.chains.combine_documents.stuff import StuffDocumentsChain | |
| # Gemini configuration | |
| from gemini_config import get_gemini_config | |
| # تنظیم لاگ | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| class DocumentProcessor: | |
| """پردازشگر پیشرفته اسناد""" | |
| def __init__(self, model_name: str = "gemini-1.5-pro", temperature: float = 0.3): | |
| """مقداردهی اولیه""" | |
| self.model_name = model_name | |
| self.temperature = temperature | |
| self.config = get_gemini_config() | |
| # ایجاد مدل‌ها | |
| self.llm = ChatGoogleGenerativeAI( | |
| model=self.model_name, | |
| temperature=self.temperature, | |
| google_api_key=self.config.api_key, | |
| convert_system_message_to_human=True | |
| ) | |
| self.embeddings = GoogleGenerativeAIEmbeddings( | |
| model="models/embedding-001", | |
| google_api_key=self.config.api_key | |
| ) | |
| # تنظیمات پردازش | |
| self.processing_settings = { | |
| "chunk_size": 1000, | |
| "chunk_overlap": 200, | |
| "max_chunks": 1000, | |
| "min_chunk_size": 100, | |
| "enable_metadata": True, | |
| "enable_summarization": True | |
| } | |
| # تاریخچه پردازش | |
| self.processing_history = [] | |
| logger.info("✅ پردازشگر اسناد ایجاد شد") | |
| def update_settings(self, **kwargs): | |
| """به‌روزرسانی تنظیمات پردازش""" | |
| for key, value in kwargs.items(): | |
| if key in self.processing_settings: | |
| self.processing_settings[key] = value | |
| logger.info("✅ تنظیمات پردازش به‌روزرسانی شد") | |
| def load_document(self, file_path: str) -> List[Document]: | |
| """بارگذاری سند تکی""" | |
| try: | |
| file_path = Path(file_path) | |
| if not file_path.exists(): | |
| raise FileNotFoundError(f"فایل یافت نشد: {file_path}") | |
| # انتخاب loader مناسب | |
| loader = self._get_loader(file_path) | |
| if loader is None: | |
| raise ValueError(f"فرمت فایل پشتیبانی نمی‌شود: {file_path.suffix}") | |
| documents = loader.load() | |
| # اضافه کردن metadata | |
| for doc in documents: | |
| doc.metadata.update({ | |
| "source": str(file_path), | |
| "file_type": file_path.suffix, | |
| "file_size": file_path.stat().st_size, | |
| "processed_at": datetime.now().isoformat(), | |
| "file_hash": self._calculate_file_hash(file_path) | |
| }) | |
| logger.info(f"📄 {len(documents)} سند از {file_path} بارگذاری شد") | |
| return documents | |
| except Exception as e: | |
| logger.error(f"❌ خطا در بارگذاری {file_path}: {e}") | |
| raise | |
| def load_directory(self, directory_path: str, glob_pattern: str = "**/*") -> List[Document]: | |
| """بارگذاری پوشه اسناد""" | |
| try: | |
| directory_path = Path(directory_path) | |
| if not directory_path.exists(): | |
| raise FileNotFoundError(f"پوشه یافت نشد: {directory_path}") | |
| documents = [] | |
| file_count = 0 | |
| for file_path in directory_path.glob(glob_pattern): | |
| if file_path.is_file(): | |
| try: | |
| docs = self.load_document(str(file_path)) | |
| documents.extend(docs) | |
| file_count += 1 | |
| except Exception as e: | |
| logger.warning(f"خطا در بارگذاری {file_path}: {e}") | |
| continue | |
| logger.info(f"📁 {file_count} فایل از {directory_path} بارگذاری شد") | |
| return documents | |
| except Exception as e: | |
| logger.error(f"❌ خطا در بارگذاری پوشه {directory_path}: {e}") | |
| raise | |
| def _get_loader(self, file_path: Path): | |
| """انتخاب loader مناسب""" | |
| suffix = file_path.suffix.lower() | |
| if suffix == '.pdf': | |
| return PyPDFLoader(str(file_path)) | |
| elif suffix == '.txt': | |
| return TextLoader(str(file_path), encoding='utf-8') | |
| elif suffix == '.csv': | |
| return CSVLoader(str(file_path), encoding='utf-8') | |
| elif suffix == '.json': | |
| return JSONLoader(str(file_path), jq_schema='.', text_content=False) | |
| elif suffix == '.md': | |
| return UnstructuredMarkdownLoader(str(file_path)) | |
| else: | |
| # تلاش برای استفاده از UnstructuredFileLoader | |
| try: | |
| return UnstructuredFileLoader(str(file_path)) | |
| except: | |
| return None | |
| def _calculate_file_hash(self, file_path: Path) -> str: | |
| """محاسبه hash فایل""" | |
| try: | |
| with open(file_path, 'rb') as f: | |
| content = f.read() | |
| return hashlib.md5(content).hexdigest() | |
| except: | |
| return "unknown" | |
| def split_documents(self, documents: List[Document], method: str = "recursive") -> List[Document]: | |
| """تقسیم اسناد به بخش‌های کوچک‌تر""" | |
| try: | |
| if method == "recursive": | |
| splitter = RecursiveCharacterTextSplitter( | |
| chunk_size=self.processing_settings["chunk_size"], | |
| chunk_overlap=self.processing_settings["chunk_overlap"], | |
| length_function=len, | |
| separators=["\n\n", "\n", " ", ""] | |
| ) | |
| elif method == "token": | |
| splitter = TokenTextSplitter( | |
| chunk_size=self.processing_settings["chunk_size"], | |
| chunk_overlap=self.processing_settings["chunk_overlap"] | |
| ) | |
| else: | |
| raise ValueError(f"روش تقسیم نامعتبر: {method}") | |
| splits = splitter.split_documents(documents) | |
| # محدود کردن تعداد chunks | |
| if len(splits) > self.processing_settings["max_chunks"]: | |
| splits = splits[:self.processing_settings["max_chunks"]] | |
| logger.warning(f"تعداد chunks به {self.processing_settings['max_chunks']} محدود شد") | |
| # فیلتر کردن chunks کوچک | |
| min_size = self.processing_settings["min_chunk_size"] | |
| splits = [doc for doc in splits if len(doc.page_content) >= min_size] | |
| logger.info(f"📝 {len(splits)} بخش از {len(documents)} سند ایجاد شد") | |
| return splits | |
| except Exception as e: | |
| logger.error(f"❌ خطا در تقسیم اسناد: {e}") | |
| raise | |
| def summarize_documents(self, documents: List[Document]) -> Dict[str, str]: | |
| """خلاصه‌سازی اسناد""" | |
| try: | |
| summaries = {} | |
| for i, doc in enumerate(documents): | |
| if len(doc.page_content) > 500: # فقط اسناد طولانی | |
| prompt = f""" | |
| خلاصه‌ای از متن زیر ارائه دهید: | |
| {doc.page_content[:2000]}... | |
| خلاصه باید شامل: | |
| 1. موضوع اصلی | |
| 2. نکات کلیدی | |
| 3. نتیجه‌گیری | |
| خلاصه: | |
| """ | |
| summary = self.llm.invoke(prompt) | |
| summaries[f"document_{i}"] = summary.content | |
| # محدود کردن تعداد خلاصه‌ها | |
| if len(summaries) >= 10: | |
| break | |
| logger.info(f"📋 {len(summaries)} خلاصه ایجاد شد") | |
| return summaries | |
| except Exception as e: | |
| logger.error(f"❌ خطا در خلاصه‌سازی: {e}") | |
| return {} | |
| def extract_keywords(self, documents: List[Document], top_k: int = 20) -> List[str]: | |
| """استخراج کلمات کلیدی""" | |
| try: | |
| # ترکیب متن همه اسناد | |
| combined_text = " ".join([doc.page_content for doc in documents[:10]]) # محدود کردن | |
| prompt = f""" | |
| کلمات کلیدی مهم متن زیر را استخراج کنید: | |
| {combined_text[:3000]}... | |
| لطفاً {top_k} کلمه کلیدی مهم را لیست کنید: | |
| """ | |
| response = self.llm.invoke(prompt) | |
| # پردازش پاسخ | |
| keywords = [] | |
| for line in response.content.split('\n'): | |
| line = line.strip() | |
| if line and not line.startswith('#'): | |
| # حذف شماره‌ها و علائم | |
| keyword = ''.join(c for c in line if c.isalnum() or c.isspace()).strip() | |
| if keyword: | |
| keywords.append(keyword) | |
| keywords = keywords[:top_k] | |
| logger.info(f"🔑 {len(keywords)} کلمه کلیدی استخراج شد") | |
| return keywords | |
| except Exception as e: | |
| logger.error(f"❌ خطا در استخراج کلمات کلیدی: {e}") | |
| return [] | |
| def analyze_sentiment(self, documents: List[Document]) -> Dict[str, Any]: | |
| """تحلیل احساسات اسناد""" | |
| try: | |
| sentiments = [] | |
| for doc in documents[:5]: # محدود کردن | |
| prompt = f""" | |
| تحلیل احساسات متن زیر را انجام دهید: | |
| {doc.page_content[:1000]} | |
| لطفاً تحلیل را در قالب JSON ارائه دهید: | |
| {{ | |
| "sentiment": "positive/negative/neutral", | |
| "confidence": 0.0-1.0, | |
| "emotions": ["list", "of", "emotions"] | |
| }} | |
| """ | |
| try: | |
| response = self.llm.invoke(prompt) | |
| sentiment_data = json.loads(response.content) | |
| sentiments.append(sentiment_data) | |
| except: | |
| continue | |
| if sentiments: | |
| avg_confidence = sum(s['confidence'] for s in sentiments) / len(sentiments) | |
| dominant_sentiment = max(set(s['sentiment'] for s in sentiments), | |
| key=[s['sentiment'] for s in sentiments].count) | |
| result = { | |
| "dominant_sentiment": dominant_sentiment, | |
| "average_confidence": avg_confidence, | |
| "total_documents": len(sentiments), | |
| "sentiments": sentiments | |
| } | |
| logger.info(f"😊 تحلیل احساسات انجام شد: {dominant_sentiment}") | |
| return result | |
| return {"error": "هیچ تحلیل احساساتی انجام نشد"} | |
| except Exception as e: | |
| logger.error(f"❌ خطا در تحلیل احساسات: {e}") | |
| return {"error": str(e)} | |
| def create_vector_store(self, documents: List[Document], store_type: str = "chroma", | |
| persist_directory: str = "./vector_store") -> Any: | |
| """ایجاد vector store""" | |
| try: | |
| if store_type.lower() == "chroma": | |
| vector_store = Chroma.from_documents( | |
| documents=documents, | |
| embedding=self.embeddings, | |
| persist_directory=persist_directory | |
| ) | |
| elif store_type.lower() == "faiss": | |
| vector_store = FAISS.from_documents( | |
| documents=documents, | |
| embedding=self.embeddings | |
| ) | |
| vector_store.save_local(persist_directory) | |
| else: | |
| raise ValueError(f"نوع vector store پشتیبانی نمی‌شود: {store_type}") | |
| logger.info(f"✅ Vector store از نوع {store_type} ایجاد شد") | |
| return vector_store | |
| except Exception as e: | |
| logger.error(f"❌ خطا در ایجاد vector store: {e}") | |
| raise | |
| def process_documents(self, input_path: str, output_dir: str = "./processed_documents") -> Dict[str, Any]: | |
| """پردازش کامل اسناد""" | |
| try: | |
| input_path = Path(input_path) | |
| output_dir = Path(output_dir) | |
| output_dir.mkdir(exist_ok=True) | |
| # بارگذاری اسناد | |
| if input_path.is_file(): | |
| documents = self.load_document(str(input_path)) | |
| else: | |
| documents = self.load_directory(str(input_path)) | |
| if not documents: | |
| return {"error": "هیچ سندی یافت نشد"} | |
| # تقسیم اسناد | |
| chunks = self.split_documents(documents) | |
| # خلاصه‌سازی (اگر فعال باشد) | |
| summaries = {} | |
| if self.processing_settings["enable_summarization"]: | |
| summaries = self.summarize_documents(documents) | |
| # استخراج کلمات کلیدی | |
| keywords = self.extract_keywords(chunks) | |
| # تحلیل احساسات | |
| sentiment_analysis = self.analyze_sentiment(documents) | |
| # ایجاد vector store | |
| vector_store = self.create_vector_store(chunks, persist_directory=str(output_dir / "vector_store")) | |
| # ذخیره نتایج | |
| results = { | |
| "processing_info": { | |
| "input_path": str(input_path), | |
| "output_dir": str(output_dir), | |
| "processed_at": datetime.now().isoformat(), | |
| "total_documents": len(documents), | |
| "total_chunks": len(chunks), | |
| "settings": self.processing_settings | |
| }, | |
| "keywords": keywords, | |
| "summaries": summaries, | |
| "sentiment_analysis": sentiment_analysis, | |
| "chunks_preview": [chunk.page_content[:200] + "..." for chunk in chunks[:5]] | |
| } | |
| # ذخیره در فایل | |
| with open(output_dir / "processing_results.json", 'w', encoding='utf-8') as f: | |
| json.dump(results, f, ensure_ascii=False, indent=2) | |
| # ذخیره chunks | |
| chunks_data = [] | |
| for i, chunk in enumerate(chunks): | |
| chunks_data.append({ | |
| "id": i, | |
| "content": chunk.page_content, | |
| "metadata": chunk.metadata | |
| }) | |
| with open(output_dir / "chunks.json", 'w', encoding='utf-8') as f: | |
| json.dump(chunks_data, f, ensure_ascii=False, indent=2) | |
| logger.info(f"✅ پردازش کامل انجام شد. نتایج در {output_dir} ذخیره شد") | |
| # اضافه کردن به تاریخچه | |
| self.processing_history.append({ | |
| "timestamp": datetime.now().isoformat(), | |
| "input_path": str(input_path), | |
| "output_dir": str(output_dir), | |
| "documents_count": len(documents), | |
| "chunks_count": len(chunks) | |
| }) | |
| return results | |
| except Exception as e: | |
| logger.error(f"❌ خطا در پردازش اسناد: {e}") | |
| return {"error": str(e)} | |
| def get_processing_history(self) -> List[Dict[str, Any]]: | |
| """دریافت تاریخچه پردازش""" | |
| return self.processing_history | |
| def export_processing_report(self, filepath: str) -> None: | |
| """صادرات گزارش پردازش""" | |
| try: | |
| report = { | |
| "processor_info": { | |
| "model": self.model_name, | |
| "temperature": self.temperature, | |
| "settings": self.processing_settings | |
| }, | |
| "processing_history": self.processing_history, | |
| "export_time": datetime.now().isoformat() | |
| } | |
| with open(filepath, 'w', encoding='utf-8') as f: | |
| json.dump(report, f, ensure_ascii=False, indent=2) | |
| logger.info(f"✅ گزارش پردازش در {filepath} ذخیره شد") | |
| except Exception as e: | |
| logger.error(f"❌ خطا در صادرات گزارش: {e}") | |
| def demo_document_processor(): | |
| """نمایش پردازشگر اسناد""" | |
| print("📄 نمایش پردازشگر اسناد") | |
| print("=" * 40) | |
| try: | |
| # ایجاد پردازشگر | |
| processor = DocumentProcessor() | |
| # ایجاد اسناد نمونه | |
| sample_dir = Path("sample_docs") | |
| sample_dir.mkdir(exist_ok=True) | |
| sample_texts = [ | |
| "هوش مصنوعی شاخه‌ای از علوم کامپیوتر است که به ایجاد سیستم‌های هوشمند می‌پردازد.", | |
| "یادگیری ماشین زیرمجموعه‌ای از هوش مصنوعی است که به ماشین‌ها امکان یادگیری می‌دهد.", | |
| "پردازش زبان طبیعی یکی از کاربردهای مهم هوش مصنوعی در دنیای امروز است." | |
| ] | |
| for i, text in enumerate(sample_texts, 1): | |
| with open(sample_dir / f"doc_{i}.txt", 'w', encoding='utf-8') as f: | |
| f.write(text) | |
| print(f"✅ {len(sample_texts)} سند نمونه ایجاد شد") | |
| # پردازش اسناد | |
| results = processor.process_documents(str(sample_dir)) | |
| if "error" not in results: | |
| print(f"📊 نتایج پردازش:") | |
| print(f" • اسناد پردازش شده: {results['processing_info']['total_documents']}") | |
| print(f" • بخش‌های ایجاد شده: {results['processing_info']['total_chunks']}") | |
| print(f" • کلمات کلیدی: {len(results['keywords'])}") | |
| print(f" • خلاصه‌ها: {len(results['summaries'])}") | |
| print(f"\n🔑 کلمات کلیدی:") | |
| for keyword in results['keywords'][:5]: | |
| print(f" • {keyword}") | |
| if results['summaries']: | |
| print(f"\n📋 نمونه خلاصه:") | |
| first_summary = list(results['summaries'].values())[0] | |
| print(f" {first_summary[:200]}...") | |
| logger.info("✅ نمایش پردازشگر اسناد با موفقیت انجام شد") | |
| except Exception as e: | |
| logger.error(f"❌ خطا در نمایش پردازشگر: {e}") | |
| if __name__ == "__main__": | |
| demo_document_processor() | |
Xet Storage Details
- Size:
- 21 kB
- Xet hash:
- 6062e793cdf7e8221f11fe269a24b89ebd06b3219e133ea53d4f6c87d7ef66fd
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.