Trafilatura - requsets를 이용한 한글깨짐 문제 해결하기

TL;DR

• Trafilatura를 이용한 한글 뉴스 기사 본문 자동 수집 시 인코딩 문제가 간헐적으로 발생함
• 한글 뉴스 기사의 약 80%는 정적인 HTML 페이지로 구성되어 requests로 수집 가능함
• requests로 HTML을 받아서 trafilatura로 넘기면 인코딩 문제 회피하면서도 빠른 처리 속도 유지 가능함
• Trafilatura + requests 조합 테스트 결과, 인코딩 처리를 위해 0.52초 → 0.61초로 다소 증가하였으나 인코딩 문제를 완전히 해결
• 이 방식을 통해 한글 뉴스 기사 수집의 안정성과 효율성을 동시에 확보할 수 있음

---

한국어 뉴스를 스크랩할 때 한글 문자가 깨지는 현상이 종종 발생한다. 이전 글에서 소개한 Trafilatura가 자동으로 문자를 변환해주지만 완벽하지는 않다. 이번 글에서는 이 문제를 해결하기 위해, requests로 HTML을 받아와 Trafilatura로 넘겨주는 방식을 다룬다. 특히, requests가 자체적으로 인코딩 문제를 완벽히 처리하지 못하므로, 이를 보완하는 별도의 문자 변환 함수도 소개한다.

 

웹 크롤링 시 한글 처리 흐름

1

HTTP 요청 및 응답 수신

requests로 서버에 요청을 보내고 response.content로 HTML 데이터를 받음

2

서버 문자 정보의 한계

서버 보낸 문자 방식을 잘못 변환하는 경우가 있음
예: 실제는 EUC-KR인데 UTF-8로 변환

3

텍스트 변환 방법

HTML 데이터를 올바른 텍스트로 변환하는 것이 핵심:

• response.encoding → requests의 자동 해석, 부정확할 수 있음
• 'euc-kr','utf-8','cp949'등을 명시 → 지정한 방식으로 변환
• guess_best_decode() → 한글 비율로 최적 방식 추정

4

정상 텍스트 처리

올바른 변환으로 한글 깨짐 없는 텍스트를 얻음
이후 trafilatura 등으로 본문 추출이 가능

---

Requests: 빠르고 안정적인 HTML 수집 도구

requests는 가장 널리 사용되는 Python의 HTTP 라이브러리로, 웹 서버로부터 HTML을 빠르게 받아올 수 있다. 특히 정적인 HTML로 구성된 뉴스 기사를 수집할 때 매우 효율적이다.

Requests 사용 시 주요 장점

1. 인코딩을 직접 지정해서 문자가 깨지는 문제를 막을 수 있음
아래의 예시를 보면 서버는 분명히 charset=EUC-KR로 보냈는데, 제목은 깨져서 나왔다. 이는 html을 받은 후 서버가 보낸 charset을 제대로 변환하지 못했기 때문이다. trafilatura를 사용하면 이부분을 바로 잡아줄 수 없지만, requests를 사용하면 이 부분을 제대로 잡아 줄 기회가 있다. 

        <title> ̱        پ  µ   ѱ     "  ġ  ׸  ָ   ⷷ"    ̵           -  Ӵ       </title>
        <meta http-equiv="content-type" content="text/html; charset=EUC-KR">

 

2. 가볍고 빠른 요청 처리
브라우저를 사용하지 않고 순수 HTML 요청만 처리하기 때문에 속도가 빠르고 리소스 사용량도 적다.이전 글에서도 언급했듯, 우리가 수집하는 한글 뉴스의 약 80%는 정적 페이지로 만들어져 있다. 따라서, requests와 문자열 보정 함수를 거치더라도, 전체적인 수집 속도는 여전히 빠르게 유지될 수 있다.

---

Requests + Trafilatura의 기본 사용법

다음과 같은 간단한 코드로 쉽게 두 기능을 결합할 수 있다. 아래는 requests로 html을 받아와서 trafilatura에 넘겨주는 기본 코드이다. 여기서는 requests가 인코딩을 추정하는 기본 기능인  apparent_encoding을 적용하였다. 

import requests
import trafilatura
from trafilatura.metadata import extract_metadata

# 대상 URL 설정
url = "https://news.mt.co.kr/mtview.php?no=2025050210033043266&VMK"

