공개 데이터 크롤러 파이프라인 설계 — BaseCrawler로 5개국 출처를 하나로 통합하기 (1)

BaseCrawler 추상 클래스로 5개 공개 데이터 출처(벌크 API·주간 XML·웹페이지 PDF)를 하나의 수집-파싱-저장 파이프라인에 통합한 설계. fetch/parse 두 메서드만 출처별로 구현하고, Pydantic 공통 모델의 원문/번역 분리, COALESCE UPSERT 중복 제거, 2단계 보강, 건별·출처별 격리, 수집과 번역 분리까지 다룬다.

시리즈를 시작하며 — "출처가 5개인데 화면은 하나여야 한다"

여러 나라가 각자 공개하는 제품 리콜 데이터를 한곳에 모아, 자동으로 번역하고, 검색·내보내기까지 되는 데스크톱 앱을 만들었다. 한 줄로 줄이면 "공개 데이터 크롤러"지만, 실제로 시간을 잡아먹은 건 크롤링 그 자체가 아니었다. 나라마다 데이터를 주는 방식이 전부 다르다는 점이 모든 설계 결정을 지배했다.

이 시리즈는 그 싸움의 기록이다. 네 편으로 나눠 쓴다.

1. (이 글) BaseCrawler 파이프라인 — 5개 출처를 하나의 추상 구조로 통합
2. 외국어 문서(이미지 PDF 포함)를 표 구조로 파싱하기
3. 여러 PC에서 재현 안 되는 버그 잡기 — 시계 오차와 동시성
4. 오번역 차단 · 다필드 검색 · 스케줄러 · 데스크톱 앱화

의뢰 맥락이나 특정 기관을 식별할 수 있는 정보는 전부 빼고, 기술적인 부분만 다룬다. 출처는 "북미 두 곳, 동아시아 한 곳, 오세아니아 한 곳, 유럽 통합 창구 한 곳" 정도로만 일반화해 부른다. 코드 속 국가 코드는 A~E처럼 익명화했다.

문제: 같은 "리콜"인데 모양이 다섯 가지

수집 대상이 5개 출처인데, 데이터 제공 방식이 제각각이었다. 대략 이런 식이다.

• 북미 한 곳은 오픈데이터 API로 수만 건을 한 번에 벌크로 내려준다. 그런데 핵심 필드(제조사·모델·연식)는 거기 없고, 별도의 캠페인 상세 API를 건당 다시 호출해야 채워진다.
• 유럽 통합 창구는 주간 단위로 묶은 XML 보고서를 통째로 던진다. 한 보고서 안에 온갖 제품 카테고리가 섞여 있어, 그중 차량 관련만 골라내야 한다. 날짜는 DD/MM/YYYY 형식이다.
• 동아시아 한 곳은 웹페이지에 보도자료 링크가 깔리고, 진짜 데이터는 그 안의 PDF에 들어 있다. 심지어 일부는 텍스트가 아니라 스캔 이미지 PDF다(이건 2편 주제다).
• 나머지 두 곳도 각각 또 다른 API·테이블 구조를 가진다.

여기서 가장 위험한 유혹은 "나라별로 크롤러를 따로따로 짜자"이다. 처음엔 그게 빨라 보인다. 하지만 그렇게 가면 번역·저장·로그·에러 처리·재시도·날짜 정규화 같은 공통 로직이 다섯 군데로 복제된다. 한 곳에서 버그를 잡아도 나머지 네 곳은 그대로다. 출처가 6개, 7개로 늘면 손댈 수 없는 코드가 된다.

그래서 첫 줄을 긋기 전에 정한 원칙이 하나 있다. "나라마다 다른 건 딱 두 가지뿐이게 만든다 — 어디서 가져오는가(collect), 그 모양을 어떻게 공통 모델로 바꾸는가(parse). 나머지는 전부 한 군데."

해법: 파이프라인을 추상 클래스로 고정

가장 먼저 한 일이 수집 → 파싱 → (날짜 거르기) → 저장 → 로그로 이어지는 파이프라인 전체를 BaseCrawler 추상 클래스의 run() 한 메서드에 박아 넣은 것이다. 나라별 크롤러는 이 run()을 건드리지 않는다. 추상 메서드 fetch()와 parse()만 자기 사정에 맞게 구현한다.

이 한 함수가 시리즈 전체의 척추다. 여기서 눈여겨볼 점이 몇 개 있다.

