import gradio as gr import os from typing import List, Dict, Tuple, Optional import pandas as pd import logging from pathlib import Path from datetime import datetime import asyncio from functools import partial from file_processor import FileProcessor from criteria_analyzer import CriteriaAnalyzer from offer_analyzer import OfferAnalyzer from report_generator import ReportGenerator from cache_manager import CacheManager from config import Config from dotenv import load_dotenv # Załaduj zmienne środowiskowe load_dotenv() # Konfiguracja logowania logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('tender_eval.log'), logging.StreamHandler() ] ) class TenderEvaluationApp: """Główna klasa aplikacji do oceny ofert przetargowych""" def __init__(self, api_key: Optional[str] = None): self.logger = logging.getLogger(__name__) # Użyj przekazanego klucza API lub pobierz z konfiguracji self.api_key = api_key or Config.get_api_config().api_key if not self.api_key: raise ValueError("Brak klucza API") self.base_url = Config.get_api_config().base_url # Utworzenie katalogów self.temp_dir = Path(Config.TEMP_DIR) self.output_dir = Path(Config.OUTPUT_DIR) self.temp_dir.mkdir(exist_ok=True) self.output_dir.mkdir(exist_ok=True) # Inicjalizacja cache self.cache = CacheManager( cache_dir="cache/main", ttl=Config.SYSTEM_CONFIG['cache_ttl'], max_size_mb=Config.SYSTEM_CONFIG['memory_limit'] ) # Inicjalizacja komponentów self.file_processor = FileProcessor(Config.FILE_CONFIG) self.criteria_analyzer = CriteriaAnalyzer( api_key=self.api_key, base_url=self.base_url ) self.offer_analyzer = OfferAnalyzer( api_key=self.api_key, base_url=self.base_url ) self.report_generator = ReportGenerator( output_dir=self.output_dir ) async def process_files( self, brief_file: str, offer_files: List[str], progress = gr.Progress() ) -> Tuple[pd.DataFrame, str, str]: """ Główna funkcja przetwarzająca pliki i generująca analizę Args: brief_file: Ścieżka do pliku z briefem/SIWZ offer_files: Lista ścieżek do plików z ofertami progress: Obiekt do raportowania postępu Returns: Tuple[pd.DataFrame, str, str]: (tabela_wyników, raport_word, status) """ status_updates = [] try: # 1. Przetwórz brief/SIWZ self.logger.info(f"Rozpoczęto przetwarzanie briefu: {brief_file}") status_updates.append("Przetwarzanie briefu...") progress(0.1, desc="Przetwarzanie briefu") brief_content = await self.file_processor.convert_to_text(brief_file) # 2. Wyodrębnij kryteria z briefu status_updates.append("Analizowanie kryteriów oceny...") progress(0.2, desc="Analizowanie kryteriów") async with self.criteria_analyzer: criteria = await self.criteria_analyzer.extract_criteria(brief_content) self.logger.info(f"Wyodrębniono {len(criteria)} kryteriów") # 3. Przetwórz oferty offers_content = {} total_offers = len(offer_files) for i, offer_file in enumerate(offer_files, 1): offer_name = Path(offer_file).name status_updates.append(f"Przetwarzanie oferty {i}/{total_offers}: {offer_name}") progress(0.2 + 0.2 * (i/total_offers), desc=f"Konwersja oferty {i}/{total_offers}") content = await self.file_processor.convert_to_text(offer_file) offers_content[f"Oferta_{i}"] = content # 4. Analizuj oferty status_updates.append("Analizowanie ofert...") progress(0.4, desc="Analiza ofert") async with self.offer_analyzer: analyses = await self.offer_analyzer.analyze_multiple_offers( offers_content, criteria, brief_content ) # 5. Porównaj oferty status_updates.append("Porównywanie ofert...") progress(0.8, desc="Porównywanie ofert") comparison = await self.offer_analyzer.compare_offers(analyses) # 6. Generuj raporty status_updates.append("Generowanie raportów końcowych...") progress(0.9, desc="Generowanie raportów") results_df = await self.report_generator.generate_excel_report( analyses, criteria ) word_report = await self.report_generator.generate_word_report( analyses, criteria, comparison ) # 7. Zapisz wyniki timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") excel_path = self.output_dir / f"wyniki_{timestamp}.xlsx" results_df.to_excel(excel_path, index=False) status_updates.append("✅ Analiza zakończona!") progress(1.0, desc="Zakończono") return results_df, word_report, "\n".join(status_updates) except Exception as e: error_msg = f"❌ Błąd podczas analizy: {str(e)}" self.logger.error(error_msg, exc_info=True) status_updates.append(error_msg) return None, None, "\n".join(status_updates) async def clear_cache(self): """Czyści cały cache aplikacji""" try: await self.cache.invalidate_all() self.logger.info("Wyczyszczono cache aplikacji") return "✅ Cache wyczyszczony pomyślnie" except Exception as e: self.logger.error(f"Błąd podczas czyszczenia cache: {str(e)}") return "❌ Błąd podczas czyszczenia cache" def create_interface() -> gr.Blocks: """ Tworzy interfejs Gradio Returns: gr.Blocks: Interfejs aplikacji """ async def process_with_api_key( api_key: str, brief_file: str, offer_files: List[str], progress=gr.Progress() ) -> Tuple[pd.DataFrame, str, str]: """Funkcja przetwarzająca pliki z walidacją klucza API""" try: # Najpierw sprawdź wprowadzony klucz if api_key: if not Config.validate_api_key(api_key): raise gr.Error("Nieprawidłowy format klucza API") active_key = api_key # Jeśli nie wprowadzono klucza, sprawdź czy jest w konfiguracji elif Config.get_api_config().api_key: active_key = Config.get_api_config().api_key else: raise gr.Error("Proszę wprowadzić klucz API") app = TenderEvaluationApp(api_key=active_key) return await app.process_files(brief_file, offer_files, progress) except Exception as e: raise gr.Error(f"Błąd: {str(e)}") async def clear_cache(): """Funkcja czyszcząca cache""" try: app = TenderEvaluationApp() return await app.clear_cache() except Exception as e: return f"❌ Błąd: {str(e)}" with gr.Blocks(title="Analiza i ocena ofert przetargowych") as interface: gr.Markdown(""" # Analiza i ocena ofert przetargowych Wgraj plik z briefem lub SIWZ oraz pliki z ofertami do oceny. System przeanalizuje dokumenty i wygeneruje tabelę ocen oraz podsumowanie. """) with gr.Row(): with gr.Column(scale=1): api_key = gr.Textbox( label="DeepSeek API Key", type="password", placeholder="Wprowadź swój klucz API...", value="" # Pole zawsze puste na starcie ) brief_input = gr.File( label="Brief/SIWZ", file_types=[".pdf", ".docx", ".txt"], type="filepath" ) offers_input = gr.File( label="Oferty", file_types=[".pptx", ".pdf", ".docx"], type="filepath", file_count="multiple" ) analyze_btn = gr.Button( "Analizuj dokumenty", variant="primary" ) clear_cache_btn = gr.Button( "Wyczyść cache", variant="secondary" ) with gr.Column(scale=2): with gr.Tab("Wyniki"): results_df = gr.DataFrame( label="Oceny", interactive=False ) word_output = gr.Markdown() with gr.Tab("Status"): status_output = gr.Markdown() cache_status = gr.Markdown() analyze_btn.click( fn=process_with_api_key, inputs=[api_key, brief_input, offers_input], outputs=[results_df, word_output, status_output] ) clear_cache_btn.click( fn=clear_cache, inputs=[], outputs=[cache_status] ) gr.Markdown(""" ### Wspierane formaty - Brief/SIWZ: PDF, DOCX, TXT - Oferty: PPTX, PDF, DOCX ### Uwaga Jeśli nie posiadasz osobistego klucza API do modelu DeepSeek, to możesz go wygenerować tutaj: https://platform.deepseek.com/ """) return interface if __name__ == "__main__": interface = create_interface() interface.launch( server_name="0.0.0.0", server_port=7860, show_api=False )