steam_data 로직 수정

2star_·2025년 1월 26일
0

최종 프로젝트

목록 보기
29/32

기존에 Selenium을 사용하던 부분을 전부 제거하고, requestsBeautifulSoup로 대체했다.
steamId항상 64비트라고 가정하여, 커스텀 URL 변환(resolve_vanity_url)이나 steam_id_str.isdigit() 같은 로직도 모두 뺐다.
또한 -게임 목록 공개 여부(check_owned_games_public)- 검사도 제거했다.

변경 코드

import environ
import requests
from django.core.management.base import BaseCommand
from django.db import transaction
from accounts.models import (
    Account,
    SteamProfile,
    SteamReview,
    SteamPlaytime,
)

from bs4 import BeautifulSoup


class Command(BaseCommand):
    help = "Steam 프로필/리뷰/플레이타임 정보를 가져와 DB에 반영"

    def handle(self, *args, **options):
        # 1) 환경 변수에서 API 키 불러오기
        env = environ.Env()
        api_key = env("STEAM_API_KEY")  # .env에 STEAM_API_KEY=... 설정

        # 2) steamId가 비어있지 않은 Account를 모두 조회
        accounts = Account.objects.exclude(steamId="")

        for account in accounts:
            steam_id_str = account.steamId.strip()

            # 3) 전체 프로필 공개 여부 API로 확인
            visibility_public = self.check_profile_public(api_key, steam_id_str)

            # 일단 기본값 False
            is_review = False
            is_playtime = False

            if visibility_public:
                # (A) 리뷰 크롤링 (최대 3개) - BeautifulSoup 이용
                review_data = self.fetch_top3_reviews(steam_id_str)
                if review_data:
                    is_review = True

                # (B) API로 플레이타임 조회 (상위 2개)
                playtime_data = self.fetch_top2_playtime_api(api_key, steam_id_str)
                if playtime_data:
                    is_playtime = True

                # (C) DB 저장
                with transaction.atomic():
                    sp, _ = SteamProfile.objects.get_or_create(account=account)
                    sp.is_review = is_review
                    sp.is_playtime = is_playtime
                    sp.save()

                    # 기존 리뷰/플레이타임 삭제 후 재생성 (중복 방지)
                    SteamReview.objects.filter(account=account).delete()
                    SteamPlaytime.objects.filter(account=account).delete()

                    if is_review:
                        for rd in review_data:
                            SteamReview.objects.create(
                                account=account,
                                app_id=rd["app_id"]
                            )
                    if is_playtime:
                        for pd in playtime_data:
                            SteamPlaytime.objects.create(
                                account=account,
                                app_id=pd["app_id"]
                            )
            else:
                # 프로필이 Private -> is_review=False, is_playtime=False
                with transaction.atomic():
                    sp, _ = SteamProfile.objects.get_or_create(account=account)
                    sp.is_review = False
                    sp.is_playtime = False
                    sp.save()

                    # 기존 리뷰/플레이타임 삭제
                    SteamReview.objects.filter(account=account).delete()
                    SteamPlaytime.objects.filter(account=account).delete()

        self.stdout.write(self.style.SUCCESS("Steam 데이터 처리 완료."))

    # -------------------------------------------------------
    # 1) 스팀 프로필이 Public인지 판별 (communityvisibilitystate == 3)
    # -------------------------------------------------------
    def check_profile_public(self, api_key, steam_id_str):
        """
        64비트 steam_id_str이 공개 상태인지 확인.
        """
        url = "https://api.steampowered.com/ISteamUser/GetPlayerSummaries/v2/"
        params = {"key": api_key, "steamids": steam_id_str}
        try:
            resp = requests.get(url, params=params).json()
        except Exception as e:
            print(f"Error checking profile public: {e}")
            return False

        players = resp.get("response", {}).get("players", [])
        if not players:
            return False

        player = players[0]
        vis_state = player.get("communityvisibilitystate", 1)
        return (vis_state == 3)

    # -------------------------------------------------------
    # 2) 리뷰 페이지 크롤링 (상위 3개 'Recommended')
    # -------------------------------------------------------
    def fetch_top3_reviews(self, steam_id_str):
        """
        Selenium 없이 HTML만 파싱하여 상위 3개의 추천 리뷰(app_id) 가져오기.
        """
        url = f"https://steamcommunity.com/profiles/{steam_id_str}/recommended"
        headers = {
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/87.0.4280.66 Safari/537.36"
            )
        }

        try:
            response = requests.get(url, headers=headers, timeout=10)
        except Exception as e:
            print(f"Error fetching reviews page: {e}")
            return []

        # 상태 코드 확인 (비공개거나 오류 발생 시)
        if response.status_code != 200:
            return []

        soup = BeautifulSoup(response.text, "html.parser")
        review_boxes = soup.select(".review_box")

        recommended = []
        for box in review_boxes:
            try:
                # '추천' 리뷰만 필터링
                title_elem = box.select_one(".vote_header .title > a")
                if not title_elem or "Recommended" not in title_elem.text:
                    continue

                href = title_elem.get("href", "")
                if "/recommended/" not in href:
                    continue

                app_id = href.split("/recommended/")[1].split("/")[0]
                recommended.append({"app_id": app_id})

                if len(recommended) >= 3:
                    break
            except Exception as e:
                print(f"리뷰 박스 처리 중 오류: {e}")
                continue

        return recommended

    # -------------------------------------------------------
    # 3) 플레이 타임 조회 (상위 2개)
    # -------------------------------------------------------
    def fetch_top2_playtime_api(self, api_key, steam_id_str):
        """
        GetOwnedGames API로 플레이타임이 가장 많은 상위 2개 게임(app_id) 반환.
        """
        url = "https://api.steampowered.com/IPlayerService/GetOwnedGames/v1/"
        params = {
            "key": api_key,
            "steamid": steam_id_str,
        }
        try:
            resp = requests.get(url, params=params).json()
        except Exception as e:
            print("Error fetching owned games:", e)
            return []

        games = resp.get("response", {}).get("games", [])
        if not games:
            return []

        game_data = []
        for g in games:
            app_id = g.get("appid")
            pt_min = g.get("playtime_forever", 0)
            pt_hr = round(pt_min / 60.0, 2)  # 시간 단위 환산 (소수점 둘째 자리)
            game_data.append({"app_id": str(app_id), "playtime": pt_hr})

        # 플레이타임 내림차순 정렬 후 상위 2개
        sorted_data = sorted(game_data, key=lambda x: x["playtime"], reverse=True)
        return sorted_data[:2]

