123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169 |
- import time
- import traceback
- import dataset
- from selenium import webdriver
- from selenium.webdriver.support.ui import WebDriverWait
- from selenium.webdriver.common.by import By
- from selenium.webdriver.support import expected_conditions as EC
- from selenium.webdriver.common.keys import Keys
- from celery import Celery
- import redis
- from celery.schedules import crontab
- import celery
- from celery.signals import celeryd_init
- from celery import signals
- import random
- from click import Option
- from selenium import webdriver
- import time
- import sys
- app = Celery('tasks', backend ='redis://172.104.92.245', broker='redis://172.104.92.245')
- qname=None
- custname=None
- app.user_options['preload'].add(Option(('-Z', '--cust'),
- default='default',
- help='Configuration template to use.'))
- @signals.user_preload_options.connect
- def on_preload_parsed(options, **kwargs):
- global custname
- # if options.get_key('template')
- print(options['cust'])
- custname=options['cust']
- # print(options)
- @celeryd_init.connect
- def configure_workers(sender=None, **kwargs):
- global app
- global qname
- worker_name = sender.split("@")[-1]
- # app.conf.update(
- # task_concurrency=1, # Use 4 threads for concurrency
- # worker_prefetch_multiplier=1 # Prefetch one task at a time
- # )
- print("******")
- print(worker_name)
- qname=worker_name
- print("******")
- app.send_task(
- 'cel_single.selenium_jared_click3',
- args=(),
- queue=qname
- )
- # app.conf.beat_schedule = {
- # # Executes every Monday morning at 7:30 a.m.
- # 'add-every-monday-morning': {
- # 'task': 'cel_single.selenium_jared_click3',
- # 'schedule': crontab(minute='*/1'),
- # 'schedule': 65.0,
- # 'args': (),
- # 'options':{'queue': qname}
- # },
- # }
- #@app.on_after_configure.connect
- #def setup_periodic_tasks(sender, **kwargs):
- # print(app.request.delivery_info['routing_key'])
- # print(json_str)
- #sender task.request.hostnam
- # sender.add_periodic_task(100.0, selenium_jared_click.s(), name='add every 2 min')
- @app.task()
- def selenium_jared_click3():
- global custname
- global app
- db = dataset.connect('postgresql://postgres:eyJhbGciOiJI@172.105.241.163:5432/postgres')
- options = webdriver.ChromeOptions()
- options.add_argument("--no-sandbox")
- options.add_argument("--headless")
- options.add_argument("--incognito")
- #options.add_argument('--blink-settings=imagesEnabled=false')
- PROXY=None
- if PROXY is not None:
- options.add_argument('--proxy-server=http://%s' % PROXY)
- mobile_emulation = {
- "deviceMetrics": { "width": 360, "height": 640, "pixelRatio": 3.0 },
- "userAgent": "Mozilla/5.0 (Linux; Android 4.2.1; en-us; Nexus 5 Build/JOP40D) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/18.0.1025.166 Mobile Safari/535.19" }
- try:
- driver = webdriver.Chrome(options=options)
- # driver = webdriver.Chrome(ChromeDriverManager().install(),options=options)
- ## Create Undetected Chromedriver with Options
- # driver = uc.Chrome(options=options)
- # options = uc.ChromeOptions()
- except:
- traceback.print_exc()
- #kw='真理大學國際生'
- #domain='cia.au.edu.tw'
- if custname is None:
- custname='真理'
- cursor=db.query("SELECT cust,plan,prefix,domain,kw,positive FROM public.seo_jobs where cust='"+custname+"' order by random() limit 1")
- kw=None
- domain=None
- for c in cursor:
- cust=c['cust']
- kw=c['kw']
- plan=c['plan']
- prefix=c['prefix']
- domain=eval(c['domain'])[0]
- positive=eval(c['positive'])
- break
- #kw='真理大學教堂'
- #domain='udn.com'
- print(kw)
- print(domain)
- #kw='真理大學校友'
- #kw='真理大學國際生'
- #kw='真理大學張聰聯'
- #domain='au.edu.tw'
- #domain='pronews.tw'
- driver.get('https://www.google.com?num=100')
- time.sleep(3)
- print(driver.current_url)
- elmt = driver.find_element(By.XPATH, "//textarea[@name='q']")
- time.sleep(1)
- elmt.send_keys(kw)
- elmt.send_keys(Keys.ENTER)
- time.sleep(4)
- elmts = driver.find_elements(By.XPATH, "//a[@jsname='UWckNb']")
- numresults=len(elmts)
- print('搜尋結果數量',numresults)
- if numresults<=0:
- print(elmts)
- for elmt in elmts:
- href=elmt.get_attribute('href')
- txt=elmt.text
- # print(txt)
- # print(href)
- if domain in href:
- webdriver.ActionChains(driver).move_to_element(elmt).perform()
- webdriver.ActionChains(driver).move_to_element(elmt).click().perform()
- print(domain)
- print(href)
- app.control.broadcast('shutdown')
- # driver.quit()
- # sys.exit()
- return txt
- break
- app.control.broadcast('shutdown')
- # driver.quit()
- # sys.exit()
- return '{empty}'
- # time.sleep(5)
|