Spaces:
Running
Running
| import json | |
| import logging | |
| from app.services.chathistory import ChatSession | |
| from app.services.llm_model import Model | |
| from app.services.environmental_condition import EnvironmentalData | |
| from app.services.prompts import SKIN_CARE_SCHEDULER, DEFAULT_SCHEDULE | |
| class SkinCareScheduler: | |
| def __init__(self, token, session_id): | |
| self.token = token | |
| self.session_id = session_id | |
| self.chat_session = ChatSession(token, session_id) | |
| self.user_city = self.chat_session.get_city() or '' | |
| self.environment_data = EnvironmentalData(self.user_city) | |
| def get_historical_data(self): | |
| """Retrieve the last 7 days of schedules.""" | |
| schedules = self.chat_session.get_last_seven_days_schedules() | |
| return [schedule["schedule_data"] for schedule in schedules] | |
| def createTable(self): | |
| """Generate and return a daily skincare schedule.""" | |
| try: | |
| # Check for an existing valid schedule | |
| existing_schedule = self.chat_session.get_today_schedule() | |
| if existing_schedule and isinstance(existing_schedule.get("schedule_data"), list): | |
| return json.dumps(existing_schedule["schedule_data"], indent=2) | |
| # Gather input data | |
| historical_data = self.get_historical_data() | |
| personalized_condition = self.chat_session.get_personalized_recommendation() or "No specific skin conditions provided" | |
| environmental_data = self.environment_data.get_environmental_data() | |
| # Format the prompt | |
| formatted_prompt = SKIN_CARE_SCHEDULER.format( | |
| personalized_condition=personalized_condition, | |
| environmental_values=json.dumps(environmental_data, indent=2), | |
| historical_data=json.dumps(historical_data, indent=2) | |
| ) | |
| # Generate schedule with the model | |
| model = Model() | |
| result = model.skinScheduler(formatted_prompt) | |
| # Handle errors by falling back to default schedule | |
| if isinstance(result, dict) and "error" in result: | |
| logging.error(f"Model error: {result['error']}") | |
| result = DEFAULT_SCHEDULE | |
| # Validate basic structure (optional, but ensures 5 entries) | |
| if not isinstance(result, list) or len(result) != 5: | |
| logging.warning("Generated schedule invalid; using default.") | |
| result = DEFAULT_SCHEDULE | |
| # Save and return the schedule | |
| self.chat_session.save_schedule(result) | |
| return json.dumps(result, indent=2) | |
| except Exception as e: | |
| logging.error(f"Schedule generation failed: {str(e)}") | |
| return json.dumps(DEFAULT_SCHEDULE, indent=2) |