주요 변경 사항

  1. Selenium 전면 제거

    • webdriver.ChromeOptions(), driver.get(), WebDriverWait 등 모든 관련 코드를 삭제했다.
    • 헤드리스 크롬을 구동하지 않으므로 CPU 부담이 크게 줄어든다.
  2. 리뷰 크롤링 로직(fetch_top3_reviews)

    • requestsBeautifulSoup만 사용한다.
    • 받은 HTML에서 .review_box를 찾아 “Recommended” 텍스트가 있는 리뷰만 걸러내고, 최대 3개까지 app_id를 추출한다.
  3. 커스텀 URL 변환 로직 제거

    • steam_id_str가 항상 64비트 ID라고 가정하므로 resolve_vanity_url 함수는 완전히 삭제했다.
    • check_profile_public 등에서 steam_id_str.isdigit() 관련 분기나 vanityurl 호출도 제거
  4. 게임 목록 공개 여부(check_owned_games_public) 제거

    • 필요 없다고 했으므로 해당 함수를 삭제했다.
  5. DB 저장 로직 동일

    • SteamProfileis_review, is_playtime 여부를 갱신하고
    • 기존 SteamReview, SteamPlaytime 전부 삭제 후 새로 생성한다(중복 방지)
profile
안녕하세요.

0개의 댓글

관련 채용 정보