368 lines
15 KiB
Python
368 lines
15 KiB
Python
|
"""
|
||
|
|
数据采集模块 - 基于基座架构
|
||
|
|
负责收集用户数据、问卷和餐食记录
|
||
|
|
"""
|
||
|
|
|
||
|
|
from typing import Dict, List, Optional, Any
|
||
|
|
import json
|
||
|
|
from datetime import datetime, date
|
||
|
|
from core.base import BaseModule, ModuleType, UserData, AnalysisResult, BaseConfig
|
||
|
|
|
||
|
|
|
||
|
|
class DataCollectionModule(BaseModule):
|
||
|
|
"""数据采集模块"""
|
||
|
|
|
||
|
|
def __init__(self, config: BaseConfig):
|
||
|
|
super().__init__(config, ModuleType.DATA_COLLECTION)
|
||
|
|
self.questionnaire_templates = self._load_questionnaire_templates()
|
||
|
|
|
||
|
|
def initialize(self) -> bool:
|
||
|
|
"""初始化模块"""
|
||
|
|
try:
|
||
|
|
self.logger.info("数据采集模块初始化中...")
|
||
|
|
self.is_initialized = True
|
||
|
|
self.logger.info("数据采集模块初始化完成")
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
self.logger.error(f"数据采集模块初始化失败: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
def process(self, input_data: Any, user_data: UserData) -> AnalysisResult:
|
||
|
|
"""处理数据采集请求"""
|
||
|
|
try:
|
||
|
|
request_type = input_data.get('type', 'unknown')
|
||
|
|
|
||
|
|
if request_type == 'questionnaire':
|
||
|
|
result = self._process_questionnaire(input_data, user_data)
|
||
|
|
elif request_type == 'meal_record':
|
||
|
|
result = self._process_meal_record(input_data, user_data)
|
||
|
|
elif request_type == 'feedback':
|
||
|
|
result = self._process_feedback(input_data, user_data)
|
||
|
|
else:
|
||
|
|
result = self._create_error_result("未知的请求类型")
|
||
|
|
|
||
|
|
return AnalysisResult(
|
||
|
|
module_type=self.module_type,
|
||
|
|
user_id=user_data.user_id,
|
||
|
|
input_data=input_data,
|
||
|
|
result=result,
|
||
|
|
confidence=0.9 if result.get('success', False) else 0.1
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
self.logger.error(f"处理数据采集请求失败: {e}")
|
||
|
|
return self._create_error_result(str(e))
|
||
|
|
|
||
|
|
def _process_questionnaire(self, input_data: Dict, user_data: UserData) -> Dict[str, Any]:
|
||
|
|
"""处理问卷数据"""
|
||
|
|
questionnaire_type = input_data.get('questionnaire_type', 'basic')
|
||
|
|
answers = input_data.get('answers', {})
|
||
|
|
|
||
|
|
# 根据问卷类型处理答案
|
||
|
|
if questionnaire_type == 'basic':
|
||
|
|
processed_data = self._process_basic_questionnaire(answers)
|
||
|
|
elif questionnaire_type == 'taste':
|
||
|
|
processed_data = self._process_taste_questionnaire(answers)
|
||
|
|
elif questionnaire_type == 'physiological':
|
||
|
|
processed_data = self._process_physiological_questionnaire(answers)
|
||
|
|
else:
|
||
|
|
processed_data = answers
|
||
|
|
|
||
|
|
# 更新用户数据
|
||
|
|
user_data.profile.update(processed_data)
|
||
|
|
user_data.updated_at = datetime.now().isoformat()
|
||
|
|
|
||
|
|
return {
|
||
|
|
'success': True,
|
||
|
|
'processed_data': processed_data,
|
||
|
|
'message': f'{questionnaire_type}问卷处理完成'
|
||
|
|
}
|
||
|
|
|
||
|
|
def _process_meal_record(self, input_data: Dict, user_data: UserData) -> Dict[str, Any]:
|
||
|
|
"""处理餐食记录"""
|
||
|
|
meal_data = {
|
||
|
|
'date': input_data.get('date', datetime.now().strftime('%Y-%m-%d')),
|
||
|
|
'meal_type': input_data.get('meal_type', 'unknown'),
|
||
|
|
'foods': input_data.get('foods', []),
|
||
|
|
'quantities': input_data.get('quantities', []),
|
||
|
|
'calories': input_data.get('calories'),
|
||
|
|
'satisfaction_score': input_data.get('satisfaction_score'),
|
||
|
|
'notes': input_data.get('notes', ''),
|
||
|
|
'timestamp': datetime.now().isoformat()
|
||
|
|
}
|
||
|
|
|
||
|
|
# 添加到用户餐食记录
|
||
|
|
user_data.meals.append(meal_data)
|
||
|
|
user_data.updated_at = datetime.now().isoformat()
|
||
|
|
|
||
|
|
return {
|
||
|
|
'success': True,
|
||
|
|
'meal_data': meal_data,
|
||
|
|
'message': '餐食记录保存成功'
|
||
|
|
}
|
||
|
|
|
||
|
|
def _process_feedback(self, input_data: Dict, user_data: UserData) -> Dict[str, Any]:
|
||
|
|
"""处理用户反馈"""
|
||
|
|
feedback_data = {
|
||
|
|
'date': input_data.get('date', datetime.now().strftime('%Y-%m-%d')),
|
||
|
|
'recommended_foods': input_data.get('recommended_foods', []),
|
||
|
|
'user_choice': input_data.get('user_choice', ''),
|
||
|
|
'feedback_type': input_data.get('feedback_type', 'unknown'),
|
||
|
|
'satisfaction_score': input_data.get('satisfaction_score'),
|
||
|
|
'notes': input_data.get('notes', ''),
|
||
|
|
'timestamp': datetime.now().isoformat()
|
||
|
|
}
|
||
|
|
|
||
|
|
# 添加到用户反馈记录
|
||
|
|
user_data.feedback.append(feedback_data)
|
||
|
|
user_data.updated_at = datetime.now().isoformat()
|
||
|
|
|
||
|
|
return {
|
||
|
|
'success': True,
|
||
|
|
'feedback_data': feedback_data,
|
||
|
|
'message': '反馈记录保存成功'
|
||
|
|
}
|
||
|
|
|
||
|
|
def _process_basic_questionnaire(self, answers: Dict) -> Dict[str, Any]:
|
||
|
|
"""处理基础信息问卷"""
|
||
|
|
return {
|
||
|
|
'name': answers.get('name', ''),
|
||
|
|
'age': int(answers.get('age', 0)),
|
||
|
|
'gender': answers.get('gender', ''),
|
||
|
|
'height': float(answers.get('height', 0)),
|
||
|
|
'weight': float(answers.get('weight', 0)),
|
||
|
|
'activity_level': answers.get('activity_level', ''),
|
||
|
|
'health_goals': answers.get('health_goals', [])
|
||
|
|
}
|
||
|
|
|
||
|
|
def _process_taste_questionnaire(self, answers: Dict) -> Dict[str, Any]:
|
||
|
|
"""处理口味偏好问卷"""
|
||
|
|
return {
|
||
|
|
'taste_preferences': {
|
||
|
|
'sweet': int(answers.get('sweet', 3)),
|
||
|
|
'salty': int(answers.get('salty', 3)),
|
||
|
|
'spicy': int(answers.get('spicy', 3)),
|
||
|
|
'sour': int(answers.get('sour', 3)),
|
||
|
|
'bitter': int(answers.get('bitter', 3)),
|
||
|
|
'umami': int(answers.get('umami', 3))
|
||
|
|
},
|
||
|
|
'dietary_preferences': answers.get('dietary_preferences', []),
|
||
|
|
'allergies': answers.get('allergies', []),
|
||
|
|
'dislikes': answers.get('dislikes', [])
|
||
|
|
}
|
||
|
|
|
||
|
|
def _process_physiological_questionnaire(self, answers: Dict) -> Dict[str, Any]:
|
||
|
|
"""处理生理信息问卷"""
|
||
|
|
return {
|
||
|
|
'is_female': answers.get('gender') == '女',
|
||
|
|
'menstrual_cycle_length': int(answers.get('menstrual_cycle_length', 28)),
|
||
|
|
'last_period_date': answers.get('last_period_date', ''),
|
||
|
|
'ovulation_symptoms': answers.get('ovulation_symptoms', []),
|
||
|
|
'zodiac_sign': answers.get('zodiac_sign', ''),
|
||
|
|
'personality_traits': answers.get('personality_traits', [])
|
||
|
|
}
|
||
|
|
|
||
|
|
def _load_questionnaire_templates(self) -> Dict[str, Dict]:
|
||
|
|
"""加载问卷模板"""
|
||
|
|
return {
|
||
|
|
'basic': {
|
||
|
|
'title': '基本信息问卷',
|
||
|
|
'questions': {
|
||
|
|
'name': {'question': '您的姓名', 'type': 'text'},
|
||
|
|
'age': {'question': '您的年龄', 'type': 'number', 'min': 1, 'max': 120},
|
||
|
|
'gender': {'question': '性别', 'type': 'select', 'options': ['男', '女']},
|
||
|
|
'height': {'question': '身高 (cm)', 'type': 'number', 'min': 100, 'max': 250},
|
||
|
|
'weight': {'question': '体重 (kg)', 'type': 'number', 'min': 30, 'max': 200},
|
||
|
|
'activity_level': {
|
||
|
|
'question': '日常活动水平',
|
||
|
|
'type': 'select',
|
||
|
|
'options': ['久坐', '轻度活动', '中度活动', '高度活动', '极高活动']
|
||
|
|
},
|
||
|
|
'health_goals': {
|
||
|
|
'question': '健康目标 (可多选)',
|
||
|
|
'type': 'checkbox',
|
||
|
|
'options': ['减肥', '增肌', '维持体重', '提高免疫力', '改善消化']
|
||
|
|
}
|
||
|
|
}
|
||
|
|
},
|
||
|
|
'taste': {
|
||
|
|
'title': '口味偏好问卷',
|
||
|
|
'questions': {
|
||
|
|
'sweet': {'question': '甜味偏好 (1-5分)', 'type': 'scale', 'min': 1, 'max': 5},
|
||
|
|
'salty': {'question': '咸味偏好 (1-5分)', 'type': 'scale', 'min': 1, 'max': 5},
|
||
|
|
'spicy': {'question': '辣味偏好 (1-5分)', 'type': 'scale', 'min': 1, 'max': 5},
|
||
|
|
'sour': {'question': '酸味偏好 (1-5分)', 'type': 'scale', 'min': 1, 'max': 5},
|
||
|
|
'bitter': {'question': '苦味偏好 (1-5分)', 'type': 'scale', 'min': 1, 'max': 5},
|
||
|
|
'umami': {'question': '鲜味偏好 (1-5分)', 'type': 'scale', 'min': 1, 'max': 5},
|
||
|
|
'dietary_preferences': {
|
||
|
|
'question': '饮食限制 (可多选)',
|
||
|
|
'type': 'checkbox',
|
||
|
|
'options': ['素食', '纯素食', '无麸质', '无乳制品', '无坚果', '低钠', '低碳水', '无糖']
|
||
|
|
},
|
||
|
|
'allergies': {
|
||
|
|
'question': '过敏食物 (可多选)',
|
||
|
|
'type': 'checkbox',
|
||
|
|
'options': ['花生', '坚果', '海鲜', '鸡蛋', '牛奶', '大豆', '小麦', '无过敏']
|
||
|
|
},
|
||
|
|
'dislikes': {
|
||
|
|
'question': '不喜欢的食物类型',
|
||
|
|
'type': 'checkbox',
|
||
|
|
'options': ['内脏', '海鲜', '蘑菇', '香菜', '洋葱', '大蒜', '辛辣食物', '甜食']
|
||
|
|
}
|
||
|
|
}
|
||
|
|
},
|
||
|
|
'physiological': {
|
||
|
|
'title': '生理信息问卷',
|
||
|
|
'questions': {
|
||
|
|
'menstrual_cycle_length': {
|
||
|
|
'question': '月经周期长度 (天)',
|
||
|
|
'type': 'number',
|
||
|
|
'min': 20,
|
||
|
|
'max': 40,
|
||
|
|
'optional': True
|
||
|
|
},
|
||
|
|
'last_period_date': {
|
||
|
|
'question': '上次月经日期',
|
||
|
|
'type': 'date',
|
||
|
|
'optional': True
|
||
|
|
},
|
||
|
|
'ovulation_symptoms': {
|
||
|
|
'question': '排卵期症状 (可多选)',
|
||
|
|
'type': 'checkbox',
|
||
|
|
'options': ['乳房胀痛', '情绪波动', '食欲变化', '疲劳', '无特殊症状'],
|
||
|
|
'optional': True
|
||
|
|
},
|
||
|
|
'zodiac_sign': {
|
||
|
|
'question': '星座',
|
||
|
|
'type': 'select',
|
||
|
|
'options': ['白羊座', '金牛座', '双子座', '巨蟹座', '狮子座', '处女座',
|
||
|
|
'天秤座', '天蝎座', '射手座', '摩羯座', '水瓶座', '双鱼座'],
|
||
|
|
'optional': True
|
||
|
|
},
|
||
|
|
'personality_traits': {
|
||
|
|
'question': '性格特征 (可多选)',
|
||
|
|
'type': 'checkbox',
|
||
|
|
'options': ['外向', '内向', '理性', '感性', '冒险', '保守', '创新', '传统'],
|
||
|
|
'optional': True
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
def get_questionnaire_template(self, questionnaire_type: str) -> Optional[Dict]:
|
||
|
|
"""获取问卷模板"""
|
||
|
|
return self.questionnaire_templates.get(questionnaire_type)
|
||
|
|
|
||
|
|
def get_all_questionnaire_types(self) -> List[str]:
|
||
|
|
"""获取所有问卷类型"""
|
||
|
|
return list(self.questionnaire_templates.keys())
|
||
|
|
|
||
|
|
def _create_error_result(self, error_message: str) -> Dict[str, Any]:
|
||
|
|
"""创建错误结果"""
|
||
|
|
return {
|
||
|
|
'success': False,
|
||
|
|
'error': error_message,
|
||
|
|
'message': f'数据采集失败: {error_message}'
|
||
|
|
}
|
||
|
|
|
||
|
|
def cleanup(self) -> bool:
|
||
|
|
"""清理资源"""
|
||
|
|
try:
|
||
|
|
self.logger.info("数据采集模块清理完成")
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
self.logger.error(f"数据采集模块清理失败: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
# 便捷函数
|
||
|
|
def collect_questionnaire_data(user_id: str, questionnaire_type: str, answers: Dict) -> bool:
|
||
|
|
"""收集问卷数据"""
|
||
|
|
from core.base import get_app_core
|
||
|
|
|
||
|
|
app = get_app_core()
|
||
|
|
input_data = {
|
||
|
|
'type': 'questionnaire',
|
||
|
|
'questionnaire_type': questionnaire_type,
|
||
|
|
'answers': answers
|
||
|
|
}
|
||
|
|
|
||
|
|
result = app.process_user_request(ModuleType.DATA_COLLECTION, input_data, user_id)
|
||
|
|
return result and result.result.get('success', False)
|
||
|
|
|
||
|
|
|
||
|
|
def record_meal(user_id: str, meal_data: Dict) -> bool:
|
||
|
|
"""记录餐食"""
|
||
|
|
from core.base import get_app_core
|
||
|
|
|
||
|
|
app = get_app_core()
|
||
|
|
input_data = {
|
||
|
|
'type': 'meal_record',
|
||
|
|
**meal_data
|
||
|
|
}
|
||
|
|
|
||
|
|
result = app.process_user_request(ModuleType.DATA_COLLECTION, input_data, user_id)
|
||
|
|
return result and result.result.get('success', False)
|
||
|
|
|
||
|
|
|
||
|
|
def record_feedback(user_id: str, feedback_data: Dict) -> bool:
|
||
|
|
"""记录反馈"""
|
||
|
|
from core.base import get_app_core
|
||
|
|
|
||
|
|
app = get_app_core()
|
||
|
|
input_data = {
|
||
|
|
'type': 'feedback',
|
||
|
|
**feedback_data
|
||
|
|
}
|
||
|
|
|
||
|
|
result = app.process_user_request(ModuleType.DATA_COLLECTION, input_data, user_id)
|
||
|
|
return result and result.result.get('success', False)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
# 测试数据采集模块
|
||
|
|
from core.base import BaseConfig, initialize_app, cleanup_app
|
||
|
|
|
||
|
|
print("测试数据采集模块...")
|
||
|
|
|
||
|
|
# 初始化应用
|
||
|
|
config = BaseConfig()
|
||
|
|
if initialize_app(config):
|
||
|
|
print("✅ 应用初始化成功")
|
||
|
|
|
||
|
|
# 测试问卷数据收集
|
||
|
|
test_user_id = "test_user_001"
|
||
|
|
questionnaire_answers = {
|
||
|
|
'name': '小美',
|
||
|
|
'age': 25,
|
||
|
|
'gender': '女',
|
||
|
|
'height': 165,
|
||
|
|
'weight': 55,
|
||
|
|
'activity_level': '中度活动',
|
||
|
|
'health_goals': ['维持体重', '提高免疫力']
|
||
|
|
}
|
||
|
|
|
||
|
|
if collect_questionnaire_data(test_user_id, 'basic', questionnaire_answers):
|
||
|
|
print("✅ 基础问卷数据收集成功")
|
||
|
|
|
||
|
|
# 测试餐食记录
|
||
|
|
meal_data = {
|
||
|
|
'date': '2024-01-15',
|
||
|
|
'meal_type': 'breakfast',
|
||
|
|
'foods': ['燕麦粥', '香蕉', '牛奶'],
|
||
|
|
'quantities': ['1碗', '1根', '200ml'],
|
||
|
|
'calories': 350.0,
|
||
|
|
'satisfaction_score': 4,
|
||
|
|
'notes': '很满意，营养均衡'
|
||
|
|
}
|
||
|
|
|
||
|
|
if record_meal(test_user_id, meal_data):
|
||
|
|
print("✅ 餐食记录成功")
|
||
|
|
|
||
|
|
# 清理应用
|
||
|
|
cleanup_app()
|
||
|
|
print("✅ 应用清理完成")
|
||
|
|
else:
|
||
|
|
print("❌ 应用初始化失败")
|
||
|
|
|
||
|
|
print("数据采集模块测试完成！")
|