# HTML 다운로드 (requests 사용)
response = requests.get(url, timeout=10)
response.encoding = response.apparent_encoding  # 인코딩 자동 감지
downloaded = response.text

# 메타데이터 및 본문 텍스트 추출
metadata = extract_metadata(downloaded)
text = trafilatura.extract(downloaded, output_format='txt', include_comments=False, favor_precision=True)

# 결과 출력
print(f"📰 제목: {metadata.title}")
print(f"📅 날짜: {metadata.date}")
print(f"📝 본문:\n{text}")

 

하지만, apparent_encoding으로는 여전히 제대로 인식이 되지 않는 웹사이트가 존재하여, 아래와 같이 별도의 문자열 인식 함수를 추가하였다. 이 함수를 간단히 설명하면, response.encoding → response.apparent_encoding → 'utf-8' → 'euc-kr' → 'cp949'을 순차적으로 테스트 해보면서 한글이 많이 나오는 인코딩 방식을 선정하는 함수다.

def guess_best_decode(data: bytes, encodings: List[str]) -> str:
    """Pick the decoding that yields the most Hangul characters."""
    best_text: Optional[str] = None
    best_score = -1
    for enc in encodings:
        if not enc:
            continue
        try:
            text = data.decode(enc, errors="replace")
        except LookupError:
            continue
        # Score by Hangul character count
        score = sum(0xAC00 <= ord(ch) <= 0xD7A3 for ch in text)
        if score > best_score:
            best_text, best_score = text, score
        # Early exit if score is very high (heuristic)
        if score > 10:
            break
    if best_text is None:
        best_text = data.decode("utf-8", errors="replace")
    return best_text
    
def fetch_html_with_requests(url: str, timeout: int = 10) -> dict:
    """
    requests로 HTML만 다운로드하는 함수 (본문 추출은 별도로 수행)
    guess_best_decode를 활용한 강력한 인코딩 자동 보정
    """
    result = {
        'url': url,
        'html': '',
        'success': False,
        'error': None,
        'method_used': 'requests_only'
    }

    try:
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36'
        }
        response = requests.get(url, headers=headers, timeout=timeout)
        response.raise_for_status()
        # ✅ guess_best_decode를 통해 인코딩 복원
        candidates = [
        response.encoding,           # 서버가 응답 헤더에 명시한 인코딩
        response.apparent_encoding,  # requests가 추정한 인코딩
        'utf-8',                     # 가장 일반적인 기본값
        'euc-kr',                    # 한국 뉴스에서 흔히 사용됨
        'cp949'                      # euc-kr의 superset, Windows 환경에서 자주 사용
        ]

        result['html'] = guess_best_decode(response.content, candidates)
        result['success'] = True

    except Exception as e:
        result['error'] = str(e)

    return result

---

Trafilatura - Requests 조합의 성능을 평가해 보자

성능 평가를 위해, 앞의 글에서 사용한 500개 URL이 담긴 파일을 그대로 사용했다. 성능 평가를 위한 최종 코드를 아래와 같이 작성하였다. 

 

📄 URL Batch Processor (Trafilatura + requests) 🔽 펼치기

import pandas as pd
import trafilatura
from trafilatura.metadata import extract_metadata
import requests
import time
from datetime import datetime
import csv
from typing import List, Optional


def guess_best_decode(data: bytes, encodings: List[str]) -> str:
    """Pick the decoding that yields the most Hangul characters."""
    best_text: Optional[str] = None
    best_score = -1
    for enc in encodings:
        if not enc:
            continue
        try:
            text = data.decode(enc, errors="replace")
        except LookupError:
            continue
        # Score by Hangul character count
        score = sum(0xAC00 <= ord(ch) <= 0xD7A3 for ch in text)
        if score > best_score:
            best_text, best_score = text, score
        # Early exit if score is very high (heuristic)
        if score > 10:
            break
    if best_text is None:
        best_text = data.decode("utf-8", errors="replace")
    return best_text

def fetch_html_with_requests(url: str, timeout: int = 10) -> dict:
    """
    requests로 HTML만 다운로드하는 함수 (본문 추출은 별도로 수행)
    guess_best_decode를 활용한 강력한 인코딩 자동 보정
    """
    result = {
        'url': url,
        'html': '',
        'success': False,
        'error': None,
        'method_used': 'requests_only'
    }

    try:
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36'
        }
        response = requests.get(url, headers=headers, timeout=timeout)
        response.raise_for_status()

        # ✅ guess_best_decode를 통해 인코딩 복원
        candidates = [
            response.encoding,
            response.apparent_encoding,
            'utf-8',
            'euc-kr',
            'cp949'
        ]
        result['html'] = guess_best_decode(response.content, candidates)
        result['success'] = True

    except Exception as e:
        result['error'] = str(e)

    return result

