| 30 | |
| 31 | |
| 32 | class CDP: |
| 33 | log = logging.getLogger("CDP") |
| 34 | |
| 35 | endpoints = CDPObject( |
| 36 | { |
| 37 | "json": "/json", |
| 38 | "protocol": "/json/protocol", |
| 39 | "list": "/json/list", |
| 40 | "new": "/json/new?{url}", |
| 41 | "activate": "/json/activate/{id}", |
| 42 | "close": "/json/close/{id}", |
| 43 | } |
| 44 | ) |
| 45 | |
| 46 | def __init__(self, options): |
| 47 | self.server_addr = "http://{0}:{1}".format( |
| 48 | *options.debugger_address.split(":") |
| 49 | ) |
| 50 | self._reqid = 0 |
| 51 | self._session = requests.Session() |
| 52 | self._last_resp = None |
| 53 | self._last_json = None |
| 54 | with requests.Session() as session: |
| 55 | resp = session.get( |
| 56 | self.server_addr + self.endpoints.json, |
| 57 | headers={"Connection": "close"}, |
| 58 | timeout=2, |
| 59 | ) |
| 60 | self.sessionId = resp.json()[0]["id"] |
| 61 | self.wsurl = resp.json()[0]["webSocketDebuggerUrl"] |
| 62 | |
| 63 | def tab_activate(self, id=None): |
| 64 | if not id: |
| 65 | active_tab = self.tab_list()[0] |
| 66 | id = active_tab.id |
| 67 | self.wsurl = active_tab.webSocketDebuggerUrl |
| 68 | with requests.Session() as session: |
| 69 | resp = session.post( |
| 70 | self.server_addr + self.endpoints["activate"].format(id=id), |
| 71 | headers={"Connection": "close"}, |
| 72 | timeout=2, |
| 73 | ) |
| 74 | return resp.json() |
| 75 | |
| 76 | def tab_list(self): |
| 77 | with requests.Session() as session: |
| 78 | resp = session.get( |
| 79 | self.server_addr + self.endpoints["list"], |
| 80 | headers={"Connection": "close"}, |
| 81 | timeout=2, |
| 82 | ) |
| 83 | retval = resp.json() |
| 84 | return [PageElement(o) for o in retval] |
| 85 | |
| 86 | def tab_new(self, url): |
| 87 | with requests.Session() as session: |
| 88 | resp = session.post( |
| 89 | self.server_addr + self.endpoints["new"].format(url=url), |