File size: 7,029 Bytes
ee4ef53 345991e ee4ef53 345991e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | """
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
|