"""
페르소나 조회 및 검색 기능
"""
from datetime import datetime
from typing import Optional, List, Dict
from sqlalchemy import func
from ...db.database import get_session
from ...db.models import Persona, FGISurvey, FGIResponse
def get_persona(persona_id: str):
"""
특정 페르소나 정보 조회
:param persona_id: 페르소나 ID
:return: 페르소나 정보
"""
try:
with get_session() as session:
persona = session.query(Persona).filter_by(id=persona_id).first()
if not persona:
return {
"error": f"페르소나 '{persona_id}'를 찾을 수 없습니다.",
"persona_id": persona_id,
"exists": False
}
# 관련 설문 수 조회
survey_count = session.query(FGISurvey).count()
response_count = session.query(FGIResponse).filter_by(persona_id=persona_id).count()
return {
"persona_id": persona.id,
"exists": True,
"basic_profile": {
"age_range": persona.age_range,
"gender": persona.gender,
"location": persona.location,
"occupation_category": persona.occupation_category,
"income_range": persona.income_range
},
"persona_profile": {
"interests": persona.interests,
"personality_traits": persona.personality_traits,
"values": persona.values,
"lifestyle": persona.lifestyle
},
"mbti_analysis": {
"mbti_type": persona.mbti_type if hasattr(persona, 'mbti_type') else None,
"mbti_dimensions": persona.mbti_dimensions if hasattr(persona, 'mbti_dimensions') else None,
"mbti_confidence": persona.mbti_confidence if hasattr(persona, 'mbti_confidence') else None
},
"stats": {
"freshness_score": persona.freshness_score,
"survey_count": response_count,
"created_at": persona.created_at.isoformat() if persona.created_at else None,
"last_updated": persona.last_updated.isoformat() if persona.last_updated else None
}
}
except Exception as e:
return {"error": f"페르소나 조회 중 오류 발생: {str(e)}"}
def list_personas(
age_range: Optional[str] = None,
gender: Optional[str] = None,
location: Optional[str] = None,
occupation_category: Optional[str] = None,
mbti_type: Optional[str] = None,
min_freshness: Optional[float] = None,
limit: int = 50
):
"""
페르소나 목록 조회 (필터링 가능)
:param age_range: 나이대 필터
:param gender: 성별 필터
:param location: 지역 필터
:param occupation_category: 직업 필터
:param mbti_type: MBTI 유형 필터
:param min_freshness: 최소 신선도 점수
:param limit: 최대 결과 수
:return: 페르소나 목록
"""
try:
with get_session() as session:
query = session.query(Persona)
# 필터 적용
if age_range:
query = query.filter(Persona.age_range == age_range)
if gender:
query = query.filter(Persona.gender == gender)
if location:
query = query.filter(Persona.location == location)
if occupation_category:
query = query.filter(Persona.occupation_category == occupation_category)
if min_freshness is not None:
query = query.filter(Persona.freshness_score >= min_freshness)
if mbti_type and hasattr(Persona, 'mbti_type'):
query = query.filter(Persona.mbti_type == mbti_type)
# 신선도 점수 내림차순 정렬
personas = query.order_by(Persona.freshness_score.desc()).limit(limit).all()
result = []
for persona in personas:
result.append({
"persona_id": persona.id,
"basic_profile": {
"age_range": persona.age_range,
"gender": persona.gender,
"location": persona.location,
"occupation_category": persona.occupation_category,
"income_range": persona.income_range
},
"mbti_type": persona.mbti_type if hasattr(persona, 'mbti_type') else None,
"freshness_score": persona.freshness_score,
"created_at": persona.created_at.isoformat() if persona.created_at else None,
"last_updated": persona.last_updated.isoformat() if persona.last_updated else None
})
return {
"count": len(result),
"personas": result
}
except Exception as e:
return {"error": f"페르소나 목록 조회 중 오류 발생: {str(e)}"}
def get_persona_survey_history(persona_id: str):
"""
특정 페르소나의 설문 히스토리 조회
:param persona_id: 페르소나 ID
:return: 설문 히스토리
"""
try:
with get_session() as session:
# 페르소나 존재 확인
persona = session.query(Persona).filter_by(id=persona_id).first()
if not persona:
return {
"error": f"페르소나 '{persona_id}'를 찾을 수 없습니다.",
"persona_id": persona_id,
"exists": False
}
# 해당 페르소나의 응답 조회
responses = session.query(FGIResponse).filter_by(persona_id=persona_id).order_by(FGIResponse.submitted_at.desc()).all()
history = []
for response in responses:
# 설문 정보 조회
survey = session.query(FGISurvey).filter_by(id=response.survey_id).first()
history.append({
"response_id": response.id,
"survey_id": response.survey_id,
"survey_category": survey.category if survey else None,
"question_count": len(survey.questions) if survey and survey.questions else 0,
"response_count": len(response.responses) if response.responses else 0,
"submitted_at": response.submitted_at.isoformat() if response.submitted_at else None,
"responses": response.responses
})
return {
"persona_id": persona_id,
"total_surveys": len(history),
"history": history
}
except Exception as e:
return {"error": f"설문 히스토리 조회 중 오류 발생: {str(e)}"}
def get_statistics():
"""
전체 통계 정보 조회
:return: 통계 정보
"""
try:
with get_session() as session:
# 페르소나 통계
total_personas = session.query(Persona).count()
avg_freshness = session.query(Persona).with_entities(
func.avg(Persona.freshness_score)
).scalar() or 0.0
# 설문 통계
total_surveys = session.query(FGISurvey).count()
total_responses = session.query(FGIResponse).count()
# 카테고리별 통계
category_stats = session.query(
FGISurvey.category,
func.count(FGISurvey.id).label('count')
).group_by(FGISurvey.category).all()
# MBTI 분포 (컬럼이 있는 경우)
mbti_distribution = {}
if hasattr(Persona, 'mbti_type'):
mbti_stats = session.query(
Persona.mbti_type,
func.count(Persona.id).label('count')
).filter(Persona.mbti_type.isnot(None)).group_by(Persona.mbti_type).all()
mbti_distribution = {mbti: count for mbti, count in mbti_stats}
# 지역 분포
location_stats = session.query(
Persona.location,
func.count(Persona.id).label('count')
).filter(Persona.location.isnot(None)).group_by(Persona.location).all()
# 직업 분포
occupation_stats = session.query(
Persona.occupation_category,
func.count(Persona.id).label('count')
).filter(Persona.occupation_category.isnot(None)).group_by(Persona.occupation_category).all()
return {
"personas": {
"total": total_personas,
"average_freshness": round(avg_freshness, 2)
},
"surveys": {
"total": total_surveys,
"total_responses": total_responses,
"by_category": {category: count for category, count in category_stats}
},
"distribution": {
"mbti": mbti_distribution,
"location": {location: count for location, count in location_stats},
"occupation": {occupation: count for occupation, count in occupation_stats}
}
}
except Exception as e:
return {"error": f"통계 조회 중 오류 발생: {str(e)}"}
def compare_personas(persona_id1: str, persona_id2: str):
"""
두 페르소나 비교
:param persona_id1: 첫 번째 페르소나 ID
:param persona_id2: 두 번째 페르소나 ID
:return: 비교 결과
"""
try:
with get_session() as session:
persona1 = session.query(Persona).filter_by(id=persona_id1).first()
persona2 = session.query(Persona).filter_by(id=persona_id2).first()
if not persona1:
return {"error": f"페르소나 '{persona_id1}'를 찾을 수 없습니다."}
if not persona2:
return {"error": f"페르소나 '{persona_id2}'를 찾을 수 없습니다."}
return {
"persona1": {
"id": persona1.id,
"basic_profile": {
"age_range": persona1.age_range,
"gender": persona1.gender,
"location": persona1.location,
"occupation_category": persona1.occupation_category
},
"mbti_type": persona1.mbti_type if hasattr(persona1, 'mbti_type') else None,
"freshness_score": persona1.freshness_score,
"interests": persona1.interests,
"personality_traits": persona1.personality_traits
},
"persona2": {
"id": persona2.id,
"basic_profile": {
"age_range": persona2.age_range,
"gender": persona2.gender,
"location": persona2.location,
"occupation_category": persona2.occupation_category
},
"mbti_type": persona2.mbti_type if hasattr(persona2, 'mbti_type') else None,
"freshness_score": persona2.freshness_score,
"interests": persona2.interests,
"personality_traits": persona2.personality_traits
},
"similarities": {
"same_age_range": persona1.age_range == persona2.age_range,
"same_gender": persona1.gender == persona2.gender,
"same_location": persona1.location == persona2.location,
"same_occupation": persona1.occupation_category == persona2.occupation_category,
"same_mbti": (hasattr(persona1, 'mbti_type') and hasattr(persona2, 'mbti_type') and
persona1.mbti_type == persona2.mbti_type) if hasattr(Persona, 'mbti_type') else False
}
}
except Exception as e:
return {"error": f"페르소나 비교 중 오류 발생: {str(e)}"}