def process_single_url(url, index, total_urls):
    """단일 URL 처리 함수"""
    start_time = time.time()
    result = {
        "index": index,
        "url": url,
        "status": "failed",
        "title": None,
        "date": None,
        "content": None,
        "processing_time": 0,
        "error_message": None,
    }
    
    try:
        print(f"Processing URL {index + 1}/{total_urls}: {url[:60]}...")
        
        # HTML 다운로드
        html_result = fetch_html_with_requests(url)
        
        if html_result['success']:
            html_content = html_result['html']
            
            # trafilatura로 메타데이터 추출
            metadata = trafilatura.extract_metadata(html_content)
            
            # trafilatura로 본문 추출
            text = trafilatura.extract(
                html_content,
                output_format="txt",
                include_comments=False,
                favor_precision=True,
            )
            
            if text and len(text.strip()) > 0:
                result.update({
                    "status": "success",
                    "title": metadata.title if metadata and metadata.title else "No title",
                    "date": str(metadata.date) if metadata and metadata.date else "No date",
                    "content": text[:200] + "..." if len(text) > 200 else text,
                })
            else:
                result["error_message"] = "Empty content extracted"
        else:
            result["error_message"] = f"Failed to download page: {html_result['error']}"
            
    except Exception as e:
        result["error_message"] = str(e)
    
    # 처리 시간 계산
    processing_time = time.time() - start_time
    result["processing_time"] = processing_time
    
    return result


def save_results_to_csv(results, output_file=None):
    """결과를 CSV 파일로 저장"""

    # 파일명이 지정되지 않으면 현재 시간을 포함한 파일명 생성
    if output_file is None:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = f"url_processing_results_{timestamp}.csv"

    try:
        with open(output_file, "w", newline="", encoding="utf-8") as csvfile:
            fieldnames = [
                "index",
                "url",
                "status",
                "title",
                "date",
                "content",
                "processing_time",
                "error_message",
            ]
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

            writer.writeheader()
            for result in results:
                writer.writerow(result)

        print(f"✅ 결과가 '{output_file}' 파일에 저장되었습니다.")
        return output_file  # 실제 저장된 파일명 반환

    except Exception as e:
        print(f"❌ 결과 저장 실패: {e}")
        return None


def print_summary(results, total_time, skipped_count=0):
    """처리 결과 요약 출력"""
    total_urls = len(results)
    successful_count = sum(1 for r in results if r["status"] == "success")
    failed_count = total_urls - successful_count
    processing_times = [r["processing_time"] for r in results]

    print("\n" + "=" * 7 + " SUMMARY " + "=" * 7)
    print(f"Total URLs processed: {total_urls}")
    print("Workers used: 1")  # 단일 스레드 처리
    print(f"Successfully decoded: {successful_count} ({successful_count/total_urls*100:.1f}%)")
    print(f"Failed to decode: {failed_count} ({failed_count/total_urls*100:.1f}%)")
    print(f"Skipped (Google News URLs): {skipped_count} ({skipped_count/(total_urls + skipped_count)*100:.1f}%)")

    if processing_times:
        avg_time = sum(processing_times) / len(processing_times)

        print("\n" + "-" * 5 + " TIMING INFORMATION " + "-" * 5)
        print(f"Total processing time: {int(total_time//60)}:{total_time%60:05.2f}")
        print(f"Average processing time per URL: {avg_time:.2f} seconds")
        print(f"Fastest URL processing time: {min(processing_times):.2f} seconds")
        print(f"Slowest URL processing time: {max(processing_times):.2f} seconds")

    print("\nProcess completed successfully. Results saved to CSV file.")


