Coverage for services / naver_finance_scraper_service.py: 93%

1# services/naver_finance_scraper_service.py 

2import aiohttp 

3import logging 

4from bs4 import BeautifulSoup 

5from typing import Optional 

6 

7class NaverFinanceScraperService: 

8 """네이버 금융 웹페이지 스크래핑을 전담하는 서비스 클래스.""" 

9 

10 def __init__(self, logger: Optional[logging.Logger] = None): 

11 self._logger = logger or logging.getLogger(__name__) 

12 # 차단 방지를 위한 User-Agent 설정 

13 self._headers = { 

14 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' 

15 } 

16 

17 async def fetch_yoy_profit_growth(self, code: str) -> float: 

18 """ 

19 특정 종목의 최근 분기(또는 예상치)와 전년 동기 영업이익을 비교하여 YoY 성장률을 반환합니다. 

20 가장 최근 분기(예상치 등) 데이터가 비어있을 경우, 직전 분기 실적으로 후퇴(Fallback)하여 계산합니다. 

21 턴어라운드(적자 -> 흑자 전환) 시그널 발생 시 999.0을 반환합니다. 

22 """ 

23 url = f"https://finance.naver.com/item/main.naver?code={code}" 

24 

25 try: 

26 async with aiohttp.ClientSession() as session: 

27 async with session.get(url, headers=self._headers, timeout=5) as response: 

28 if response.status != 200: 

29 self._logger.warning({"event": "scraping_http_error", "code": code, "status": response.status}) 

30 return 0.0 

31 html = await response.text() 

32 

33 soup = BeautifulSoup(html, 'html.parser') 

34 

35 div_cop = soup.find('div', class_='cop_analysis') 

36 if not div_cop: return 0.0 

37 

38 tbody = div_cop.find('tbody') 

39 if not tbody: return 0.0 39 ↛ exitline 39 didn't return from function 'fetch_yoy_profit_growth' because the return on line 39 wasn't executed

40 

41 op_row = None 

42 for tr in tbody.find_all('tr'): 42 ↛ 48line 42 didn't jump to line 48 because the loop on line 42 didn't complete

43 th = tr.find('th') 

44 if th and '영업이익' in th.text: 44 ↛ 42line 44 didn't jump to line 42 because the condition on line 44 was always true

45 op_row = tr 

46 break 

47 

48 if not op_row: return 0.0 48 ↛ exitline 48 didn't return from function 'fetch_yoy_profit_growth' because the return on line 48 wasn't executed

49 

50 tds = op_row.find_all('td') 

51 quarterly_tds = tds[4:] # 앞 4열은 연간, 뒤 6열은 최근 분기 

52 

53 # 최소한 후퇴(Fallback) 로직을 처리할 수 있을 만큼의 열이 있는지 확인 

54 if len(quarterly_tds) < 6: return 0.0 54 ↛ exitline 54 didn't return from function 'fetch_yoy_profit_growth' because the return on line 54 wasn't executed

55 

56 # 1. 우선 가장 최근 분기(-1)와 전년 동기(-5) 데이터 추출 시도 

57 latest_str = quarterly_tds[-1].text.replace(',', '').strip() 

58 yoy_str = quarterly_tds[-5].text.replace(',', '').strip() 

59 

60 # 2. 대안(Fallback) 탐색: 최신 데이터가 비어있거나 '-'인 경우, 직전 확정 분기(-2)로 후퇴 

61 if not latest_str or latest_str == '-': 

62 latest_str = quarterly_tds[-2].text.replace(',', '').strip() 

63 yoy_str = quarterly_tds[-6].text.replace(',', '').strip() 

64 self._logger.debug({"event": "fallback_to_previous_quarter", "code": code}) 

65 

66 # 3. 후퇴 후에도 데이터가 유효하지 않으면 0.0 반환 

67 if not latest_str or latest_str == '-' or not yoy_str or yoy_str == '-': 

68 return 0.0 

69 

70 latest_op = float(latest_str) 

71 yoy_op = float(yoy_str) 

72 

73 # --- 턴어라운드 (적자 -> 흑자) 예외 처리 --- 

74 if yoy_op <= 0 and latest_op > 0: 

75 self._logger.debug({"event": "turnaround_detected", "code": code, "yoy": yoy_op, "latest": latest_op}) 

76 return 999.0 

77 

78 # --- 일반 YoY 성장률 계산 --- 

79 if yoy_op > 0 and latest_op > 0: 

80 return ((latest_op - yoy_op) / yoy_op) * 100.0 

81 

82 return 0.0 

83 

84 except Exception as e: 

85 self._logger.warning({"event": "scraping_failed", "code": code, "error": str(e)}) 

86 return 0.0