bca8032b79
쿠팡이 봇으로 감지(Akamai 차단 페이지)하면, 같은 IP로 재시도하던 것을 '브라우저 끄고 새 IP로 켜서 재시도'로 전환. 감지 패턴도 축적한다. - DecodoProxy.rotate(): 시간창 무관 즉시 다음 포트(=새 IP). current_port 노출 - coupang/adapter: 0건+차단마커 감지 시 → 감지기록 → proxy.rotate()+강제재기동 → 인라인 재시도(max_block_retries=1). IP당 요청수(ip_request_no) 추적, 재기동 시 리셋 - bot_detection 테이블 + crud: source/query/ip_request_no/proxy_port/elapsed_sec/marker/ headless/html_len 기록 → 'IP당 몇 요청 만에 감지되나' 분석 가능 - worker_main: on_detect=BotDetectionLog.record 주입 - 로그: [coupang][BOT-DETECTED] ip_req#N port=... elapsed=...s marker=... - tests: proxy rotate 즉시회전 + 감지기록 CRUD → 전체 47/47 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
35 lines
1.2 KiB
Python
35 lines
1.2 KiB
Python
"""봇 감지 이력 기록 CRUD 테스트 (실 lps_db)."""
|
|
|
|
import pytest_asyncio
|
|
from sqlalchemy import text
|
|
|
|
from crud.bot_detection import BotDetectionLog
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def bd(db_engine):
|
|
async with db_engine.begin() as conn:
|
|
await conn.execute(text("TRUNCATE bot_detection"))
|
|
return BotDetectionLog()
|
|
|
|
|
|
async def test_record_persists_event(bd, db_engine):
|
|
await bd.record({
|
|
"source": "coupang", "query": "맥심 커피", "ip_request_no": 7,
|
|
"proxy_port": 10003, "elapsed_sec": 42, "marker": "/akam/",
|
|
"headless": False, "html_len": 2604,
|
|
})
|
|
async with db_engine.begin() as conn:
|
|
row = (await conn.execute(text(
|
|
"SELECT source, query, ip_request_no, proxy_port, marker FROM bot_detection"
|
|
))).first()
|
|
assert row.source == "coupang" and row.query == "맥심 커피"
|
|
assert row.ip_request_no == 7 and row.proxy_port == 10003 and row.marker == "/akam/"
|
|
|
|
|
|
async def test_record_tolerates_missing_fields(bd, db_engine):
|
|
await bd.record({"source": "coupang"}) # 나머지는 None 허용
|
|
async with db_engine.begin() as conn:
|
|
cnt = (await conn.execute(text("SELECT count(*) FROM bot_detection"))).scalar()
|
|
assert cnt == 1
|