kw_tools/similar_server/sel_server.py

from datetime import datetime
import os
import sys
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait, Select
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.remote.webdriver import WebDriver
import time
import rpyc
from rpyc.utils.server import ThreadedServer # or ForkingServer
from bs4 import BeautifulSoup
import json



def send(driver, cmd, params={}):
    resource = "/session/%s/chromium/send_command_and_get_result" % driver.session_id
    url = driver.command_executor._url + resource
    body = json.dumps({'cmd': cmd, 'params': params})
    response = driver.command_executor._request('POST', url, body)
#    if response['status']:
#        raise Exception(response.get('value'))
    return response.get('value')

def add_script(driver, script):
    send(driver, "Page.addScriptToEvaluateOnNewDocument", {"source": script})

def init_webdriver():
    WebDriver.add_script = add_script
    dir_path = os.path.dirname(os.path.realpath(__file__))
    options = webdriver.ChromeOptions()
    driver = webdriver.Chrome(  chrome_options=options)
    return driver

#global driver
class MyService(rpyc.Service):
    def process(self,url):
        self.driver.add_script('const setProperty = () => {     Object.defineProperty(navigator, "webdriver", {       get: () => false,     }); }; setProperty();')
        print('add url.............')
        print(self.driver)

    def __init__(self):
        self.driver = None
        try:
            self.driver = init_webdriver()
            print(self.driver )
        except Exception as e:
            raise e
        finally:
            True
        pass
    
    def exposed_get_url(self,url):
        self.driver.add_script('const setProperty = () => {     Object.defineProperty(navigator, "webdriver", {       get: () => false,     }); }; setProperty();')
        print('add url.............')
        print(self.driver)
        self.driver.get(url)
        
#        txt=self.driver.find_element_by_xpath("//div[@id='json']").text
        print(self.driver.page_source)
        soup = BeautifulSoup(self.driver.page_source,features="lxml")
        dict_from_json = json.loads(soup.find("body").text)
        return dict_from_json
#        return json.loads(self.driver.page_source)



if __name__ == "__main__":
    server = ThreadedServer(MyService(), port = 12345,protocol_config={
    'allow_public_attrs': True,
})
    server.start()