browser-use-oauth/run.py

import argparse
import os
import signal
import subprocess
import sys
from datetime import datetime

import requests

#!/usr/bin/env python3

# ── 설정 부분 ──
PYTHON_SCRIPT = "./src/main.py"
DOMAIN_FILE = "./data/domains.txt"
# ─────────────


def download_domains():
    """도메인 파일 다운로드"""
    try:
        print("도메인 파일 다운로드 중...")
        response = requests.get(
            "https://f.imnya.ng/.whs/tp-domains/data/domains/latest.txt"
        )
        response.raise_for_status()

        # 디렉토리가 없으면 생성
        os.makedirs(os.path.dirname("./data"), exist_ok=True)

        with open(DOMAIN_FILE, "w", encoding="utf-8") as f:
            f.write(response.text)
        print("도메인 파일 다운로드 완료")
    except requests.RequestException as e:
        print(f"도메인 파일 다운로드 실패: {e}")
        sys.exit(1)


def run_script(start_line, end_line, skh_option):
    """Python 스크립트 실행"""
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{current_time}] Processing lines {start_line} to {end_line}...")

    process = None
    signal_handled = False

    def signal_handler(sig, frame):
        nonlocal signal_handled
        if signal_handled:
            return
        signal_handled = True

        print("\n🛑 종료 신호를 받았습니다. 정리 작업을 진행합니다...")
        
        if process:
            try:
                # 자식 프로세스에 SIGTERM 전송
                print("📤 서브프로세스에 종료 신호를 전달합니다...")
                process.terminate()
                # 5초간 대기
                process.wait(timeout=5)
                print("✅ 서브프로세스가 정상적으로 종료되었습니다.")
            except subprocess.TimeoutExpired:
                print("⚠️ 서브프로세스가 응답하지 않아 강제 종료합니다...")
                process.kill()
                try:
                    process.wait(timeout=3)
                    print("✅ 서브프로세스가 강제 종료되었습니다.")
                except subprocess.TimeoutExpired:
                    print("❌ 서브프로세스 강제 종료 실패")
            except Exception as e:
                print(f"❌ 프로세스 종료 중 오류: {e}")

        print("✅ 런처 종료 완료.")
        sys.exit(0)

    # 원래 시그널 핸들러 저장
    original_sigint = signal.signal(signal.SIGINT, signal_handler)
    original_sigterm = signal.signal(signal.SIGTERM, signal_handler)

    try:
        command = [
            "uv",
            "run",
            PYTHON_SCRIPT,
            "-f",
            DOMAIN_FILE,
            "-s",
            str(start_line),
            "-e",
            str(end_line),
        ]
        if skh_option:
            command.append("--skip-html-check")

        process = subprocess.Popen(command)
        returncode = process.wait()

        if returncode != 0:
            print(f"❌ Python 스크립트가 오류 코드 {returncode}로 종료되었습니다.")
            sys.exit(returncode)

    except KeyboardInterrupt:
        signal_handler(signal.SIGINT, None)
    except Exception as e:
        print(f"❌ 스크립트 실행 중 오류: {e}")
        if process:
            try:
                process.terminate()
                process.wait(timeout=3)
            except subprocess.TimeoutExpired:
                process.kill()
        sys.exit(1)
    finally:
        # 시그널 핸들러 복원
        signal.signal(signal.SIGINT, original_sigint)
        signal.signal(signal.SIGTERM, original_sigterm)


def main():
    parser = argparse.ArgumentParser(
        description="도메인 처리 스크립트 실행기",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
사용 예시:
  uv run run.py 10000 11000           # 10000~11000 라인 처리
  uv run run.py 10000 11000 --skh     # SKH 옵션 활성화
  uv run run.py 10000 11000 --no-download  # 다운로드 생략
        """,
    )

    parser.add_argument("start_line", type=int, help="시작 라인 번호")
    parser.add_argument("end_line", type=int, help="종료 라인 번호")
    parser.add_argument("--skh", action="store_true", help="SKH 옵션 활성화")
    parser.add_argument(
        "--no-download", action="store_true", help="도메인 파일 다운로드 생략"
    )

    args = parser.parse_args()

    # 라인 범위 검증
    if args.start_line < 0 or args.end_line < 0:
        print("라인 번호는 0 이상이어야 합니다.")
        sys.exit(1)

    if args.start_line > args.end_line:
        print("시작 라인은 종료 라인보다 크거나 같아야 합니다.")
        sys.exit(1)

    # 도메인 파일 다운로드
    if not args.no_download:
        download_domains()
    elif not os.path.exists(DOMAIN_FILE):
        print(
            f"도메인 파일({DOMAIN_FILE})이 존재하지 않습니다. --no-download 옵션을 제거하거나 파일을 준비해주세요."
        )
        sys.exit(1)

    # 스크립트 실행
    run_script(args.start_line, args.end_line, args.skh)

    print("처리 완료.")


if __name__ == "__main__":
    main()