• 수집(fetch)과 변환(parse)만 추상이고, 저장·로그·날짜 거르기·정지 처리·집계는 베이스가 전부 책임진다.
• 루프 안에서 parse()를 try로 감쌌다. 원본 한 건이 깨져도(빈 필드, 형식 이탈) 로그만 남기고 다음 건으로 넘어간다. 5,000건 중 3건이 깨졌다고 수집 전체가 멈추면 안 된다.
• 신규 건수(new)가 있으면 상태를 translating으로 둔다. 번역은 이 run() 안에서 하지 않는다. 수집과 번역을 분리한 건 의도적인 결정인데, 뒤에서 따로 설명한다.
• "fetch 몇 건 → 저장 몇 건 → 날짜범위 밖 제외 몇 건 → 파싱불가 몇 건"을 한 줄로 남긴다. "수집은 됐는데 화면에 안 보인다"는 신고가 들어왔을 때, 데이터가 어느 단계에서 탈락했는지 로그만 보면 즉시 안다. 이 분리 로깅 한 줄이 나중에 디버깅 시간을 크게 줄였다.

나라별 크롤러는 그래서 놀랄 만큼 얇다.

새 출처를 추가할 때 답해야 할 질문이 정확히 두 개로 좁혀진다. "이 출처는 데이터를 어떻게 주는가(fetch)"와 "그 모양을 공통 키로 어떻게 바꾸는가(parse)". 번역도, 저장도, 재시도도, 로그도 다시 짤 일이 없다. 출처가 5개에서 6개가 되어도 추가 비용은 이 클래스 하나뿐이다.

공통 모델: 파싱 단계에서 스키마 위반을 조기에 차단

다섯 출처가 토해내는 제각각의 원본을 하나의 화면에서 다루려면, 모든 크롤러가 같은 모양의 딕셔너리를 내놓아야 한다. 이 "공통 모델"을 Pydantic으로 못 박았다. 제조사·모델·연식·결함요약·결과·시정조치·영향대수 등, 리콜이라면 나라를 불문하고 공통으로 가지는 의미 단위를 필드로 정의한다.

설계에서 두 가지를 의식적으로 정했다.

첫째, 원문과 번역을 _original / _ko 쌍으로 둔다. 제조사명을 manufacturer_original과 manufacturer_ko 두 칸으로 나눠 저장한다. 번역은 깨질 수 있고(특히 고유명사), 오번역이 의심되면 사용자가 원문을 봐야 한다. 한 칸에 번역을 덮어쓰면 원문이 영영 사라진다. 그래서 원문은 불변으로 두고, 번역은 그 옆에 채운다. 이 구조가 4편의 "오번역 차단"으로 이어진다.

둘째, recall_date만 필수(Optional 아님)다. 다른 필드는 출처에 따라 비어 있을 수 있다. 하지만 날짜가 없으면 시계열 정렬도, 증분 수집(마지막 수집일 이후만 가져오기)도 불가능하다. 그래서 날짜만은 어떻게든 채우게 강제하고, 정 안 되면 파싱 단계에서 1970-01-01 같은 안전한 sentinel로 떨어뜨린다(이 sentinel을 어떻게 다루는지가 3편의 주제 중 하나다).

출처마다 다른 "원본 한 건"을 같은 모양으로

추상 구조의 진가는 실제로 모양이 다른 두 출처를 나란히 놓고 볼 때 드러난다. 둘 다 결국 같은 키의 딕셔너리를 반환하지만, 거기까지 가는 길은 완전히 다르다.

벌크 API 출처는 평평한 JSON을 받아 키만 매핑하면 되지만, 그 과정에 도메인 지식이 박힌 작은 정리들이 들어간다. 예를 들어 이 출처는 부품·악세서리 리콜의 연식을 9999라는 sentinel로 응답한다. 이걸 그대로 두면 화면에 "9999년식"이 떠 버린다.

def parse(self, raw: dict) -> dict:
    _year = (raw.get("_detail") or {}).get("year")
    # "연식 미지정" sentinel(9999/0) 정리 — 사용자에게 9999가 보이지 않도록 빈값 처리
    if _year is not None and str(_year).strip() in {"9999", "0", ""}:
        _year = None
    return {
        "country": self.country,
        "recall_number": str(raw.get("id", "")),
        "manufacturer_original": raw.get("manufacturer"),
        "model_year": _year,
        "summary_original": raw.get("subject"),
        "defect_cause_original": raw.get("defect_summary"),
        "consequence_original": raw.get("consequence_summary"),
        "remedy_original": raw.get("corrective_action"),
        "recall_date": self._parse_date(str(raw.get("date", ""))),
    }

