[2026.01.19] 요약 모델 수정 및 1차 요약api 구현
[2026.01.20] PR리뷰 반영 / 비동기 처리(Celery) / 욕설 1차 필터링
[2026.01.21] 욕설 필터링 완성 / 댓글 등록,조회 api 구현
[2026.01.22] 댓글 수정,삭제api 구현 / swagger.yaml파일 수정
[2026.01.23] getattr 수정
2026.01.19 ✅
컬럼 비교(게임/요약)
| 비교 항목 | ForeignKey (현재 코드) | OneToOneField (추천) |
|---|---|---|
| 관계 (Relationship) | 1 : N (일대다) | 1 : 1 (일대일) |
| 설명 | 하나의 게임(Game)에 여러 개의 요약(Summary) 데이터가 쌓일 수 있습니다. | 하나의 게임(Game)에는 단 하나의 요약(Summary)만 존재합니다. |
| DB 동작 | 부모 ID가 중복되어 저장될 수 있습니다. (히스토리 저장 가능) | 부모 ID에 UNIQUE 제약 조건이 걸려 중복 저장이 불가능합니다. |
| 데이터 접근법 | game.summaries.all() (리스트 반환) | game.summaries (객체 바로 반환) |
| 주요 사용처 | 댓글, 리뷰, 게시글 히스토리 등 | 프로필 확장, 설정 정보, 현재 상태 캐싱 |
FK
- 게임을 통해 요약내용을 가져올때, 목록(QuerySet)으로 가져오기 때문에
.first()나 루프를 통해 하나를 꺼내야 함
장점
- 과거에 요약했던 기록(히스토리)을 모두 남겨둘 수 있음
단점
- 단순히 "현재 요약"을 보여줄 때 매번
.first()를 호출해야 해서 코드가 길어짐
- 단순히 "현재 요약"을 보여줄 때 매번
# 모델 정의
game = models.ForeignKey(Game, related_name="summaries", ...)
# 사용 예시
my_game = Game.objects.get(id=1)
# 여러 개일 수 있으므로 .all() 혹은 .filter() 사용
summaries = my_game.summaries.all()
# 가장 최신 거 하나만 가져오려면 추가 로직 필요
latest_summary = my_game.summaries.order_by('-created_at').first()
print(latest_summary.text)OneToOneField
- 게임을 통해 요약을 가져올 때, 객체 그 자체를 바로 가져옴
장점
- 코드가 직관적
단점
- 과거 기록을 남길 수 없음
- 새로운 요약을 만들면 기존 요약은 덮어씌워지거나 삭제 됨
# 모델 정의
game = models.OneToOneField(Game, related_name="summary", ...)
# related_name을 복수형(summaries)이 아닌 단수형(summary)으로 짓는 것이 관례입니다.
# 사용 예시
my_game = Game.objects.get(id=1)
# 바로 접근 가능 (없으면 Error 발생 가능성 있음 -> hasattr로 체크 권장)
if hasattr(my_game, 'summary'):
print(my_game.summary.text)요약 모델 수정
수정 이유
- 진행했던 요약내용은 굳이 저장할 필요가 없기 때문에 FK를
OneToOneField로 변경OneToOneField로 변경하면서 1대1에 대한 고민
is_deleted의 필요 이유- 어차피 새로운 요약이 생성되면 기존의 요약을 삭제되거나 덮어씌워짐
- 불필요하다는 생각이 들어 제거
- 어차피 새로운 요약이 생성되면 기존의 요약을 삭제되거나 덮어씌워짐
updated_at필요 이유is_deleted와 동일한 이유로 요약내용이 업데이트 될 경우 덮어씌워 질텐데updated_at이 없는건 말이 안됨created_at를 지우고TimeStampedModel상속
수정 결과
# 현재
class GameReviewSummary(models.Model):
game = models.ForeignKey(
Game,
on_delete=models.CASCADE,
related_name="summaries",
)
text = models.TextField(verbose_name="AI가 생성한 본문")
is_deleted = models.BooleanField(default=False, verbose_name="삭제 여부")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="생성된 시간")
class Meta:
db_table = "game_review_summarise"
——————————————————————————————————————[비교]—————————————————————————————————————————
# 변경후
class GameReviewSummary(TimeStampedModel):
game = models.OneToOneField(
Game,
on_delete=models.CASCADE,
related_name="summary",
)
text = models.TextField(verbose_name="AI가 생성한 본문")
class Meta:
db_table = "game_review_summarise"2026.01.20 ✅
PR 리뷰 반영
- 최소 리뷰 조건 / 기간 30일 제한 은 환경변수로 빼기
- 리뷰 5개 가져오는 로직 수정 요망
- 글자수 제한 / 가져오는 개수 늘리기
환경변수 설정
# settings.py
AI_SUMMARY_MIN_REVIEW_COUNT = 10 # 요약 생성 최소 리뷰 수
AI_SUMMARY_UPDATE_INTERVAL_DAYS = 30 # 요약 갱신 주기 (일)
AI_REVIEW_MIN_LENGTH = 10 # 요약에 사용할 리뷰의 최소 글자 수
AI_SUMMARY_MIN_VALID_REVIEWS = 3 # 유효한 리뷰의 개수
# service.py
class ReviewSummaryService:
def __init__(self):
"""
서비스 초기화: Client 생성 및 공통 설정 정의
"""
api_key = settings.GEMINI_API_KEY
self.client = genai.Client(api_key=api_key)
self.model_name = "gemini-flash-latest"
# 환경변수 혹은 settings에서 값을 가져옴, 없을 경우 기본값(default)을 사용
self.min_review_count = getattr(settings, 'AI_SUMMARY_MIN_REVIEW_COUNT', 10)
self.update_interval_days = getattr(settings, 'AI_SUMMARY_UPDATE_INTERVAL_DAYS', 30)
self.min_review_length = getattr(settings, 'AI_REVIEW_MIN_LENGTH', 10)
self.min_valid_reviews = getattr(settings, 'AI_SUMMARY_MIN_VALID_REVIEWS', 3)
self.system_instruction = (
...
)최소 리뷰 개수
# 이전
def get_summary(self, game_id: int) -> dict:
try:
game = Game.objects.select_related("summary").get(id=game_id)
except Game.DoesNotExist:
raise GameNotFound()
review_count = game.reviews.filter(is_deleted=False).count() # type: ignore
if review_count < 10:
raise NotEnoughReviews()
# exception
class NotEnoughReviews(APIException):
status_code = status.HTTP_400_BAD_REQUEST
default_detail = "리뷰가 부족하여 요약을 생성할 수 없습니다. (최소 10개 필요)"
default_code = "not_enough_reviews"
——————————————————————————————————————[비교]—————————————————————————————————————————
# 이후
def get_summary(self, game_id: int) -> dict:
try:
game = Game.objects.select_related("summary").get(id=game_id)
except Game.DoesNotExist:
raise GameNotFound()
review_count = game.reviews.filter(is_deleted=False).count() # type: ignore
if review_count < self.min_review_count:
raise NotEnoughReviews()
# exception
MIN_COUNT = getattr(settings, 'AI_SUMMARY_MIN_REVIEW_COUNT', 10)
class NotEnoughReviews(APIException):
status_code = status.HTTP_400_BAD_REQUEST
default_detail = f"리뷰가 부족하여 요약을 생성할 수 없습니다. (최소 {MIN_COUNT}개 필요)"
default_code = "not_enough_reviews"요약 갱신 주기
# 이전
def _update_and_parse(self, summary_obj) -> bool:
# 기존 요약이 없으면 무조건 갱신(생성)
if not summary_obj:
return True
# 현재 시간과 마지막 수정 시간(updated_at)의 차이를 계산
update_time_difference = timezone.now() - summary_obj.updated_at
# 마지막 수정일로부터 30일이 지났으면 True를 반환
return update_time_difference > timedelta(days=30)
——————————————————————————————————————[비교]—————————————————————————————————————————
# 이후
def _update_and_parse(self, summary_obj) -> bool:
if not summary_obj:
return True
update_time_difference = timezone.now() - summary_obj.updated_at
return update_time_difference > timedelta(days=self.update_interval_days)
요약에 사용할 리뷰의 최소 글자 수
- 슬라이싱이기 때문에 최대 5개를 가져오기는 하지만
- 5개가 꼭 채워져야 요약을 진행하지는 않음
- 즉, 비어있지만 않으면 1개든 2개든 요약은 진행 됨
# 이전
def _generate_and_save(self, game: Game, summary_obj) -> dict:
"""
AI 요약 생성 및 DB 저장
"""
# 최신 리뷰 5개를 가져오기
reviews = game.reviews.filter(is_deleted=False).order_by("-created_at")[:5]
# 리뷰 내용들을 줄바꿈 문자로 연결하여 하나의 문자열로 만듬
reviews_text = "\n".join([f"- {r.content}" for r in reviews])
——————————————————————————————————————[비교]—————————————————————————————————————————
# 이후
def _generate_and_save(self, game: Game, summary_obj) -> dict:
# 글자 수 필터링 및 유효 리뷰 개수 확인
reviews = (
game.reviews.annotate(text_len=Length("content"))
.filter(is_deleted=False, text_len__gte=self.min_review_length)
.order_by("-created_at")[:5]
)
# 유효한 리뷰가 설정된 개수(예: 3개) 미만이면 중단
if len(reviews) < self.min_valid_reviews:
logger.warning(
f"Game({game.id}) has enough raw reviews but VALID reviews({len(reviews)}) "
f"are less than {self.min_valid_reviews}."
)
raise NotEnoughValidReviews()
reviews_text = "\n".join([f"- {r.content}" for r in reviews])
# exception
MIN_VALID_COUNT = getattr(settings, 'AI_SUMMARY_MIN_VALID_REVIEWS', 3)
class NotEnoughValidReviews(APIException):
"""
글자 수 조건을 만족하는 '유효한 리뷰'가 부족할 때 발생
"""
status_code = status.HTTP_400_BAD_REQUEST
default_detail = f"요약에 적합한 긴 리뷰가 부족합니다. (최소{MIN_VALID_COUNT}개 필요)"
default_code = "not_enough_valid_reviews"비동기 세팅
환경설정(settings.py)
- Django가 어떤 브로커를 통해 메시지를 주고받을지 설정하는 곳
# ... (생략)
CELERY_BROKER_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
"""
1. Celery 브로커 설정
- Django(Producer)와 Celery Worker(Consumer) 사이의 중개자 역할을 하는 브로커 URL
- 여기서는 Redis를 사용 (redis://localhost:6379/0)
"""
CELERY_RESULT_BACKEND = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
"""
2. 결과 백엔드
- 작업이 끝난 후 그 결과(리턴값, 성공/실패 여부)를 저장할 곳
- 역시 Redis를 사용하여 결과를 저장하도록 설정
"""
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
"""
3. 직렬화(Serializer)
- Python 객체(함수 인자 등)를 Redis가 이해할 수 있는 문자열(JSON)로 변환하는 방식
"""celery / redis
poetry add celery redispoetry add "celery[redis]"- Redis 서버도 컴퓨터에 설치되어 있어야 함(mac)
- brew install redis
- brew services start redis
- Redis 서버도 컴퓨터에 설치되어 있어야 함(mac)
config/celery.p
- Celery 설정 파일 생성
- settings.py가 있는 폴더(프로젝트 설정 폴더) 안에 celery.py 파일을 새로 만듬
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '프로젝트명.settings')
"""
- 1. Django 설정 파일 지정 (Django 설정 로드)
- Celery는 독립적인 프로그램이라 Django의 설정(DB 정보 등)을 모름
- 이 코드를 통해 "Django 설정은 config.settings에 있어"라고 알려줌
"""
app = Celery('oz_playtype')
"""
- 2. Celery 인스턴스 생성
- 작업을 관리할 Celery 객체(app)를 만듬
- 이 객체가 나중에 워커(Worker)가 됨
- celery.py 파일이 config 폴더 안에 들어있기 때문에,
- Celery에게 "너의 설정과 기준점은 config 폴더야"라고 알려줌
"""
app.config_from_object('django.conf:settings', namespace='CELERY')
"""
- 3. settings.py에서 'CELERY_'로 시작하는 설정을 불러옴
- 설정 통합
- Celery 설정을 별도 파일이 아닌,
- Django의 settings.py에서 관리하겠다는 뜻 (CELERY_로 시작하는 설정들)
"""
app.autodiscover_tasks()
"""
4. 등록된 앱(apps)에서 자동으로 tasks.py를 찾음
- 자동 탐색
- 등록된 Django 앱(INSTALLED_APPS)들을 돌면서
- tasks.py라는 파일이 있는지 자동으로 찾아 등록해 줌
- 일일이 등록할 필요가 없어짐
"""
@app.task(bind=True)
def debug_task(self):
print(f'Request: {self.request!r}')config/__init__.py
# settings 파일이 있는 폴더의 __init__.py
from .celery import app as celery_app
__all__ = ('celery_app',)
"""
- 1. Django가 켜질 때, celery.py에서 만든 app을 메모리에 올리는 역할
- celery_app을 임포트하지 않으면,
- Django가 실행되어도 Celery 앱이 로드되지 않아
- @shared_task 데코레이터가 제대로 작동하지 않을 수 있음
- 즉, "Django가 시작될 때 Celery도 같이 준비시켜 줘"라고 명령하는 것
"""워커(tasks.py)
- 비동기 작업 생성
- 이 코드는 웹 서버(Django)가 아닌, 별도의 Celery Worker 프로세스에서 실행됨
@shared_task- 이 함수가 Celery가 처리할 수 있는 '작업(Task)'임을 명시
- 이 데코레이터가 있어야 Redis 큐에 작업을 등록하고 워커가 이를 인식할 수 있음
from celery import shared_task
from apps.ai.services import ReviewSummaryService
import logging
logger = logging.getLogger(__name__)
@shared_task
def run_ai_summary(game_id):
"""
Celery가 실행할 백그라운드 작업
"""
logger.info(f"Start AI Summary Task for Game ID: {game_id}")
try:
service = ReviewSummaryService()
# 서비스의 get_summary 호출 (내부적으로 저장 로직 포함됨)
# 여기서 반환값은 필요 없으므로 무시
service.get_summary(game_id)
logger.info(f"Successfully finished summary for Game ID: {game_id}")
except Exception as e:
logger.error(f"Error in AI Summary Task: {e}")
"""
백그라운드 작업은 실패해도 사용자에게 바로 에러 페이지를 보여줄 수 없음
반드시 로그(logger)를 남겨서 개발자가 나중에 확인할 수 있게 해야 함
"""Signal 생성
- 리뷰가 저장될 때 조건을 체크하고 Celery Task를 호출하는 감시자 코드를 작성
.delay()- 일반적인 함수 호출인 run_ai_summary(game.id)였다면,
- AI 응답이 올 때까지 사용자의 화면(브라우저)은 멈춰 있게 됨
.delay()를 쓰면 "이 작업을 큐(Redis)에 넣어줘"라고 명령만 하고- 즉시 다음 코드로 넘어가므로 사용자는 기다리지 않음
- 일반적인 함수 호출인 run_ai_summary(game.id)였다면,
game = instance.game- "방금 저장된 리뷰가 어떤 게임에 달린 것인지 알아내서, 그 게임 정보를 가져오는 코드"
instance- 방금 DB에 저장된 데이터 그 자체
- sender=Review로 지정되어 있음
- 누군가 리뷰를 쓰고 저장하면 이 함수가 호출되는데,
- 이때 "그 저장된 리뷰 내용"이 instance라는 변수에 담겨서 넘어옴
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.conf import settings
from django.utils import timezone
from datetime import timedelta
from apps.community.models.reviews import Review
from apps.ai.tasks import run_ai_summary
@receiver(post_save, sender=Review)
def trigger_ai_summary(sender, instance, created, **kwargs):
"""
리뷰가 저장(post_save)될 때마다 실행되는 함수
"""
# 1. 수정된 리뷰는 무시하고, 새로 생성된 리뷰일 때만 체크 (필요시 로직 변경 가능)
if not created:
return
game = instance.game
# 설정값 로드
MIN_COUNT = getattr(settings, 'AI_SUMMARY_MIN_REVIEW_COUNT', 10)
UPDATE_DAYS = getattr(settings, 'AI_SUMMARY_UPDATE_INTERVAL_DAYS', 30)
# 2. 현재 유효 리뷰 개수 확인 (삭제되지 않은 리뷰)
current_count = game.reviews.filter(is_deleted=False).count()
# 조건 A: 리뷰 개수가 기준(10개) 이상인가?
if current_count >= MIN_COUNT:
summary_obj = getattr(game, 'summary', None)
should_run = False
# 조건 B-1: 요약본이 아예 없으면 -> 실행
if not summary_obj:
should_run = True
# 조건 B-2: 요약본이 있지만, 마지막 업데이트로부터 30일 지났으면 -> 실행
else:
time_diff = timezone.now() - summary_obj.updated_at
if time_diff > timedelta(days=UPDATE_DAYS):
should_run = True
# 실행 조건 만족 시 Celery Task 호출 (.delay() 사용)
if should_run:
print(f"[Signal] Triggering AI Summary for Game {game.id}")
run_ai_summary.delay(game.id)Signal 등록(apps.py)
- Django가 시작될 때 signals.py를 읽어들이도록
- apps.py의 ready() 메서드를 수정해야 함
from django.apps import AppConfig
class AiConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = "apps.ai"
def ready(self):
# Django 시작 시 signals 모듈을 임포트하여 작동하게 함
import apps.ai.signals사용
- 1번 터미널에
- celery -A config worker --loglevel=info
- 2번 터미널에
- python manage.py runserver
욕설 1차 필터링
선택지
- 시리얼라이저에서 리뷰를 등록할때 욕설이 포함되어 있으면 등록 불가하게 만들기
- 간단한데 욕설은 리뷰에 등록하지 못하는건 말이 안됨
- 욕설이 포함된 리뷰는 생략하는 방향으로 가고 싶음
- 리뷰모델에 욕설여부 컬럼을 추가하여 이 컬럼이 True인 경우
- 요약에 필요한 데이터로 가져가지 않게 하기
- 단점: 일이 너무 커짐, 수정해야할 부분이 한트럭이 됨
- AI 호출 전 Python 서비스 계층에서의 1차 필터링 ✅
- 시리얼라이저에서 리뷰를 등록할때 욕설이 포함되어 있으면 등록 불가하게 만들기
정규표현식(Regex)
.*?- "그 사이에 뭐가 들어가든(숫자, 특수문자, 공백 등) 다 무시하고 잡아라"
- 이걸 사용하면 "시간이 지나서 발견" 같은 단어도 잡힘
- "그 사이에 뭐가 들어가든(숫자, 특수문자, 공백 등) 다 무시하고 잡아라"
ex. r"시.*?발"
- 시발 (기본)
- 시1발 (숫자 섞음)
- 시...발 (특수문자)
- 시 발 (공백)
- 시@#$%발 (이상한 문자)
- 시123123발 (긴 숫자)
- "시"로 시작해서 "발"로 끝나는 모든 단어를 잡음[^가-힣]₩추가- "중간에 한글이 끼면 봐주자"
ex. r"시[^가-힣]*?발"
- 잡음: 시1발, 시.발, 시..발, 시 ^& 발
- 시린발, 시골발, 시간이 지나서 발견safety_settings
- Google Gemini AI가 유해한 콘텐츠를 생성하거나 입력받았을 때
- 얼마나 엄격하게 차단할지 결정하는 안전장치 설정
4가지 유해 카테고리
HARM_CATEGORY_HATE_SPEECH(혐오 발언)- 인종, 종교, 성별 등에 대한 차별적 발언.
HARM_CATEGORY_HARASSMENT(괴롭힘)- 특정 인물을 공격하거나 괴롭히는 내용.
HARM_CATEGORY_SEXUALLY_EXPLICIT(선정성)- 성적인 묘사나 음란한 내용.
HARM_CATEGORY_DANGEROUS_CONTENT(위험 콘텐츠)- 자해, 폭력 조장, 위험한 행동 유도 등.
threshold(차단 기준)
BLOCK_LOW_AND_ABOVE(매우 엄격)- 조금이라도(Low) 위험해 보이면 즉시 차단 (기본값에 가까움)
BLOCK_MEDIUM_AND_ABOVE(보통)- 중간(Medium) 정도 위험 수위부터 차단
BLOCK_ONLY_HIGH(최소 차단)- 확실히 위험하고 심각한(High) 수준일 때만 차단, 경미한 욕설 등은 허용
BLOCK_NONE(차단 안 함)- 필터를 끔 (일부 계정에서는 사용 불가할 수 있음)
safety_settings = [
types.SafetySetting(
category=types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH,
),
types.SafetySetting(
category=types.HarmCategory.HARM_CATEGORY_HARASSMENT,
threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH,
),
types.SafetySetting(
category=types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH,
),
types.SafetySetting(
category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH,
),
]코드 리뷰
re.compile("|".join(BAD_PATTERNS), re.IGNORECASE)- 리스트에 있는 수많은 욕설 패턴들을 하나로 합쳐서
- 한 번의 스캔으로 빠르게 잡아내는 검색 도구를 만드는 과정
"|".join(BAD_PATTERNS)- join은 리스트 안에 있는 문자열들을 이어주는 함수
- 여기서 이어주는 역할을 하는 것이 바로
| (파이프)기호- 정규표현식에서
|는"또는(OR)"을 의미
- 정규표현식에서
re.compile(...)- re.compile을 하면
- 파이썬이 이 복잡한 패턴을 기계가 이해하기 쉬운 형태로 미리 최적화해줌
- re.compile을 하면
re.IGNORECASE- "영어 대소문자를 구분하지 말라"
2026.01.21 ✅
index
review
class Meta:
db_table = "reviews"
indexes = [
models.Index(fields=["game"], name="idx_reviews_best"),
"""
1. 특정 게임의 상세 페이지에서 그 게임에 달린 리뷰만 빠르게 불러오기 위한 인덱스
- 이 인덱스가 없으면, 리뷰가 수만 개일 때 특정 게임의 리뷰를 찾기 위해
- 전체를 훑어야(Full Scan) 해서 느려짐
"""
models.Index(fields=["like_count"], name="idx_reviews_best_like_count"),
"""
2. '베스트 리뷰'나 '인기 리뷰' 순으로 정렬할 때 빠르게 처리하기 위한 인덱스
좋아요 수(like_count)가 많은 순서대로 데이터를 가져오는 쿼리 속도를 최적화
"""
]comments
class Meta:
db_table = "comments"
indexes = [
models.Index(fields=["review", "created_at"]),
"""
1. 복합 인덱스(Composite Index)
- 'review': 특정 리뷰에 달린 댓글들을 먼저 찾고,
- 'created_at': 그 댓글들을 작성된 시간 순서대로 정렬해서 보여줄 때 사용
- 즉, "이 리뷰의 댓글을 최신순(혹은 과거순)으로 가져와라"라는
- 쿼리에 최적화되어 있음
"""
]작동(review기준)
models.Index(fields=["game"], name="idx_reviews_best")"리뷰가 달릴 때"
- 사용자가 리뷰를 작성해서 저장(save())하면 DB에서는 두 가지 작업을 동시에 수행함
- 실제 데이터 저장 (reviews 테이블)
- 리뷰의 내용, 별점, 작성자 등의 모든 정보를
- 실제 테이블의 맨 뒤(또는 빈 공간)에 넣음 (여기는 순서가 뒤죽박죽일 수 있음)
- 인덱스 업데이트 (idx_reviews_best 인덱스)
- DB 엔진이 방금 저장된 리뷰의 game_id를 확인
- idx_reviews_best라는 인덱스 공간(B-Tree 구조)에서
- 해당 game_id가 들어가야 할 정확한 위치를 찾아서 끼워 넣고 정렬 상태를 유지
- 실제 데이터 저장 (reviews 테이블)
- 사용자가 리뷰를 작성해서 저장(save())하면 DB에서는 두 가지 작업을 동시에 수행함
장점
- 나중에 "A 게임(id=10)의 리뷰를 다 보여줘"라고 요청하면, DB는 이렇게 행동함
- 인덱스가 없다면
- 실제 데이터 테이블을 처음부터 끝까지 다 뒤져야 함
- 인덱스가 존재한다면
- idx_reviews_best를 봄
- 여기는 이미 게임 ID 별로 예쁘게 정렬되어 있으므로,
- game_id=10이 모여 있는 구간을 바로 찾을 수 있음
- 그 구간에 적힌 주소("이 리뷰는 실제 테이블 5번째 줄에 있어")를 보고
- 바로 데이터를 가져옴
- 그 구간에 적힌 주소("이 리뷰는 실제 테이블 5번째 줄에 있어")를 보고
- 인덱스가 없다면
- 나중에 "A 게임(id=10)의 리뷰를 다 보여줘"라고 요청하면, DB는 이렇게 행동함
단점
- 조회속도는 엄청난 이득을 보지만 저장시에 DB에 데이터를 저장할 때 조금 시간이 더 걸림
복합 인덱스
models.Index(fields=["review", "created_at"])- "먼저 리뷰(Review) 별로 그룹을 짓고,
- 그 그룹 안에서 작성일(Created_at) 순서대로 줄을 세워둔 상태
- 1차 정렬: 일단 리뷰 ID가 같은 것끼리 뭉쳐 있음
- 2차 정렬: 같은 리뷰 ID 안에서는 날짜 순서대로 정렬함
제약 조건
constraints
- "한 명의 유저(user)는 한 개의 리뷰(review)에 딱 한 번만 '좋아요'를 누를 수 있다"
- 만약 똑같은 유저가 똑같은 리뷰에 또 save()를 시도하면,
- DB 차원에서 에러(IntegrityError)를 뱉어내며 저장을 거부
class Meta:
db_table = "review_like"
constraints = [
models.UniqueConstraint(
fields=["review", "user"], name="uk_review_like_user"
)
]comment(등록)
코드
# view
def post(self, request, review_id):
# 1. 입력 데이터 검증
serializer = ReviewCommentCreateSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 2. 유저 타입 캐스팅 (Type Hinting)
user = cast(User, request.user)
# 3. 서비스 레이어 호출
comment = create_comment(
author=user,
review_id=review_id,
validated_data=serializer.validated_data,
)
# 4. 응답 반환
return Response(
{"id": comment.id, "message": "댓글이 등록되었습니다."},
status=status.HTTP_201_CREATED,
)
# service
def create_comment(
*, author: User, review_id: int, validated_data: dict[str, Any]
) -> ReviewComment:
"""
댓글 생성 비즈니스 로직
"""
try:
review = Review.objects.get(pk=review_id)
except Review.DoesNotExist:
raise ReviewNotFound()
comment = ReviewComment.objects.create(
author=author,
review=review,
content=validated_data["content"],
)
return comment
# serializer
class ReviewCommentCreateSerializer(serializers.ModelSerializer):
class Meta:
model = ReviewComment
fields = ["content"]리뷰
- 삭제된 리뷰에도 댓글이 달리는 문제
- 전: review = Review.objects.get(pk=review_id)
- 후: review = Review.objects.get(pk=review_id, is_deleted=False)
comment(조회)
코드
# service
def get_review_comment_detail(*, review_id: int) -> QuerySet[Review]:
"""
특정 게임의 리뷰 목록과 리뷰에 작성된 댓글들을 가져옵니다.
"""
# 1. Selector 호출
review = get_review_detail_queryset(review_id)
# 2. 결과 검증
if not review:
raise ReviewNotFound()
return review
# selector
def get_review_detail_queryset(review_id: int) -> Review | None:
"""
리뷰 상세 정보와 해당 리뷰의 댓글 목록을 함께 조회합니다.
"""
return (
Review.objects.filter(id=review_id, is_deleted=False)
.select_related("user") # 리뷰 작성자
.prefetch_related(
Prefetch(
"comments", # ReviewComment 역참조 이름
queryset=ReviewComment.objects.filter(is_deleted=False)
.select_related("user") # 댓글 작성자
.order_by("-created_at"),
)
)
.first() # 없으면 None 반환
)
# view
def get(self, request, review_id):
# 1. 서비스 호출 (리뷰 + 댓글 목록 조회)
review = get_review_comment_detail(review_id=review_id)
# 2. 직렬화
serializer = ReviewCommentListSerializer(review)
# 3. 응답 반환
return Response(serializer.data, status=status.HTTP_200_OK)
# serializer
class CommentListSerializer(serializers.ModelSerializer):
author = AuthorSerializer(source="user", read_only=True)
class Meta:
model = ReviewComment
fields = [
"id",
"author",
"content",
"created_at",
]
class ReviewCommentListSerializer(serializers.ModelSerializer):
author = AuthorSerializer(source="user", read_only=True)
comments = CommentListSerializer(many=True, read_only=True)
class Meta:
model = Review
fields = [
"id",
"author",
"content",
"rating",
"like_count",
"created_at",
"comments",
]2026.01.22 ✅
2026.01.23 ✅
ai
signal
@receiver(post_save, sender=Review)
def trigger_ai_summary(sender, instance, created, **kwargs):
"""
리뷰가 저장(post_save)될 때마다 실행되는 함수
"""
# 1. 수정된 리뷰는 무시하고, 새로 생성된 리뷰일 때만 체크
if not created:
return
game = instance.game
# 설정값 로드
MIN_COUNT = getattr(settings, "AI_SUMMARY_MIN_REVIEW_COUNT", 10)
UPDATE_DAYS = getattr(settings, "AI_SUMMARY_UPDATE_INTERVAL_DAYS", 30)
# 2. 현재 유효 리뷰 개수 확인
current_count = game.reviews.filter(is_deleted=False).count()
# 조건1: 리뷰 개수가 기준(10개) 이상인가?
if current_count >= MIN_COUNT:
summary_obj = getattr(game, "summary", None)
should_run = False
# 조건1-1: 요약본이 아예 없으면 -> 실행
if not summary_obj:
should_run = True
# 조건1-2: 요약본이 있지만, 마지막 업데이트로부터 30일 지났으면 -> 실행
else:
time_diff = timezone.now() - summary_obj.updated_at
if time_diff > timedelta(days=UPDATE_DAYS):
should_run = True
# 실행 조건 만족 시 Celery Task 호출
if should_run:
print(f"[Signal] Triggering AI Summary for Game {game.id}")
run_ai_summary.delay(game.id)
——————————————————————————————————————[비교]—————————————————————————————————————————
# 예외 처리를 위해 모델 임포트 필요 (GameReviewSummary가 정의된 곳)
from apps.ai.models import GameReviewSummary
@receiver(post_save, sender=Review)
def trigger_ai_summary(sender, instance, created, **kwargs):
if not created:
return
game = instance.game
# [수정 1] Settings 값 가져오기 (try-except 또는 직접 접근)
try:
MIN_COUNT = settings.AI_SUMMARY_MIN_REVIEW_COUNT
except AttributeError:
MIN_COUNT = 10
try:
UPDATE_DAYS = settings.AI_SUMMARY_UPDATE_INTERVAL_DAYS
except AttributeError:
UPDATE_DAYS = 30
...
if current_count >= MIN_COUNT:
try:
summary_obj = game.summary
except GameReviewSummary.DoesNotExist:
summary_obj = None
should_run = False
if not summary_obj:
should_run = True
else:
# ... (나머지 로직 동일)service
logger = logging.getLogger(__name__)
class ReviewSummaryService:
def __init__(self):
"""
서비스 초기화: Client 생성 및 공통 설정 정의
"""
api_key = settings.GEMINI_API_KEY
# 최신 SDK의 Client 인스턴스 생성
self.client = genai.Client(api_key=api_key)
self.model_name = "gemini-flash-latest"
# 환경변수 혹은 settings에서 값을 가져옴, 없을 경우 기본값(default)을 사용
self.min_review_count = getattr(settings, "AI_SUMMARY_MIN_REVIEW_COUNT", 10)
self.update_interval_days = getattr(
settings, "AI_SUMMARY_UPDATE_INTERVAL_DAYS", 30
)
self.min_review_length = getattr(settings, "AI_REVIEW_MIN_LENGTH", 10)
self.min_valid_reviews = getattr(settings, "AI_SUMMARY_MIN_VALID_REVIEWS", 3)
self.summary_review_count = getattr(settings, "AI_SUMMARY_REVIEW_COUNT", 5)
self.profanity_pattern = re.compile("|".join(BAD_PATTERNS))
# AI의 페르소나(역할)를 정의
self.system_instruction = (
"당신은 20년 경력의 베테랑 게임 전문 리뷰 분석가입니다. "
"사용자의 리뷰 데이터를 객관적으로 분석하여, "
"게임의 장단점과 핵심적인 특징을 명확하게 요약해야 합니다. "
# 2차 필터링
"리뷰 원문에 비속어, 은어, 공격적인 표현이 포함되어 있더라도 "
"절대로 요약 결과에는 그대로 포함하지 마세요. "
"해당 표현은 무시하거나, 격식 있고 정중한 표현으로 순화하여 요약해야 합니다. "
"무조건 JSON 형식으로만 응답하세요."
)
def get_summary(self, game_id: int) -> dict:
"""
외부에서 호출하는 요약 조회 메서드
"""
try:
game = Game.objects.select_related("summary").get(id=game_id)
except Game.DoesNotExist:
raise GameNotFound()
review_count = game.reviews.filter(is_deleted=False).count() # type: ignore
if review_count < self.min_review_count:
raise NotEnoughReviews()
summary_obj = getattr(game, "summary", None)
# 갱신이 필요한지 확인(필요OX)
if self._update_and_parse(summary_obj):
# 갱신이 필요하면 AI 생성 및 저장을 수행하고 결과를 반환
return self._generate_and_save(game, summary_obj)
# 갱신이 필요 없으면 DB에 저장된 JSON 텍스트를 파싱하여 반환
return json.loads(summary_obj.text) # type: ignore
def _update_and_parse(self, summary_obj) -> bool:
"""
요약 데이터 갱신 여부 판단
"""
# 기존 요약이 없으면 무조건 갱신(생성)
if not summary_obj:
return True
# 현재 시간과 마지막 수정 시간(updated_at)의 차이를 계산
update_time_difference = timezone.now() - summary_obj.updated_at
# 마지막 수정일로부터 30일이 지났으면 True를 반환
return update_time_difference > timedelta(days=self.update_interval_days)
def _generate_and_save(self, game: Game, summary_obj) -> dict:
"""
AI 요약 생성 및 DB 저장
"""
# 글자 수 필터링 및 유효 리뷰 개수 확인
bring_reviews = (
game.reviews.annotate(text_len=Length("content")) # type: ignore
.filter(is_deleted=False, text_len__gte=self.min_review_length)
.order_by("-created_at")[:20]
)
clean_reviews = []
for review in bring_reviews:
if self.profanity_pattern.search(review.content):
continue # 욕설 발견 시 건너뜀
clean_reviews.append(review)
if len(clean_reviews) >= self.summary_review_count:
break
# 유효한 리뷰가 설정된 개수(예: 3개) 미만이면 중단
if len(clean_reviews) < self.min_valid_reviews:
logger.warning(
f"Game({game.id}) has enough raw reviews but VALID reviews({len(clean_reviews)}) "
f"are less than {self.min_valid_reviews}."
)
raise NotEnoughValidReviews()
# 리뷰 내용들을 줄바꿈 문자로 연결하여 하나의 문자열로 만듬
reviews_text = "\n".join([f"- {r.content}" for r in clean_reviews])
# AI에게 보낼 사용자 프롬프트를 구성
user_prompt = f"""
게임명: {game.name}
아래 유저 리뷰들을 분석해서 지정된 JSON 스키마에 맞춰 요약해줘.
[Review Data]
{reviews_text}
"""
try:
response = self.client.models.generate_content(
model=self.model_name,
contents=user_prompt,
# 생성 설정 객체
config=types.GenerateContentConfig(
system_instruction=self.system_instruction, # 페르소나 적용
response_mime_type="application/json", # 응답 형식을 JSON으로 강제
response_schema=GameSummary, # Pydantic 모델을 스키마로 전달하여 구조 강제
safety_settings=SAFETY_SETTINGS, # 안전 설정 적용
),
)
# 생성된 JSON 문자열을 가져옴
result_json_str = response.text
with transaction.atomic():
if summary_obj:
# 기존 요약이 있다면 내용을 수정
summary_obj.text = result_json_str
summary_obj.save()
else:
# 없다면 생성
GameReviewSummary.objects.create(game=game, text=result_json_str)
# 저장된 JSON 문자열을 딕셔너리로 변환하여 반환
return json.loads(result_json_str)
except Exception as e:
# 에러 발생 시 로그를 남김
logger.error(f"Game({game.id}) AI Summary Generation Failed: {e}")
# 생성에 실패했더라도 기존 캐시가 있다면 반환
if summary_obj:
return json.loads(summary_obj.text)
# 기존 데이터도 없고 생성도 실패했다면 503 예외 발생
raise AiGenerationFailed()
——————————————————————————————————————[비교]—————————————————————————————————————————
# [ai/services.py]
class ReviewSummaryService:
def __init__(self):
api_key = settings.GEMINI_API_KEY
self.client = genai.Client(api_key=api_key)
self.model_name = "gemini-flash-latest"
# [수정 1] Settings 값을 안전하게 가져오기
# 방법 A: settings.py에 해당 값이 무조건 있다고 가정 (가장 깔끔함)
# self.min_review_count = settings.AI_SUMMARY_MIN_REVIEW_COUNT
# 방법 B: 값이 없을 경우를 대비한 안전한 코드 (getattr 제거)
try:
self.min_review_count = settings.AI_SUMMARY_MIN_REVIEW_COUNT
except AttributeError:
self.min_review_count = 10
try:
self.update_interval_days = settings.AI_SUMMARY_UPDATE_INTERVAL_DAYS
except AttributeError:
self.update_interval_days = 30
# ... 나머지 설정들도 동일한 방식으로 변경 ...
def get_summary(self, game_id: int) -> dict:
try:
# select_related를 사용했으므로 DB 쿼리 레벨에서 join됩니다.
# 하지만 Python 객체에서 접근할 때 없으면 에러가 납니다.
game = Game.objects.select_related("summary").get(id=game_id)
except Game.DoesNotExist:
raise GameNotFound()
# ... (중략) ...
# [수정 2] 요약본 존재 여부 확인
# summary_obj = getattr(game, "summary", None) 대신:
if hasattr(game, "summary"):
summary_obj = game.summary
else:
summary_obj = None
# 갱신 로직 등은 그대로 유지
if self._update_and_parse(summary_obj):
return self._generate_and_save(game, summary_obj)
return json.loads(summary_obj.text)ai_exceptions
from rest_framework.exceptions import APIException
from rest_framework import status
from django.conf import settings
MIN_COUNT = getattr(settings, "AI_SUMMARY_MIN_REVIEW_COUNT", 10)
MIN_VALID_COUNT = getattr(settings, "AI_SUMMARY_MIN_VALID_REVIEWS", 3)
class NotEnoughReviews(APIException):
"""
리뷰 개수가 요약 생성 기준(10개) 미만일 때 발생
"""
status_code = status.HTTP_400_BAD_REQUEST
default_detail = (
f"리뷰가 부족하여 요약을 생성할 수 없습니다. (최소 {MIN_COUNT}개 필요)"
)
default_code = "not_enough_reviews"
class NotEnoughValidReviews(APIException):
"""
글자 수 조건을 만족하는 '유효한 리뷰'가 부족할 때 발생
"""
status_code = status.HTTP_400_BAD_REQUEST
default_detail = (
f"요약에 적합한 긴 리뷰가 부족합니다. (최소 {MIN_VALID_COUNT}개 필요)"
)
default_code = "not_enough_valid_reviews"
class AiGenerationFailed(APIException):
"""
Gemini API 호출 실패 또는 로직 에러 시 발생
"""
status_code = status.HTTP_503_SERVICE_UNAVAILABLE
default_detail = "AI 요약 생성 중 일시적인 오류가 발생했습니다."
default_code = "ai_generation_failed"
class GameNotFound(APIException):
"""
게임 ID가 존재하지 않을 때 발생
(Community 앱에 같은 예외가 존재하지만 의존성 분리를 위해 따로 정의)
"""
status_code = status.HTTP_404_NOT_FOUND
default_detail = "존재하지 않는 게임입니다."
default_code = "game_not_found"
——————————————————————————————————————[비교]—————————————————————————————————————————
# [ai/exceptions/ai_exceptions.py]
from rest_framework.exceptions import APIException
from rest_framework import status
from django.conf import settings
# [수정] getattr 제거
try:
MIN_COUNT = settings.AI_SUMMARY_MIN_REVIEW_COUNT
except AttributeError:
MIN_COUNT = 10
try:
MIN_VALID_COUNT = settings.AI_SUMMARY_MIN_VALID_REVIEWS
except AttributeError:
MIN_VALID_COUNT = 3
# ... 이후 클래스 정의 동일 ...
안녕하세요.