| """ |
| PostProcessor handles cleaning and normalizing the generated nutrition plan. |
| """ |
| import logging |
| import re |
|
|
| logger = logging.getLogger(__name__) |
|
|
| class PostProcessor: |
|
|
| @staticmethod |
| def create_processor() -> "PostProcessor": |
| return PostProcessor() |
|
|
| def validate_and_clean(self, generated_text: str) -> str: |
| if "DISCLAIMER" not in generated_text: |
| generated_text += "\n\nDISCLAIMER\nThis nutrition plan is AI-generated guidance and not a medical prescription. Consult a veterinarian for medical conditions or major diet changes." |
|
|
| generated_text = self._ensure_plain_text(generated_text) |
|
|
| logger.info("Post-processing completed") |
| return generated_text |
| |
| def extract_nutrition_values(self, final_plan: str) -> dict: |
| values = {} |
| lines = final_plan.split('\n') |
| for line in lines: |
| line = line.strip() |
| if line.startswith('RER:'): |
| try: |
| values['rer'] = float(line.split('kcal/day')[0].split('RER:')[1].strip()) |
| except: |
| pass |
| elif line.startswith('MER:'): |
| try: |
| values['mer'] = float(line.split('kcal/day')[0].split('MER:')[1].strip()) |
| except: |
| pass |
| elif line.startswith('Daily Calorie Target:'): |
| try: |
| values['daily_calorie_target'] = float(line.split('kcal/day')[0].split('Daily Calorie Target:')[1].strip()) |
| except: |
| pass |
| elif line.startswith('Protein:') and 'g/day' in line: |
| try: |
| values['protein_g'] = float(line.split('g/day')[0].split('Protein:')[1].strip()) |
| except: |
| pass |
| elif line.startswith('Fat:') and 'g/day' in line: |
| try: |
| values['fat_g'] = float(line.split('g/day')[0].split('Fat:')[1].strip()) |
| except: |
| pass |
| elif line.startswith('Carbohydrates:') and 'g/day' in line: |
| try: |
| values['carbohydrates_g'] = float(line.split('g/day')[0].split('Carbohydrates:')[1].strip()) |
| except: |
| pass |
| elif line.startswith('Water Intake:'): |
| try: |
| values['water_ml'] = float(line.split('ml/day')[0].split('Water Intake:')[1].strip()) |
| except: |
| pass |
| return values |
|
|
| def parse_plan_sections(self, final_plan: str) -> dict: |
| sections = {} |
| lines = final_plan.split('\n') |
| current_section = None |
| section_content = [] |
|
|
| for line in lines: |
| line = line.strip() |
| if line.startswith('PET NUTRITION PLAN'): |
| current_section = 'header' |
| section_content = [line] |
| elif line.startswith('DAILY NUTRITION TARGETS'): |
| if current_section: |
| sections[current_section] = '\n'.join(section_content).strip() |
| current_section = 'nutrition_targets' |
| section_content = [line] |
| elif line.startswith('RECOMMENDED MEAL PLAN'): |
| if current_section: |
| sections[current_section] = '\n'.join(section_content).strip() |
| current_section = 'meal_plan' |
| section_content = [line] |
| elif line.startswith('FEEDING INSTRUCTIONS'): |
| if current_section: |
| sections[current_section] = '\n'.join(section_content).strip() |
| current_section = 'feeding_instructions' |
| section_content = [line] |
| elif line.startswith('FOOD CATEGORIES'): |
| if current_section: |
| sections[current_section] = '\n'.join(section_content).strip() |
| current_section = 'food_categories' |
| section_content = [line] |
| elif line.startswith('SAFETY NOTES'): |
| if current_section: |
| sections[current_section] = '\n'.join(section_content).strip() |
| current_section = 'safety_notes' |
| section_content = [line] |
| elif line.startswith('DISCLAIMER'): |
| if current_section: |
| sections[current_section] = '\n'.join(section_content).strip() |
| current_section = 'disclaimer' |
| section_content = [line] |
| else: |
| if current_section: |
| section_content.append(line) |
|
|
| if current_section: |
| sections[current_section] = '\n'.join(section_content).strip() |
|
|
| return sections |
|
|
| def _ensure_plain_text(self, text: str) -> str: |
| text = re.sub(r'^#+\s+', '', text, flags=re.MULTILINE) |
| text = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', text) |
| text = re.sub(r'\*\*([^\*]+)\*\*', r'\1', text) |
| text = re.sub(r'\*([^\*]+)\*', r'\1', text) |
| return text |
| |
| def format_sections_for_api(self, sections: dict) -> dict: |
| formatted = {} |
| for key, value in sections.items(): |
| if key == 'food_categories': |
| formatted[key] = self._parse_food_categories(value) |
| else: |
| formatted[key] = [line for line in value.split('\n') if line.strip()] |
| return formatted |
| |
| def _parse_food_categories(self, text: str) -> dict: |
| |
| result = { |
| "dangerous": [], |
| "avoid": [], |
| "safe": [] |
| } |
| |
| lines = text.split('\n') |
| current_category = None |
| |
| for line in lines: |
| line = line.strip() |
| if not line: |
| continue |
| |
| if line.lower().startswith('dangerous:'): |
| current_category = 'dangerous' |
| items_text = line.split(':', 1)[1].strip() |
| if items_text: |
| items = [item.strip() for item in items_text.split(',') if item.strip()] |
| result['dangerous'].extend(items) |
| elif line.lower().startswith('avoid:'): |
| current_category = 'avoid' |
| items_text = line.split(':', 1)[1].strip() |
| if items_text: |
| items = [item.strip() for item in items_text.split(',') if item.strip()] |
| result['avoid'].extend(items) |
| elif line.lower().startswith('safe:'): |
| current_category = 'safe' |
| items_text = line.split(':', 1)[1].strip() |
| if items_text: |
| items = [item.strip() for item in items_text.split(',') if item.strip()] |
| result['safe'].extend(items) |
| else: |
| if current_category: |
| items = [item.strip() for item in line.split(',') if item.strip()] |
| result[current_category].extend(items) |
| |
| return result |
|
|
