data-management/c2m/scrape/c2m_scrape_app.py

from playwright.sync_api import Playwright, sync_playwright
import re
import time
import csv
import json
import os
import pandas as pd
from datetime import datetime


def run(playwright: Playwright, DATA_PATH, ERROR_PATH, CONFIG, is_headless) -> None:
    try:
        dt = datetime.now()
        date = dt.strftime('%Y-%m-%d')
        hour = dt.hour

        RANGE = CONFIG['range']
        PRODUCT_NAME = CONFIG['product_name']
        SITE_NAME = CONFIG['site_name']

        browser = playwright.chromium.launch(headless=True)
        context = browser.new_context()
        page = context.new_page()

        page.goto(CONFIG["base_url"])
        page.click("text=选择语言")

        with page.expect_navigation():
            page.click("text=English (US)")

        page.click("text=Please login to your account or register an account for free. >> a")
        page.click("input[name=\"login_id\"]")
        page.fill("input[name=\"login_id\"]", "yamzhenglim@gmail.com")
        page.click("input[name=\"login_pass\"]")
        page.fill("input[name=\"login_pass\"]", "Abc_123456")

        with page.expect_navigation():
            page.click("button:has-text(\"Login\")")

        page.click(":nth-match(:text(\"Request Service\"), 2)")

        time.sleep(2)
        page.eval_on_selector('.btn-agree', '$(".btn-agree").button( "option", "disabled", true | false )')
        page.click("text=I Agree")

        for option in CONFIG['options']: # Not Implemented  "ALLINK", "GH", "NH", "OB"
            try:

                option_selector = page.locator("select[name=\"reload[acc_type]\"]")
                option_selector.select_option(option)

                data = []
                for i in range (RANGE['start'], RANGE['stop']+1, RANGE['step']):
                    d = {}
                    myr_selector = '#reload_myr_amount'
                    myr_locator = page.locator(myr_selector)
                    myr_locator.clear()
                    myr_locator.fill(f'{i}')

                    page.press("text=Transaction Amount", 'Tab')
                    time.sleep(0.5)

                    myr = page.evaluate('$("#reload_myr_amount").val()')
                    cny = page.evaluate('$("#reload_cny_amount").val()')

                    q_staff = page.query_selector('.lime')
                    if q_staff:
                        staff_status = q_staff.inner_text().strip()
                    else:
                        staff_status = 'Offline'

                    orders_selector = page.query_selector('.in-processing')
                    if orders_selector:
                        orders = re.findall('\d+', orders_selector.inner_text())

                    svc = page.evaluate('$("#service_charge").val()')
                    total_payable = page.evaluate('$("#total_amount").val()')
                    current_rate = page.query_selector('#ex_rate').text_content()
                    # today_rate = page.query_selector('.today_rate').text_content()
                    # last_rate_update = page.query_selector('span:below(.today_rate)').text_content()
                    last_rate_update = page.query_selector('.text-muted.help-inline').text_content().strip()

                    d['product_name'] = PRODUCT_NAME
                    d['site_name'] = SITE_NAME
                    d['service_type'] = option  # need confirmation / add to config
                    d['pending_orders'] = int(orders[0]) 
                    d['CNY'] = cny
                    d['MYR'] = myr
                    d['handeling_fee'] = svc
                    d['payable'] = total_payable
                    d['current_rate'] = current_rate
                    # d['day_rate'] = today_rate
                    d['last_rate_update'] = last_rate_update
                    d['staff_status'] = staff_status
                    d['scrape_date'] = datetime.now().strftime('%Y-%m-%d')
                    d['scrape_hour'] = datetime.now().strftime('%H')
                
                    data.append(d)

                df = pd.DataFrame(data)

            except Exception as e:
                print(e)
                stamp = datetime.now().strftime('%Y_%m_%d')
                page.screenshot(path = f"{ERROR_PATH}/ERROR_{stamp}.png")
                
            df.to_csv(f"{DATA_PATH}/{option}_{date}_{hour}.csv", index=False)

    except Exception as e:
        print(e)
        stamp = datetime.now().strftime('%Y_%m_%d')
        page.screenshot(path = f"{ERROR_PATH}/ERROR_{stamp}.png")

if __name__ == '__main__':

    # Define a dictionary
    conf_dic = {
        "base_url": "https://www.c2m.my/reload.php",
        "product_name": "Exchange",
        "site_name": "c2m",
        "range":{
            "start":1,
            "stop":100001,
            "step":1000
        },
        "options":["TBDF", "ALTR"]
    }

    # Convert the dictionary to JSON format
    conf = json.dumps(conf_dic)

    CONFIG = json.loads(conf)

    DATA_PATH = "./c2m/scrape/local_run_files/"
    ERROR_PATH ="./c2m/scrape/local_run_files/error/"
    os.makedirs(DATA_PATH, exist_ok=True)
        
    with sync_playwright() as playwright:
        run(playwright, DATA_PATH, ERROR_PATH, CONFIG, is_headless = False)

else:
    def run_app(C2M_FILES_TEMP_STORAGE_PATH, C2M_CONFIG):
        # dt = datetime.now()
        # date = dt.strftime('%Y-%m-%d')
        # hour = dt.hour

        DATA_PATH = C2M_FILES_TEMP_STORAGE_PATH
        ERROR_PATH = C2M_FILES_TEMP_STORAGE_PATH
        os.makedirs(DATA_PATH, exist_ok=True)
            
        with sync_playwright() as playwright:
            run(playwright, DATA_PATH, ERROR_PATH, C2M_CONFIG, is_headless = True)