from uuid import UUID import uuid from sqlalchemy import select from src.repositories import BaseRepository from src.models import QueryType, CreateOrUpdateType, ProposalAIAnalysis, ProposalStrength, ProposalWeakness, ProposalRejectionReason, ProposalImprovement, Category class ProposalAIAnalysisClient: def __init__(self): self.repository = BaseRepository self.tables_mapping = { "STRENGTHS": ProposalStrength, "WEAKNESSES": ProposalWeakness, "REJECTION_REASONS": ProposalRejectionReason, "IMPROVEMENTS": ProposalImprovement, "AI_SCORE": ProposalAIAnalysis } async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_value, traceback): pass async def get_all_data(self, query_type: QueryType = QueryType.ALL, id: str = None, proposal_id: str = None, category: Category = None): async with self.repository(model=self.tables_mapping[query_type.name]).get_session() as session: query = select(self.tables_mapping[query_type.name]) if id: query = query.where(self.tables_mapping[query_type.name].id == id) if proposal_id: query = query.where(self.tables_mapping[query_type.name].proposal_id == proposal_id) if category: query = query.where(self.tables_mapping[query_type.name].category == category) result = await session.execute(query) all_result = result.scalars().all() final_result_dict = [{k: v for k, v in r.__dict__.items() if not k.startswith("_")} for r in all_result] return final_result_dict async def create_entry(self, query_type: QueryType = QueryType.ALL, data: dict = None): data["id"] = str(uuid.uuid4()) async with self.repository(model=self.tables_mapping[query_type.name]).get_session() as session: entry = self.tables_mapping[query_type.name](**data) session.add(entry) await session.commit() await session.refresh(entry) return entry async def create_data(self, query_type: QueryType = QueryType.ALL, create_or_update_type: CreateOrUpdateType = CreateOrUpdateType.ALL, data: dict = None): if create_or_update_type == CreateOrUpdateType.ID: result = await self.create_entry(query_type=query_type, data=data) return await self.get_all_data(query_type=query_type, id=result.id) else: for d in data: if d != "proposal_id" and d != "category": for point in data[d]: entry_data = {"proposal_id": data["proposal_id"], "category": data["category"], d: point} await self.create_entry(query_type=query_type, data=entry_data) return await self.get_all_data(query_type=query_type, proposal_id=data["proposal_id"], category=data["category"]) async def update_data(self, query_type: QueryType = QueryType.ALL, create_or_update_type: CreateOrUpdateType = CreateOrUpdateType.ALL, data: dict = None): if query_type == QueryType.AI_SCORE: if create_or_update_type == CreateOrUpdateType.ID: async with self.repository(model=self.tables_mapping[query_type.name]).get_session() as session: result = await session.execute(select(self.tables_mapping[query_type.name]).where(self.tables_mapping[query_type.name].id == data["id"])) all_result = result.scalars().first() for key, value in data.items(): setattr(all_result, key, value) await session.commit() await session.refresh(all_result) return await self.get_all_data(query_type=query_type, id=data["id"]) else: async with self.repository(model=self.tables_mapping[query_type.name]).get_session() as session: result = await session.execute(select(self.tables_mapping[query_type.name]).where(self.tables_mapping[query_type.name].proposal_id == data["proposal_id"])) all_result = result.scalars().first() for key, value in data.items(): setattr(all_result, key, value) await session.commit() await session.refresh(all_result) return await self.get_all_data(query_type=query_type, proposal_id=data["proposal_id"]) else: if create_or_update_type == CreateOrUpdateType.ID: async with self.repository(model=self.tables_mapping[query_type.name]).get_session() as session: result = await session.execute(select(self.tables_mapping[query_type.name]).where(self.tables_mapping[query_type.name].id == data["id"])) all_result = result.scalars().all() for r in all_result: for key, value in data.items(): setattr(r, key, value) await session.commit() await session.refresh(r) return await self.get_all_data(query_type=query_type, id=data["id"]) else: await self.delete_data(query_type=query_type, proposal_id=data["proposal_id"], category=data["category"]) return await self.create_data(query_type=query_type, create_or_update_type=create_or_update_type, data=data) async def delete_data(self, query_type: QueryType = QueryType.ALL, id: str = None, proposal_id: str = None, category: Category = None): async with self.repository(model=self.tables_mapping[query_type.name]).get_session() as session: query = select(self.tables_mapping[query_type.name]) if id: query = query.where(self.tables_mapping[query_type.name].id == id) if proposal_id: query = query.where(self.tables_mapping[query_type.name].proposal_id == proposal_id) if category: query = query.where(self.tables_mapping[query_type.name].category == category) result = await session.execute(query) all_result = result.scalars().all() for r in all_result: await session.delete(r) await session.commit() return True async def get_complete_analysis(self, proposal_id: str = None): analysis = { "ai_score": 0, "technical": { "strengths": [], "weaknesses": [], "rejection_reasons": [], "improvements": [] }, "management": { "strengths": [], "weaknesses": [], "rejection_reasons": [], "improvements": [] }, "past_performance": { "strengths": [], "weaknesses": [], "rejection_reasons": [], "improvements": [] }, "price": { "strengths": [], "weaknesses": [], "rejection_reasons": [], "improvements": [] } } result = await self.get_all_data(query_type=QueryType.AI_SCORE, proposal_id=proposal_id) analysis["ai_score"] = result[0]["ai_score"] for category in Category: strengths = await self.get_all_data(query_type=QueryType.STRENGTHS, proposal_id=proposal_id, category=category) analysis[category.value.lower()]["strengths"] = [s["strength_point"] for s in strengths] if strengths else [] weaknesses = await self.get_all_data(query_type=QueryType.WEAKNESSES, proposal_id=proposal_id, category=category) analysis[category.value.lower()]["weaknesses"] = [w["weakness_point"] for w in weaknesses] if weaknesses else [] rejection_reason = await self.get_all_data(query_type=QueryType.REJECTION_REASONS, proposal_id=proposal_id, category=category) analysis[category.value.lower()]["rejection_reasons"] = [r["reason"] for r in rejection_reason] if rejection_reason else [] improvements = await self.get_all_data(query_type=QueryType.IMPROVEMENTS, proposal_id=proposal_id, category=category) analysis[category.value.lower()]["improvements"] = [i["improvement_point"] for i in improvements] if improvements else [] # strengths = await self.get_all_data(query_type=QueryType.STRENGTHS, proposal_id=proposal_id) # analysis["strengths"] = [s["strength_point"] for s in strengths] # weakness = await self.get_all_data(query_type=QueryType.WEAKNESSES, proposal_id=proposal_id) # analysis["weaknesses"] = [w["weakness_point"] for w in weakness] # rejection_reason = await self.get_all_data(query_type=QueryType.REJECTION_REASONS, proposal_id=proposal_id) # analysis["rejection_reasons"] = [r["reason"] for r in rejection_reason] # improvements = await self.get_all_data(query_type=QueryType.IMPROVEMENTS, proposal_id=proposal_id) # analysis["improvements"] = [i["improvement_point"] for i in improvements] return analysis async def create_complete_analysis(self, data: dict = None): if "ai_score" in data: await self.create_data(query_type=QueryType.AI_SCORE, create_or_update_type=CreateOrUpdateType.ID, data={"proposal_id": data["proposal_id"], "ai_score": data["ai_score"]}) for category in Category: if "strengths" in data[category.value.lower()]: await self.create_data(query_type=QueryType.STRENGTHS, create_or_update_type=CreateOrUpdateType.ALL, data={"proposal_id": data["proposal_id"], "category": category, "strength_point": data[category.value.lower()]["strengths"]}) if "weaknesses" in data[category.value.lower()]: await self.create_data(query_type=QueryType.WEAKNESSES, create_or_update_type=CreateOrUpdateType.ALL, data={"proposal_id": data["proposal_id"], "category": category, "weakness_point": data[category.value.lower()]["weaknesses"]}) if "rejection_reasons" in data[category.value.lower()]: await self.create_data(query_type=QueryType.REJECTION_REASONS, create_or_update_type=CreateOrUpdateType.ALL, data={"proposal_id": data["proposal_id"], "category": category, "reason": data[category.value.lower()]["rejection_reasons"]}) if "improvements" in data[category.value.lower()]: await self.create_data(query_type=QueryType.IMPROVEMENTS, create_or_update_type=CreateOrUpdateType.ALL, data={"proposal_id": data["proposal_id"], "category": category, "improvement_point": data[category.value.lower()]["improvements"]}) # if "strengths" in data: # await self.create_data(query_type=QueryType.STRENGTHS, create_or_update_type=CreateOrUpdateType.ALL, data={"proposal_id": data["proposal_id"], "strength_point": data["strengths"]}) # if "weaknesses" in data: # await self.create_data(query_type=QueryType.WEAKNESSES, create_or_update_type=CreateOrUpdateType.ALL, data={"proposal_id": data["proposal_id"], "weakness_point": data["weaknesses"]}) # if "rejection_reasons" in data: # await self.create_data(query_type=QueryType.REJECTION_REASONS, create_or_update_type=CreateOrUpdateType.ALL, data={"proposal_id": data["proposal_id"], "reason": data["rejection_reasons"]}) # if "improvements" in data: # await self.create_data(query_type=QueryType.IMPROVEMENTS, create_or_update_type=CreateOrUpdateType.ALL, data={"proposal_id": data["proposal_id"], "improvement_point": data["improvements"]}) return await self.get_complete_analysis(proposal_id=data["proposal_id"]) async def delete_complete_analysis(self, proposal_id: str = None): await self.delete_data(query_type=QueryType.AI_SCORE, proposal_id=proposal_id) await self.delete_data(query_type=QueryType.STRENGTHS, proposal_id=proposal_id) await self.delete_data(query_type=QueryType.WEAKNESSES, proposal_id=proposal_id) await self.delete_data(query_type=QueryType.REJECTION_REASONS, proposal_id=proposal_id) await self.delete_data(query_type=QueryType.IMPROVEMENTS, proposal_id=proposal_id) return True async def update_complete_analysis(self, data: dict = None): old_data = await self.get_complete_analysis(proposal_id=data["proposal_id"]) if "ai_score" in data: old_data["ai_score"] = data["ai_score"] for category in Category: if category.value.lower() not in data: continue if "strengths" in data[category.value.lower()]: old_data[category.value.lower()]["strengths"] = data[category.value.lower()]["strengths"] if "weaknesses" in data[category.value.lower()]: old_data[category.value.lower()]["weaknesses"] = data[category.value.lower()]["weaknesses"] if "rejection_reasons" in data[category.value.lower()]: old_data[category.value.lower()]["rejection_reasons"] = data[category.value.lower()]["rejection_reasons"] if "improvements" in data[category.value.lower()]: old_data[category.value.lower()]["improvements"] = data[category.value.lower()]["improvements"] # if "strengths" in data: # old_data["strengths"] = data["strengths"] # if "weaknesses" in data: # old_data["weaknesses"] = data["weaknesses"] # if "rejection_reasons" in data: # old_data["rejection_reasons"] = data["rejection_reasons"] # if "improvements" in data: # old_data["improvements"] = data["improvements"] await self.delete_complete_analysis(proposal_id=data["proposal_id"]) old_data["proposal_id"] = data["proposal_id"] return await self.create_complete_analysis(data=old_data)