tixcraft_bot/nodriver_tixcraft.py

1836 lines
68 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
#encoding=utf-8
#執行方式python chrome_tixcraft.py 或 python3 chrome_tixcraft.py
import argparse
import base64
import json
import logging
import os
import pathlib
import platform
import random
import ssl
import subprocess
import sys
import threading
import time
import warnings
import webbrowser
from datetime import datetime
import nodriver as uc
from nodriver import cdp
from nodriver.core.config import Config
from urllib3.exceptions import InsecureRequestWarning
import util
from NonBrowser import NonBrowser
try:
import ddddocr
except Exception as exc:
print(exc)
pass
2024-04-07 13:24:21 +00:00
CONST_APP_VERSION = "MaxBot (2024.03.26)"
CONST_MAXBOT_ANSWER_ONLINE_FILE = "MAXBOT_ONLINE_ANSWER.txt"
CONST_MAXBOT_CONFIG_FILE = "settings.json"
CONST_MAXBOT_EXTENSION_NAME = "Maxbotplus_1.0.0"
CONST_MAXBOT_INT28_FILE = "MAXBOT_INT28_IDLE.txt"
CONST_MAXBOT_LAST_URL_FILE = "MAXBOT_LAST_URL.txt"
CONST_MAXBOT_QUESTION_FILE = "MAXBOT_QUESTION.txt"
CONST_MAXBLOCK_EXTENSION_NAME = "Maxblockplus_1.0.0"
CONST_MAXBLOCK_EXTENSION_FILTER =[
"*google-analytics.com/*",
"*googletagmanager.com/*",
"*googletagservices.com/*",
"*lndata.com/*",
"*a.amnet.tw/*",
"*adx.c.appier.net/*",
"*clarity.ms/*",
"*cloudfront.com/*",
"*cms.analytics.yahoo.com/*",
"*doubleclick.net/*",
"*e2elog.fetnet.net/*",
"*fundingchoicesmessages.google.com/*",
"*ghtinc.com/*",
"*match.adsrvr.org/*",
"*onead.onevision.com.tw/*",
"*popin.cc/*",
"*rollbar.com/*",
"*sb.scorecardresearch.com/*",
"*tagtoo.co/*",
"*.ssp.hinet.net/*",
"*ticketmaster.sg/js/adblock*",
"*.googlesyndication.com/*",
"*treasuredata.com/*",
"*play.google.com/log?*",
"*www.youtube.com/youtubei/v1/player/heartbeat*",
"*tixcraft.com/js/analytics.js*",
"*ticketmaster.sg/js/adblock.js*",
"*img.uniicreative.com/*",
"*cdn.cookielaw.org/*",
"*tixcraft.com/js/custom.js*",
"*tixcraft.com/js/common.js*",
"*cdnjs.cloudflare.com/ajax/libs/clipboard.js/*"]
CONST_CITYLINE_SIGN_IN_URL = "https://www.cityline.com/Login.html?targetUrl=https%3A%2F%2Fwww.cityline.com%2FEvents.html"
CONST_FAMI_SIGN_IN_URL = "https://www.famiticket.com.tw/Home/User/SignIn"
CONST_HKTICKETING_SIGN_IN_URL = "https://premier.hkticketing.com/Secure/ShowLogin.aspx"
CONST_KHAM_SIGN_IN_URL = "https://kham.com.tw/application/UTK13/UTK1306_.aspx"
CONST_KKTIX_SIGN_IN_URL = "https://kktix.com/users/sign_in?back_to=%s"
CONST_TICKET_SIGN_IN_URL = "https://ticket.com.tw/application/utk13/utk1306_.aspx"
CONST_URBTIX_SIGN_IN_URL = "https://www.urbtix.hk/member-login"
CONST_FROM_TOP_TO_BOTTOM = "from top to bottom"
CONST_FROM_BOTTOM_TO_TOP = "from bottom to top"
CONST_CENTER = "center"
CONST_RANDOM = "random"
CONST_SELECT_ORDER_DEFAULT = CONST_FROM_TOP_TO_BOTTOM
CONT_STRING_1_SEATS_REMAINING = ['@1 seat(s) remaining','剩餘 1@','@1 席残り']
CONST_OCR_CAPTCH_IMAGE_SOURCE_NON_BROWSER = "NonBrowser"
CONST_OCR_CAPTCH_IMAGE_SOURCE_CANVAS = "canvas"
CONST_WEBDRIVER_TYPE_NODRIVER = "nodriver"
CONST_CHROME_FAMILY = ["chrome","edge","brave"]
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36"
warnings.simplefilter('ignore',InsecureRequestWarning)
ssl._create_default_https_context = ssl._create_unverified_context
logging.basicConfig()
logger = logging.getLogger('logger')
def get_config_dict(args):
app_root = util.get_app_root()
config_filepath = os.path.join(app_root, CONST_MAXBOT_CONFIG_FILE)
# allow assign config by command line.
if not args.input is None:
if len(args.input) > 0:
config_filepath = args.input
config_dict = None
if os.path.isfile(config_filepath):
# start to overwrite config settings.
with open(config_filepath) as json_data:
config_dict = json.load(json_data)
if not args.headless is None:
config_dict["advanced"]["headless"] = util.t_or_f(args.headless)
if not args.homepage is None:
if len(args.homepage) > 0:
config_dict["homepage"] = args.homepage
2024-04-06 14:53:42 +00:00
if not args.ticket_number is None:
2024-04-04 17:27:01 +00:00
if args.ticket_number > 0:
config_dict["ticket_number"] = args.ticket_number
2024-04-06 14:53:42 +00:00
if not args.browser is None:
if len(args.browser) > 0:
config_dict["browser"] = args.browser
if not args.tixcraft_sid is None:
if len(args.tixcraft_sid) > 0:
config_dict["advanced"]["tixcraft_sid"] = args.tixcraft_sid
if not args.ibonqware is None:
if len(args.ibonqware) > 0:
config_dict["advanced"]["ibonqware"] = args.ibonqware
if not args.kktix_account is None:
if len(args.kktix_account) > 0:
config_dict["advanced"]["kktix_account"] = args.kktix_account
if not args.kktix_password is None:
if len(args.kktix_password) > 0:
config_dict["advanced"]["kktix_password_plaintext"] = args.kktix_password
2024-04-06 14:53:42 +00:00
if not args.proxy_server is None:
if len(args.proxy_server) > 2:
config_dict["advanced"]["proxy_server_port"] = args.proxy_server
2024-04-04 05:58:16 +00:00
if not args.window_size is None:
if len(args.window_size) > 2:
config_dict["advanced"]["window_size"] = args.window_size
# special case for headless to enable away from keyboard mode.
is_headless_enable_ocr = False
if config_dict["advanced"]["headless"]:
# for tixcraft headless.
#print("If you are runnig headless mode on tixcraft, you need input your cookie SID.")
if len(config_dict["advanced"]["tixcraft_sid"]) > 1:
is_headless_enable_ocr = True
if is_headless_enable_ocr:
config_dict["ocr_captcha"]["enable"] = True
config_dict["ocr_captcha"]["force_submit"] = True
return config_dict
def write_question_to_file(question_text):
working_dir = os.path.dirname(os.path.realpath(__file__))
target_path = os.path.join(working_dir, CONST_MAXBOT_QUESTION_FILE)
util.write_string_to_file(target_path, question_text)
def write_last_url_to_file(url):
working_dir = os.path.dirname(os.path.realpath(__file__))
target_path = os.path.join(working_dir, CONST_MAXBOT_LAST_URL_FILE)
util.write_string_to_file(target_path, url)
def read_last_url_from_file():
ret = ""
with open(CONST_MAXBOT_LAST_URL_FILE, "r") as text_file:
ret = text_file.readline()
return ret
def play_sound_while_ordering(config_dict):
app_root = util.get_app_root()
captcha_sound_filename = os.path.join(app_root, config_dict["advanced"]["play_sound"]["filename"].strip())
util.play_mp3_async(captcha_sound_filename)
2024-04-06 14:53:42 +00:00
async def nodriver_facebook_login(tab, facebook_account, facebook_password):
if tab:
try:
account = await tab.query_selector("#email")
if account:
await account.send_keys(facebook_account)
else:
print("account not found")
password = await tab.query_selector("#pass")
if password:
await password.send_keys(facebook_password)
await tab.send(cdp.input_.dispatch_key_event("keyDown", code="Enter", key="Enter", text="\r", windows_virtual_key_code=13))
await tab.send(cdp.input_.dispatch_key_event("keyUp", code="Enter", key="Enter", text="\r", windows_virtual_key_code=13))
time.sleep(2)
else:
print("password not found")
except Exception as e:
print("send_keys fail.")
print(e)
pass
async def nodriver_kktix_signin(tab, url, config_dict):
kktix_account = config_dict["advanced"]["kktix_account"]
kktix_password = config_dict["advanced"]["kktix_password_plaintext"].strip()
if kktix_password == "":
kktix_password = util.decryptMe(config_dict["advanced"]["kktix_password"])
if len(kktix_account) > 4:
2024-04-06 14:53:42 +00:00
try:
account = await tab.query_selector("#user_login")
await account.send_keys(kktix_account)
2024-04-06 14:53:42 +00:00
password = await tab.query_selector("#user_password")
await password.send_keys(kktix_password)
2024-04-06 14:53:42 +00:00
submit = await tab.query_selector("input[type='submit'][name]")
await submit.click()
time.sleep(0.2)
except Exception as e:
print(e)
pass
async def nodriver_kktix_paused_main(tab, url, config_dict, kktix_dict):
is_url_contain_sign_in = False
# fix https://kktix.com/users/sign_in?back_to=https://kktix.com/events/xxxx and registerStatus: SOLD_OUT cause page refresh.
if '/users/sign_in?' in url:
await nodriver_kktix_signin(tab, url, config_dict)
is_url_contain_sign_in = True
return kktix_dict
async def nodriver_goto_homepage(driver, config_dict):
homepage = config_dict["homepage"]
if 'kktix.c' in homepage:
if len(config_dict["advanced"]["kktix_account"])>0:
if not 'https://kktix.com/users/sign_in?' in homepage:
homepage = CONST_KKTIX_SIGN_IN_URL % (homepage)
2024-04-06 14:53:42 +00:00
if 'famiticket.com' in homepage:
if len(config_dict["advanced"]["fami_account"])>0:
homepage = CONST_FAMI_SIGN_IN_URL
if 'kham.com' in homepage:
if len(config_dict["advanced"]["kham_account"])>0:
homepage = CONST_KHAM_SIGN_IN_URL
if 'ticket.com.tw' in homepage:
if len(config_dict["advanced"]["ticket_account"])>0:
homepage = CONST_TICKET_SIGN_IN_URL
if 'urbtix.hk' in homepage:
if len(config_dict["advanced"]["urbtix_account"])>0:
homepage = CONST_URBTIX_SIGN_IN_URL
if 'cityline.com' in homepage:
if len(config_dict["advanced"]["cityline_account"])>0:
homepage = CONST_CITYLINE_SIGN_IN_URL
if 'hkticketing.com' in homepage:
if len(config_dict["advanced"]["hkticketing_account"])>0:
homepage = CONST_HKTICKETING_SIGN_IN_URL
if 'ticketplus.com.tw' in homepage:
if len(config_dict["advanced"]["ticketplus_account"]) > 1:
homepage = "https://ticketplus.com.tw/"
tab = await driver.get(homepage)
time.sleep(1)
tixcraft_family = False
if 'tixcraft.com' in homepage:
tixcraft_family = True
if 'indievox.com' in homepage:
tixcraft_family = True
if 'ticketmaster.' in homepage:
tixcraft_family = True
if tixcraft_family:
tixcraft_sid = config_dict["advanced"]["tixcraft_sid"]
if len(tixcraft_sid) > 1:
cookies = await driver.cookies.get_all()
for cookie in cookies:
if cookie.name=='SID':
cookie.value=tixcraft_sid
break
await driver.cookies.set_all(cookies)
if 'ibon.com' in homepage:
ibonqware = config_dict["advanced"]["ibonqware"]
if len(ibonqware) > 1:
cookies = await driver.cookies.get_all()
for cookie in cookies:
if cookie.name=='ibonqware':
cookie.value=ibonqware
break
await driver.cookies.set_all(cookies)
return tab
async def nodriver_kktix_travel_price_list(tab, config_dict, kktix_area_auto_select_mode, kktix_area_keyword):
show_debug_message = True # debug.
show_debug_message = False # online
if config_dict["advanced"]["verbose"]:
show_debug_message = True
ticket_number = config_dict["ticket_number"]
areas = None
is_ticket_number_assigned = False
ticket_price_list = None
try:
ticket_price_list = await tab.select_all('div.display-table-row')
except Exception as exc:
ticket_price_list = None
print("find ticket-price Exception:")
print(exc)
pass
is_dom_ready = True
price_list_count = 0
if not ticket_price_list is None:
price_list_count = len(ticket_price_list)
if show_debug_message:
print("found price count:", price_list_count)
else:
is_dom_ready = False
print("find ticket-price fail")
if price_list_count > 0:
areas = []
kktix_area_keyword_array = kktix_area_keyword.split(' ')
kktix_area_keyword_1 = kktix_area_keyword_array[0]
kktix_area_keyword_1_and = ""
if len(kktix_area_keyword_array) > 1:
kktix_area_keyword_1_and = kktix_area_keyword_array[1]
# clean stop word.
kktix_area_keyword_1 = util.format_keyword_string(kktix_area_keyword_1)
kktix_area_keyword_1_and = util.format_keyword_string(kktix_area_keyword_1_and)
if show_debug_message:
print('kktix_area_keyword_1:', kktix_area_keyword_1)
print('kktix_area_keyword_1_and:', kktix_area_keyword_1_and)
for row in ticket_price_list:
row_text = ""
row_html = ""
row_input = None
current_ticket_number = "0"
try:
js_attr = await row.get_js_attributes()
row_html = js_attr["innerHTML"]
row_text = js_attr["innerText"]
row_input = await row.query_selector("input")
if not row_input is None:
js_attr_input = await row_input.get_js_attributes()
current_ticket_number = js_attr_input["value"]
except Exception as exc:
is_dom_ready = False
if show_debug_message:
print(exc)
# error, exit loop
break
if len(row_text) > 0:
if '未開賣' in row_text:
row_text = ""
if '暫無票' in row_text:
row_text = ""
if '已售完' in row_text:
row_text = ""
if 'Sold Out' in row_text:
row_text = ""
if '完売' in row_text:
row_text = ""
if not('<input type=' in row_html):
row_text = ""
if len(row_text) > 0:
if util.reset_row_text_if_match_keyword_exclude(config_dict, row_text):
row_text = ""
if len(row_text) > 0:
# clean stop word.
row_text = util.format_keyword_string(row_text)
if len(row_text) > 0:
if ticket_number > 1:
# start to check danger notice.
# 剩 n 張票 / n Left / 残り n 枚
ticket_count = 999
# for cht.
if ' danger' in row_html and '' in row_text and '' in row_text:
tmp_array = row_html.split('')
tmp_array = tmp_array[1].split('')
if len(tmp_array) > 0:
tmp_ticket_count = tmp_array[0].strip()
if tmp_ticket_count.isdigit():
ticket_count = int(tmp_ticket_count)
if show_debug_message:
print("found ticket 剩:", tmp_ticket_count)
# for ja.
if ' danger' in row_html and '残り' in row_text and '' in row_text:
tmp_array = row_html.split('残り')
tmp_array = tmp_array[1].split('')
if len(tmp_array) > 0:
tmp_ticket_count = tmp_array[0].strip()
if tmp_ticket_count.isdigit():
ticket_count = int(tmp_ticket_count)
if show_debug_message:
print("found ticket 残り:", tmp_ticket_count)
# for en.
if ' danger' in row_html and ' Left ' in row_html:
tmp_array = row_html.split(' Left ')
tmp_array = tmp_array[0].split('>')
if len(tmp_array) > 0:
tmp_ticket_count = tmp_array[len(tmp_array)-1].strip()
if tmp_ticket_count.isdigit():
if show_debug_message:
print("found ticket left:", tmp_ticket_count)
ticket_count = int(tmp_ticket_count)
if ticket_count < ticket_number:
# skip this row, due to no ticket remaining.
if show_debug_message:
print("found ticket left:", tmp_ticket_count, ",but target ticket:", ticket_number)
row_text = ""
if not row_input is None:
# check ticket input textbox.
if len(current_ticket_number) > 0:
if current_ticket_number != "0":
is_ticket_number_assigned = True
if is_ticket_number_assigned:
# no need to travel
break
if len(kktix_area_keyword_1) == 0:
# keyword #1, empty, direct add to list.
is_match_area = True
match_area_code = 1
else:
# MUST match keyword #1.
if kktix_area_keyword_1 in row_text:
#print('match keyword#1')
# because of logic between keywords is AND!
if len(kktix_area_keyword_1_and) == 0:
#print('keyword#2 is empty, directly match.')
# keyword #2 is empty, direct append.
is_match_area = True
match_area_code = 2
else:
if kktix_area_keyword_1_and in row_text:
#print('match keyword#2')
is_match_area = True
match_area_code = 3
else:
#print('not match keyword#2')
pass
else:
#print('not match keyword#1')
pass
if show_debug_message:
print("is_match_area:", is_match_area)
print("match_area_code:", match_area_code)
if is_match_area:
areas.append(row_input)
# from top to bottom, match first to break.
if kktix_area_auto_select_mode == CONST_FROM_TOP_TO_BOTTOM:
break
if not is_dom_ready:
# not sure to break or continue..., maybe break better.
break
else:
if show_debug_message:
print("no any price list found.")
pass
return is_dom_ready, is_ticket_number_assigned, areas
async def nodriver_kktix_assign_ticket_number(tab, config_dict, kktix_area_keyword):
show_debug_message = True # debug.
show_debug_message = False # online
if config_dict["advanced"]["verbose"]:
show_debug_message = True
ticket_number_str = str(config_dict["ticket_number"])
auto_select_mode = config_dict["area_auto_select"]["mode"]
is_ticket_number_assigned = False
matched_blocks = None
is_dom_ready = True
is_dom_ready, is_ticket_number_assigned, matched_blocks = await nodriver_kktix_travel_price_list(tab, config_dict, auto_select_mode, kktix_area_keyword)
target_area = None
is_need_refresh = False
if is_dom_ready:
if not is_ticket_number_assigned:
target_area = util.get_target_item_from_matched_list(matched_blocks, auto_select_mode)
if not matched_blocks is None:
if len(matched_blocks) == 0:
is_need_refresh = True
if show_debug_message:
print("matched_blocks is empty, is_need_refresh")
if not target_area is None:
current_ticket_number = ""
if show_debug_message:
print("try to get input box value.")
try:
current_ticket_number = str(target_area.attribute('value')).strip()
except Exception as exc:
pass
if len(current_ticket_number) > 0:
if current_ticket_number == "0":
try:
print("asssign ticket number:%s" % ticket_number_str)
target_area.clear_input()
target_area.send_keys(ticket_number_str)
is_ticket_number_assigned = True
except Exception as exc:
print("asssign ticket number to ticket-price field Exception:")
print(exc)
try:
target_area.clear()
target_area.send_keys("1")
is_ticket_number_assigned = True
except Exception as exc2:
print("asssign ticket number to ticket-price still failed.")
pass
else:
if show_debug_message:
print("value already assigned.")
# already assigned.
is_ticket_number_assigned = True
return is_dom_ready, is_ticket_number_assigned, is_need_refresh
async def nodriver_kktix_reg_captcha(tab, config_dict, fail_list, registrationsNewApp_div):
show_debug_message = True # debug.
show_debug_message = False # online
if config_dict["advanced"]["verbose"]:
show_debug_message = True
answer_list = []
is_question_popup = False
2024-04-06 14:53:42 +00:00
question_text = await nodriver_get_text_by_selector(tab, 'div.custom-captcha-inner p', 'innerText')
if len(question_text) > 0:
is_question_popup = True
write_question_to_file(question_text)
answer_list = util.get_answer_list_from_user_guess_string(config_dict, CONST_MAXBOT_ANSWER_ONLINE_FILE)
if len(answer_list)==0:
if config_dict["advanced"]["auto_guess_options"]:
#answer_list = util.get_answer_list_from_question_string(registrationsNewApp_div, question_text)
# due to selenium forat.
answer_list = util.get_answer_list_from_question_string(None, question_text)
inferred_answer_string = ""
for answer_item in answer_list:
if not answer_item in fail_list:
inferred_answer_string = answer_item
break
if show_debug_message:
print("inferred_answer_string:", inferred_answer_string)
print("question_text:", question_text)
print("answer_list:", answer_list)
print("fail_list:", fail_list)
# PS: auto-focus() when empty inferred_answer_string with empty inputed text value.
input_text_css = 'div.custom-captcha-inner > div > div > input'
next_step_button_css = ''
submit_by_enter = False
check_input_interval = 0.2
#is_answer_sent, fail_list = fill_common_verify_form(tab, config_dict, inferred_answer_string, fail_list, input_text_css, next_step_button_css, submit_by_enter, check_input_interval)
if len(answer_list) > 0:
2024-04-06 14:53:42 +00:00
input_text = await tab.query_selector(input_text_css)
if not input_text is None:
await input_text.send_keys(answer_list[0])
# due multi next buttons(pick seats/best seats)
await nodriver_kktix_press_next_button(tab)
return fail_list, is_question_popup
# : This is for case-2 next button.
async def nodriver_kktix_press_next_button(tab):
ret = False
css_select = "div.register-new-next-button-area > button"
but_button_list = None
try:
but_button_list = await tab.select_all(css_select)
except Exception as exc:
print(exc)
pass
if not but_button_list is None:
button_count = len(but_button_list)
#print("button_count:",button_count)
if button_count > 0:
try:
#print("click on last button")
await but_button_list[button_count-1].click()
ret = True
except Exception as exc:
print(exc)
pass
return ret
async def nodriver_kktix_reg_new_main(tab, config_dict, fail_list, played_sound_ticket):
show_debug_message = True # debug.
show_debug_message = False # online
if config_dict["advanced"]["verbose"]:
show_debug_message = True
# read config.
area_keyword = config_dict["area_auto_select"]["area_keyword"].strip()
# part 1: check div.
registrationsNewApp_div = None
try:
2024-04-06 14:53:42 +00:00
registrationsNewApp_div = await tab.query_selector('#registrationsNewApp')
except Exception as exc:
pass
#print("find input fail:", exc)
# part 2: assign ticket number
is_ticket_number_assigned = False
if not registrationsNewApp_div is None:
is_dom_ready = True
is_need_refresh = False
if len(area_keyword) > 0:
area_keyword_array = []
try:
area_keyword_array = json.loads("["+ area_keyword +"]")
except Exception as exc:
area_keyword_array = []
# default refresh
is_need_refresh_final = True
for area_keyword_item in area_keyword_array:
is_need_refresh_tmp = Falses
is_dom_ready, is_ticket_number_assigned, is_need_refresh_tmp = await nodriver_kktix_assign_ticket_number(tab, config_dict, area_keyword_item)
if not is_dom_ready:
# page redirecting.
break
# one of keywords not need to refresh, final is not refresh.
if not is_need_refresh_tmp:
is_need_refresh_final = False
if is_ticket_number_assigned:
break
else:
if show_debug_message:
print("is_need_refresh for keyword:", area_keyword_item)
if not is_ticket_number_assigned:
is_need_refresh = is_need_refresh_final
else:
# empty keyword, match all.
# TODO:
is_dom_ready, is_ticket_number_assigned, is_need_refresh = await nodriver_kktix_assign_ticket_number(tab, config_dict, "")
pass
if is_dom_ready:
# part 3: captcha
if is_ticket_number_assigned:
if config_dict["advanced"]["play_sound"]["ticket"]:
if not played_sound_ticket:
play_sound_while_ordering(config_dict)
played_sound_ticket = True
# whole event question.
fail_list, is_question_popup = await nodriver_kktix_reg_captcha(tab, config_dict, fail_list, registrationsNewApp_div)
2024-04-06 14:53:42 +00:00
# single option question
if not is_question_popup:
# no captcha text popup, goto next page.
2024-04-06 14:53:42 +00:00
control_text = await nodriver_get_text_by_selector(tab, 'div > div.code-input > div.control-group > label.control-label', 'innerText')
if show_debug_message:
print("control_text:", control_text)
2024-04-06 14:53:42 +00:00
if len(control_text) == 0:
click_ret = await nodriver_kktix_press_next_button(tab)
else:
# input by maxbox plus extension.
is_fill_at_webdriver = False
if not config_dict["browser"] in CONST_CHROME_FAMILY:
is_fill_at_webdriver = True
else:
if not config_dict["advanced"]["chrome_extension"]:
is_fill_at_webdriver = True
2024-04-06 14:53:42 +00:00
# TODO: not implement in extension, so force to fill in webdriver.
is_fill_at_webdriver = True
if is_fill_at_webdriver:
#TODO:
#set_kktix_control_label_text(driver, config_dict)
pass
else:
if is_need_refresh:
# reset to play sound when ticket avaiable.
played_sound_ticket = False
try:
print("no match any price, start to refresh page...")
await tab.reload()
except Exception as exc:
#print("refresh fail")
pass
if config_dict["advanced"]["auto_reload_page_interval"] > 0:
time.sleep(config_dict["advanced"]["auto_reload_page_interval"])
return fail_list, played_sound_ticket
async def nodriver_kktix_main(tab, url, config_dict, kktix_dict):
is_url_contain_sign_in = False
# fix https://kktix.com/users/sign_in?back_to=https://kktix.com/events/xxxx and registerStatus: SOLD_OUT cause page refresh.
if '/users/sign_in?' in url:
await nodriver_kktix_signin(tab, url, config_dict)
is_url_contain_sign_in = True
if not is_url_contain_sign_in:
if '/registrations/new' in url:
kktix_dict["start_time"] = time.time()
is_dom_ready = False
try:
html_body = await tab.get_content()
#print("html_body:",len(html_body))
if len(html_body) > 10240:
if "registrationsNewApp" in html_body:
if not "{{'new.i_read_and_agree_to'" in html_body:
is_dom_ready = True
except Exception as exc:
print(exc)
pass
if not is_dom_ready:
# reset answer fail list.
kktix_dict["fail_list"] = []
kktix_dict["played_sound_ticket"] = False
else:
is_finish_checkbox_click = False
#TODO: check checkbox here.
# check is able to buy.
if config_dict["kktix"]["auto_fill_ticket_number"]:
kktix_dict["fail_list"], kktix_dict["played_sound_ticket"] = await nodriver_kktix_reg_new_main(tab, config_dict, kktix_dict["fail_list"], kktix_dict["played_sound_ticket"])
kktix_dict["done_time"] = time.time()
else:
is_event_page = False
if '/events/' in url:
# ex: https://xxx.kktix.cc/events/xxx-copy-1
if len(url.split('/'))<=5:
is_event_page = True
if is_event_page:
if config_dict["kktix"]["auto_press_next_step_button"]:
# pass switch check.
#print("should press next here.")
#kktix_events_press_next_button(driver)
pass
# reset answer fail list.
kktix_dict["fail_list"] = []
kktix_dict["played_sound_ticket"] = False
is_kktix_got_ticket = False
if '/events/' in url and '/registrations/' in url and "-" in url:
if not '/registrations/new' in url:
if not 'https://kktix.com/users/sign_in?' in url:
is_kktix_got_ticket = True
if is_kktix_got_ticket:
if '/events/' in config_dict["homepage"] and '/registrations/' in config_dict["homepage"] and "-" in config_dict["homepage"]:
# do nothing when second time come in.
if len(url.split('/'))>=7:
if len(config_dict["homepage"].split('/'))>=7:
# match event code.
if url.split('/')[4]==config_dict["homepage"].split('/')[4]:
# break loop.
is_kktix_got_ticket = False
2024-04-06 14:53:42 +00:00
is_quit_bot = False
if is_kktix_got_ticket:
2024-04-06 14:53:42 +00:00
if not kktix_dict["start_time"] is None:
if not kktix_dict["done_time"] is None:
bot_elapsed_time = kktix_dict["done_time"] - kktix_dict["start_time"]
if kktix_dict["elapsed_time"] != bot_elapsed_time:
print("bot elapsed time:", "{:.3f}".format(bot_elapsed_time))
kktix_dict["elapsed_time"] = bot_elapsed_time
if config_dict["advanced"]["play_sound"]["order"]:
if not kktix_dict["played_sound_order"]:
play_sound_while_ordering(config_dict)
kktix_dict["played_sound_order"] = True
if config_dict["advanced"]["headless"]:
if not kktix_dict["is_popup_checkout"]:
kktix_account = config_dict["advanced"]["kktix_account"]
kktix_password = config_dict["advanced"]["kktix_password_plaintext"].strip()
if kktix_password == "":
kktix_password = util.decryptMe(config_dict["advanced"]["kktix_password"])
print("基本資料(或實名制)網址:", url)
if len(kktix_account) > 0:
print("搶票成功, 帳號:", kktix_account)
script_name = "chrome_tixcraft"
if config_dict["advanced"]["webdriver_type"] == CONST_WEBDRIVER_TYPE_NODRIVER:
script_name = "nodriver_tixcraft"
threading.Thread(target=util.launch_maxbot, args=(script_name,"", url, kktix_account, kktix_password,"","false",)).start()
#driver.quit()
#sys.exit()
is_event_page = False
if len(url.split('/'))>=7:
is_event_page = True
if is_event_page:
confirm_clicked = False
2024-04-06 14:53:42 +00:00
try:
submit = await tab.query_selector("div.form-actions a.btn-primary")
await submit.click()
confirm_clicked = True
except Exception as exc:
print(exc)
if confirm_clicked:
domain_name = url.split('/')[2]
checkout_url = "https://%s/account/orders" % (domain_name)
print("搶票成功, 請前往該帳號訂單查看: %s" % (checkout_url))
webbrowser.open_new(checkout_url)
kktix_dict["is_popup_checkout"] = True
is_quit_bot = True
else:
kktix_dict["is_popup_checkout"] = False
kktix_dict["played_sound_order"] = False
2024-04-06 14:53:42 +00:00
return kktix_dict, is_quit_bot
async def nodriver_tixcraft_home_close_window(tab):
accept_all_cookies_btn = None
try:
accept_all_cookies_btn = await tab.query_selector('#onetrust-accept-btn-handler')
if accept_all_cookies_btn:
accept_all_cookies_btn.click()
except Exception as exc:
#print(exc)
pass
async def nodriver_get_text_by_selector(tab, my_css_selector, attribute='innerHTML'):
div_text = ""
try:
div_element = await tab.query_selector(my_css_selector)
if div_element:
js_attr = await div_element.get_js_attributes()
if js_attr:
div_text = js_attr[attribute]
except Exception as exc:
print("find verify textbox fail")
pass
return div_text
async def nodriver_tixcraft_redirect(tab, url):
ret = False
game_name = ""
url_split = url.split("/")
if len(url_split) >= 6:
game_name = url_split[5]
if len(game_name) > 0:
if "/activity/detail/%s" % (game_name,) in url:
entry_url = url.replace("/activity/detail/","/activity/game/")
print("redirec to new url:", entry_url)
try:
await tab.get(entry_url)
ret = True
except Exception as exec1:
pass
return ret
async def nodriver_ticketmaster_promo(tab, config_dict, fail_list):
question_selector = '#promoBox'
return nodriver_tixcraft_input_check_code(tab, config_dict, fail_list, question_selector)
async def nodriver_tixcraft_verify(tab, config_dict, fail_list):
question_selector = '.zone-verify'
return nodriver_tixcraft_input_check_code(tab, config_dict, fail_list, question_selector)
async def nodriver_tixcraft_input_check_code(tab, config_dict, fail_list, question_selector):
show_debug_message = True # debug.
show_debug_message = False # online
if config_dict["advanced"]["verbose"]:
show_debug_message = True
answer_list = []
question_text = await nodriver_get_text_by_selector(tab, question_selector, 'innerText')
if len(question_text) > 0:
write_question_to_file(question_text)
answer_list = util.get_answer_list_from_user_guess_string(config_dict, CONST_MAXBOT_ANSWER_ONLINE_FILE)
if len(answer_list)==0:
if config_dict["advanced"]["auto_guess_options"]:
answer_list = util.guess_tixcraft_question(driver, question_text)
inferred_answer_string = ""
for answer_item in answer_list:
if not answer_item in fail_list:
inferred_answer_string = answer_item
break
if show_debug_message:
print("inferred_answer_string:", inferred_answer_string)
print("answer_list:", answer_list)
# PS: auto-focus() when empty inferred_answer_string with empty inputed text value.
input_text_css = "input[name='checkCode']"
next_step_button_css = ""
submit_by_enter = True
check_input_interval = 0.2
is_answer_sent, fail_list = fill_common_verify_form(driver, config_dict, inferred_answer_string, fail_list, input_text_css, next_step_button_css, submit_by_enter, check_input_interval)
return fail_list
async def nodriver_tixcraft_main(tab, url, config_dict, tixcraft_dict, ocr, Captcha_Browser):
await nodriver_tixcraft_home_close_window(tab)
# special case for same event re-open, redirect to user's homepage.
if 'https://tixcraft.com/' == url or 'https://tixcraft.com/activity' == url:
if "/ticket/area/" in config_dict["homepage"]:
if len(config_dict["homepage"].split('/'))==7:
try:
await tab.get(config_dict["homepage"])
except Exception as e:
pass
if "/activity/detail/" in url:
tixcraft_dict["start_time"] = time.time()
is_redirected = await nodriver_tixcraft_redirect(tab, url)
is_date_selected = False
if "/activity/game/" in url:
tixcraft_dict["start_time"] = time.time()
if config_dict["date_auto_select"]["enable"]:
domain_name = url.split('/')[2]
# TODO:
#is_date_selected = tixcraft_date_auto_select(driver, url, config_dict, domain_name)
pass
if '/artist/' in url and 'ticketmaster.com' in url:
tixcraft_dict["start_time"] = time.time()
if len(url.split('/'))==6:
if config_dict["date_auto_select"]["enable"]:
domain_name = url.split('/')[2]
# TODO:
#is_date_selected = ticketmaster_date_auto_select(driver, url, config_dict, domain_name)
pass
# choose area
if '/ticket/area/' in url:
domain_name = url.split('/')[2]
if config_dict["area_auto_select"]["enable"]:
if not 'ticketmaster' in domain_name:
# for tixcraft
# TODO:
#tixcraft_area_auto_select(driver, url, config_dict)
pass
tixcraft_dict["area_retry_count"]+=1
#print("count:", tixcraft_dict["area_retry_count"])
if tixcraft_dict["area_retry_count"] >= (60 * 15):
# Cool-down
tixcraft_dict["area_retry_count"] = 0
time.sleep(5)
else:
# area auto select is too difficult, skip in this version.
# TODO:
#tixcraft_dict["fail_promo_list"] = ticketmaster_promo(driver, config_dict, tixcraft_dict["fail_promo_list"])
#ticketmaster_assign_ticket_number(driver, config_dict)
pass
else:
tixcraft_dict["fail_promo_list"] = []
tixcraft_dict["area_retry_count"]=0
# https://ticketmaster.sg/ticket/check-captcha/23_blackpink/954/5/75
if '/ticket/check-captcha/' in url:
domain_name = url.split('/')[2]
# TODO:
#ticketmaster_captcha(driver, config_dict, ocr, Captcha_Browser, domain_name)
pass
if '/ticket/verify/' in url:
# TODO:
#tixcraft_dict["fail_list"] = tixcraft_verify(driver, config_dict, tixcraft_dict["fail_list"])
pass
else:
tixcraft_dict["fail_list"] = []
# main app, to select ticket number.
if '/ticket/ticket/' in url:
domain_name = url.split('/')[2]
# TODO:
#tixcraft_ticket_main(driver, config_dict, ocr, Captcha_Browser, domain_name)
tixcraft_dict["done_time"] = time.time()
if config_dict["advanced"]["play_sound"]["ticket"]:
if not tixcraft_dict["played_sound_ticket"]:
play_sound_while_ordering(config_dict)
tixcraft_dict["played_sound_ticket"] = True
else:
tixcraft_dict["played_sound_ticket"] = False
if '/ticket/order' in url:
tixcraft_dict["done_time"] = time.time()
is_quit_bot = False
if '/ticket/checkout' in url:
if not tixcraft_dict["start_time"] is None:
if not tixcraft_dict["done_time"] is None:
bot_elapsed_time = tixcraft_dict["done_time"] - tixcraft_dict["start_time"]
if tixcraft_dict["elapsed_time"] != bot_elapsed_time:
print("bot elapsed time:", "{:.3f}".format(bot_elapsed_time))
tixcraft_dict["elapsed_time"] = bot_elapsed_time
if config_dict["advanced"]["headless"]:
if not tixcraft_dict["is_popup_checkout"]:
domain_name = url.split('/')[2]
checkout_url = "https://%s/ticket/checkout" % (domain_name)
print("搶票成功, 請前往該帳號訂單查看: %s" % (checkout_url))
webbrowser.open_new(checkout_url)
tixcraft_dict["is_popup_checkout"] = True
is_quit_bot = True
if config_dict["advanced"]["play_sound"]["order"]:
if not tixcraft_dict["played_sound_order"]:
play_sound_while_ordering(config_dict)
tixcraft_dict["played_sound_order"] = True
else:
tixcraft_dict["is_popup_checkout"] = False
tixcraft_dict["played_sound_order"] = False
return tixcraft_dict, is_quit_bot
async def nodriver_ticketplus_account_sign_in(tab, config_dict):
print("nodriver_ticketplus_account_sign_in")
is_filled_form = False
is_submited = False
ticketplus_account = config_dict["advanced"]["ticketplus_account"]
ticketplus_password = config_dict["advanced"]["ticketplus_password_plaintext"].strip()
if ticketplus_password == "":
ticketplus_password = util.decryptMe(config_dict["advanced"]["ticketplus_password"])
# manually keyin verify code.
country_code = ""
try:
my_css_selector = 'input[placeholder="區碼"]'
el_country = await tab.query_selector(my_css_selector)
if el_country:
country_code = await el_country.apply('function (element) { return element.value; } ')
print("country_code", country_code)
except Exception as exc:
print(exc)
is_account_assigned = False
try:
my_css_selector = 'input[placeholder="手機號碼 *"]'
el_account = await tab.query_selector(my_css_selector)
if el_account:
await el_account.click()
await el_account.apply('function (element) {element.value = ""; } ')
await el_account.send_keys(ticketplus_account);
is_account_assigned = True
except Exception as exc:
print(exc)
if is_account_assigned:
try:
my_css_selector = 'input[type="password"]'
el_password = await tab.query_selector(my_css_selector)
if el_password:
print("ticketplus_password:", ticketplus_password)
await el_password.click()
await el_password.apply('function (element) {element.value = ""; } ')
await el_password.send_keys(ticketplus_password);
time.sleep(0.1)
is_filled_form = True
if country_code=="+886":
# only this case to auto sumbmit.
print("press enter")
await tab.send(cdp.input_.dispatch_key_event("keyDown", code="Enter", key="Enter", text="\r", windows_virtual_key_code=13))
await tab.send(cdp.input_.dispatch_key_event("keyUp", code="Enter", key="Enter", text="\r", windows_virtual_key_code=13))
time.sleep(1)
# PS: ticketplus country field may not located at your target country.
is_submited = True
except Exception as exc:
print(exc)
pass
return is_filled_form, is_submited
async def nodriver_ticketplus_is_signin(tab):
is_user_signin = False
try:
cookies = await tab.browser.cookies.get_all()
for cookie in cookies:
if cookie.name=='user':
if '%22account%22:%22' in cookie.value:
is_user_signin = True
cookies = None
except Exception as exc:
print(exc)
pass
return is_user_signin
async def nodriver_ticketplus_account_auto_fill(tab, config_dict):
global is_filled_ticketplus_singin_form
if not 'is_filled_ticketplus_singin_form' in globals():
is_filled_ticketplus_singin_form = False
# auto fill account info.
is_user_signin = False
if len(config_dict["advanced"]["ticketplus_account"]) > 0:
is_user_signin = await nodriver_ticketplus_is_signin(tab)
print("is_user_signin:", is_user_signin)
if not is_user_signin:
time.sleep(0.1)
if not is_filled_ticketplus_singin_form:
is_sign_in_btn_pressed = False
try:
# full screen mode.
my_css_selector = 'button.v-btn > span.v-btn__content > i.mdi-account'
sign_in_btn = await tab.query_selector(my_css_selector)
if sign_in_btn:
await sign_in_btn.click()
is_sign_in_btn_pressed = True
time.sleep(0.2)
except Exception as exc:
print(exc)
pass
print("is_sign_in_btn_pressed", is_sign_in_btn_pressed)
if not is_sign_in_btn_pressed:
#print("rwd mode")
action_btns = None
try:
my_css_selector = 'div.px-4.py-3.drawerItem.cursor-pointer'
action_btns = await tab.query_selector_all(my_css_selector)
except Exception as exc:
print(exc)
pass
if action_btns:
print("len:", len(action_btns))
if len(action_btns) >= 4:
try:
await action_btns[3].click()
except Exception as exc:
print(exc)
pass
is_filled_form, is_submited = await nodriver_ticketplus_account_sign_in(tab, config_dict)
if is_filled_form:
is_filled_ticketplus_singin_form = True
return is_user_signin
async def nodriver_ticketplus_main(tab, url, config_dict, ocr, Captcha_Browser, ticketplus_dict):
home_url = 'https://ticketplus.com.tw/'
is_user_signin = False
if home_url == url.lower():
if config_dict["ocr_captcha"]["enable"]:
domain_name = url.split('/')[2]
if not Captcha_Browser is None:
# TODO:
#Captcha_Browser.Set_cookies(driver.get_cookies())
Captcha_Browser.Set_Domain(domain_name)
is_user_signin = await nodriver_ticketplus_account_auto_fill(tab, config_dict)
if is_user_signin:
# only sign in on homepage.
if url != config_dict["homepage"]:
try:
await tab.get(config_dict["homepage"])
except Exception as e:
pass
# https://ticketplus.com.tw/activity/XXX
if '/activity/' in url.lower():
is_event_page = False
if len(url.split('/'))==5:
is_event_page = True
if is_event_page:
# TODO:
#is_button_pressed = ticketplus_accept_realname_card(driver)
#print("is accept button pressed:", is_button_pressed)
# TODO:
#is_button_pressed = ticketplus_accept_other_activity(driver)
#print("is accept button pressed:", is_button_pressed)
if config_dict["date_auto_select"]["enable"]:
# TODO:
#ticketplus_date_auto_select(driver, config_dict)
pass
#https://ticketplus.com.tw/order/XXX/OOO
if '/order/' in url.lower():
is_event_page = False
if len(url.split('/'))==6:
is_event_page = True
if is_event_page:
# TODO:
#is_button_pressed = ticketplus_accept_realname_card(driver)
#is_button_pressed = ticketplus_accept_order_fail(driver)
is_reloading = False
is_reload_at_webdriver = False
if not config_dict["browser"] in CONST_CHROME_FAMILY:
is_reload_at_webdriver = True
else:
if not config_dict["advanced"]["chrome_extension"]:
is_reload_at_webdriver = True
if is_reload_at_webdriver:
# move below code to chrome extension.
# TODO:
#is_reloading = ticketplus_order_auto_reload_coming_soon(driver)
pass
if not is_reloading:
# TODO:
# is_captcha_sent, ticketplus_dict = ticketplus_order(driver, config_dict, ocr, Captcha_Browser, ticketplus_dict)
pass
else:
ticketplus_dict["fail_list"]=[]
#https://ticketplus.com.tw/confirm/xx/oo
if '/confirm/' in url.lower() or '/confirmseat/' in url.lower():
is_event_page = False
if len(url.split('/'))==6:
is_event_page = True
if is_event_page:
#print("is_popup_confirm",ticketplus_dict["is_popup_confirm"])
if not ticketplus_dict["is_popup_confirm"]:
ticketplus_dict["is_popup_confirm"] = True
play_sound_while_ordering(config_dict)
# TODO:
#ticketplus_confirm(driver, config_dict)
else:
ticketplus_dict["is_popup_confirm"] = False
else:
ticketplus_dict["is_popup_confirm"] = False
return ticketplus_dict
def get_nodriver_browser_args():
browser_args = [
"--user-agent=%s" % (USER_AGENT),
"--disable-2d-canvas-clip-aa",
"--disable-3d-apis",
"--disable-animations",
"--disable-app-info-dialog-mac",
"--disable-background-networking",
"--disable-backgrounding-occluded-windows",
"--disable-bookmark-reordering",
"--disable-boot-animation",
"--disable-breakpad",
"--disable-canvas-aa",
"--disable-client-side-phishing-detection",
"--disable-cloud-import",
"--disable-component-cloud-policy",
"--disable-component-update",
"--disable-composited-antialiasing",
"--disable-default-apps",
"--disable-dev-shm-usage",
"--disable-device-discovery-notifications",
"--disable-dinosaur-easter-egg",
"--disable-domain-reliability",
"--disable-features=IsolateOrigins,site-per-process,TranslateUI",
"--disable-infobars",
"--disable-logging",
"--disable-login-animations",
"--disable-login-screen-apps",
"--disable-notifications",
"--disable-office-editing-component-extension",
"--disable-password-generation",
"--disable-popup-blocking",
"--disable-renderer-backgrounding",
"--disable-session-crashed-bubble",
"--disable-smooth-scrolling",
2024-04-04 22:40:56 +00:00
"--disable-suggestions-ui",
"--disable-sync",
"--disable-translate",
2024-04-04 22:40:56 +00:00
"--hide-crash-restore-bubble",
"--homepage=about:blank",
"--no-default-browser-check",
"--no-first-run",
"--no-pings",
"--no-sandbox"
"--no-service-autorun",
"--password-store=basic",
"--remote-allow-origins=*",
"--lang=zh-TW",
2024-04-04 22:40:56 +00:00
#"--disable-remote-fonts",
]
return browser_args
def get_maxbot_plus_extension_path():
extension_path = "webdriver/Maxbotplus_1.0.0/"
if platform.system() == 'Windows':
extension_path = extension_path.replace("/","\\")
2024-04-06 14:53:42 +00:00
app_root = util.get_app_root()
config_filepath = os.path.join(app_root, extension_path)
#print("config_filepath:", config_filepath)
# check extesion mainfest
path = pathlib.Path(config_filepath)
if path.exists():
if path.is_dir():
#print("found extension dir")
for item in path.rglob("manifest.*"):
path = item.parent
#print("final path:", path)
return config_filepath
2024-04-07 13:24:21 +00:00
def get_maxbot_block_extension_path():
extension_path = "webdriver/Maxblockplus_1.0.0/"
if platform.system() == 'Windows':
extension_path = extension_path.replace("/","\\")
app_root = util.get_app_root()
config_filepath = os.path.join(app_root, extension_path)
#print("config_filepath:", config_filepath)
# check extesion mainfest
path = pathlib.Path(config_filepath)
if path.exists():
if path.is_dir():
#print("found extension dir")
for item in path.rglob("manifest.*"):
path = item.parent
#print("final path:", path)
return config_filepath
def get_extension_config(config_dict):
default_lang = "zh-TW"
2024-04-04 05:58:16 +00:00
no_sandbox=True
conf = Config(browser_args=get_nodriver_browser_args(), lang=default_lang, no_sandbox=no_sandbox, headless=config_dict["advanced"]["headless"])
2024-04-06 14:53:42 +00:00
if config_dict["advanced"]["chrome_extension"]:
conf.add_extension(get_maxbot_plus_extension_path())
2024-04-07 13:24:21 +00:00
conf.add_extension(get_maxbot_block_extension_path())
return conf
async def nodrver_block_urls(tab, config_dict):
NETWORK_BLOCKED_URLS = ['*/adblock.js'
,'*/google_ad_block.js'
,'*google-analytics.*'
,'*googletagmanager.*'
,'*googletagservices.*'
,'*googlesyndication.*'
,'*play.google.com/*'
,'*cdn.cookielaw.org/*'
,'*fundingchoicesmessages.google.com/*'
,'*.doubleclick.net/*'
,'*.rollbar.com/*'
,'*.cloudfront.com/*'
,'*.lndata.com/*'
,'*.twitter.com/i/*'
,'*platform.twitter.com/*'
,'*syndication.twitter.com/*'
,'*youtube.com/*'
,'*player.youku.*'
,'*.clarity.ms/*'
,'*img.uniicreative.com/*'
,'*e2elog.fetnet.net*']
2024-04-06 14:53:42 +00:00
if config_dict["advanced"]["hide_some_image"]:
NETWORK_BLOCKED_URLS.append('*.woff')
NETWORK_BLOCKED_URLS.append('*.woff2')
NETWORK_BLOCKED_URLS.append('*.ttf')
NETWORK_BLOCKED_URLS.append('*.otf')
NETWORK_BLOCKED_URLS.append('*fonts.googleapis.com/earlyaccess/*')
NETWORK_BLOCKED_URLS.append('*/ajax/libs/font-awesome/*')
NETWORK_BLOCKED_URLS.append('*.ico')
NETWORK_BLOCKED_URLS.append('*ticketimg2.azureedge.net/image/ActivityImage/*')
NETWORK_BLOCKED_URLS.append('*static.tixcraft.com/images/activity/*')
NETWORK_BLOCKED_URLS.append('*static.ticketmaster.sg/images/activity/*')
NETWORK_BLOCKED_URLS.append('*static.ticketmaster.com/images/activity/*')
NETWORK_BLOCKED_URLS.append('*ticketimg2.azureedge.net/image/ActivityImage/ActivityImage_*')
NETWORK_BLOCKED_URLS.append('*.azureedge.net/QWARE_TICKET//images/*')
NETWORK_BLOCKED_URLS.append('*static.ticketplus.com.tw/event/*')
NETWORK_BLOCKED_URLS.append('https://kktix.cc/change_locale?locale=*')
NETWORK_BLOCKED_URLS.append('https://t.kfs.io/assets/logo_*.png')
NETWORK_BLOCKED_URLS.append('https://t.kfs.io/assets/icon-*.png')
NETWORK_BLOCKED_URLS.append('https://t.kfs.io/upload_images/*.jpg')
2024-04-06 14:53:42 +00:00
if config_dict["advanced"]["block_facebook_network"]:
NETWORK_BLOCKED_URLS.append('*facebook.com/*')
NETWORK_BLOCKED_URLS.append('*.fbcdn.net/*')
await tab.send(cdp.network.enable())
# set_blocked_ur_ls is author's typo..., waiting author to chagne.
await tab.send(cdp.network.set_blocked_ur_ls(NETWORK_BLOCKED_URLS))
return tab
2024-04-04 05:58:16 +00:00
async def nodriver_resize_window(tab, config_dict):
if len(config_dict["advanced"]["window_size"]) > 0:
if "," in config_dict["advanced"]["window_size"]:
size_array = config_dict["advanced"]["window_size"].split(",")
position_left = 0
if len(size_array) >= 3:
position_left = int(size_array[0]) * int(size_array[2])
#tab = await driver.main_tab()
if tab:
await tab.set_window_size(left=position_left, top=30, width=int(size_array[0]), height=int(size_array[1]))
2024-04-04 12:23:42 +00:00
async def nodriver_current_url(tab):
2024-04-06 14:53:42 +00:00
is_quit_bot = False
exit_bot_error_strings = [
"server rejected WebSocket connection: HTTP 500",
"[Errno 61] Connect call failed ('127.0.0.1',",
]
2024-04-04 12:23:42 +00:00
url = ""
if tab:
url_dict = {}
try:
url_dict = await tab.js_dumps('window.location.href')
except Exception as exc:
print(exc)
2024-04-06 14:53:42 +00:00
str_exc = ""
try:
str_exc = str(exc)
except Exception as exc2:
pass
if len(str_exc) > 0:
for each_error_string in exit_bot_error_strings:
if each_error_string in str_exc:
#print('quit bot by error:', each_error_string, driver)
is_quit_bot = True
2024-04-04 12:23:42 +00:00
url_array = []
if url_dict:
2024-04-06 14:53:42 +00:00
for k in url_dict:
2024-04-04 12:23:42 +00:00
if k.isnumeric():
if "0" in url_dict[k]:
url_array.append(url_dict[k]["0"])
url = ''.join(url_array)
2024-04-06 14:53:42 +00:00
return url, is_quit_bot
2024-04-04 12:23:42 +00:00
2024-04-04 22:40:56 +00:00
def nodriver_overwrite_prefs(conf, prefs_dict={}):
#print(conf.user_data_dir)
prefs_filepath = os.path.join(conf.user_data_dir,"Default")
if not os.path.exists(prefs_filepath):
os.mkdir(prefs_filepath)
prefs_filepath = os.path.join(prefs_filepath,"Preferences")
2024-04-07 13:24:21 +00:00
prefs_dict["in_product_help"]={}
prefs_dict["in_product_help"]["snoozed_feature"]={}
prefs_dict["in_product_help"]["snoozed_feature"]["IPH_LiveCaption"]={}
prefs_dict["in_product_help"]["snoozed_feature"]["IPH_LiveCaption"]["is_dismissed"]=True
prefs_dict["in_product_help"]["snoozed_feature"]["IPH_LiveCaption"]["last_dismissed_by"]=4
prefs_dict["media_router"]={}
prefs_dict["media_router"]["show_cast_sessions_started_by_other_devices"]={}
prefs_dict["media_router"]["show_cast_sessions_started_by_other_devices"]["enabled"]=False
prefs_dict["net"]={}
prefs_dict["net"]["network_prediction_options"]=3
prefs_dict["privacy_guide"]={}
prefs_dict["privacy_guide"]["viewed"]=True
prefs_dict["privacy_sandbox"]={}
prefs_dict["privacy_sandbox"]["first_party_sets_enabled"]=False
2024-04-04 22:40:56 +00:00
prefs_dict["profile"]={}
2024-04-07 13:24:21 +00:00
prefs_dict["profile"]["cookie_controls_mode"]=1
prefs_dict["profile"]["default_content_setting_values"]={}
prefs_dict["profile"]["default_content_setting_values"]["notifications"]=2
prefs_dict["profile"]["default_content_setting_values"]["sound"]=2
2024-04-04 22:40:56 +00:00
prefs_dict["profile"]["name"]=CONST_APP_VERSION
prefs_dict["profile"]["password_manager_enabled"]=False
2024-04-07 13:24:21 +00:00
prefs_dict["safebrowsing"]={}
prefs_dict["safebrowsing"]["enabled"]=False
prefs_dict["safebrowsing"]["enhanced"]=False
prefs_dict["sync"]={}
prefs_dict["sync"]["autofill_wallet_import_enabled_migrated"]=False
2024-04-04 22:40:56 +00:00
json_str = json.dumps(prefs_dict)
with open(prefs_filepath, 'w') as outfile:
outfile.write(json_str)
state_filepath = os.path.join(conf.user_data_dir,"Local State")
state_dict = {}
state_dict["performance_tuning"]={}
state_dict["performance_tuning"]["high_efficiency_mode"]={}
state_dict["performance_tuning"]["high_efficiency_mode"]["state"]=1
state_dict["browser"]={}
state_dict["browser"]["enabled_labs_experiments"]=[
"memory-saver-multi-state-mode@1",
"modal-memory-saver@1"
]
2024-04-07 13:24:21 +00:00
state_dict["dns_over_https"]={}
state_dict["dns_over_https"]["mode"]="off"
2024-04-04 22:40:56 +00:00
json_str = json.dumps(state_dict)
with open(state_filepath, 'w') as outfile:
outfile.write(json_str)
async def main(args):
config_dict = get_config_dict(args)
driver = None
tab = None
if not config_dict is None:
2024-04-04 05:58:16 +00:00
sandbox = False
conf = get_extension_config(config_dict)
2024-04-04 22:40:56 +00:00
prefs = {"credentials_enable_service": False,
"ack_existing_ntp_extensions": False,
"translate":{"enabled": False}}
nodriver_overwrite_prefs(conf, prefs)
2024-04-04 05:58:16 +00:00
# PS: nodrirver run twice always cause error:
# Failed to connect to browser
# One of the causes could be when you are running as root.
# In that case you need to pass no_sandbox=True
#driver = await uc.start(conf, sandbox=sandbox, headless=config_dict["advanced"]["headless"])
driver = await uc.start(conf)
if not driver is None:
tab = await nodriver_goto_homepage(driver, config_dict)
tab = await nodrver_block_urls(tab, config_dict)
if not config_dict["advanced"]["headless"]:
2024-04-04 05:58:16 +00:00
await nodriver_resize_window(tab, config_dict)
else:
print("無法使用nodriver程式無法繼續工作")
sys.exit()
else:
print("Load config error!")
# internal variable. 說明:這是一個內部變數,請略過。
url = ""
last_url = ""
# for tixcraft
tixcraft_dict = {}
tixcraft_dict["fail_list"]=[]
tixcraft_dict["fail_promo_list"]=[]
tixcraft_dict["start_time"]=None
tixcraft_dict["done_time"]=None
tixcraft_dict["elapsed_time"]=None
tixcraft_dict["is_popup_checkout"] = False
tixcraft_dict["area_retry_count"]=0
# for kktix
kktix_dict = {}
kktix_dict["fail_list"]=[]
kktix_dict["start_time"]=None
kktix_dict["done_time"]=None
kktix_dict["elapsed_time"]=None
kktix_dict["is_popup_checkout"] = False
kktix_dict["played_sound_ticket"] = False
kktix_dict["played_sound_order"] = False
fami_dict = {}
fami_dict["fail_list"] = []
fami_dict["last_activity"]=""
ibon_dict = {}
ibon_dict["fail_list"]=[]
ibon_dict["start_time"]=None
ibon_dict["done_time"]=None
ibon_dict["elapsed_time"]=None
hkticketing_dict = {}
hkticketing_dict["is_date_submiting"] = False
hkticketing_dict["fail_list"]=[]
ticketplus_dict = {}
ticketplus_dict["fail_list"]=[]
ticketplus_dict["is_popup_confirm"] = False
ocr = None
Captcha_Browser = None
try:
if config_dict["ocr_captcha"]["enable"]:
ocr = ddddocr.DdddOcr(show_ad=False, beta=config_dict["ocr_captcha"]["beta"])
Captcha_Browser = NonBrowser()
if len(config_dict["advanced"]["tixcraft_sid"]) > 1:
2024-04-06 14:53:42 +00:00
#set_non_browser_cookies(driver, config_dict["homepage"], Captcha_Browser)
pass
except Exception as exc:
print(exc)
pass
maxbot_last_reset_time = time.time()
2024-04-06 14:53:42 +00:00
is_quit_bot = False
while True:
time.sleep(0.05)
# pass if driver not loaded.
if driver is None:
print("nodriver not accessible!")
break
2024-04-06 14:53:42 +00:00
if not is_quit_bot:
url, is_quit_bot = await nodriver_current_url(tab)
if is_quit_bot:
try:
await driver.stop()
driver = None
except Exception as e:
pass
break
if url is None:
continue
else:
if len(url) == 0:
continue
is_maxbot_paused = False
if os.path.exists(CONST_MAXBOT_INT28_FILE):
is_maxbot_paused = True
if len(url) > 0 :
if url != last_url:
print(url)
write_last_url_to_file(url)
if is_maxbot_paused:
print("MAXBOT Paused.")
last_url = url
if is_maxbot_paused:
if 'kktix.c' in url:
kktix_dict = await nodriver_kktix_paused_main(tab, url, config_dict, kktix_dict)
# sleep more when paused.
time.sleep(0.1)
continue
# for kktix.cc and kktix.com
if 'kktix.c' in url:
2024-04-06 14:53:42 +00:00
kktix_dict, is_quit_bot = await nodriver_kktix_main(tab, url, config_dict, kktix_dict)
pass
2024-04-04 22:40:56 +00:00
tixcraft_family = False
if 'tixcraft.com' in url:
tixcraft_family = True
if 'indievox.com' in url:
tixcraft_family = True
if 'ticketmaster.' in url:
tixcraft_family = True
if tixcraft_family:
2024-04-06 14:53:42 +00:00
tixcraft_dict, is_quit_bot = await nodriver_tixcraft_main(driver, url, config_dict, tixcraft_dict, ocr, Captcha_Browser)
2024-04-04 22:40:56 +00:00
pass
if 'famiticket.com' in url:
#fami_dict = famiticket_main(driver, url, config_dict, fami_dict)
pass
if 'ibon.com' in url:
#ibon_dict = ibon_main(driver, url, config_dict, ibon_dict, ocr, Captcha_Browser)
pass
kham_family = False
if 'kham.com.tw' in url:
kham_family = True
if 'ticket.com.tw' in url:
kham_family = True
if 'tickets.udnfunlife.com' in url:
kham_family = True
if kham_family:
#kham_main(driver, url, config_dict, ocr, Captcha_Browser)
pass
if 'ticketplus.com' in url:
2024-04-06 14:53:42 +00:00
ticketplus_dict = await nodriver_ticketplus_main(tab, url, config_dict, ocr, Captcha_Browser, ticketplus_dict)
2024-04-04 22:40:56 +00:00
pass
if 'urbtix.hk' in url:
#urbtix_main(driver, url, config_dict)
pass
if 'cityline.com' in url:
#cityline_main(driver, url, config_dict)
pass
softix_family = False
if 'hkticketing.com' in url:
softix_family = True
if 'galaxymacau.com' in url:
softix_family = True
if 'ticketek.com' in url:
softix_family = True
if softix_family:
#hkticketing_dict = softix_powerweb_main(driver, url, config_dict, hkticketing_dict)
pass
# for facebook
facebook_login_url = 'https://www.facebook.com/login.php?'
if url[:len(facebook_login_url)]==facebook_login_url:
facebook_account = config_dict["advanced"]["facebook_account"].strip()
facebook_password = config_dict["advanced"]["facebook_password_plaintext"].strip()
if facebook_password == "":
facebook_password = util.decryptMe(config_dict["advanced"]["facebook_password"])
if len(facebook_account) > 4:
2024-04-06 14:53:42 +00:00
await nodriver_facebook_login(tab, facebook_account, facebook_password)
2024-04-04 22:40:56 +00:00
def cli():
parser = argparse.ArgumentParser(
description="MaxBot Aggument Parser")
parser.add_argument("--input",
help="config file path",
type=str)
parser.add_argument("--homepage",
help="overwrite homepage setting",
type=str)
parser.add_argument("--ticket_number",
help="overwrite ticket_number setting",
type=int)
parser.add_argument("--tixcraft_sid",
help="overwrite tixcraft sid field",
type=str)
parser.add_argument("--kktix_account",
help="overwrite kktix_account field",
type=str)
parser.add_argument("--kktix_password",
help="overwrite kktix_password field",
type=str)
parser.add_argument("--ibonqware",
help="overwrite ibonqware field",
type=str)
#default="False",
parser.add_argument("--headless",
help="headless mode",
type=str)
parser.add_argument("--browser",
help="overwrite browser setting",
default='',
choices=['chrome','firefox','edge','safari','brave'],
type=str)
2024-04-04 05:58:16 +00:00
parser.add_argument("--window_size",
help="Window size",
type=str)
parser.add_argument("--proxy_server",
help="overwrite proxy server, format: ip:port",
type=str)
args = parser.parse_args()
uc.loop().run_until_complete(main(args))
if __name__ == "__main__":
cli()
2024-04-06 14:53:42 +00:00