Spaces:
Running
Running
| """ | |
| Service pour exécution des analyses | |
| """ | |
| from sqlalchemy.orm import Session | |
| from uuid import UUID | |
| from app.db.repositories.analysis_repo import AnalysisRepository | |
| from app.db.models.analysis_run import AnalysisRun | |
| from datetime import datetime | |
| import json | |
| from app.core.logging import logger | |
| class AnalysisService: | |
| """Service pour orchestration des analyses K2 Think""" | |
| def __init__(self, db: Session): | |
| self.analysis_repo = AnalysisRepository(db) | |
| def create_analysis_run( | |
| self, | |
| project_id: UUID, | |
| model_used: str | |
| ): | |
| """Crée une nouvelle run d' d'analyse""" | |
| analysis = AnalysisRun( | |
| project_id=project_id, | |
| model_used=model_used, | |
| status="PENDING", | |
| started_at=datetime.utcnow() | |
| ) | |
| self.analysis_repo.create(analysis) | |
| return analysis | |
| def complete_analysis( | |
| self, | |
| analysis_id: UUID, | |
| result: dict = None, | |
| status: str = "COMPLETED" | |
| ): | |
| """Marque une analyse comme complétée""" | |
| analysis = self.analysis_repo.get_by_id(analysis_id) | |
| if analysis: | |
| analysis.status = status | |
| analysis.completed_at = datetime.utcnow() | |
| if result: | |
| analysis.result_data = result # SQLAlchemy handles JSON serialization | |
| self.analysis_repo.db.add(analysis) | |
| self.analysis_repo.db.commit() | |
| self.analysis_repo.db.refresh(analysis) | |
| return analysis | |
| async def process_analysis(self, analysis_id: str, request): | |
| """Tâche de fond pour traiter l'analyse""" | |
| from app.services.k2_think_engine import K2ThinkEngine | |
| from app.db.repositories.paper_repo import PaperRepository | |
| from app.db.models.research_paper import ResearchPaper | |
| from app.models.schemas import AnalysisRequest as K2AnalysisRequest, ScientificDocument | |
| from app.db.models.reasoning_trace import ReasoningTrace | |
| from app.services.arxiv_service import ArXivService | |
| from app.services.doaj_service import DOAJService | |
| from app.services.pubmed_service import PubMedService | |
| from app.services.openalex_service import OpenAlexService | |
| db = self.analysis_repo.db | |
| try: | |
| # 1. Fetch papers from DB or Remote | |
| paper_repo = PaperRepository(db) | |
| analysis_run = self.analysis_repo.get_by_id(UUID(analysis_id)) | |
| project_id = analysis_run.project_id | |
| docs = [] | |
| for pid in request.paper_ids: | |
| # Try UUID search first | |
| paper = None | |
| try: | |
| paper_uuid = UUID(pid) | |
| paper = paper_repo.get_by_id(paper_uuid) | |
| except (ValueError, AttributeError): | |
| # Try Remote ID search in DB | |
| paper = db.query(ResearchPaper).filter( | |
| ResearchPaper.remote_id == pid, | |
| ResearchPaper.project_id == project_id | |
| ).first() | |
| if not paper: | |
| # FETCH FROM REMOTE AND SAVE | |
| logger.info(f"Paper {pid} not in DB. Fetching metadata on-demand...") | |
| remote_data = None | |
| try: | |
| if "doaj_" in pid: | |
| svc = DOAJService(); remote_data = svc.fetch_papers(pid.replace("doaj_", ""), 1) | |
| elif "pubmed_" in pid: | |
| svc = PubMedService(); remote_data = svc.fetch_papers(pid.replace("pubmed_", ""), 1) | |
| elif "openalex_" in pid: | |
| svc = OpenAlexService(); remote_data = svc.fetch_papers(pid.replace("openalex_", ""), 1) | |
| elif "arxiv_" in pid: | |
| svc = ArXivService(); remote_data = svc.fetch_papers(pid.replace("arxiv_", ""), 1) | |
| else: | |
| # Fallback to ArXiv if no prefix (for old saved papers without prefix) | |
| svc = ArXivService(); remote_data = svc.fetch_papers(pid, 1) | |
| if remote_data: | |
| p_info = remote_data[0] | |
| snippet = p_info.get("content", p_info.get("summary", ""))[:10000] | |
| paper = ResearchPaper( | |
| project_id=project_id, | |
| remote_id=pid, | |
| title=p_info.get("title", "Unknown"), | |
| authors=", ".join(p_info.get("authors", [])), | |
| summary=snippet, | |
| publication_year=datetime.now().year # Fallback | |
| ) | |
| db.add(paper) | |
| db.commit() | |
| db.refresh(paper) | |
| logger.info(f"Saved remote paper {pid} to project {project_id}") | |
| else: | |
| logger.warning(f"No remote data found for {pid}") | |
| except Exception as fetch_err: | |
| logger.error(f"Failed to fetch/save remote paper {pid}: {fetch_err}") | |
| if paper: | |
| # Convert authors string to list | |
| author_list = [a.strip() for a in paper.authors.split(",")] if paper.authors else ["Unknown"] | |
| # Determine document type based on remote_id prefix | |
| dtype = "custom" | |
| if paper.remote_id: | |
| if "pubmed_" in paper.remote_id: dtype = "pubmed" | |
| elif "arxiv_" in paper.remote_id or "." in paper.remote_id: dtype = "arxiv" | |
| else: dtype = "arxiv" # default for most research platforms | |
| docs.append(ScientificDocument( | |
| id=str(paper.id), | |
| title=paper.title, | |
| authors=author_list, | |
| abstract=paper.summary or "", | |
| content=(paper.summary or "")[:10000], | |
| document_type=dtype, | |
| url=paper.pdf_path or "" | |
| )) | |
| if not docs: | |
| logger.error(f"No documents found for analysis {analysis_id}") | |
| raise ValueError("Zero documents identified for analysis. Aborting.") | |
| # 2. Prepare K2 Request | |
| k2_req = K2AnalysisRequest( | |
| documents=docs, | |
| user_id=request.user_id, | |
| user_profile=request.user_profile, | |
| reasoning_depth=request.reasoning_depth, | |
| ethics_rigor=request.ethics_rigor, | |
| info_density=request.info_density | |
| ) | |
| # 3. Call K2 Engine | |
| engine = K2ThinkEngine() | |
| result = await engine.process_analysis_request(k2_req) | |
| # 4. Save Reasoning Trace (Detailed audit log) | |
| trace = ReasoningTrace( | |
| analysis_id=UUID(analysis_id), | |
| step_number=1, | |
| reasoning=result.reasoning_summary or "Analysis complete", | |
| source_chunks=result.reasoning_trace # JSON field | |
| ) | |
| db.add(trace) | |
| # 5. Complete Analysis | |
| # result is a Pydantic model, convert to dict for storage | |
| self.complete_analysis(UUID(analysis_id), result=result.model_dump()) | |
| db.commit() | |
| except Exception as e: | |
| db.rollback() # CLEAN TRANSACTION | |
| logger.error(f"FATAL ERROR in process_analysis for {analysis_id}: {str(e)}") | |
| import traceback | |
| logger.error(traceback.format_exc()) | |
| try: | |
| error_data = { | |
| "status": "FAILED", | |
| "reasoning_summary": f"Erreur technique lors de l'analyse : {str(e)}\n\nTrace: {traceback.format_exc()[:500]}...", | |
| "confidence_overall": 0 | |
| } | |
| # Use a safe UUID conversion | |
| try: | |
| target_uuid = UUID(analysis_id) if isinstance(analysis_id, str) else analysis_id | |
| self.complete_analysis(target_uuid, status="FAILED", result=error_data) | |
| db.commit() | |
| logger.info(f"Marked analysis {analysis_id} as FAILED in DB") | |
| except Exception as uuid_err: | |
| logger.error(f"Could not convert {analysis_id} to UUID or save failure: {uuid_err}") | |
| except Exception as final_err: | |
| logger.error(f"Failed to even mark analysis as FAILED: {final_err}") | |
| db.rollback() | |