실전 자동화 파이썬 코드의 모든 함수와 라이브러리, 역할·사용 예시를 한눈에!
os
, json
: 파일 경로 관리 및 설정 데이터 저장/불러오기requests
: 업비트/텔레그램 HTTP 요청BeautifulSoup
: HTML 공지 파싱 (태그 제거, 텍스트 추출)cloudscraper
: 403 에러 우회용 클라우드플레어 대응 HTTP 요청re
: 정규표현식 기반 문자열 파싱datetime
, time
, random
: 시간계산 및 루프 딜레이 처리💡 설치 명령: pip install requests beautifulsoup4 cloudscraper
같은 에러는 5분 이내 중복 알림 방지
def send_error_once(key, message):
now = time.time()
if key not in last_error_messages or (now - last_error_messages[key]) > ERROR_COOLDOWN_SECONDS:
send_telegram_message(message, chat_id=ADMIN_CHAT_ID)
last_error_messages[key] = now
403 차단시 cloudscraper로 재요청
def get_with_fallback(url):
try:
res = session.get(url, timeout=5)
if res.status_code == 403:
scraper = cloudscraper.create_scraper()
res = scraper.get(url, timeout=5)
return res
except Exception as e:
send_error_once("요청실패", f"[요청 예외] {e}")
return None
텔레그램 메시지 전송
def send_telegram_message(msg, bot_token=TELEGRAM_BOT_TOKEN, chat_id=TELEGRAM_CHAT_ID):
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
requests.post(url, data={"chat_id": chat_id, "text": msg, "parse_mode": "HTML"})
마지막 감지된 공지를 JSON으로 저장/불러오기
def load_last_notice(path=LAST_NOTICE_PATH):
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
def save_last_notice(notice, path=LAST_NOTICE_PATH):
with open(path, 'w', encoding='utf-8') as f:
json.dump(notice, f, ensure_ascii=False, indent=2)
공지 HTML을 순수 텍스트로 변환
def html_to_text(html):
soup = BeautifulSoup(html, "html.parser")
return soup.get_text(separator='\n', strip=True)
표 안에서 상장시간 정보 추출
def extract_all_trade_times_table(html):
soup = BeautifulSoup(html, "html.parser")
table = soup.find("table")
# 코인명, 시간 추출 로직 작성
공지 제목에서 코인명만 추출
def extract_coin_name_from_title(title):
match = re.match(r'([^(]+)\(', title)
if match:
return match.group(1).strip()
공지 본문에서 상장 시간 추출 (표 or 텍스트)
def extract_trade_times(text, html, pattern_type, title="상장코인"):
# 유형별로 파싱하는 구조
타임존 제거 후 비교 가능하게 처리
def to_naive(dt):
if dt and dt.tzinfo is not None:
return dt.replace(tzinfo=None)
return dt
상장 관련 키워드 포함 여부 확인
def is_listing_notice(title, config):
return any(kw in title for kw in config["listing_keywords"])
‘5월 30일 17시’ → datetime 객체로 변환
def parse_trade_time(trade_time_str):
hangul_match = re.search(r'(\d{1,2})월\s*(\d{1,2})일\s*(\d{1,2})시', trade_time_str)
if hangul_match:
month = int(hangul_match.group(1))
day = int(hangul_match.group(2))
hour = int(hangul_match.group(3))
now = datetime.now()
return datetime(now.year, month, day, hour)
📌 실제 자동화의 핵심 함수
상장공지 확인 → 중복방지 → 텔레그램 알림 전송 → 마지막 공지 저장
def process_notice_by_id(notice_id):
# 상세조회, 상장공지 여부, 중복확인, 거래시간 추출, 알림전송 등
3~10초 랜덤 대기하며 무한반복 감시
while True:
# 1. 업비트 공지 확인
# 2. 신규 공지면 상세 정보 파싱 → 알림 전송
# 3. 3~10초 랜덤 대기 후 반복