ys/GenericAgent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: newtab机制支持 & 相关优化

Browse Source
This commit is contained in:
Liang Jiaqing
2026-02-22 13:19:17 +08:00
parent bf5554e7a8
commit a2c6bd9d90
4 changed files with 54 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
TMWebDriver.py
View File

@@ -39,7 +39,6 @@ class TMWebDriver:
self.sessions, self.results, self.acks = {}, {}, {}
self.default_session_id = None
self.latest_session_id = None
self.last_cmd_time = 0
self.is_remote = socket.socket().connect_ex((host, port+1)) == 0
if not self.is_remote:
self.start_ws_server()
@@ -94,12 +93,11 @@ class TMWebDriver:
session_id = data.get('sessionId')
code = data.get('code')
timeout = float(data.get('timeout', 10.0))
auto_switch_newtab = data.get('auto_switch_newtab', False)
detect_newtab = data.get('detect_newtab', False)
try:
result = self.execute_js(code, timeout=timeout, session_id=session_id, auto_switch_newtab=auto_switch_newtab)
print('[remote result]', str(result)[:500].replace('\n', ' '))
newTabs = result.get('newTabs', []) if isinstance(result, dict) else []
return json.dumps({'result': result, 'newTabs': newTabs}, ensure_ascii=False)
result = self.execute_js(code, timeout=timeout, session_id=session_id, detect_newtab=detect_newtab)
print('[remote result]', str(code)[:50] + ' RESULT:' +str(result)[:50].replace('\n', ' '))
return json.dumps({'r': result}, ensure_ascii=False)
except Exception as e:
return json.dumps({'error': str(e)}, ensure_ascii=False)
return 'ok'
@@ -161,12 +159,7 @@ class TMWebDriver:
print(f"Tab reconnected: {session.url} (Session: {session_id})")
self.latest_session_id = session_id
if self.default_session_id is None:
self.default_session_id = session_id
elif is_new_session:
if time.time() - self.last_cmd_time < 5.0:
print(f"检测到脚本触发的新窗口，自动切换焦点: {session_id}")
self.default_session_id = session_id
if self.default_session_id is None: self.default_session_id = session_id
def _unregister_client(self, client: WebSocket) -> None:
@@ -175,21 +168,15 @@ class TMWebDriver:
session.mark_disconnected()
break
def execute_js(self, code, timeout=15, session_id=None, auto_switch_newtab=False) -> Any:
def execute_js(self, code, timeout=15, session_id=None, detect_newtab=False) -> Any:
if session_id is None: session_id = self.default_session_id
if self.is_remote:
print('remote_execute_js')
response = self._remote_cmd({"cmd": "execute_js", "sessionId": session_id,
"code": code, "timeout": str(timeout),
"auto_switch_newtab": auto_switch_newtab})
"detect_newtab": detect_newtab}).get('r', {})
if response.get('error'): raise Exception(response['error'])
if auto_switch_newtab and 'newTabs' in response:
newtabs = response.get('newTabs', [])
if len(newtabs) > 0:
new_session_id = newtabs[0]['sessionId']
self.default_session_id = new_session_id
print(f"自动切换到新标签会话: {new_session_id}")
return response.get('result', None)
return response
session = self.sessions.get(session_id)
if not session or not session.is_active():
@@ -207,12 +194,10 @@ class TMWebDriver:
tp = session.type
assert tp in ['ws', 'http'], f"Unsupported session type: {tp}"
exec_id = str(uuid.uuid4())
payload = json.dumps({'id': exec_id, 'code': code, 'auto_switch_newtab': auto_switch_newtab})
payload = json.dumps({'id': exec_id, 'code': code, 'detect_newtab': detect_newtab})
if tp == 'ws':
session.ws_client.send_message(payload)
elif tp == 'http':
session.http_queue.put(payload)
if tp == 'ws': session.ws_client.send_message(payload)
elif tp == 'http': session.http_queue.put(payload)
start_time = time.time()
self.clean_sessions()
@@ -225,11 +210,10 @@ class TMWebDriver:
if tp == 'ws':
if not session.is_active(): hasjump = True
if hasjump and session.is_active():
if not self.is_remote and auto_switch_newtab: self.last_cmd_time = time.time()
return {"result": f"Session {session_id} reloaded.", "closed":1}
return {'result': f"Session {session_id} reloaded.", "closed":1}
if time.time() - start_time > timeout:
if tp == 'ws':
if hasjump: return {"result": f"Session {session_id} reloaded and new page is loading...", "closed":1}
if hasjump: return {'result': f"Session {session_id} reloaded and new page is loading...", 'closed':1}
if acked: return {"result": f"No response data in {timeout}s (ACK received, script may still be running)"}
return {"result": f"No response data in {timeout}s (no ACK, script may not have been delivered)"}
elif tp == 'http':
@@ -239,15 +223,10 @@ class TMWebDriver:
result = self.results.pop(exec_id)
if exec_id in self.acks: self.acks.pop(exec_id)
if not result['success']: raise Exception(result['data'])
if not self.is_remote and auto_switch_newtab:
newtabs = result.get('newTabs', [])
if len(newtabs) > 0:
new_session_id = newtabs[0]['sessionId']
self.default_session_id = new_session_id
print(f"自动切换到新标签会话: {new_session_id}")
elif not self.is_remote:
self.last_cmd_time = time.time()
return result['data']
rr = {'data': result['data']}
newtabs = result.get('newTabs', []); [x.pop('ts', None) for x in newtabs]
if newtabs: rr['newTabs'] = newtabs
return rr
def _remote_cmd(self, cmd):
return requests.post(self.remote, headers={"Content-Type": "application/json"}, json=cmd).json()
@@ -259,7 +238,7 @@ class TMWebDriver:
if session.is_active()]
def get_session_dict(self):
return {session.id: session.url for session in self.sessions.values() if session.is_active()}
return {session['id']: session['url'] for session in self.get_all_sessions()}
def find_session(self, url_pattern: str):
if url_pattern == '':
@@ -279,17 +258,14 @@ class TMWebDriver:
matched = self.find_session(url_pattern)
if not matched: return print(f"警告: 未找到URL包含 '{url_pattern}' 的会话")
if len(matched) > 1: print(f"警告: 找到多个URL包含 '{url_pattern}' 的会话，选择第一个")
self.last_cmd_time = 0
self.default_session_id, info = matched[0]
print(f"成功设置默认会话: {self.default_session_id}: {info['url']}")
return self.default_session_id
def jump(self, url, timeout=10): self.execute_js(f"window.location.href='{url}'", timeout=timeout)
def page_source(self): return self.execute_js("document.documentElement.outerHTML")
def body(self): return self.execute_js("document.body.outerHTML")
def newtab(self, url=None):
if url is None: url = "http://www.baidu.com/robots.txt"
return self.execute_js(f'GM_openInTab("{url}");', auto_switch_newtab=True)
return self.execute_js(f'GM_openInTab("{url}");', detect_newtab=True)
if __name__ == "__main__":
driver = TMWebDriver(host='localhost', port=18765)

