"""
샘플 데이터 생성기 - 비즈니스 데모 및 테스트용
"""
import os
import json
import uuid
from datetime import datetime, timedelta
from typing import List, Dict
from groq import Groq
from ...db.database import get_session
from ...db.models import Persona, FGISurvey, FGIResponse
# 샘플 기본 프로필 템플릿
SAMPLE_PROFILES = [
{"age_range": "20대 초반", "gender": "여성", "location": "서울", "occupation_category": "학생", "income_range": "200만원 미만"},
{"age_range": "20대 후반", "gender": "남성", "location": "서울", "occupation_category": "IT/개발", "income_range": "300-500만원"},
{"age_range": "20대 후반", "gender": "여성", "location": "경기", "occupation_category": "디자인", "income_range": "200-300만원"},
{"age_range": "30대 초반", "gender": "남성", "location": "서울", "occupation_category": "마케팅", "income_range": "500만원 이상"},
{"age_range": "30대 초반", "gender": "여성", "location": "부산", "occupation_category": "교육", "income_range": "300-500만원"},
{"age_range": "20대 중반", "gender": "여성", "location": "서울", "occupation_category": "학생", "income_range": "200만원 미만"},
{"age_range": "30대 후반", "gender": "남성", "location": "경기", "occupation_category": "IT/개발", "income_range": "500만원 이상"},
{"age_range": "20대 초반", "gender": "남성", "location": "인천", "occupation_category": "서비스", "income_range": "200-300만원"},
]
CATEGORIES = ["technology", "lifestyle", "culture", "fashion", "food", "general"]
def generate_sample_personas(count: int = 50, use_llm: bool = True):
"""
샘플 페르소나 생성
:param count: 생성할 페르소나 수
:param use_llm: LLM을 사용해서 더 자연스러운 프로필 생성 (False면 템플릿만 사용)
:return: 생성된 페르소나 목록
"""
try:
groq_api_key = os.getenv("GROQ_API_KEY") if use_llm else None
client = Groq(api_key=groq_api_key) if groq_api_key else None
created_personas = []
with get_session() as session:
for i in range(count):
# 기본 프로필 선택 (순환)
base_profile = SAMPLE_PROFILES[i % len(SAMPLE_PROFILES)].copy()
persona_id = f"sample_persona_{uuid.uuid4().hex[:12]}"
# LLM으로 더 자연스러운 프로필 생성 (선택적)
if client and use_llm and i % 3 == 0: # 1/3만 LLM 사용 (비용 절감)
try:
prompt = f"""다음 기본 정보를 바탕으로 페르소나 프로필을 생성해주세요:
나이: {base_profile['age_range']}
성별: {base_profile['gender']}
지역: {base_profile['location']}
직업: {base_profile['occupation_category']}
JSON 형식:
{{
"interests": ["관심사1", "관심사2", "관심사3"],
"personality_traits": ["성격특성1", "성격특성2"],
"values": ["가치관1", "가치관2"],
"lifestyle": "라이프스타일 설명 (한 문장)",
"mbti_type": "MBTI 유형 (예: INTJ, ENFP 등)",
"mbti_confidence": 70-90
}}"""
response = client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
max_tokens=400,
temperature=0.8
)
profile_data = json.loads(response.choices[0].message.content)
except:
profile_data = {}
else:
profile_data = {}
# 페르소나 생성
persona = Persona(
id=persona_id,
age_range=base_profile["age_range"],
gender=base_profile["gender"],
location=base_profile["location"],
occupation_category=base_profile["occupation_category"],
income_range=base_profile.get("income_range"),
interests=profile_data.get("interests", []),
personality_traits=profile_data.get("personality_traits", []),
values=profile_data.get("values", []),
lifestyle=profile_data.get("lifestyle", ""),
mbti_type=profile_data.get("mbti_type") if hasattr(Persona, 'mbti_type') else None,
mbti_confidence=profile_data.get("mbti_confidence") if hasattr(Persona, 'mbti_confidence') else None,
freshness_score=float(50 + (i % 100)), # 50-150점 범위
created_at=datetime.now() - timedelta(days=i % 30),
last_updated=datetime.now() - timedelta(days=i % 7)
)
session.add(persona)
created_personas.append(persona_id)
session.commit()
return {
"success": True,
"count": len(created_personas),
"persona_ids": created_personas,
"message": f"{len(created_personas)}개의 샘플 페르소나가 생성되었습니다."
}
except Exception as e:
return {"error": f"샘플 페르소나 생성 중 오류: {str(e)}", "success": False}
def generate_sample_surveys_and_responses(
persona_count: int = 20,
responses_per_persona: int = 2,
use_llm: bool = True
):
"""
샘플 설문 및 응답 생성
:param persona_count: 응답을 생성할 페르소나 수
:param responses_per_persona: 페르소나당 응답 수
:param use_llm: LLM을 사용해서 자연스러운 응답 생성
:return: 생성 결과
"""
try:
groq_api_key = os.getenv("GROQ_API_KEY") if use_llm else None
client = Groq(api_key=groq_api_key) if groq_api_key else None
with get_session() as session:
# 페르소나 목록 가져오기
personas = session.query(Persona).limit(persona_count).all()
if not personas:
return {
"error": "페르소나가 없습니다. 먼저 샘플 페르소나를 생성해주세요.",
"success": False
}
created_surveys = []
created_responses = []
# 각 페르소나에 대해 설문 및 응답 생성
import random
for persona in personas:
for resp_idx in range(responses_per_persona):
# 카테고리 랜덤 선택 (안정적인 방법)
category = CATEGORIES[(hash(persona.id) % 1000 + resp_idx) % len(CATEGORIES)]
# 설문 생성
survey_id = f"survey_{uuid.uuid4().hex[:12]}"
# 오프닝 질문
opening_questions = {
"technology": "요즘 기술 분야에서 가장 관심있는 주제나 배우고 싶은 것이 있나요?",
"lifestyle": "평소 여가 시간을 어떻게 보내시나요? 가장 좋아하는 활동은?",
"fashion": "당신의 패션 스타일을 어떻게 설명하시겠어요? 어떤 옷을 즐겨 입나요?",
"food": "평소 식사는 주로 어떻게 해결하시나요? 좋아하는 음식 종류는?",
"culture": "최근에 본 영화, 드라마, 책 중에 인상 깊었던 게 있나요?",
"general": "요즘 당신의 일상이나 삶에서 가장 중요한 것은 무엇인가요?"
}
opening_question = {
"id": "q1",
"type": "open",
"text": opening_questions.get(category, opening_questions["general"])
}
# 추가 질문 생성 (LLM 사용)
questions = [opening_question]
responses_data = []
if client and use_llm:
try:
# 페르소나 정보 기반으로 자연스러운 응답 생성
persona_context = f"""
나이: {persona.age_range}
성별: {persona.gender}
지역: {persona.location}
직업: {persona.occupation_category}
관심사: {', '.join(persona.interests) if persona.interests else '없음'}
"""
# 첫 질문에 대한 응답 생성
prompt1 = f"""다음 페르소나 정보를 바탕으로 자연스러운 설문 응답을 생성해주세요.
페르소나 정보:
{persona_context}
질문: {opening_question['text']}
요구사항:
- 자연스럽고 진솔한 답변
- 페르소나의 특성을 반영
- 구체적이고 상세한 내용
- 2-3문장 정도
응답만 작성해주세요 (질문 없이):"""
response1 = client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=[{"role": "user", "content": prompt1}],
max_tokens=200,
temperature=0.8
)
answer1 = response1.choices[0].message.content.strip()
responses_data.append({
"question_id": "q1",
"answer": answer1
})
# 다음 질문 생성 (2-3개)
for q_num in range(2, 4):
qa_context = "\n".join([
f"Q{i+1}: {q['text']}\nA{i+1}: {resp['answer']}"
for i, (q, resp) in enumerate(zip(questions, responses_data))
])
prompt_q = f"""다음은 설문 진행 상황입니다:
페르소나 정보:
{persona_context}
지금까지의 대화:
{qa_context}
카테고리: {category}
다음 질문을 1개 생성해주세요. 이전 답변과 자연스럽게 연결되도록 하세요.
질문만 작성해주세요:"""
response_q = client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=[{"role": "user", "content": prompt_q}],
max_tokens=150,
temperature=0.8
)
next_question_text = response_q.choices[0].message.content.strip()
next_question = {
"id": f"q{q_num}",
"type": "open",
"text": next_question_text
}
questions.append(next_question)
# 응답 생성
prompt_a = f"""다음 페르소나 정보를 바탕으로 자연스러운 설문 응답을 생성해주세요.
페르소나 정보:
{persona_context}
질문: {next_question_text}
이전 답변:
{qa_context}
요구사항:
- 자연스럽고 진솔한 답변
- 이전 답변과 일관성 유지
- 구체적이고 상세한 내용
- 2-3문장 정도
응답만 작성해주세요:"""
response_a = client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=[{"role": "user", "content": prompt_a}],
max_tokens=200,
temperature=0.8
)
answer = response_a.choices[0].message.content.strip()
responses_data.append({
"question_id": f"q{q_num}",
"answer": answer
})
except Exception as e:
# LLM 실패 시 기본 응답
responses_data = [
{"question_id": "q1", "answer": f"{category}에 관심이 있습니다."}
]
else:
# LLM 없이 기본 응답
responses_data = [
{"question_id": "q1", "answer": f"{category} 카테고리에 대한 답변입니다."}
]
# 설문 저장
days_ago_survey = (hash(persona.id) % 30) if persona.id else 0
survey = FGISurvey(
id=survey_id,
category=category,
questions=questions,
news_source={"type": "sample", "category": category},
created_at=datetime.now() - timedelta(days=days_ago_survey)
)
session.add(survey)
created_surveys.append(survey_id)
# 응답 저장
response_id = f"resp_{uuid.uuid4().hex[:12]}"
days_ago = (hash(persona.id) % 7) if persona.id else 0
fgi_response = FGIResponse(
id=response_id,
persona_id=persona.id,
survey_id=survey_id,
responses=responses_data,
submitted_at=datetime.now() - timedelta(days=days_ago)
)
session.add(fgi_response)
created_responses.append(response_id)
# 페르소나 신선도 점수 업데이트
quality_score = len(responses_data) * 10 # 응답 수에 비례
persona.freshness_score = (persona.freshness_score or 0) + quality_score
persona.last_updated = datetime.now()
session.commit()
return {
"success": True,
"surveys_created": len(created_surveys),
"responses_created": len(created_responses),
"personas_updated": len(personas),
"message": f"{len(created_surveys)}개 설문, {len(created_responses)}개 응답이 생성되었습니다."
}
except Exception as e:
return {"error": f"샘플 설문/응답 생성 중 오류: {str(e)}", "success": False}
def generate_all_sample_data(
persona_count: int = 50,
responses_per_persona: int = 2,
use_llm: bool = True
):
"""
전체 샘플 데이터 생성 (페르소나 + 설문 + 응답)
:param persona_count: 생성할 페르소나 수
:param responses_per_persona: 페르소나당 응답 수
:param use_llm: LLM 사용 여부
:return: 생성 결과
"""
try:
# 1. 페르소나 생성
personas_result = generate_sample_personas(count=persona_count, use_llm=use_llm)
if not personas_result.get("success"):
return personas_result
# 2. 설문 및 응답 생성
surveys_result = generate_sample_surveys_and_responses(
persona_count=persona_count,
responses_per_persona=responses_per_persona,
use_llm=use_llm
)
return {
"success": True,
"personas": personas_result,
"surveys_and_responses": surveys_result,
"message": f"샘플 데이터 생성 완료! 페르소나 {persona_count}개, 설문 및 응답 다수 생성됨."
}
except Exception as e:
return {"error": f"샘플 데이터 생성 중 오류: {str(e)}", "success": False}