반면 XML 보고서 출처는 한 보고서 안에 수십 개 카테고리가 섞여 있어, parse 이전에 "차량 관련만 골라내는" 필터가 필요하다. 단순히 문자열에 car가 들어갔는지로 거르면 carrier, childcare 같은 단어에 오매칭된다. 그래서 단어 경계를 살린 정규식으로 카테고리를 판별한다.

# 'car' 단독 단어만 — carrier/childcare 오매칭 방지
_VEHICLE_CAT_RE = re.compile(
    r'motor|vehicle|autom|\bcar\b|\bcars\b|truck|tyre|tire'
    r'|motorcycle|\blorry\b|\bbus\b|\bvan\b|\btrailer\b',
    re.IGNORECASE,
)

for notif in root.findall("notifications"):
    category = notif.findtext("category", "")
    if not _VEHICLE_CAT_RE.search(category):
        continue                       # 차량 외 카테고리는 건너뜀
    # ... 차량 알림만 공통 키로 변환

핵심은, 이런 출처별 특수 처리가 전부 fetch/parse 안에 갇힌다는 점이다. 베이스의 run()은 "원본 리스트를 받아 한 건씩 parse해서 저장한다"만 알 뿐, 어느 출처가 XML인지 PDF인지 전혀 모른다. 출처의 복잡함이 베이스로 새어 나오지 않게 막은 것 — 이게 추상화가 실제로 일하는 지점이다.

보강(backfill): 한 출처가 두 번에 나눠 데이터를 줄 때

가장 까다로운 출처는 "데이터를 한 번에 다 주지 않는" 경우였다. 벌크 API는 수만 건의 목록을 빠르게 주지만, 정작 화면에 꼭 필요한 제조사·모델·연식은 그 목록에 없다. 그 값들은 건별 상세 API를 다시 호출해야 나온다. 수만 건을 전부 상세 호출하면 수집이 끝나지 않는다.

그래서 2단계로 나눴다. 먼저 벌크로 목록을 통째로 저장하고(빠름), 그다음 제조사가 비어 있는 레코드만 골라 상세 API로 채운다. 이 보강은 베이스의 run()이 끝난 뒤, 해당 크롤러가 run()을 오버라이드해 한 번 더 호출하는 식으로 끼워 넣었다. 그리고 상세 호출은 건별로 독립적이라 병렬화가 자연스럽다.

UPDATE에 COALESCE를 쓴 게 중요하다. 상세 API가 일부 필드만 채워 줄 때, 이미 들어 있던 값을 NULL로 덮어쓰지 않게 한다. "있으면 유지, 없으면 새 값"이라는 보강의 의미를 SQL 한 줄로 강제했다. 보강 대상을 최신 500건으로 제한한 것도 의도적이다 — 사용자가 가장 먼저 보는 건 최신 리콜이니, 거기부터 완전하게 채운다.

저장은 UPSERT로: 같은 리콜을 두 번 봐도 한 줄

증분 수집을 돌리면 어제 본 리콜을 오늘 또 보게 된다. 같은 리콜이 두 줄로 쌓이면 검색 결과가 더러워진다. 그래서 저장은 단순 INSERT가 아니라 (country, recall_number) 기준 UPSERT다. 같은 키가 이미 있으면 새로 들어온 값으로 갱신하되, 여기서도 COALESCE로 빈 값이 기존 값을 지우지 못하게 막는다.

이 UPSERT 한 덩어리가 여러 문제를 동시에 푼다. 중복이 안 생기고, 출처가 나중에 보강해 준 필드(예: 늦게 채워지는 모델명)가 자연스럽게 합쳐지며, 한 번 채워진 값이 다음 수집의 빈 응답에 지워지지 않는다. UPSERT가 동작하려면 (country, recall_number)에 UNIQUE 제약이 반드시 있어야 하는데, 구버전 DB에는 이 제약이 없을 수 있어 초기화 시 인덱스를 보장하는 마이그레이션을 따로 넣었다(3편에서 다룬다).

끝의 for attempt in range(5) 재시도는 SQLite 특유의 사정이다. 여러 출처를 거의 동시에 돌리면 쓰기가 겹쳐 database is locked가 난다. WAL 모드와 busy_timeout으로 대부분 흡수하지만, 그래도 잠기면 점증 대기로 다섯 번까지 재시도한다.

한 건 실패가 전체를 멈추지 않는다 — 격리의 두 층위

run()의 가장 중요한 불변식은 "한 건의 실패가 수집 전체를 멈추지 않는다"이다. 크롤링은 원본 한 건이 깨지는 일이 일상이다. 필드가 비고, 날짜 형식이 어긋나고, 표가 한 칸 밀린다. 이걸 격리하지 않으면 5,000건 중 단 한 건의 예외가 나머지 4,999건을 통째로 날린다.