13
assets/ljq_web_driver.user.js
View File

@@ -1,7 +1,7 @@
// ==UserScript==
// @name ljq_web_driver
// @namespace http://tampermonkey.net/
// @version 0.2
// @version 0.3
// @description Execute JS via ljq_web_driver
// @require https://code.jquery.com/jquery-3.6.0.min.js
// @author You
@@ -38,8 +38,15 @@
}
let ws;
let sid = (window.name && window.name.startsWith('ljq_')) ?
let sid;
if (window.opener && window.name && window.name.startsWith('ljq_')) {
sid = null;
console.log(log_prefix + `检测到opener，丢弃继承的window.name: ${window.name}`);
window.name = '';
} else {
sid = (window.name && window.name.startsWith('ljq_')) ?
window.name : window.sessionStorage.getItem('ljq_driver_sid');
}
if (!sid) {
sid = `ljq_${Date.now().toString().slice(-2)}${Math.random().toString(36).slice(2, 4)}`;
window.sessionStorage.setItem('ljq_driver_sid', sid);
@@ -369,7 +376,7 @@
ws.send(JSON.stringify({type: 'ack',id: data.id}));
let startTime = Date.now();
let newTabs = [];
let checkNewTab = data.auto_switch_newtab === true;
let checkNewTab = data.detect_newtab === true;
GM_setValue('new_tab_report', null);
const response = executeCode(data);

2
ga.py
View File

@@ -158,7 +158,7 @@ def web_execute_js(script, switch_tab_id=None):
"error": error_msg,
"transients": transients,
"environment": {
"new_tab": new_tab,
"newTabs": [],
"reloaded": reloaded
},
"diff": diff_summary,

46
simphtml.py
View File

@@ -802,20 +802,17 @@ def get_temp_texts(driver):
}
stopStrMonitor();
"""
try: return set(driver.execute_js(js))
try: return list(set(driver.execute_js(js).get('data', [])))
except Exception as e:
print(e)
return set()
return []
import time
def get_main_block(driver):
html = driver.execute_js(js_optHTML)
html = driver.execute_js(js_optHTML).get('data', '')
if type(html) is not str:
time.sleep(2)
html = driver.execute_js(js_optHTML)
if type(html) is not str:
print('[STRANGE TYPE]', type(html), str(html)[:500])
return html
html = driver.execute_js(js_optHTML).get('data', '')
return html
@@ -854,7 +851,7 @@ def get_html(driver, cutlist=False, maxchars=28000, instruction=""):
html = str(soup)
if not cutlist or len(html) <= maxchars: return html
rr = driver.execute_js(js_findMainList + js_findMainContent + """
return findMainList(findMainContent(document.body));""")
return findMainList(findMainContent(document.body));""").get('data', {})
sel = rr.get("selector", None) if isinstance(rr, dict) else None
if sel:
s = BeautifulSoup(str(soup), "html.parser"); items = s.select(sel)
@@ -870,37 +867,40 @@ def get_html(driver, cutlist=False, maxchars=28000, instruction=""):
def execute_js_rich(script, driver):
try: start_temp_monitor(driver)
except: pass
curr_session = driver.default_session_id
try: last_html = get_html(driver, cutlist=False)
except: last_html = None
result = None; error_msg = None
new_tab = False; reloaded = False
result = None; error_msg = None; newTabs = []; reloaded = False
before_sids = set(driver.get_session_dict().keys())
try:
print(f"Executing: {script[:250]} ...")
result = driver.execute_js(script, auto_switch_newtab=True)
if type(result) is dict and result.get('closed', 0) == 1: reloaded = True
response = driver.execute_js(script, detect_newtab=True)
result = response.get('data') or response.get('result')
newTabs = response.get('newTabs', [])
if response.get('closed', 0) == 1: reloaded = True
time.sleep(2)
except Exception as e:
error = e.args[0] if e.args else str(e)
if isinstance(error, dict): error.pop('stack', None)
error_msg = str(error)
print(f"Error: {error_msg}")
if driver.default_session_id != curr_session:
print('Session changed')
new_tab = True
rr = {
"status": "failed" if error_msg else "success",
"js_return": result,
"environment": {
"new_tab": new_tab,
"reloaded": reloaded
}
"environment": {"newTabs": newTabs, "reloaded": reloaded}
}
print(reloaded, newTabs)
if reloaded and len(newTabs) == 0:
after = driver.get_session_dict()
new_sids = {k: v for k, v in after.items() if k not in before_sids}
if new_sids:
newTabs = [{'id': k, 'url': v} for k, v in new_sids.items()]
rr['environment']['newTabs'] = newTabs
rr['suggestion'] = "页面已刷新，以上新标签页在执行期间连接。"
if error_msg: rr['error'] = error_msg
if not reloaded:
try: rr['transients'] = get_temp_texts(driver)
except: rr['transients'] = []
if not reloaded and not new_tab:
if not reloaded and len(newTabs) == 0:
try:
current_html = get_html(driver, cutlist=False)
if last_html is None: raise Exception("no baseline")
@@ -910,11 +910,9 @@ def execute_js_rich(script, driver):
diff_summary = f"DOM变化量: {change_count}"
if top_change: diff_summary += f"\n最显著变化:\n{top_change}"
transients = rr.get('transients', [])
if change_count == 0 and not transients and not new_tab:
if change_count == 0 and not transients and len(newTabs) == 0:
diff_summary += " (页面无变化)"
rr['suggestion'] = "页面无明显变化"
else:
rr['suggestion'] = ""
except:
diff_summary = "页面变化监控不可用"
rr['diff'] = diff_summary