Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
918ac92230 | ||
|
|
c3d07f7325 |
+43
@@ -0,0 +1,43 @@
|
||||
# Python
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
*.so
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
|
||||
# Virtual Environment
|
||||
venv/
|
||||
env/
|
||||
ENV/
|
||||
.venv
|
||||
|
||||
# IDE
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
*.swo
|
||||
*~
|
||||
|
||||
# Logs
|
||||
*.log
|
||||
log/
|
||||
|
||||
# Data files (optional - uncomment if you don't want to commit these)
|
||||
# cookie_data.json
|
||||
# cookies.json
|
||||
# config.json
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+189
-72
@@ -71,29 +71,34 @@ class BrowserLogin:
|
||||
parsed = urlparse(url)
|
||||
return parsed.netloc
|
||||
|
||||
def _get_cookies_dict(self) -> Dict[str, List[Dict]]:
|
||||
def get_all_cookies(self) -> List[Dict]:
|
||||
"""
|
||||
获取当前浏览器的所有 cookie,按域名分组
|
||||
获取当前浏览器实例的所有 cookie
|
||||
|
||||
Returns:
|
||||
按域名分组的 cookie 字典
|
||||
所有 cookie 的列表
|
||||
"""
|
||||
cookies = self.tab.cookies.as_dict
|
||||
# 先获取 cookies 对象,然后调用 as_dict 方法
|
||||
cookies_obj = self.tab.cookies
|
||||
cookies = cookies_obj() if callable(cookies_obj) else cookies_obj
|
||||
cookies = cookies.as_dict() if hasattr(cookies, 'as_dict') else cookies
|
||||
|
||||
# 按域名分组
|
||||
cookie_dict = {}
|
||||
# 确保返回完整的 cookie 列表,包含所有属性
|
||||
formatted_cookies = []
|
||||
for cookie in cookies:
|
||||
domain = cookie.get('domain', '')
|
||||
# 统一域名格式(去掉开头的点)
|
||||
if domain.startswith('.'):
|
||||
domain = domain[1:]
|
||||
|
||||
if domain not in cookie_dict:
|
||||
cookie_dict[domain] = []
|
||||
|
||||
cookie_dict[domain].append(cookie)
|
||||
formatted_cookie = {
|
||||
'name': cookie.get('name', ''),
|
||||
'value': cookie.get('value', ''),
|
||||
'domain': cookie.get('domain', ''),
|
||||
'path': cookie.get('path', '/'),
|
||||
'expires': cookie.get('expirationDate', None),
|
||||
'secure': cookie.get('secure', False),
|
||||
'httpOnly': cookie.get('httpOnly', False),
|
||||
'sameSite': cookie.get('sameSite', 'Lax')
|
||||
}
|
||||
formatted_cookies.append(formatted_cookie)
|
||||
|
||||
return cookie_dict
|
||||
return formatted_cookies
|
||||
|
||||
def check_login(self, url: str, check_selector: str, success_text: str) -> bool:
|
||||
"""
|
||||
@@ -146,8 +151,11 @@ class BrowserLogin:
|
||||
fail_count = 0
|
||||
|
||||
logger.info(f"开始导入 {len(cookies)} 个 cookie")
|
||||
for cookie in cookies:
|
||||
try:
|
||||
|
||||
try:
|
||||
# 准备所有 cookie 数据
|
||||
formatted_cookies = []
|
||||
for cookie in cookies:
|
||||
cookie_dict = {
|
||||
'name': cookie.get('name', ''),
|
||||
'value': cookie.get('value', ''),
|
||||
@@ -160,11 +168,23 @@ class BrowserLogin:
|
||||
cookie_dict['httpOnly'] = True
|
||||
if cookie.get('sameSite'):
|
||||
cookie_dict['sameSite'] = cookie.get('sameSite')
|
||||
|
||||
self.tab.set.cookies(cookie_dict)
|
||||
success_count += 1
|
||||
except Exception as e:
|
||||
fail_count += 1
|
||||
formatted_cookies.append(cookie_dict)
|
||||
|
||||
# 根据 DrissionPage 文档,set.cookies() 方法可以直接接收列表格式的多个 cookie
|
||||
logger.info("使用一次性注入方式导入 cookie")
|
||||
self.tab.set.cookies(formatted_cookies)
|
||||
success_count = len(formatted_cookies)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"导入 cookie 失败: {e}")
|
||||
# 回退到逐个注入
|
||||
logger.info("回退到逐个注入方式导入 cookie")
|
||||
for cookie_dict in formatted_cookies:
|
||||
try:
|
||||
self.tab.set.cookies(cookie_dict)
|
||||
success_count += 1
|
||||
except Exception as e:
|
||||
fail_count += 1
|
||||
|
||||
logger.info(f"Cookie 导入完成: 成功 {success_count}, 失败 {fail_count}")
|
||||
return fail_count == 0
|
||||
@@ -309,9 +329,19 @@ class BrowserLogin:
|
||||
time.sleep(3)
|
||||
|
||||
# 获取 cookies
|
||||
cookies = self.tab.cookies.as_dict
|
||||
cookies_obj = self.browser.cookies
|
||||
cookies = cookies_obj() if callable(cookies_obj) else cookies_obj
|
||||
cookies = cookies.as_dict() if hasattr(cookies, 'as_dict') else cookies
|
||||
|
||||
return cookies
|
||||
# 确保返回的是列表
|
||||
if isinstance(cookies, list):
|
||||
return cookies
|
||||
elif isinstance(cookies, dict):
|
||||
# 如果是字典,转换为列表
|
||||
return [cookies]
|
||||
else:
|
||||
# 如果是其他类型,返回空列表
|
||||
return []
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"刷新页面并保存 cookie 失败: {e}")
|
||||
@@ -323,7 +353,7 @@ class BrowserLogin:
|
||||
|
||||
|
||||
class CookieManager:
|
||||
"""Cookie 管理器"""
|
||||
"""Cookie 管理器 - 实现浏览器实例级别的 Cookie 共享"""
|
||||
|
||||
def __init__(self, cookie_file: str):
|
||||
"""
|
||||
@@ -335,14 +365,25 @@ class CookieManager:
|
||||
self.cookie_file = cookie_file
|
||||
self.cookies = self._load_cookies()
|
||||
|
||||
def _load_cookies(self) -> Dict[str, Dict[str, List[Dict]]]:
|
||||
def _load_cookies(self) -> Dict[str, List[Dict]]:
|
||||
"""加载 cookie 文件"""
|
||||
import os
|
||||
|
||||
if os.path.exists(self.cookie_file):
|
||||
try:
|
||||
with open(self.cookie_file, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
data = json.load(f)
|
||||
# 兼容旧格式
|
||||
if isinstance(data, dict) and any(isinstance(v, dict) for v in data.values()):
|
||||
# 转换旧格式为新格式
|
||||
new_data = {}
|
||||
for user, websites in data.items():
|
||||
all_cookies = []
|
||||
for site_cookies in websites.values():
|
||||
all_cookies.extend(site_cookies)
|
||||
new_data[user] = all_cookies
|
||||
return new_data
|
||||
return data
|
||||
except Exception as e:
|
||||
logger.error(f"加载 cookie 文件失败: {e}")
|
||||
|
||||
@@ -356,76 +397,152 @@ class CookieManager:
|
||||
except Exception as e:
|
||||
logger.error(f"保存 cookie 文件失败: {e}")
|
||||
|
||||
def get_cookies(self, user_name: str, website_name: str) -> List[Dict]:
|
||||
def get_all_cookies(self, user_name: str) -> List[Dict]:
|
||||
"""
|
||||
获取指定用户和网站的 cookie
|
||||
获取指定用户的所有 cookie
|
||||
|
||||
Args:
|
||||
user_name: 用户名
|
||||
website_name: 网站名
|
||||
|
||||
Returns:
|
||||
cookie 列表
|
||||
"""
|
||||
if user_name not in self.cookies:
|
||||
return []
|
||||
|
||||
if website_name not in self.cookies[user_name]:
|
||||
return []
|
||||
|
||||
return self.cookies[user_name][website_name]
|
||||
return self.cookies.get(user_name, [])
|
||||
|
||||
def save_cookies(self, user_name: str, website_name: str, cookies: List[Dict]):
|
||||
def save_all_cookies(self, user_name: str, cookies: List[Dict]):
|
||||
"""
|
||||
保存指定用户和网站的 cookie
|
||||
保存指定用户的所有 cookie
|
||||
|
||||
Args:
|
||||
user_name: 用户名
|
||||
website_name: 网站名
|
||||
cookies: cookie 列表
|
||||
"""
|
||||
if user_name not in self.cookies:
|
||||
self.cookies[user_name] = {}
|
||||
|
||||
self.cookies[user_name][website_name] = cookies
|
||||
self.cookies[user_name] = cookies
|
||||
self._save_cookies()
|
||||
|
||||
def get_cookies_for_domain(self, domain: str) -> List[Dict]:
|
||||
def save_cookies(self, user_name: str, website_name: str, cookies: List[Dict]):
|
||||
"""
|
||||
获取指定域名的 cookie(从所有用户的 cookie 中查找)
|
||||
保存指定用户和网站的 cookie(兼容旧接口)
|
||||
|
||||
Args:
|
||||
user_name: 用户名
|
||||
website_name: 网站名(未使用,保持兼容)
|
||||
cookies: cookie 列表
|
||||
"""
|
||||
# 保持向后兼容,将网站的cookie添加到用户的所有cookie中
|
||||
if user_name not in self.cookies:
|
||||
self.cookies[user_name] = []
|
||||
|
||||
# 合并新cookie到现有cookie中
|
||||
existing_cookies = self.cookies[user_name]
|
||||
|
||||
# 创建现有cookie的索引,用于快速查找
|
||||
cookie_index = {}
|
||||
for i, cookie in enumerate(existing_cookies):
|
||||
if isinstance(cookie, dict):
|
||||
key = (cookie.get('name'), cookie.get('domain'), cookie.get('path'))
|
||||
cookie_index[key] = i
|
||||
|
||||
# 添加或更新cookie
|
||||
if isinstance(cookies, list):
|
||||
valid_cookies = []
|
||||
for cookie in cookies:
|
||||
if isinstance(cookie, dict):
|
||||
valid_cookies.append(cookie)
|
||||
|
||||
for cookie in valid_cookies:
|
||||
key = (cookie.get('name'), cookie.get('domain'), cookie.get('path'))
|
||||
if key in cookie_index:
|
||||
# 更新现有cookie
|
||||
existing_cookies[cookie_index[key]] = cookie
|
||||
else:
|
||||
# 添加新cookie
|
||||
existing_cookies.append(cookie)
|
||||
|
||||
if valid_cookies:
|
||||
logger.info(f" 已保存 {len(valid_cookies)} 个新 cookie")
|
||||
|
||||
self._save_cookies()
|
||||
|
||||
def add_cookie(self, user_name: str, cookie: Dict):
|
||||
"""
|
||||
为指定用户添加单个 cookie
|
||||
|
||||
Args:
|
||||
user_name: 用户名
|
||||
cookie: cookie 字典
|
||||
"""
|
||||
if user_name not in self.cookies:
|
||||
self.cookies[user_name] = []
|
||||
|
||||
# 检查是否已存在相同的 cookie
|
||||
existing_index = -1
|
||||
for i, existing_cookie in enumerate(self.cookies[user_name]):
|
||||
if (existing_cookie.get('name') == cookie.get('name') and
|
||||
existing_cookie.get('domain') == cookie.get('domain') and
|
||||
existing_cookie.get('path') == cookie.get('path')):
|
||||
existing_index = i
|
||||
break
|
||||
|
||||
if existing_index >= 0:
|
||||
# 更新现有 cookie
|
||||
self.cookies[user_name][existing_index] = cookie
|
||||
else:
|
||||
# 添加新 cookie
|
||||
self.cookies[user_name].append(cookie)
|
||||
|
||||
self._save_cookies()
|
||||
|
||||
def remove_cookie(self, user_name: str, cookie_name: str, domain: str):
|
||||
"""
|
||||
从指定用户中移除指定的 cookie
|
||||
|
||||
Args:
|
||||
user_name: 用户名
|
||||
cookie_name: cookie 名称
|
||||
domain: cookie 域名
|
||||
"""
|
||||
if user_name in self.cookies:
|
||||
self.cookies[user_name] = [
|
||||
cookie for cookie in self.cookies[user_name]
|
||||
if not (cookie.get('name') == cookie_name and cookie.get('domain') == domain)
|
||||
]
|
||||
self._save_cookies()
|
||||
|
||||
def clear_cookies(self, user_name: str):
|
||||
"""
|
||||
清空指定用户的所有 cookie
|
||||
|
||||
Args:
|
||||
user_name: 用户名
|
||||
"""
|
||||
if user_name in self.cookies:
|
||||
self.cookies[user_name] = []
|
||||
self._save_cookies()
|
||||
|
||||
def get_cookies_for_domain(self, user_name: str, domain: str) -> List[Dict]:
|
||||
"""
|
||||
获取指定用户的指定域名的 cookie
|
||||
|
||||
Args:
|
||||
user_name: 用户名
|
||||
domain: 域名
|
||||
|
||||
Returns:
|
||||
cookie 列表
|
||||
"""
|
||||
all_cookies = []
|
||||
user_cookies = self.get_all_cookies(user_name)
|
||||
filtered_cookies = []
|
||||
|
||||
for user_name, websites in self.cookies.items():
|
||||
for website_name, cookies in websites.items():
|
||||
for cookie in cookies:
|
||||
cookie_domain = cookie.get('domain', '')
|
||||
# 统一域名格式
|
||||
if cookie_domain.startswith('.'):
|
||||
cookie_domain = cookie_domain[1:]
|
||||
|
||||
if domain in cookie_domain or cookie_domain in domain:
|
||||
all_cookies.append(cookie)
|
||||
for cookie in user_cookies:
|
||||
cookie_domain = cookie.get('domain', '')
|
||||
# 统一域名格式
|
||||
if cookie_domain.startswith('.'):
|
||||
cookie_domain = cookie_domain[1:]
|
||||
|
||||
if domain in cookie_domain or cookie_domain in domain:
|
||||
filtered_cookies.append(cookie)
|
||||
|
||||
return all_cookies
|
||||
return filtered_cookies
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 测试代码
|
||||
manager = CookieManager("cookies.json")
|
||||
|
||||
# 保存测试 cookie
|
||||
test_cookies = [
|
||||
{"name": "test", "value": "123", "domain": "example.com", "path": "/"}
|
||||
]
|
||||
manager.save_cookies("测试用户", "测试网站", test_cookies)
|
||||
|
||||
# 获取 cookie
|
||||
cookies = manager.get_cookies("测试用户", "测试网站")
|
||||
logger.info(f"获取到的 cookie: {cookies}")
|
||||
+179
@@ -0,0 +1,179 @@
|
||||
import tkinter as tk
|
||||
from tkinter import ttk, messagebox
|
||||
from DrissionPage import ChromiumPage, ChromiumOptions
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
|
||||
class BrowserTool:
|
||||
def __init__(self, root):
|
||||
self.root = root
|
||||
self.root.title('浏览器Cookie管理工具')
|
||||
self.root.geometry('500x200')
|
||||
self.root.resizable(False, False)
|
||||
|
||||
self.page = None
|
||||
self.cookie_file = 'd:\\test\\openUrl\\cookies.json'
|
||||
|
||||
self.setup_ui()
|
||||
|
||||
def setup_ui(self):
|
||||
frame = ttk.Frame(self.root, padding=20)
|
||||
frame.pack(fill=tk.BOTH, expand=True)
|
||||
|
||||
ttk.Label(frame, text='网址:').grid(row=0, column=0, sticky=tk.W, pady=5)
|
||||
self.url_entry = ttk.Entry(frame, width=60)
|
||||
self.url_entry.grid(row=0, column=1, columnspan=3, pady=5, sticky=tk.W)
|
||||
self.url_entry.insert(0, 'https://yunlu.sinopec.com/AddressBook/AddressBookCheck')
|
||||
|
||||
btn_frame = ttk.Frame(frame)
|
||||
btn_frame.grid(row=1, column=0, columnspan=4, pady=20)
|
||||
|
||||
self.open_btn = ttk.Button(btn_frame, text='打开', width=15, command=self.open_browser)
|
||||
self.open_btn.pack(side=tk.LEFT, padx=10)
|
||||
|
||||
self.save_btn = ttk.Button(btn_frame, text='保存', width=15, command=self.save_cookies)
|
||||
self.save_btn.pack(side=tk.LEFT, padx=10)
|
||||
|
||||
self.import_btn = ttk.Button(btn_frame, text='导入', width=15, command=self.import_cookies)
|
||||
self.import_btn.pack(side=tk.LEFT, padx=10)
|
||||
|
||||
self.status_var = tk.StringVar(value='就绪')
|
||||
ttk.Label(frame, textvariable=self.status_var).grid(row=2, column=0, columnspan=4, pady=10)
|
||||
|
||||
def log(self, msg):
|
||||
self.status_var.set(msg)
|
||||
self.root.update()
|
||||
|
||||
def init_browser(self):
|
||||
co = ChromiumOptions()
|
||||
co.set_browser_path(r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe')
|
||||
self.page = ChromiumPage(addr_or_opts=co)
|
||||
|
||||
def wait_page_loaded(self, timeout=30):
|
||||
if self.page:
|
||||
self.page.wait.doc_loaded(timeout=timeout)
|
||||
self.page.ele('tag:body', timeout=timeout)
|
||||
|
||||
def open_browser(self):
|
||||
def run():
|
||||
try:
|
||||
self.log('正在启动浏览器...')
|
||||
self.init_browser()
|
||||
|
||||
url = self.url_entry.get().strip()
|
||||
if not url:
|
||||
messagebox.showerror('错误', '请输入网址')
|
||||
return
|
||||
|
||||
self.log(f'正在打开: {url}')
|
||||
self.page.get(url)
|
||||
self.wait_page_loaded()
|
||||
self.log('页面加载完成')
|
||||
|
||||
except Exception as e:
|
||||
self.log(f'错误: {str(e)}')
|
||||
messagebox.showerror('错误', str(e))
|
||||
|
||||
threading.Thread(target=run, daemon=True).start()
|
||||
|
||||
def save_cookies(self):
|
||||
def run():
|
||||
try:
|
||||
if not self.page:
|
||||
messagebox.showwarning('提示', '请先打开浏览器')
|
||||
return
|
||||
|
||||
self.log('正在保存Cookies...')
|
||||
cookies = self.page.cookies()
|
||||
|
||||
with open(self.cookie_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(cookies, f, ensure_ascii=False, indent=2)
|
||||
|
||||
self.log(f'Cookies已保存到 {self.cookie_file}')
|
||||
|
||||
self.log('正在关闭浏览器...')
|
||||
self.page.quit()
|
||||
self.page = None
|
||||
self.log('浏览器已关闭')
|
||||
|
||||
messagebox.showinfo('成功', 'Cookies保存成功,浏览器已关闭')
|
||||
|
||||
except Exception as e:
|
||||
self.log(f'错误: {str(e)}')
|
||||
messagebox.showerror('错误', str(e))
|
||||
|
||||
threading.Thread(target=run, daemon=True).start()
|
||||
|
||||
def import_cookies(self):
|
||||
def run():
|
||||
try:
|
||||
import os
|
||||
if not os.path.exists(self.cookie_file):
|
||||
messagebox.showwarning('提示', 'Cookie文件不存在,请先保存Cookies')
|
||||
return
|
||||
|
||||
self.log('正在启动浏览器...')
|
||||
self.init_browser()
|
||||
|
||||
self.log('正在加载Cookies...')
|
||||
with open(self.cookie_file, 'r', encoding='utf-8') as f:
|
||||
cookies = json.load(f)
|
||||
|
||||
url = self.url_entry.get().strip()
|
||||
if not url:
|
||||
messagebox.showerror('错误', '请输入网址')
|
||||
return
|
||||
|
||||
from urllib.parse import urlparse
|
||||
parsed = urlparse(url)
|
||||
domain = parsed.netloc
|
||||
|
||||
self.page.get('about:blank')
|
||||
|
||||
for cookie in cookies:
|
||||
cookie_dict = {
|
||||
'name': cookie.get('name'),
|
||||
'value': cookie.get('value'),
|
||||
'domain': cookie.get('domain', domain),
|
||||
'path': cookie.get('path', '/')
|
||||
}
|
||||
try:
|
||||
self.page.set.cookies(cookie_dict)
|
||||
except Exception as ce:
|
||||
self.log(f'设置Cookie失败: {cookie.get("name")} - {str(ce)}')
|
||||
|
||||
self.log(f'已导入 {len(cookies)} 个Cookie')
|
||||
|
||||
self.log(f'正在打开: {url}')
|
||||
self.page.get(url)
|
||||
self.wait_page_loaded()
|
||||
|
||||
self.log('刷新页面验证登录状态...')
|
||||
for i in range(3):
|
||||
self.page.refresh()
|
||||
self.wait_page_loaded()
|
||||
time.sleep(1)
|
||||
self.log(f'刷新第 {i+1} 次...')
|
||||
|
||||
self.log('页面加载完成')
|
||||
|
||||
except Exception as e:
|
||||
self.log(f'错误: {str(e)}')
|
||||
messagebox.showerror('错误', str(e))
|
||||
|
||||
threading.Thread(target=run, daemon=True).start()
|
||||
|
||||
def on_closing(self):
|
||||
if self.page:
|
||||
try:
|
||||
self.page.quit()
|
||||
except:
|
||||
pass
|
||||
self.root.destroy()
|
||||
|
||||
if __name__ == '__main__':
|
||||
root = tk.Tk()
|
||||
app = BrowserTool(root)
|
||||
root.protocol('WM_DELETE_WINDOW', app.on_closing)
|
||||
root.mainloop()
|
||||
+1
-1
@@ -20,7 +20,7 @@
|
||||
"name": "网站A",
|
||||
"url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll",
|
||||
"login_check_selector": "",
|
||||
"success_text": ""
|
||||
"success_text": "彭峰"
|
||||
}
|
||||
]
|
||||
},
|
||||
|
||||
+105
-23
@@ -1,4 +1,4 @@
|
||||
"""Cookie Cloud 模块 - 从 Cookie Cloud 服务器获取 cookie 数据"""
|
||||
"""Cookie Cloud 模块 - 从 Cookie Cloud 服务器获取并处理 cookie 数据"""
|
||||
import json
|
||||
import hashlib
|
||||
import base64
|
||||
@@ -6,11 +6,21 @@ from typing import Dict, List, Optional, Tuple
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from Crypto.Cipher import AES
|
||||
from dataclasses import dataclass
|
||||
from applogger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CookieCloudConfig:
|
||||
"""CookieCloud 配置"""
|
||||
api_url: str
|
||||
uuid: str
|
||||
password: str
|
||||
timeout: int = 30
|
||||
|
||||
|
||||
class CookieCloud:
|
||||
"""Cookie Cloud 客户端"""
|
||||
|
||||
@@ -23,13 +33,16 @@ class CookieCloud:
|
||||
uuid: 用户 UUID
|
||||
password: 用户密码
|
||||
"""
|
||||
self.api_url = api_url.rstrip('/')
|
||||
self.uuid = uuid
|
||||
self.password = password
|
||||
self.config = CookieCloudConfig(
|
||||
api_url=api_url.rstrip('/'),
|
||||
uuid=uuid,
|
||||
password=password,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
def _get_crypt_key(self) -> bytes:
|
||||
"""生成加密密钥"""
|
||||
combined_string = f"{self.uuid}-{self.password}"
|
||||
combined_string = f"{self.config.uuid}-{self.config.password}"
|
||||
return hashlib.md5(combined_string.encode('utf-8')).hexdigest()[:16].encode("utf-8")
|
||||
|
||||
def _bytes_to_key(self, data: bytes, salt: bytes, output: int = 48) -> bytes:
|
||||
@@ -58,10 +71,10 @@ class CookieCloud:
|
||||
padding_length = ord(padding_length)
|
||||
return decrypted_padded[:-padding_length]
|
||||
|
||||
def _download_data(self) -> Tuple[Optional[Dict], str]:
|
||||
"""下载并解密所有数据"""
|
||||
def _download_all(self) -> Tuple[Optional[Dict], str]:
|
||||
"""下载所有cookie和local storage数据"""
|
||||
try:
|
||||
url = f"{self.api_url}/get/{self.uuid}"
|
||||
url = f"{self.config.api_url}/get/{self.config.uuid}"
|
||||
request = urllib.request.Request(
|
||||
url,
|
||||
headers={
|
||||
@@ -71,7 +84,7 @@ class CookieCloud:
|
||||
method='GET'
|
||||
)
|
||||
|
||||
response = urllib.request.urlopen(request, timeout=30)
|
||||
response = urllib.request.urlopen(request, timeout=self.config.timeout)
|
||||
|
||||
if response.status != 200:
|
||||
return None, f"服务器返回错误状态码: {response.status}"
|
||||
@@ -98,7 +111,10 @@ class CookieCloud:
|
||||
return result, ""
|
||||
|
||||
except urllib.error.HTTPError as e:
|
||||
return None, f"HTTP错误: {e.code} {e.reason}"
|
||||
if e.code == 401:
|
||||
return None, "认证失败,请检查用户名和密码"
|
||||
else:
|
||||
return None, f"HTTP错误: {e.code} {e.reason}"
|
||||
except urllib.error.URLError as e:
|
||||
return None, f"网络连接失败: {e.reason}"
|
||||
except Exception as e:
|
||||
@@ -112,7 +128,7 @@ class CookieCloud:
|
||||
按域名分组的 cookie 数据字典
|
||||
格式: {"domain.com": [{"name": "cookie1", "value": "val1", ...}, ...], ...}
|
||||
"""
|
||||
data, error = self._download_data()
|
||||
data, error = self._download_all()
|
||||
|
||||
if error:
|
||||
logger.error(error)
|
||||
@@ -139,8 +155,32 @@ class CookieCloud:
|
||||
cookie['sameSite'] = 'Lax'
|
||||
processed_cookies[domain].append(cookie)
|
||||
|
||||
logger.info(f"获取到 {len(processed_cookies)} 个域名的 cookies")
|
||||
return processed_cookies
|
||||
|
||||
def get_local_storage(self) -> Dict[str, Dict]:
|
||||
"""
|
||||
从 Cookie Cloud 服务器获取 local storage 数据
|
||||
|
||||
Returns:
|
||||
按域名分组的 local storage 数据字典
|
||||
格式: {"domain.com": {"key1": "value1", ...}, ...}
|
||||
"""
|
||||
data, error = self._download_all()
|
||||
|
||||
if error:
|
||||
logger.error(error)
|
||||
return {}
|
||||
|
||||
# 处理数据结构
|
||||
local_storage_data = {}
|
||||
|
||||
if isinstance(data, dict) and data.get('local_storage_data'):
|
||||
local_storage_data = data.get('local_storage_data', {})
|
||||
|
||||
logger.info(f"获取到 {len(local_storage_data)} 个域名的 local storage")
|
||||
return local_storage_data
|
||||
|
||||
def get_cookies_for_domain(self, domain: str) -> List[Dict]:
|
||||
"""
|
||||
获取指定域名的 cookie
|
||||
@@ -159,18 +199,60 @@ class CookieCloud:
|
||||
if domain in clean_domain or clean_domain in domain:
|
||||
matched_cookies.extend(cookies)
|
||||
|
||||
logger.info(f"获取到 {len(matched_cookies)} 个 {domain} 的 cookies")
|
||||
return matched_cookies
|
||||
|
||||
def get_local_storage_for_domain(self, domain: str) -> Dict:
|
||||
"""
|
||||
获取指定域名的 local storage
|
||||
|
||||
Args:
|
||||
domain: 域名
|
||||
|
||||
Returns:
|
||||
local storage 字典
|
||||
"""
|
||||
all_local_storage = self.get_local_storage()
|
||||
|
||||
for ls_domain, data in all_local_storage.items():
|
||||
clean_domain = ls_domain.lstrip('.')
|
||||
if domain in clean_domain or clean_domain in domain:
|
||||
logger.info(f"获取到 {domain} 的 local storage")
|
||||
return data
|
||||
|
||||
return {}
|
||||
|
||||
def get_all_data(self) -> Dict:
|
||||
"""
|
||||
获取所有数据(包括 cookies 和 local storage)
|
||||
|
||||
Returns:
|
||||
包含所有数据的字典
|
||||
"""
|
||||
data, error = self._download_all()
|
||||
|
||||
if error:
|
||||
logger.error(error)
|
||||
return {}
|
||||
|
||||
result = {
|
||||
'cookies': {},
|
||||
'local_storage': {}
|
||||
}
|
||||
|
||||
# 处理 cookies
|
||||
if isinstance(data, dict):
|
||||
if not data.get('cookie_data'):
|
||||
result['cookies'] = data
|
||||
elif data.get('cookie_data'):
|
||||
result['cookies'] = data.get('cookie_data', {})
|
||||
|
||||
# 处理 local storage
|
||||
if data.get('local_storage_data'):
|
||||
result['local_storage'] = data.get('local_storage_data', {})
|
||||
|
||||
logger.info(f"获取到所有数据: {len(result['cookies'])} 个域名的 cookies, {len(result['local_storage'])} 个域名的 local storage")
|
||||
return result
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
cc = CookieCloud(
|
||||
api_url="https://movie-pilot.org/cookiecloud",
|
||||
uuid="hu6n2vcUmzpu7mqUN2rVCg",
|
||||
password="jtDM5dV9AyqXkZdQVeA9f6"
|
||||
)
|
||||
|
||||
cookies = cc.get_cookies()
|
||||
logger.info(f"获取到 {len(cookies)} 个域名的 cookies")
|
||||
|
||||
lmkbi_cookies = cc.get_cookies_for_domain('lmkbi.95155.com')
|
||||
logger.info(f"获取到 {len(lmkbi_cookies)} 个 lmkbi.95155.com 的 cookies")
|
||||
|
||||
@@ -0,0 +1,160 @@
|
||||
{
|
||||
"cookies": [
|
||||
{
|
||||
"name": "JSESSIONID",
|
||||
"value": "DD8A9B4F6E06DF35AE1EC2339DF0FA36",
|
||||
"domain": "10.190.1.205"
|
||||
},
|
||||
{
|
||||
"name": "__rubyUX",
|
||||
"value": "false",
|
||||
"domain": "ntp.msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "pglt-edgeChromium-dhp",
|
||||
"value": "2083",
|
||||
"domain": ".msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "pglt-edgeChromium-ntp",
|
||||
"value": "2083",
|
||||
"domain": ".msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "_C_Auth",
|
||||
"value": "",
|
||||
"domain": "ntp.msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "MicrosoftApplicationsTelemetryDeviceId",
|
||||
"value": "da0bf04e-2115-4952-a012-fc29b4954ec0",
|
||||
"domain": "ntp.msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "USRLOC",
|
||||
"value": "",
|
||||
"domain": ".msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "MUID",
|
||||
"value": "15CB8BCF863F643B03449CDE8778658F",
|
||||
"domain": ".msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "v1",
|
||||
"value": "ExJ_flr=b!>Rb_?o_sb/",
|
||||
"domain": ".mediav.com"
|
||||
},
|
||||
{
|
||||
"name": "MUID",
|
||||
"value": "15CB8BCF863F643B03449CDE8778658F",
|
||||
"domain": ".bing.com"
|
||||
},
|
||||
{
|
||||
"name": "MR",
|
||||
"value": "0",
|
||||
"domain": ".c.bing.com"
|
||||
},
|
||||
{
|
||||
"name": "SRM_B",
|
||||
"value": "15CB8BCF863F643B03449CDE8778658F",
|
||||
"domain": ".c.bing.com"
|
||||
},
|
||||
{
|
||||
"name": "MR",
|
||||
"value": "0",
|
||||
"domain": ".c.msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "MUIDB",
|
||||
"value": "15CB8BCF863F643B03449CDE8778658F",
|
||||
"domain": "ntp.msn.cn"
|
||||
},
|
||||
{
|
||||
"name": ".ASPXANONYMOUS",
|
||||
"value": "21JPwTAtPyn84bHA6aNjbcWhPsVrLyiIZ-4mnJ6_dMtvLzgXbw3J9nyy0dxCZj0b8vOEHXZpgfYoPylWhp5s3AY2GwcwdYg9JI6VqZ6YK3uRkwMf0",
|
||||
"domain": "i3.sinopec.com"
|
||||
},
|
||||
{
|
||||
"name": "x",
|
||||
"value": "x",
|
||||
"domain": "auth.siam.sinopec.com"
|
||||
},
|
||||
{
|
||||
"name": "msnup",
|
||||
"value": "%7B%22cnex%22%3A%22no%22%7D",
|
||||
"domain": ".msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "msaoptout",
|
||||
"value": "0",
|
||||
"domain": "ntp.msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "OptanonAlertBoxClosed",
|
||||
"value": "2026-03-04T08:18:01.713Z",
|
||||
"domain": ".msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "eupubconsent-v2",
|
||||
"value": "CQghjcAQghjcAAcABBZHBfFsAP_gAELgACiQK4tX_G__bXlr8X73aftkeY1f99h77sQxBhaJk-4FzJvW_JwX32E7NAz6tqYKmRIAu3TBAQNlHJDURVCgaIgVqSDMaEyUoTNKJ6BkiFMRY2dYCFxvm4tjeQCY5vr991d52B-t7dr83dzyy4hHv3a5_0S1WAAAAYCNDfv9bROb-9IO9_x8v4v4_F7pE2_eS1l_tWvp7D9-cts_9XW99_bbff9Pn_-uF_-_X_vf_H37v9oK5AACAAAARgkCAAAgAAAAABAAAAAQAAAAAQBgAAARBAEAAgAQGEgAEAAAAIAAAAAgSAAAAAJAAgAAAABQAAAIBAAAAwAIBgAAGAAACABACAAAAgEgYJgQQCBAABCYBAAgQgAAAACwECAABAgiACEQcAQAAAAAAAAEAAoAAEAgDAQkkBCxIIAuIJoAACAAAIIAChFJ2YAggDNlqrwZNoytMCwfMFzymAZAEQRk5JsQmsAoJA1AAQAAuACgAKgAcAA8ACCAGQAagA8ACIAEwAKoAbwA9AB-AEJAIYAiQBHACWAE0AK0AYYAywBsgDvgHsAfEA-wD9AIBARcBGACNAFBAKgAVcAuYBigDaAG4AOIAkQBOwChwFHgKRAWwAuQBd4DDQGSAMnAZcAzmBrAGsgNvAeOEAMAAOAHOAQcAn4CPQEigJWATaAsIBeQDEAGLQMhAyMBowDUwG0ANuAboA8oB8gD9wICAQMAgiOATgAIgAcAB4AFwAfgBoAHOAO4AgEBBwEIAJ-AVAAvQB0gEegJFASsAmIBMoCbQFIAKTAWoAvoBiADFgGQgMmAaMA00BqYDXgG0ANuAeUA-IB9sD9gP3AgeBBEdBAAAXABQAFQAOAAgABdADIANQAeABEACYAFWALgAugBiADeAHoAP0AhgCJAEsAJoAUYArQBhgDKAGiANkAd4A9oB9gH7ARYBGACggFXALEAXMAvIBigDaAG4AOIAdQBF4CRAEyAJ2AUOAo8BTQCxQFsALgAXIAu0Bd4DDQGPAMkAZOAyqBlgGXAM5AaqA1gBt4DxwH1gQBIAHAAEABoAHOAWIBB4CPQE2gKTAVKAvIBqYDbAG3APKAfEA_YCB4EGAINgQrAimBGkCN8EkQSSAqmBWkCtsFcQVyIQIwAFgAUABcAFUALgAYgA3gB6AEcAO8AigBKQCggFXALmAYoA2gB1IFNAU2AqwBYoC0QFwALkAZOAzkBqoDxwH9gQtAh6BIoCSAFSgKvkoD4ACAAFgAUAA4ADwAIgATAAqgBcADFAIYAiQBHACjAFaANkAd4A_ACrgGKAOoAi8BIgCjwFigLYAZOAywBnIDWAG3gQPJADQALgDuAIAAVABHoCRQErAJtAUmAxYB5QD9wIIlIFwAC4AKAAqABwAEEAMgA0AB4AEQAJgAVQAxAB-gEMARIAowBWgDKAGiANkAd8A-wD9AIsARgAoIBVwC5gF5AMUAbQA3ACLwEiAJ2AUOApoBYoC2AFwALkAXaAw0BkgDJwGXAM5gawBrIDbwHjlADYAFwBHADnAHcAQAAkQBYgDXgHbAP-Aj0BIoCYgE2gKQAU-AvIBfQDFgGTANTAa8A8oB8UD9gP3AgYBA8.f_wACFwAAAAA",
|
||||
"domain": ".msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "MSFPC",
|
||||
"value": "GUID=139384d4545d4f4583ff3ef161faee39&HASH=1393&LV=202603&V=4&LU=1772612281994",
|
||||
"domain": "ntp.msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "UID",
|
||||
"value": "1862498edfbe5a24e12c8d21772613043",
|
||||
"domain": ".scorecardresearch.com"
|
||||
},
|
||||
{
|
||||
"name": "XID",
|
||||
"value": "1862498edfbe5a24e12c8d21772613043",
|
||||
"domain": ".scorecardresearch.com"
|
||||
},
|
||||
{
|
||||
"name": "SM",
|
||||
"value": "C",
|
||||
"domain": ".c.msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "JSESSIONID",
|
||||
"value": "DD8A9B4F6E06DF35AE1EC2339DF0FA36",
|
||||
"domain": "10.190.1.205"
|
||||
},
|
||||
{
|
||||
"name": "language",
|
||||
"value": "zh-CN",
|
||||
"domain": "192.168.8.147"
|
||||
},
|
||||
{
|
||||
"name": "fnos-token",
|
||||
"value": "n4fYErf2p2lHSucAt8uYAo+M+JDy354smCjF98dHtJM=",
|
||||
"domain": "192.168.8.147"
|
||||
},
|
||||
{
|
||||
"name": "fnos-long-token",
|
||||
"value": "07MRUOgDAAC3g89pAAAAAMrPwB8hdcC8nqwi0bqBWfNDO0uPBM7fPQ==",
|
||||
"domain": "192.168.8.147"
|
||||
},
|
||||
{
|
||||
"name": "_C_ETH",
|
||||
"value": "1",
|
||||
"domain": ".msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "_EDGE_S",
|
||||
"value": "\"SID=217608637BE1667B39501F717A966764\"",
|
||||
"domain": ".msn.cn"
|
||||
},
|
||||
{
|
||||
"name": "OptanonConsent",
|
||||
"value": "isGpcEnabled=0&datestamp=Wed+Mar+04+2026+17%3A09%3A17+GMT%2B0800+(%E4%B8%AD%E5%9B%BD%E6%A0%87%E5%87%86%E6%97%B6%E9%97%B4)&version=202501.2.0&browserGpcFlag=0&isIABGlobal=false&hosts=&consentId=3599d387-f02a-409c-910d-0a6caea03c05&interactionCount=1&isAnonUser=1&landingPath=NotLandingPage&groups=C0001%3A1%2CC0002%3A1%2CC0004%3A1%2CC0008%3A1%2CV2STACK42%3A1&AwaitingReconsent=false&intType=1&geolocation=%3B",
|
||||
"domain": ".msn.cn"
|
||||
}
|
||||
],
|
||||
"url": "http://192.168.8.147:5666/login"
|
||||
}
|
||||
@@ -0,0 +1,263 @@
|
||||
"""Cookie提取器 - 使用DrissionPage库"""
|
||||
import tkinter as tk
|
||||
from tkinter import ttk, messagebox
|
||||
import json
|
||||
import os
|
||||
import logging
|
||||
from DrissionPage import Chromium, ChromiumOptions
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||||
handlers=[
|
||||
logging.FileHandler('cookie_extractor.log'),
|
||||
logging.StreamHandler()
|
||||
]
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class CookieExtractorApp:
|
||||
"""Cookie提取器应用"""
|
||||
|
||||
def __init__(self, root):
|
||||
self.root = root
|
||||
self.root.title("Cookie提取器")
|
||||
self.root.geometry("600x250")
|
||||
self.root.resizable(True, True)
|
||||
|
||||
self.browser = None
|
||||
self.tab = None
|
||||
|
||||
self._create_widgets()
|
||||
|
||||
def _create_widgets(self):
|
||||
main_frame = ttk.Frame(self.root, padding="20")
|
||||
main_frame.pack(fill=tk.BOTH, expand=True)
|
||||
|
||||
url_frame = ttk.LabelFrame(main_frame, text="网址输入", padding="10")
|
||||
url_frame.pack(fill=tk.X, pady=10)
|
||||
|
||||
ttk.Label(url_frame, text="网址:").pack(side=tk.LEFT, padx=5)
|
||||
self.url_var = tk.StringVar(value="http://192.168.8.147:5666/login")
|
||||
self.url_entry = ttk.Entry(url_frame, textvariable=self.url_var, width=50)
|
||||
self.url_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)
|
||||
|
||||
button_frame = ttk.Frame(main_frame)
|
||||
button_frame.pack(fill=tk.X, pady=10)
|
||||
|
||||
self.open_button = ttk.Button(button_frame, text="打开", command=self.open_website, width=15)
|
||||
self.open_button.pack(side=tk.LEFT, padx=5)
|
||||
|
||||
self.save_button = ttk.Button(button_frame, text="保存", command=self.save_cookies, width=15, state=tk.DISABLED)
|
||||
self.save_button.pack(side=tk.LEFT, padx=5)
|
||||
|
||||
self.import_button = ttk.Button(button_frame, text="导入", command=self.import_cookies, width=15)
|
||||
self.import_button.pack(side=tk.LEFT, padx=5)
|
||||
|
||||
self.close_button = ttk.Button(button_frame, text="关闭", command=self.close_app, width=15)
|
||||
self.close_button.pack(side=tk.RIGHT, padx=5)
|
||||
|
||||
self.status_var = tk.StringVar(value="就绪")
|
||||
status_frame = ttk.Frame(main_frame)
|
||||
status_frame.pack(fill=tk.X, pady=10)
|
||||
|
||||
ttk.Label(status_frame, text="状态:").pack(side=tk.LEFT, padx=5)
|
||||
ttk.Label(status_frame, textvariable=self.status_var, foreground="blue").pack(side=tk.LEFT, padx=5)
|
||||
|
||||
def open_website(self):
|
||||
"""打开指定网站"""
|
||||
url = self.url_var.get().strip()
|
||||
|
||||
if not url:
|
||||
messagebox.showerror("错误", "请输入网址")
|
||||
return
|
||||
|
||||
if not (url.startswith("http://") or url.startswith("https://")):
|
||||
url = "https://" + url
|
||||
self.url_var.set(url)
|
||||
|
||||
try:
|
||||
self._cleanup_browser()
|
||||
|
||||
self.status_var.set("正在打开浏览器...")
|
||||
self.root.update()
|
||||
|
||||
co = ChromiumOptions()
|
||||
co.set_browser_path(r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe")
|
||||
co.set_argument('--no-sandbox')
|
||||
co.set_argument('--disable-dev-shm-usage')
|
||||
|
||||
self.browser = Chromium(co)
|
||||
self.tab = self.browser.new_tab()
|
||||
|
||||
self.status_var.set(f"正在访问: {url}")
|
||||
self.root.update()
|
||||
|
||||
self.tab.get(url)
|
||||
|
||||
self.status_var.set("等待页面加载...")
|
||||
self.root.update()
|
||||
|
||||
self.tab.wait.load_start()
|
||||
self.tab.wait.doc_loaded()
|
||||
|
||||
self.status_var.set(f"已打开: {url}")
|
||||
self.save_button.config(state=tk.NORMAL)
|
||||
messagebox.showinfo("成功", f"已成功打开网址: {url}\n\n请在浏览器中完成登录操作后,点击'保存'按钮")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"打开网站失败: {e}")
|
||||
messagebox.showerror("错误", f"打开网站失败: {str(e)}")
|
||||
self.status_var.set("就绪")
|
||||
self._cleanup_browser()
|
||||
|
||||
def save_cookies(self):
|
||||
"""保存当前的cookie信息,并强制关闭浏览器"""
|
||||
if not self.browser or not self.tab:
|
||||
messagebox.showerror("错误", "请先打开网站")
|
||||
return
|
||||
|
||||
try:
|
||||
self.status_var.set("正在提取cookies...")
|
||||
self.root.update()
|
||||
|
||||
cookies = self.browser.cookies()
|
||||
logger.info(f"成功提取 {len(cookies)} 个 cookies")
|
||||
|
||||
data = {
|
||||
"cookies": cookies,
|
||||
"url": self.url_var.get()
|
||||
}
|
||||
|
||||
output_file = "cookie_data.json"
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
|
||||
self.status_var.set("正在关闭浏览器...")
|
||||
self.root.update()
|
||||
|
||||
self._cleanup_browser()
|
||||
|
||||
self.status_var.set(f"数据已保存到: {output_file}")
|
||||
messagebox.showinfo("成功", f"已成功保存 {len(cookies)} 个cookie到: {output_file}\n\n浏览器已关闭")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"保存cookies失败: {e}")
|
||||
messagebox.showerror("错误", f"保存cookies失败: {str(e)}")
|
||||
self.status_var.set("就绪")
|
||||
self._cleanup_browser()
|
||||
|
||||
def import_cookies(self):
|
||||
"""打开浏览器,导入cookie信息,然后打开指定网站"""
|
||||
try:
|
||||
output_file = "cookie_data.json"
|
||||
if not os.path.exists(output_file):
|
||||
messagebox.showerror("错误", f"文件不存在: {output_file}")
|
||||
return
|
||||
|
||||
self.status_var.set("正在读取cookies...")
|
||||
self.root.update()
|
||||
|
||||
with open(output_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
|
||||
url = data.get("url", "")
|
||||
cookies = data.get("cookies", [])
|
||||
|
||||
if not url:
|
||||
url = self.url_var.get().strip()
|
||||
|
||||
if not url:
|
||||
messagebox.showerror("错误", "没有有效的网址")
|
||||
return
|
||||
|
||||
if not (url.startswith("http://") or url.startswith("https://")):
|
||||
url = "https://" + url
|
||||
|
||||
self._cleanup_browser()
|
||||
|
||||
self.status_var.set("正在打开浏览器...")
|
||||
self.root.update()
|
||||
|
||||
co = ChromiumOptions()
|
||||
co.set_browser_path(r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe")
|
||||
co.set_argument('--no-sandbox')
|
||||
co.set_argument('--disable-dev-shm-usage')
|
||||
|
||||
self.browser = Chromium(co)
|
||||
self.tab = self.browser.new_tab()
|
||||
|
||||
self.status_var.set("正在准备环境...")
|
||||
self.root.update()
|
||||
self.tab.get("about:blank")
|
||||
|
||||
if cookies:
|
||||
self.status_var.set("正在导入cookies...")
|
||||
self.root.update()
|
||||
|
||||
success_count = 0
|
||||
for cookie in cookies:
|
||||
try:
|
||||
if not cookie.get('name') or not cookie.get('value'):
|
||||
continue
|
||||
|
||||
cookie.setdefault('path', '/')
|
||||
cookie.setdefault('httpOnly', False)
|
||||
cookie.setdefault('secure', False)
|
||||
|
||||
self.browser.set.cookies([cookie])
|
||||
success_count += 1
|
||||
except Exception as e:
|
||||
logger.error(f"导入cookie失败: {cookie.get('name')} - {e}")
|
||||
|
||||
logger.info(f"成功导入 {success_count} 个 cookies")
|
||||
|
||||
self.status_var.set(f"正在访问: {url}")
|
||||
self.root.update()
|
||||
self.tab.get(url)
|
||||
|
||||
self.status_var.set("等待页面加载...")
|
||||
self.root.update()
|
||||
self.tab.wait.load_start()
|
||||
self.tab.wait.doc_loaded()
|
||||
|
||||
self.url_var.set(url)
|
||||
self.status_var.set(f"已导入并打开: {url}")
|
||||
self.save_button.config(state=tk.NORMAL)
|
||||
messagebox.showinfo("成功", f"已成功导入 {success_count} 个cookie并打开网址: {url}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"导入cookies失败: {e}")
|
||||
messagebox.showerror("错误", f"导入cookies失败: {str(e)}")
|
||||
self.status_var.set("就绪")
|
||||
self._cleanup_browser()
|
||||
|
||||
def close_app(self):
|
||||
self._cleanup_browser()
|
||||
self.root.destroy()
|
||||
|
||||
def _cleanup_browser(self):
|
||||
if self.browser:
|
||||
try:
|
||||
self.browser.quit()
|
||||
except Exception as e:
|
||||
logger.error(f"关闭浏览器失败: {e}")
|
||||
finally:
|
||||
self.browser = None
|
||||
self.tab = None
|
||||
self.save_button.config(state=tk.DISABLED)
|
||||
self.status_var.set("就绪")
|
||||
|
||||
def main():
|
||||
root = tk.Tk()
|
||||
app = CookieExtractorApp(root)
|
||||
|
||||
def on_closing():
|
||||
app._cleanup_browser()
|
||||
root.destroy()
|
||||
|
||||
root.protocol("WM_DELETE_WINDOW", on_closing)
|
||||
root.mainloop()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+12
-1
@@ -1 +1,12 @@
|
||||
{}
|
||||
[
|
||||
{
|
||||
"name": "language",
|
||||
"value": "zh-CN",
|
||||
"domain": "192.168.8.156"
|
||||
},
|
||||
{
|
||||
"name": "fnos-token",
|
||||
"value": "sxRTKZf7p2lA4txCiYB1dvB0xT+sVmDkrmxaX3L61a4=",
|
||||
"domain": "192.168.8.156"
|
||||
}
|
||||
]
|
||||
-992
File diff suppressed because one or more lines are too long
@@ -1,394 +0,0 @@
|
||||
"""
|
||||
MoviePilot Cookie功能示例程序
|
||||
从CookieCloud下载数据并注入浏览器访问网站
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
import hashlib
|
||||
import base64
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
from dataclasses import dataclass
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from Crypto.Cipher import AES
|
||||
|
||||
|
||||
@dataclass
|
||||
class CookieCloudConfig:
|
||||
"""CookieCloud配置"""
|
||||
server: str
|
||||
username: str
|
||||
password: str
|
||||
timeout: int = 30
|
||||
|
||||
|
||||
class CookieCloudDownloader:
|
||||
"""从CookieCloud服务器下载cookie数据"""
|
||||
|
||||
def __init__(self, config: CookieCloudConfig):
|
||||
self.config = config
|
||||
|
||||
def _get_crypt_key(self) -> bytes:
|
||||
combined_string = f"{self.config.username}-{self.config.password}"
|
||||
return hashlib.md5(combined_string.encode('utf-8')).hexdigest()[:16].encode("utf-8")
|
||||
|
||||
def _bytes_to_key(self, data: bytes, salt: bytes, output: int = 48) -> bytes:
|
||||
assert len(salt) == 8, len(salt)
|
||||
data += salt
|
||||
key = hashlib.md5(data).digest()
|
||||
final_key = key
|
||||
while len(final_key) < output:
|
||||
key = hashlib.md5(key + data).digest()
|
||||
final_key += key
|
||||
return final_key[:output]
|
||||
|
||||
def _decrypt(self, encrypted: str, passphrase: bytes) -> bytes:
|
||||
encrypted_bytes = base64.b64decode(encrypted)
|
||||
assert encrypted_bytes.startswith(b"Salted__"), "Invalid encrypted data format"
|
||||
salt = encrypted_bytes[8:16]
|
||||
key_iv = self._bytes_to_key(passphrase, salt, 32 + 16)
|
||||
key = key_iv[:32]
|
||||
iv = key_iv[32:]
|
||||
aes = AES.new(key, AES.MODE_CBC, iv)
|
||||
decrypted_padded = aes.decrypt(encrypted_bytes[16:])
|
||||
padding_length = decrypted_padded[-1]
|
||||
if isinstance(padding_length, str):
|
||||
padding_length = ord(padding_length)
|
||||
return decrypted_padded[:-padding_length]
|
||||
|
||||
def _get_url_domain(self, domain: str) -> str:
|
||||
if not domain:
|
||||
return ""
|
||||
domain = domain.lstrip('.')
|
||||
if ":" in domain:
|
||||
domain = domain.split(":")[0]
|
||||
parts = domain.split(".")
|
||||
if all(part.isdigit() for part in parts):
|
||||
return domain
|
||||
if len(parts) >= 2:
|
||||
return ".".join(parts[-2:])
|
||||
return domain
|
||||
|
||||
def download_all(self) -> Tuple[Optional[Dict], str]:
|
||||
"""下载所有cookie和local storage数据"""
|
||||
try:
|
||||
url = f"{self.config.server}/get/{self.config.username}"
|
||||
request = urllib.request.Request(
|
||||
url,
|
||||
headers={
|
||||
'Content-Type': 'application/json',
|
||||
'User-Agent': 'MoviePilot-Cookie-Client/1.0'
|
||||
},
|
||||
method='GET'
|
||||
)
|
||||
|
||||
response = urllib.request.urlopen(request, timeout=self.config.timeout)
|
||||
|
||||
if response.status != 200:
|
||||
return None, f"服务器返回错误状态码: {response.status}"
|
||||
|
||||
result = json.loads(response.read().decode('utf-8'))
|
||||
|
||||
if not result:
|
||||
return None, "服务器返回数据为空"
|
||||
|
||||
encrypted = result.get("encrypted")
|
||||
if not encrypted:
|
||||
return None, "未获取到cookie密文"
|
||||
|
||||
crypt_key = self._get_crypt_key()
|
||||
try:
|
||||
decrypted_data = self._decrypt(encrypted, crypt_key)
|
||||
result = json.loads(decrypted_data.decode("utf-8"))
|
||||
except Exception as e:
|
||||
return None, f"cookie解密失败: {str(e)}"
|
||||
|
||||
if not result:
|
||||
return None, "cookie解密为空"
|
||||
|
||||
cookie_data = result.get("cookie_data", {})
|
||||
local_storage_data = result.get("local_storage_data", {})
|
||||
|
||||
total_cookies = sum(len(v) for v in cookie_data.values())
|
||||
total_ls = sum(len(v) for v in local_storage_data.values())
|
||||
print(f" 下载完成: {len(cookie_data)} 个域名的Cookie({total_cookies}个), {len(local_storage_data)} 个域名的Local Storage({total_ls}个)")
|
||||
|
||||
return {"cookies": cookie_data, "local_storage": local_storage_data}, ""
|
||||
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 401:
|
||||
return None, "认证失败,请检查用户名和密码"
|
||||
else:
|
||||
return None, f"HTTP错误: {e.code} {e.reason}"
|
||||
except urllib.error.URLError as e:
|
||||
return None, f"网络连接失败: {e.reason}"
|
||||
except Exception as e:
|
||||
return None, f"下载失败: {str(e)}"
|
||||
|
||||
|
||||
class BrowserController:
|
||||
"""使用Playwright控制浏览器"""
|
||||
|
||||
def __init__(self, headless: bool = False, use_edge: bool = False):
|
||||
self.headless = headless
|
||||
self.use_edge = use_edge
|
||||
self.browser = None
|
||||
self.context = None
|
||||
self.page = None
|
||||
self.playwright = None
|
||||
|
||||
def start(self) -> bool:
|
||||
try:
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
self.playwright = sync_playwright().start()
|
||||
|
||||
if self.use_edge:
|
||||
self.browser = self.playwright.chromium.launch(
|
||||
channel="msedge",
|
||||
headless=self.headless,
|
||||
args=['--disable-blink-features=AutomationControlled', '--disable-dev-shm-usage', '--no-sandbox']
|
||||
)
|
||||
else:
|
||||
self.browser = self.playwright.chromium.launch(
|
||||
headless=self.headless,
|
||||
args=['--disable-blink-features=AutomationControlled', '--disable-dev-shm-usage', '--no-sandbox']
|
||||
)
|
||||
|
||||
self.context = self.browser.new_context(
|
||||
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
||||
)
|
||||
|
||||
self.page = self.context.new_page()
|
||||
|
||||
print(f"✓ 浏览器启动成功 ({'Edge' if self.use_edge else 'Chromium'})")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ 浏览器启动失败: {str(e)}")
|
||||
return False
|
||||
|
||||
def inject_cookies(self, cookies: Dict):
|
||||
"""注入cookie到浏览器"""
|
||||
if not self.context or not cookies:
|
||||
return
|
||||
|
||||
try:
|
||||
playwright_cookies = []
|
||||
for domain_key, cookie_list in cookies.items():
|
||||
for cookie in cookie_list:
|
||||
cookie_domain = cookie.get("domain", "").lstrip('.')
|
||||
if not cookie_domain:
|
||||
cookie_domain = domain_key.lstrip('.')
|
||||
|
||||
pw_cookie = {
|
||||
"name": cookie.get("name", ""),
|
||||
"value": cookie.get("value", ""),
|
||||
"domain": cookie_domain,
|
||||
"path": cookie.get("path", "/"),
|
||||
}
|
||||
if cookie.get("secure"):
|
||||
pw_cookie["secure"] = True
|
||||
if cookie.get("httpOnly"):
|
||||
pw_cookie["httpOnly"] = True
|
||||
if cookie.get("expirationDate"):
|
||||
pw_cookie["expires"] = int(cookie.get("expirationDate"))
|
||||
playwright_cookies.append(pw_cookie)
|
||||
|
||||
self.context.add_cookies(playwright_cookies)
|
||||
print(f"✓ Cookie注入成功 ({len(playwright_cookies)} 个)")
|
||||
except Exception as e:
|
||||
print(f"✗ Cookie注入失败: {str(e)}")
|
||||
|
||||
def navigate(self, url: str, wait_time: int = 5, timeout: int = 60) -> bool:
|
||||
if not self.page:
|
||||
return False
|
||||
|
||||
try:
|
||||
print(f"正在访问: {url}")
|
||||
|
||||
self.page.goto(url, timeout=timeout * 1000)
|
||||
|
||||
try:
|
||||
self.page.wait_for_load_state("networkidle", timeout=timeout * 1000)
|
||||
except Exception:
|
||||
print(" networkidle超时,尝试load状态...")
|
||||
try:
|
||||
self.page.wait_for_load_state("load", timeout=30000)
|
||||
except Exception:
|
||||
print(" load超时,尝试domcontentloaded状态...")
|
||||
self.page.wait_for_load_state("domcontentloaded", timeout=10000)
|
||||
|
||||
time.sleep(wait_time)
|
||||
|
||||
print(f"✓ 页面加载完成")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ 页面访问失败: {str(e)}")
|
||||
return False
|
||||
|
||||
def check_login_status(self, login_indicator: str) -> bool:
|
||||
if not self.page:
|
||||
return False
|
||||
|
||||
try:
|
||||
html_content = self.page.content()
|
||||
|
||||
if login_indicator:
|
||||
if login_indicator in html_content:
|
||||
print(f"✓ 检测到登录指示器: {login_indicator}")
|
||||
return True
|
||||
else:
|
||||
print(f"✗ 未检测到登录指示器: {login_indicator}")
|
||||
return False
|
||||
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ 登录状态检查失败: {str(e)}")
|
||||
return False
|
||||
|
||||
def take_screenshot(self, filepath: str):
|
||||
if not self.page:
|
||||
return
|
||||
|
||||
try:
|
||||
self.page.screenshot(path=filepath)
|
||||
print(f"✓ 截图已保存: {filepath}")
|
||||
except Exception as e:
|
||||
print(f"✗ 截图失败: {str(e)}")
|
||||
|
||||
def close(self):
|
||||
try:
|
||||
if self.browser:
|
||||
self.browser.close()
|
||||
|
||||
if self.playwright:
|
||||
self.playwright.stop()
|
||||
|
||||
print("✓ 浏览器已关闭")
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ 关闭浏览器失败: {str(e)}")
|
||||
|
||||
|
||||
class CookieDemo:
|
||||
"""Cookie功能示例"""
|
||||
|
||||
def __init__(self, cookiecloud_config: CookieCloudConfig):
|
||||
self.downloader = CookieCloudDownloader(cookiecloud_config)
|
||||
self.browser = None
|
||||
|
||||
def run_sites(self, sites: list, headless: bool = False, use_edge: bool = False):
|
||||
"""下载一次数据,依次访问多个网站"""
|
||||
# 下载数据
|
||||
print("=" * 60)
|
||||
print("从CookieCloud下载数据")
|
||||
print("=" * 60)
|
||||
|
||||
data, error = self.downloader.download_all()
|
||||
|
||||
if error:
|
||||
print(f"✗ 数据下载失败: {error}")
|
||||
return
|
||||
|
||||
# 启动浏览器
|
||||
print(f"\n启动浏览器")
|
||||
print("-" * 60)
|
||||
|
||||
self.browser = BrowserController(headless=headless, use_edge=use_edge)
|
||||
|
||||
if not self.browser.start():
|
||||
return
|
||||
|
||||
# 注入所有cookie
|
||||
if data and data.get("cookies"):
|
||||
print(f"\n注入所有cookie到浏览器")
|
||||
print("-" * 60)
|
||||
self.browser.inject_cookies(data["cookies"])
|
||||
|
||||
# 依次访问每个网站
|
||||
for idx, site in enumerate(sites, 1):
|
||||
print(f"\n\n{'=' * 60}")
|
||||
print(f"访问网站 {idx}/{len(sites)}: {site['name']}")
|
||||
print(f"URL: {site['url']}")
|
||||
print("=" * 60)
|
||||
|
||||
if not self.browser.navigate(site['url']):
|
||||
print(f"✗ 页面访问失败")
|
||||
continue
|
||||
|
||||
# 验证登录状态
|
||||
from urllib.parse import urlparse
|
||||
parsed = urlparse(site['url'])
|
||||
target_domain = parsed.netloc
|
||||
|
||||
login_indicator = site.get('login_indicator')
|
||||
is_logged_in = self.browser.check_login_status(login_indicator)
|
||||
|
||||
if is_logged_in:
|
||||
print("✓ Cookie验证成功:已登录状态")
|
||||
else:
|
||||
print("⚠ Cookie验证失败:未检测到登录状态")
|
||||
|
||||
# 截图
|
||||
screenshot_path = f"{target_domain.replace('.', '_').replace(':', '_')}_screenshot.png"
|
||||
self.browser.take_screenshot(screenshot_path)
|
||||
|
||||
if idx < len(sites):
|
||||
print("\n等待3秒后继续下一个网站...")
|
||||
time.sleep(3)
|
||||
|
||||
# 关闭浏览器
|
||||
print(f"\n关闭浏览器")
|
||||
print("-" * 60)
|
||||
self.browser.close()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("所有网站访问完成")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
def main():
|
||||
# 配置CookieCloud
|
||||
config = CookieCloudConfig(
|
||||
server="http://192.168.1.100:3000/cookiecloud",
|
||||
username="qyZpTrgiP5mVwiRZwfBhjz",
|
||||
password="iGn1B4FnWftp3oj4Ko3jxA",
|
||||
timeout=30
|
||||
)
|
||||
|
||||
# 创建示例程序
|
||||
demo = CookieDemo(config)
|
||||
|
||||
# 测试网站列表
|
||||
test_sites = [
|
||||
{
|
||||
"url": "https://lmkbi.95155.com/bi-system/#/login",
|
||||
"name": "BI系统",
|
||||
"login_indicator": "彭峰"
|
||||
},
|
||||
{
|
||||
"url": "http://192.168.1.100:3000/#/subscribe/movie",
|
||||
"name": "MoviePilot订阅",
|
||||
"login_indicator": "搜索"
|
||||
},
|
||||
{
|
||||
"url": "http://192.168.1.102:8418/bwadmin/QueryCarInfo2",
|
||||
"name": "QueryCarInfo2",
|
||||
"login_indicator": "工单管理"
|
||||
}
|
||||
]
|
||||
|
||||
# 下载一次数据,依次访问所有网站
|
||||
demo.run_sites(
|
||||
sites=test_sites,
|
||||
headless=False,
|
||||
use_edge=True
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,4 +0,0 @@
|
||||
{
|
||||
"用户A_网站A": 0,
|
||||
"用户B_网站A": 1
|
||||
}
|
||||
Reference in New Issue
Block a user