격리를 두 층위로 깔았다. 건별 격리는 위에서 본 대로 parse()를 try로 감싼 것이다. 그 위에 출처별 격리가 있다. 5개 출처를 한꺼번에 "전체 수집"할 때, 한 출처의 크롤러가 통째로 터져도 나머지 출처는 계속 돈다. 각 출처 실행을 독립 작업으로 띄우고, 실패는 그 출처의 로그에만 failed로 남긴다.

여기서 한 가지 균형을 잡아야 했다. 출처들은 논리적으로는 병렬(서로 독립)이지만, 단일 PC의 SQLite에 동시에 쓰면 잠김이 잦다. 그래서 수집 실행 자체는 런-락(_crawl_run_lock)으로 직렬화하되, 대기 중인 출처를 "대기" 상태로 UI에 보여 줘 멈춘 것처럼 보이지 않게 했다. "논리적 병렬 + 물리적 직렬"이라는 절충이다. 출처가 5개뿐이고 하루 한 번 도는 작업이라, 직렬로 줄 세워도 전체 수집 시간이 문제 되지 않았다.

수집과 번역을 떼어 놓은 이유

마지막으로, 처음 봤을 때 의아할 수 있는 결정 하나. 수집과 번역을 한 파이프라인 안에서 같이 하지 않는다. run()은 원문을 저장하고 끝나고(상태 translating), 번역은 별도의 백그라운드 스레드가 미번역 레코드를 배치로 집어 처리한다.

이렇게 나눈 이유는 세 가지다.

첫째, 수집은 빨라야 한다. 번역은 외부 호출이 끼어 느리고 실패도 잦다. 수집 루프 안에서 매 건 번역을 기다리면, 한 건의 번역 지연이 전체 수집을 늘어뜨린다. 수집은 원문만 빠르게 떨궈 끝내고, 번역은 뒤에서 따라잡게 했다.

둘째, 스케줄러를 막지 않기 위해서다. 매일 정해진 시각에 자동 수집이 도는데, 수집 작업이 번역까지 붙들고 몇 시간씩 늘어지면 다음 날 스케줄이 "아직 어제 것이 안 끝났다"며 건너뛰어진다. 수집 스레드가 즉시 반환되어야 스케줄이 매일 정확히 발동한다.

셋째, 번역은 재개 가능해야 한다. translated_at IS NULL이라는 조건 하나가 "아직 번역 안 된 것"을 정의한다. 번역 도중 앱이 꺼져도, 다시 켜면 그 조건으로 남은 것만 이어서 번역한다. 성공이든 실패든 translated_at을 찍어 두기 때문에 같은 레코드를 무한히 다시 잡는 일이 없다. 상태를 별도 큐가 아니라 DB 컬럼 하나로 표현한 덕에, 중단·재개가 공짜로 된다.

정리하며

1편의 핵심은 다섯 출처의 다양성을 추상 클래스 하나로 흡수한 것이다.

• 파이프라인(수집→파싱→저장→번역→로그)을 BaseCrawler.run()에 고정 → 출처별 차이를 fetch/parse 두 메서드로 국한했다. 새 출처는 얇은 클래스 하나면 끝.
• 공통 모델을 Pydantic으로 정의하고 원문/번역을 _original·_ko 쌍으로 분리 → 원문 보존과 오번역 대응의 토대를 깔았다. 날짜만 필수로 강제해 시계열·증분 수집을 보장.
• 출처별 특수 처리(sentinel 정리, XML 카테고리 필터, 2단계 보강)는 전부 parse/fetch 안에 가뒀다 → 복잡함이 베이스로 새지 않게 했다.
• 저장은 (country, recall_number) UPSERT + COALESCE로 중복·덮어쓰기·증분 보강을 한 번에 해결.
• 격리를 두 층(건별 try, 출처별 작업 분리)으로 깔아 "일부 실패가 전체를 멈추지 않게" 했고, 수집과 번역을 분리해 수집 속도·스케줄 정확성·번역 재개 가능성을 동시에 얻었다.

출처가 늘어도 베이스 위에 크롤러 하나만 더 얹으면 되는 구조라, 확장 부담이 작다. 다만 이 깔끔한 그림은 출처들이 텍스트 데이터를 순순히 줄 때 이야기다. 다음 2편에서는 가장 험했던 구간 — 외국어로 적힌, 그것도 일부는 스캔 이미지인 PDF에서 표 구조를 뜯어내는 일을 다룬다.