세이브

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}")

출처: 하량이