내 소식

오르비

태그

내 태그
입시
학습
생활
클럽
직업·취업
Epioptimus
Centurion
오르비 랭킹
XDK 누적 복권
XDK 경매
RARE

수능한자루로컴싸보기 [1275747] · MS 2023 · 쪽지

2025-11-07 02:23:35
조회수 184

세이브

게시글 주소: https://dev.orbi.kr/00075398714


import asyncio

import aiohttp

import urllib.parse

import logging

from typing import Dict, Any


logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')

logger = logging.getLogger(__name__)


USERNAME = ''  

PASSWORD = ''  


URL_TARGET = 'https://orbi.kr/amusement/lottery/buy_ticket'

URL_LOGIN = 'https://login.orbi.kr/login'

BATCH_SIZE = 30

WAIT_AFTER_BATCH_SEC = 70

LOGIN_REDIRECT_URL = 'https://orbi.kr/amusement/lottery'


batching_enabled = True

batch_counter = 0

global_session: aiohttp.ClientSession = None


async def perform_login(session: aiohttp.ClientSession) -> bool:

    logger.info("Attempting login...")


    login_data = {

        'url': LOGIN_REDIRECT_URL,

        'username': USERNAME,

        'password': PASSWORD,

    }


    headers = {

        'Accept': '*/*',

        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',

        'Referer': f'https://login.orbi.kr/login?url={urllib.parse.quote(LOGIN_REDIRECT_URL, safe="")}',

        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36',

    }


    try:

        async with session.post(URL_LOGIN, data=login_data, headers=headers) as response:

            logger.info(f"Login response status: {response.status}")

            if response.status in [200, 302]:

                logger.info("Login successful.")

                return True

            else:

                logger.error(f"Login failed. Status code: {response.status}")

                return False

    except Exception as e:

        logger.error(f"Error during login: {e}")

        return False



async def delay_next_cycle(from_batch: int):

    global batching_enabled

    if not batching_enabled:

        return


    logger.info(f"[CYCLE {from_batch}] Waiting {WAIT_AFTER_BATCH_SEC} seconds...")

    await asyncio.sleep(WAIT_AFTER_BATCH_SEC)


    if not batching_enabled:

        return


    logger.info(f"[CYCLE {from_batch}] Wait complete. Continuing...")



async def send_batch(session: aiohttp.ClientSession):

    global batch_counter

    global batching_enabled


    if not batching_enabled:

        return


    batch_counter += 1

    this_batch = batch_counter

    logger.info(f"\n[CYCLE {this_batch}] Sending {BATCH_SIZE} parallel requests.")


    request_headers = {

        'Accept': '*/*',

        'Referer': 'https://orbi.kr/amusement/lottery',

        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36',

    }


    async def make_request(i):

        try:

            async with session.post(URL_TARGET, headers=request_headers, data={}) as resp:

                return {'ok': resp.ok, 'status': resp.status}

        except Exception as err:

            return {'error': str(err)}


    tasks = [make_request(i) for i in range(BATCH_SIZE)]

    results = await asyncio.gather(*tasks, return_exceptions=False)


    success_count = sum(1 for r in results if 'ok' in r and r['status'] == 200)

    failed_count = sum(1 for r in results if 'error' in r or ('ok' in r and r['status'] != 200))

    summary = {'Total': len(results), 'Success (200)': success_count, 'Failed/Error': failed_count}

    logger.info(f"[CYCLE {this_batch}] Batch complete. Summary: {summary}")


    await delay_next_cycle(this_batch)



async def main_execute():

    global global_session


    async with aiohttp.ClientSession() as session:

        global_session = session


        if not await perform_login(session):

            logger.error("Login failed. Stopping routine.")

            return


        logger.info("Login successful. Starting loop.")

        while batching_enabled:

            await send_batch(session)


        logger.info("Routine stopped by user command.")



if __name__ == "__main__":

    try:

        logger.info("sc/2 loaded. Starting routine.")

        asyncio.run(main_execute())

    except KeyboardInterrupt:

        logger.info("Routine stopped by keyboard interrupt.")

    except Exception as e:

        logger.error(f"Unexpected error: {e}")



출처: 하량이

0 XDK (+0)

  1. 유익한 글을 읽었다면 작성자에게 XDK를 선물하세요.

수능한자루로컴싸보기 [1275747]

쪽지 보내기

글쓰기
오르비 1455013번째 회원 되기
아이디 찾기 비밀번호 찾기

오늘의 추천 글

오르비 과외시장

2027 수능 D - 158

오르비 플레이
오르비 RARE
Move
our corporate site
Orbi Class
on-line classes
Orbi Books
educational books & resources
Gae9
humor and fun

© 2000-2026 Move Inc.

orbisoptimus (v17-5e1090)