[혼만파] 3주차_혼자 만들면서 공부하는 파이썬

 

3주차부터 드디어 정상 진도? 의 궤도에 진입하였습니다.

근데 이제 진도가 +4 챕터가 되어버린...

방학 반납하고 복습 예정입니다. 5주차가 Chapter 13인 것을 보니 예습도 예정이어야 되겠군요...

 

ModuleNotFoundError: No module named 'playwright'

오류가 나는 경우 playwright부터 설치해야합니다.

pip install playwright
python -m playwright install

 

 

ModuleNotFoundError: No module named 'pandas'

오류가 나는 경우 pandas도 설치 해야합니다.

pip install pandas

 

Chapter 07. 종목별 시가총액 데이터 수집하기

 

페이지 열기

from playwright.sync_api import Page  # Playwright의 동기 API에서 Page 객체 임포트

from step_1_2 import run_playwright   # playwright를 실행시키는 함수(직접 구현) 임포트
from step_1_3 import goto_market_cap  # 특정 페이지(시가총액 페이지)로 이동하는 함수 임포트

def fetch_total_page(page: Page) -> int:
    # "페이지 네비게이션"이라는 텍스트를 가진 table 태그를 찾음
    table = page.locator("table", has_text="페이지 네비게이션")
    # 해당 table에서 tbody > tr > td 중 마지막 td 태그를 찾음
    td = table.locator("tbody > tr > td").last
    # 마지막 td 안의 <a> 태그의 href 속성값 추출 (예: '?page=15')
    href = td.locator("a").get_attribute("href")
    # href에서 '=' 이후 숫자만 추출해서 int로 변환 (총 페이지 수)
    return int(href.split("=")[-1])

if __name__ == "__main__":
    # playwright를 실행(브라우저, 페이지 객체 등 반환)
    play, browser, page = run_playwright(slow_mo=1000)
    # 시가총액 페이지로 이동
    goto_market_cap(page)
    # 총 페이지 수 추출
    total_page = fetch_total_page(page)
    print(f"{total_page=}")

    # 브라우저 및 playwright 세션 종료
    browser.close()
    play.stop()

 

시가 총액 페이지 까지 정상적으로 이동하는 지 확인

예제 코드와 달리 네이버 증권 내 시가총액 text링크가 두개라 아래와 같이 약간 수정이 필요합니다.

import json
from pathlib import Path

from playwright.sync_api import Page

from step_1_1 import OUT_DIR  # 이전에 작성한 모듈을 불러옵니다.
from step_1_2 import run_playwright

OUT_1_3 = OUT_DIR / f"{Path(__file__).stem}.json"

def goto_market_cap(page: Page):
    page.goto("https://finance.naver.com")
    page.get_by_role("link", name="국내증시").click()
    # "시가총액"이라는 링크가 여러 개일 때 첫 번째만 클릭
    page.get_by_role("link", name="시가총액", exact=True).first.click()

def parse_table_kospi(page: Page) -> tuple[list, list]:
    tag_table = page.locator("table", has_text="코스피")  # 코스피 시가총액 표
    tag_thead = tag_table.locator("thead > tr > th")      # 헤더 열
    header = tag_thead.all_inner_texts()
    tag_tbody = tag_table.locator("tbody > tr")           # 보디 행
    body = [tr.locator("td").all_inner_texts() for tr in tag_tbody.all()]
    return header, body

if __name__ == "__main__":
    play, browser, page = run_playwright(slow_mo=1000)
    goto_market_cap(page)  # 시가총액 페이지로 이동
    header, body = parse_table_kospi(page)  # 코스피 시가총액 데이터 수집
    dumped = json.dumps(dict(header=header, body=body), ensure_ascii=False, indent=2)
    OUT_1_3.write_text(dumped, encoding="utf-8")  # JSON으로 저장

    browser.close()
    play.stop()

 

csv파일로 저장하기

"""
1. 네이버 증권 페이지에 접근하여 Playwright로 시가총액(코스피) 테이블 크롤링 및 저장
2. JSON 파일에서 표 데이터를 불러옴
3. 불필요한 공백 및 결측값, 빈 열 등을 정재
4. 정제된 데이터를 CSV 파일로 저장
"""
import json
from pathlib import Path

