88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
import os
|
|
import asyncio
|
|
import logging
|
|
from http import HTTPMethod
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
REQUEST_TIMEOUT = 60
|
|
|
|
|
|
|
|
def get_env(key: str) -> str:
|
|
v = os.environ.get(key, "")
|
|
if not v:
|
|
raise EnvironmentError(f"Missing env: {key}")
|
|
return v
|
|
|
|
async def http_request(
|
|
method: HTTPMethod,
|
|
url: str,
|
|
*,
|
|
label: str,
|
|
headers: dict | None = None,
|
|
params: dict | None = None,
|
|
json_body: dict | None = None,
|
|
timeout: int = REQUEST_TIMEOUT,
|
|
max_retries: int = 0,
|
|
) -> httpx.Response | None:
|
|
async with httpx.AsyncClient() as client:
|
|
for attempt in range(max_retries + 1):
|
|
try:
|
|
resp = await client.request(method, url, headers=headers, params=params, json=json_body, timeout=timeout)
|
|
return resp
|
|
except httpx.RequestError as e:
|
|
if attempt < max_retries:
|
|
print(f" [retry] {label} → {e}, attempt {attempt + 1}")
|
|
await asyncio.sleep((attempt + 1) * 2)
|
|
else:
|
|
print(f" [error] {label} → {e}")
|
|
return None
|
|
return None
|
|
|
|
|
|
async def _run_optional_step(coro, label: str) -> None:
|
|
"""부가 단계 실행 헬퍼: 예외를 삼키고 경고 로그만 남겨 호출측 흐름이 멈추지 않게 격리."""
|
|
try:
|
|
await coro
|
|
except Exception as e:
|
|
logger.warning("%s 실패 (무시하고 진행): %s", label, e)
|
|
|
|
|
|
def _normalize_homepage(url: str) -> str:
|
|
"""URL을 scheme/www/끝슬래시 제거 + 소문자로 정규화 (homepage 매칭용)."""
|
|
u = (url or "").strip().lower()
|
|
for p in ("https://", "http://"):
|
|
if u.startswith(p):
|
|
u = u[len(p):]
|
|
if u.startswith("www."):
|
|
u = u[4:]
|
|
return u.rstrip("/")
|
|
|
|
|
|
# SSL 인증서가 www.* 에만 유효한 도메인 — bare 도메인이면 사용자 클릭 시 브라우저 SSL warning 뜸.
|
|
_WWW_REQUIRED = ("gangnamunni.com", "facebook.com", "instagram.com")
|
|
|
|
|
|
def _with_scheme(u: str | None) -> str | None:
|
|
"""scheme 없는 URL에 https:// 보정 (수집기/링크 표시용). 빈 값은 None.
|
|
+ 중첩된 https://가 끼어있으면 마지막 URL만 추출 (LLM이 가끔 'https://www.X/https://Y' 같이 만듦).
|
|
+ SSL 엄격 도메인(gangnamunni/facebook/instagram)은 www. 자동 보강."""
|
|
if not u:
|
|
return None
|
|
u = u.strip()
|
|
# 'https://www.facebook.com/https://facebook.com/X' 같은 중첩 → 마지막 'http(s)://' 부터 잘라 사용
|
|
last = max(u.rfind("https://"), u.rfind("http://"))
|
|
if last > 0:
|
|
u = u[last:]
|
|
if "://" not in u:
|
|
u = "https://" + u
|
|
# scheme 뒤가 www. 없이 SSL 엄격 도메인이면 www. 추가
|
|
for dom in _WWW_REQUIRED:
|
|
for scheme in ("https://", "http://"):
|
|
if u.startswith(scheme + dom):
|
|
u = scheme + "www." + u[len(scheme):]
|
|
break
|
|
return u
|