Spaces:
Runtime error
Runtime error
| 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) |