import pandas as pd  # 데이터프레임 처리를 위한 pandas 라이브러리

from step_1_1 import OUT_DIR  # 결과 파일을 저장할 폴더 경로
from step_1_3 import OUT_1_3  # 이전 단계에서 저장한 JSON 파일 경로

def clean_white_space(text: str) -> str:
    # 문자열 내 불필요한 공백(여러 칸, 줄바꿈 등)을 하나의 공백으로 정제
    return " ".join(text.split())

def table_to_dataframe(header: list, body: list) -> pd.DataFrame:
    # body(2차원 리스트)와 header(컬럼명 리스트)로 DataFrame 생성
    df_raw = pd.DataFrame(body, columns=header)
    # 한 열이라도 결측값이 있으면 해당 행 삭제
    df_raw = df_raw.dropna(how="any")
    # 마지막 열(보통 불필요한 빈 열) 제거
    df_raw = df_raw.iloc[:, :-1]
    # 모든 컬럼에 대해 공백 문자 정제 함수 적용
    for col in df_raw.columns:
        df_raw[col] = df_raw[col].apply(clean_white_space)
    return df_raw

if __name__ == "__main__":
    # 1_3단계에서 저장한 JSON 파일을 불러와서 파싱
    parsed = json.loads(OUT_1_3.read_text(encoding="utf8"))
    header, body = parsed["header"], parsed["body"]
    # 표 데이터를 DataFrame으로 변환 및 정제
    df_raw = table_to_dataframe(header, body)
    # DataFrame을 CSV 파일로 저장 (파일명은 현재 .py 파일명 기준)
    df_raw.to_csv(OUT_DIR / f"{Path(__file__).stem}.csv", index=False)

 

csv로 현재 기준 데이터를 제대로 추출하는지 확인

 

증권화면

 

 

크롤링 마지막 페이지 체크하기

if __name__ == "__main__":
    play, browser, page = run_playwright(slow_mo=1000)
    goto_market_cap(page)  # 시가총액 페이지로 이동
    total_page = fetch_total_page(page)  # 총 페이지 개수 추출
    print(f"{total_page=}")

    browser.close()
    play.stop()

첫 페이지부터 읽어서 마지막 페이지까지 페이지네이션 이동하기

from pathlib import Path

import pandas as pd
from playwright.sync_api import Page

from step_1_1 import OUT_DIR  # 이전에 작성한 모듈을 불러옵니다.
from step_1_2 import run_playwright
from step_1_3 import goto_market_cap, parse_table_kospi
from step_1_4 import table_to_dataframe
from step_2_1 import fetch_total_page

OUT_2_2 = OUT_DIR / f"{Path(__file__).stem}.csv"


def goto_page(page: Page, to: int):
    page.goto(f"https://finance.naver.com/sise/sise_market_sum.naver?&page={to}")


def fetch_market_cap(page: Page) -> pd.DataFrame:
    total_page = fetch_total_page(page)  # 총 페이지 개수 추출
    result = []
    for to in range(1, total_page + 1):  # 1부터 total_page까지 반복
        goto_page(page, to)  # 페이지 이동
        header, body = parse_table_kospi(page)  # 코스피 시가총액 표 추출
        df_raw = table_to_dataframe(header, body)  # 데이터 정제 및 DataFrame 객체 생성
        result.append(df_raw)  # DataFrame 객체를 리스트에 임시 저장
    return pd.concat(result)  # DataFrame 객체 결합


if __name__ == "__main__":
    play, browser, page = run_playwright(slow_mo=1000)
    goto_market_cap(page)  # 시가총액 페이지로 이동
    df_result = fetch_market_cap(page)  # 코스피 시가총액 표 추출
    df_result.to_csv(OUT_2_2, index=False)  # CSV 파일로 저장

    browser.close()
    play.stop()

 

단계별로 설명해줘서 좋습니다.

구몬학습처럼 단계별 지속학습이라 생각외로 덜 까먹는 것 같습니다.

 

파이썬으로 뭐 하나 만들 수 있는 날이 올 때 까지!