배경: 콘텐츠 수집의 비효율성
숏폼 콘텐츠가 폭발적으로 증가하는 현재, 특정 플랫폼의 영상을 체계적으로 보관하고 활용하는 것은 개인 사용자부터 연구자까지 폭넓은 니즈를 가진 과제입니다. 수동으로 영상을 저장하는 방식은 시간 소모가 크고, 메타데이터 손실, 중복 저장, 파일 관리 혼란 등의 문제를 야기합니다.
이 글에서는 Python 기반 오픈소스 도구를 활용하여 Douyin 영상을 자동으로 수집하고, 메타데이터를 보존하며, 계적인 폴더 구조로 관리하는 방법을 다룹니다.
핵심 요구사항 분석
효과적인 영상 수집 시스템은 다음 기능을 충족해야 합니다:
- 자동화된 일괄 수집: 개별 영상뿐 아니라 특정 계정의 전체 콘텐츠를 자동으로 다운로드
- 메타데이터 보존: 작성자, 게시일, 조회수, 좋아요 수 등 구조화된 정보 저장
- 중복 제거: SQLite 기반 데이터베이스로 이미 수집한 콘텐츠 필터링
- 유연한 출력 형식: 영상, 표지 이미지, 오디오 트랙 분리 저장
- 재시도 및 예외 처리: 네트워크 불안정 상황에서의 안정적 동작
시스템 설치 및 초기 설정
환경 구성
Python 3.8 이상이 설치된 환경에서 다음 명령을 실행합니다:
# 저장소 복제
git clone https://gitcode.com/GitHub_Trending/do/douyin-downloader.git
cd douyin-downloader
# 의존성 패키지 설치
pip install -r requirements.txt인증 정보 획득
Douyin의 API는 인증된 세션이 필요합니다. 다음 스크립트로 브라우저 쿠키를 자동 추출합니다:
python cookie_extractor.py또는 수동으로 config_downloader.yml 파일을 편집할 수 있습니다:
session:
msToken: "추출된_토큰_값"
ttwid: "추출된_식별자"
odin_tt: "추출된_세션_값"기본 사용법
개별 영상 수집
공유 링크를 기반으로 단일 영상을 다운로드합니다:
python DouYinCommand.py --url "https://v.douyin.com/xxxxx/"계정 전체 콘텐츠 수집
특정 사용자의 모든 게시물을 일괄 다운로드합니다:
python downloader.py --user "https://www.douyin.com/user/MS4wLjABAAAAxxxxx"라이브 스트림 녹화
실시간 방송 콘텐츠를 저장합니다:
python DouYinCommand.py --live "https://live.douyin.com/12345678" --quality 1080p고급 설정: 설정 파일 최적화
config_downloader.yml 파일을 통해 세밀한 동작 제어가 가능합니다:
# 병렬 처리 설정
concurrency:
workers: 12 # 동시 다운로드 스레드 수
bandwidth_limit: 0 # 0은 무제한, 단위: KB/s
# 출력 형식 제어
output:
video_format: true # 영상 파일 저장
thumbnail: true # 표지 이지 저장
audio_track: true # 오디오 분리 저장
metadata: true # JSON 메타데이터 생성
# 필터 조건
filter:
date_range:
start: "2024-01-01"
end: "2024-12-31"
content_types: ["post"] # post: 게시물, like: 좋아요 표시한 콘텐츠
# 저장 경로 템플릿
storage:
base_path: "./archives/{creator}/{year}-{month}/"
naming_pattern: "{date}_{title}_{id}"프로그래밍 방식 통합
사용자 정의 스크립트에서 모듈을 직접 임포트하여 활용할 수 있습니다:
from apiproxy.douyin.douyin import DouyinCollector
from datetime import datetime, timedelta
# 수집기 인스턴스 생성
collector = DouyinCollector(config="config_downloader.yml")
# 특정 기간의 콘텐츠만 필터링
start_date = datetime.now() - timedelta(days=30)
# 사용자 콘텐츠 조회 및 다운로드
content_list = collector.fetch_user_content(
user_id="target_account_id",
max_items=500,
since=start_date
)
for item in content_list:
# 중복 확인
if not collector.is_duplicate(item.content_id):
result = collector.download(item)
print(f"저장 완료: {result.local_path}")
# 메타데이터 추출 예시
stats = {
'views': item.statistics.play_count,
'likes': item.statistics.digg_count,
'shares': item.statistics.share_count,
'comments': item.statistics.comment_count
}
collector.save_metadata(item.content_id, stats)데이터베이스 스키마 및 중복 관리
내부적으로 SQLite를 사용하여 수집 이력을 관리합니다:
-- 자동 생성되는 테이블 구조
CREATE TABLE downloaded_content (
content_id TEXT PRIMARY KEY,
creator_id TEXT NOT NULL,
original_url TEXT UNIQUE,
local_path TEXT,
download_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
file_size_bytes INTEGER,
checksum TEXT,
metadata_json TEXT
);
-- 중복 검색 인덱스
CREATE INDEX idx_creator_time ON downloaded_content(creator_id, download_timestamp);이 구조를 통해 동일 콘텐츠의 재다운로드를 방지하고, 수집된 자료의 체계적인 추적이 가능합니다.
실전 활용 시나리오
시나리오 A: 콘텐츠 크리에이터의 레퍼런스 아카이브
주간 단위로 특정 분야의 우수 사례를 자동 수집하는 워크플로우:
#!/usr/bin/env python3
import schedule
import time
from reference_archiver import ReferenceCollector
def weekly_collection_job():
targets = [
"https://www.douyin.com/user/edu_creator_01",
"https://www.douyin.com/user/tech_explainer_02",
"https://www.douyin.com/user/design_inspiration_03"
]
archiver = ReferenceCollector(output_base="./weekly_refs/")
for profile_url in targets:
archiver.collect_new_only(profile_url, max_per_user=20)
# 자동 리포트 생성
archiver.generate_summary_report()
# 매주 월요일 새벽 2시 실행
schedule.every().monday.at("02:00").do(weekly_collection_job)
while True:
schedule.run_pending()
time.sleep(60)시나리오 B: 학술 연구용 대규모 데이터셋 구축
정량적 분석을 위한 구조화된 데이터 수집:
from research_dataset_builder import DatasetBuilder
builder = DatasetBuilder(
storage_path="./research_data/cultural_trends_2024/",
required_fields=['desc', 'create_time', 'region', 'music_info']
)
# 다중 계정 병렬 수집
source_accounts = [...] # 50개 연구 대상 계정
for batch in chunked(source_accounts, 5): # 5개씩 배치 처리
builder.ingest_accounts(batch, delay_between=30) # 계정 간 30초 간격
# 진행 상황 자동 저장
builder.checkpoint()문제 해결 및 안정성 강화
속도 최적화
네트워크 환경에 따른 조정 가이드:
| 환경 | workers | delay_ms | 예상 처리량 |
|---|---|---|---|
| 고속 유선 | 15 | 100 | ~120개/분 |
| 일반 와이파이 | 8 | 200 | ~50개/분 |
| 불안정 연결 | 4 | 500 | ~20개/분 |
일반적인 오류 대응
증명 오류 (401/403):
- 쿠키 만료 여부 확인 및 재추출
- 요청 헤더의 User-Agent 최신화
속도 제한 (429):
- workers 값 절반으로 감소
- 요청 간 지연 시간 2배 증가
- 시간대 변경하여 재시도
파일 손상:
- 체크섬 검증 활성화:
verify_checksum: true - ffmpeg 재인코딩 옵션 적용
확장 및 커스터마이징
핵심 모듈을 상속하여 사용자 정의 동작을 구현할 수 있습니다:
from apiproxy.douyin.download import BaseDownloader
from pathlib import Path
class CustomArchiver(BaseDownloader):
def __init__(self, config):
super().__init__(config)
self.processed_tags = set()
def post_download_hook(self, local_path: Path, metadata: dict):
"""다운로드 후 자동 태깅"""
# AI 기반 콘텐츠 분석 연동
tags = self.analyze_content(local_path)
# 태그 기반 서브폴더 이동
for tag in tags:
tag_dir = local_path.parent / f"tagged_{tag}"
tag_dir.mkdir(exist_ok=True)
# 하드링크 생성 (중복 저장 방지)
(tag_dir / local_path.name).hardlink_to(local_path)
return {"tags": tags, "original_path": str(local_path)}
def analyze_content(self, video_path: Path) -> list:
# 외부 분석 서비스 연동 위치
# 예: 장면 인식, 오디오 전사, 객체 탐지 등
return ["placeholder_tag"]아키텍처 개요
도구의 내부 구조는 다음 레이어로 구성됩니다:
douyin-downloader/ ├── apiproxy/ │ └── douyin/ │ ├── douyinapi.py # REST API 래퍼 │ ├── download.py # 코어 다운로드 엔진 │ ├── database.py # SQLite 관리 │ └── core/ │ ├── queue_manager.py # 작업 큐 조정 │ ├── retry_handler.py # 재시도 로직 │ └── rate_limiter.py # 속도 제어 ├── cookie_extractor.py # 인증 정보 획득 └── DouYinCommand.py # CLI 진입점
이 모듈형 설계는 특정 컴포넌트만 선택적으로 재사용하거나, 전체 파이프라인을 다른 프로젝트에 임베드하는 것을 용이하게 합니다.