def process_urls_from_csv(csv_file_path, url_column="decoded_url"):
    """CSV 파일에서 URL들을 읽어서 순차 처리"""

    print("=" * 50)
    print("🚀 URL 배치 처리 시작")
    print("=" * 50)

    # CSV 파일 읽기
    try:
        df = pd.read_csv(csv_file_path)
        print(f"📋 CSV 파일 컬럼들: {list(df.columns)}")
        print(f"📋 총 행 수: {len(df)}")

        if url_column not in df.columns:
            raise ValueError(f"Column '{url_column}' not found in CSV file")

        print(f"📋 '{url_column}' 컬럼의 NULL이 아닌 값 개수: {df[url_column].notna().sum()}")

        all_urls = df[url_column].dropna().tolist()
        print(f"📋 첫 번째 URL 샘플: {all_urls[0] if all_urls else 'None'}")

        # news.google.com이 포함되지 않은 URL만 필터링
        urls = [url for url in all_urls if "news.google.com" not in str(url)]

        total_urls = len(urls)
        skipped_urls = len(all_urls) - total_urls

        print(f"📊 전체 URL: {len(all_urls)}개")
        print(f"📊 처리 대상 URL (decoded URLs): {total_urls}개")
        print(f"📊 건너뛴 URL (Google News URLs): {skipped_urls}개")

        if urls:
            print(f"📋 첫 번째 디코딩된 URL 샘플: {urls[0]}")

        print("-" * 30)

    except Exception as e:
        print(f"❌ CSV 파일 읽기 실패: {e}")
        return None

    total_start_time = time.time()
    results = []
    successful_count = 0
    failed_count = 0

    for i, url in enumerate(urls):
        result = process_single_url(url, i, total_urls)
        results.append(result)

        if result["status"] == "success":
            successful_count += 1
        else:
            failed_count += 1

        if (i + 1) % 10 == 0 or (i + 1) == total_urls:
            print(
                f"진행: {i + 1}/{total_urls} "
                f"({(i + 1)/total_urls*100:.1f}%) "
                f"성공: {successful_count}, 실패: {failed_count}"
            )

    total_processing_time = time.time() - total_start_time
    print_summary(results, total_processing_time, skipped_urls)

    saved_file = save_results_to_csv(results)
    return results, saved_file


def main():
    csv_file_path = r"C:\Users\yhsur\Downloads\특징주\sample_data\Combined_sample_data_500_decoded_2025-05-20_224740.csv"
    results, saved_file = process_urls_from_csv(csv_file_path, url_column="decoded_url")

    if results:
        print(f"\n📁 저장된 파일: {saved_file}")

        print("\n🔍 처리 결과 샘플:")
        for i, result in enumerate(results[:3]):
            print(f"\n[{i+1}] {result['url'][:50]}...")
            print(f"    상태: {result['status']}")
            print(f"    제목: {result['title']}")
            print(f"    처리시간: {result['processing_time']:.2f}초")
            if result["status"] == "failed":
                print(f"    오류: {result['error_message']}")


if __name__ == "__main__":
    main()

🖨️ 실행결과

진행: 500/500 (100.0%) 성공: 407, 실패: 93

======= SUMMARY =======
Total URLs processed: 500
Workers used: 1
Successfully decoded: 407 (81.4%)
Failed to decode: 93 (18.6%)
Skipped (Google News URLs): 0 (0.0%)

----- TIMING INFORMATION -----
Total processing time: 5:05.95
Average processing time per URL: 0.61 seconds
Fastest URL processing time: 0.06 seconds
Slowest URL processing time: 3.96 seconds

Process completed successfully. Results saved to CSV file.
✅ 결과가 'url_processing_results_20250531_004542.csv' 파일에 저장되었습니다.

📁 저장된 파일: url_processing_results_20250531_004542.csv

---

📊 성능 평가 및 결과

항목	trafilatura	requests + trafilatura
문자 인코딩 처리	제한적 (간헐적 오류)	완전 제어 가능
처리 속도	빠름 (예: 0.52초)	약간 느림 (예: 0.61초)
설정 유연성	낮음	높음
디버깅/오류 추적	어려움	용이

 

---

마무리

이번 전략은 뉴스 수집의 일관성과 품질을 크게 향상시킨다.일관성이 중요한 이유는, 추후 문제가 발견되었을 때 수정과 재수집에 소요되는 비용을 줄일 수 있기 때문이다. 수집 대상 URL의 80% 이상이 정적 HTML이라면, requests와 trafilatura 조합이 지금까지 시도한 방법 중 가장 실용적이다.

다음 글에서는 이 전략을 더욱 발전시켜, 병렬 처리를 적용하여 처리 속도를 더욱 단축하는 방법을 시도하겠다.