bwadmin/cookie
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add-files

Browse Source
This commit is contained in:
bwadmin
2026-03-03 08:58:16 +08:00
parent 0fbbbce2ee
commit 894acc3191
23 changed files with 2492 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cookiecloud_client/README.md
+248
View File
@@ -0,0 +1,248 @@
# CookieCloud Client
一个独立的、可复用的CookieCloud服务器客户端Python模块。
## 特性
- ✅ **完全独立** - 无外部依赖，仅使用Python标准库
- ✅ **类型安全** - 完整的类型提示支持
- ✅ **异常处理** - 完善的异常体系
- ✅ **易于使用** - 简洁的API设计
- ✅ **可扩展** - 支持自定义配置
- ✅ **文档完善** - 详细的使用文档和示例
## 安装
将 `cookiecloud_client` 目录复制到您的项目中即可使用。
## 快速开始
### 基本使用
```python
from cookiecloud_client import CookieCloudClient, CookieConfig
# 创建配置
config = CookieConfig(
server="https://cookiecloud.example.com",
username="your_username",
password="your_password"
)
# 创建客户端
client = CookieCloudClient(config)
# 下载所有cookies
result = client.download()
if result.success:
print(f"下载成功！")
print(f"总域名数: {result.total_domains}")
print(f"总Cookie数: {result.total_cookies}")
print(f"下载耗时: {result.download_time:.2f}秒")
# 获取指定域名的cookie
cookie_str = result.get_cookie_string("example.com")
print(f"example.com的cookie: {cookie_str}")
else:
print(f"下载失败: {result.error_message}")
```
### 下载指定域名的Cookie
```python
# 下载单个域名
cookie_str = client.download_for_domain("baidu.com")
if cookie_str:
print(f"baidu.com的cookie: {cookie_str}")
# 批量下载多个域名
domains = ["baidu.com", "google.com", "github.com"]
cookies = client.download_for_domains(domains)
for domain, cookie in cookies.items():
print(f"{domain}: {cookie}")
```
### 测试连接
```python
success, message = client.test_connection()
if success:
print("连接成功！")
else:
print(f"连接失败: {message}")
```
## 高级用法
### 自定义配置
```python
config = CookieConfig(
server="https://cookiecloud.example.com",
username="your_username",
password="your_password",
timeout=60, # 超时时间（秒）
verify_ssl=False, # 是否验证SSL证书
ignore_cookies=[ # 忽略的cookie名称
"CookieAutoDeleteBrowsingDataCleanup",
"CookieAutoDeleteCleaningDiscarded"
]
)
```
### 异常处理
```python
from cookiecloud_client import (
CookieCloudClient,
CookieConfig,
CookieCloudError,
ConnectionError,
AuthenticationError,
DataParseError,
NetworkError
)
try:
client = CookieCloudClient(config)
result = client.download()
except AuthenticationError as e:
print(f"认证失败: {e.message}")
print(f"用户名: {e.details.get('username')}")
except ConnectionError as e:
print(f"连接失败: {e.message}")
print(f"服务器: {e.details.get('server')}")
print(f"状态码: {e.details.get('status_code')}")
except DataParseError as e:
print(f"数据解析失败: {e.message}")
except NetworkError as e:
print(f"网络错误: {e.message}")
print(f"原始错误: {e.details.get('original_error')}")
except CookieCloudError as e:
print(f"CookieCloud错误: {e.message}")
```
### 使用Cookie数据对象
```python
result = client.download()
# 获取所有域名
domains = result.get_domains()
print(f"所有域名: {domains}")
# 遍历所有cookie
for domain, collection in result.cookies.items():
print(f"\n域名: {domain}")
for cookie in collection.cookies:
print(f" - {cookie.name}={cookie.value[:20]}...")
print(f" 路径: {cookie.path}")
print(f" 安全: {cookie.secure}")
print(f" HttpOnly: {cookie.http_only}")
```
## API文档
### CookieConfig
配置类，用于存储CookieCloud服务器配置。
**参数:**
- `server` (str): CookieCloud服务器地址
- `username` (str): 用户名
- `password` (str): 密码
- `timeout` (int): 请求超时时间（秒），默认30
- `verify_ssl` (bool): 是否验证SSL证书，默认True
- `ignore_cookies` (List[str]): 忽略的cookie名称列表
### CookieCloudClient
客户端类，用于与CookieCloud服务器交互。
**方法:**
#### `download() -> DownloadResult`
下载所有cookies
**返回:** DownloadResult对象
#### `download_for_domain(domain: str) -> Optional[str]`
下载指定域名的cookie字符串
**参数:**
- `domain` (str): 目标域名
**返回:** cookie字符串或None
#### `download_for_domains(domains: List[str]) -> Dict[str, Optional[str]]`
批量下载多个域名的cookie字符串
**参数:**
- `domains` (List[str]): 目标域名列表
**返回:** {domain: cookie_string}字典
#### `test_connection() -> Tuple[bool, str]`
测试与CookieCloud服务器的连接
**返回:** (是否成功, 消息)元组
### DownloadResult
下载结果类。
**属性:**
- `success` (bool): 是否成功
- `cookies` (Dict[str, CookieCollection]): cookie集合字典
- `error_message` (str): 错误信息
- `total_domains` (int): 总域名数
- `total_cookies` (int): 总cookie数
- `download_time` (float): 下载耗时（秒）
**方法:**
- `get_cookie_string(domain: str) -> Optional[str]`: 获取指定域名的cookie字符串
- `get_domains() -> List[str]`: 获取所有域名列表
## 异常类
### CookieCloudError
基础异常类，所有其他异常都继承自此类。
### ConfigurationError
配置错误异常
### ConnectionError
连接错误异常
### AuthenticationError
认证错误异常
### DataParseError
数据解析错误异常
### NetworkError
网络错误异常
## 依赖
- Python 3.7+
- 仅使用Python标准库
## 许可证
MIT License
## 更新日志
### v1.0.0 (2026-03-02)
- 初始版本发布
- 完整的CookieCloud客户端功能
- 独立模块，无外部依赖
- 完善的异常处理
- 详细的文档和示例
cookiecloud_client/__init__.py
+33
View File
@@ -0,0 +1,33 @@
"""
CookieCloud客户端模块
一个独立的、可复用的CookieCloud服务器客户端
作者: CookieManager Team
版本: 1.0.0
许可证: MIT
"""
from .client import CookieCloudClient
from .exceptions import (
CookieCloudError,
ConnectionError,
AuthenticationError,
DataParseError,
ConfigurationError,
NetworkError
)
from .models import CookieData, CookieConfig
from .version import __version__
__all__ = [
'CookieCloudClient',
'CookieCloudError',
'ConnectionError',
'AuthenticationError',
'DataParseError',
'ConfigurationError',
'NetworkError',
'CookieData',
'CookieConfig',
'__version__'
]
cookiecloud_client/__pycache__/__init__.cpython-310.pyc
BIN
View File
Binary file not shown.
cookiecloud_client/__pycache__/client.cpython-310.pyc
BIN
View File
Binary file not shown.
cookiecloud_client/__pycache__/exceptions.cpython-310.pyc
BIN
View File
Binary file not shown.
cookiecloud_client/__pycache__/models.cpython-310.pyc
BIN
View File
Binary file not shown.
cookiecloud_client/__pycache__/tests.cpython-310.pyc
BIN
View File
Binary file not shown.
cookiecloud_client/__pycache__/version.cpython-310.pyc
BIN
View File
Binary file not shown.
cookiecloud_client/client.py
+347
View File
@@ -0,0 +1,347 @@
"""
CookieCloud客户端核心实现
"""
import json
import time
import urllib.request
import urllib.error
from typing import Dict, List, Optional, Tuple
from .models import CookieConfig, CookieData, CookieCollection, DownloadResult
from .exceptions import (
CookieCloudError,
ConfigurationError,
ConnectionError,
AuthenticationError,
DataParseError,
NetworkError
)
class CookieCloudClient:
"""
CookieCloud客户端
用于从CookieCloud服务器下载和管理cookies的独立客户端
示例:
>>> config = CookieConfig(
... server="https://cookiecloud.example.com",
... username="your_username",
... password="your_password"
... )
>>> client = CookieCloudClient(config)
>>> result = client.download()
>>> if result.success:
... print(f"下载成功，共{result.total_domains}个域名")
... cookie_str = result.get_cookie_string("example.com")
"""
def __init__(self, config: CookieConfig):
"""
初始化客户端
Args:
config: CookieCloud配置对象
Raises:
ConfigurationError: 配置验证失败
"""
if not isinstance(config, CookieConfig):
raise ConfigurationError("配置参数必须是CookieConfig类型")
self.config = config
self._last_download_time = None
self._download_count = 0
def download(self) -> DownloadResult:
"""
从CookieCloud服务器下载所有cookies
Returns:
DownloadResult: 下载结果对象
Raises:
ConnectionError: 连接服务器失败
AuthenticationError: 认证失败
DataParseError: 数据解析失败
NetworkError: 网络错误
"""
start_time = time.time()
try:
raw_data = self._fetch_data()
cookies = self._parse_cookies(raw_data)
download_time = time.time() - start_time
self._last_download_time = time.time()
self._download_count += 1
total_cookies = sum(len(c.cookies) for c in cookies.values())
return DownloadResult(
success=True,
cookies=cookies,
total_domains=len(cookies),
total_cookies=total_cookies,
download_time=download_time
)
except CookieCloudError:
raise
except Exception as e:
raise CookieCloudError(f"下载cookies失败: {str(e)}")
def download_for_domain(self, domain: str) -> Optional[str]:
"""
下载指定域名的cookie字符串
Args:
domain: 目标域名
Returns:
Optional[str]: cookie字符串，如果不存在则返回None
Raises:
ConnectionError: 连接服务器失败
AuthenticationError: 认证失败
DataParseError: 数据解析失败
"""
result = self.download()
return result.get_cookie_string(domain)
def download_for_domains(self, domains: List[str]) -> Dict[str, Optional[str]]:
"""
批量下载多个域名的cookie字符串
Args:
domains: 目标域名列表
Returns:
Dict[str, Optional[str]]: {domain: cookie_string}
"""
result = self.download()
return {domain: result.get_cookie_string(domain) for domain in domains}
def test_connection(self) -> Tuple[bool, str]:
"""
测试与CookieCloud服务器的连接
Returns:
Tuple[bool, str]: (是否成功, 消息)
"""
try:
self._fetch_data()
return True, "连接成功"
except AuthenticationError as e:
return False, f"认证失败: {e.message}"
except ConnectionError as e:
return False, f"连接失败: {e.message}"
except Exception as e:
return False, f"测试失败: {str(e)}"
def _fetch_data(self) -> Dict:
"""
从服务器获取原始数据
Returns:
Dict: 原始JSON数据
Raises:
ConnectionError: 连接失败
AuthenticationError: 认证失败
NetworkError: 网络错误
"""
url = f"{self.config.server}/get/{self.config.username}"
try:
data = json.dumps({"password": self.config.password}).encode('utf-8')
request = urllib.request.Request(
url,
data=data,
headers={
'Content-Type': 'application/json',
'User-Agent': 'CookieCloudClient/1.0'
},
method='POST'
)
response = urllib.request.urlopen(
request,
timeout=self.config.timeout
)
if response.status != 200:
if response.status == 401:
raise AuthenticationError(
"认证失败，请检查用户名和密码",
username=self.config.username
)
elif response.status == 404:
raise ConnectionError(
"用户不存在，请检查用户名",
server=self.config.server,
status_code=response.status
)
else:
raise ConnectionError(
f"服务器返回错误状态码: {response.status}",
server=self.config.server,
status_code=response.status
)
result = json.loads(response.read().decode('utf-8'))
if not result:
raise DataParseError("服务器返回数据为空")
return result
except urllib.error.HTTPError as e:
if e.code == 401:
raise AuthenticationError(
"认证失败，请检查用户名和密码",
username=self.config.username
)
else:
raise ConnectionError(
f"HTTP错误: {e.code} {e.reason}",
server=self.config.server,
status_code=e.code
)
except urllib.error.URLError as e:
raise NetworkError(
f"网络连接失败: {e.reason}",
original_error=e
)
except json.JSONDecodeError as e:
raise DataParseError(
f"JSON解析失败: {str(e)}"
)
except Exception as e:
if isinstance(e, CookieCloudError):
raise
raise NetworkError(
f"请求失败: {str(e)}",
original_error=e
)
def _parse_cookies(self, raw_data: Dict) -> Dict[str, CookieCollection]:
"""
解析原始cookie数据
Args:
raw_data: 原始JSON数据
Returns:
Dict[str, CookieCollection]: {domain: CookieCollection}
"""
if raw_data.get("cookie_data"):
contents = raw_data.get("cookie_data")
else:
contents = raw_data
domain_groups = self._group_by_domain(contents)
cookies = {}
for domain, cookie_list in domain_groups.items():
if not cookie_list:
continue
if self._is_cloudflare_only(cookie_list):
continue
collection = CookieCollection(domain=domain)
for cookie_data in cookie_list:
cookie = CookieData(
domain=cookie_data.get('domain', ''),
name=cookie_data.get('name', ''),
value=cookie_data.get('value', ''),
path=cookie_data.get('path', '/'),
secure=cookie_data.get('secure', False),
http_only=cookie_data.get('httpOnly', False)
)
collection.add_cookie(cookie)
cookies[domain] = collection
return cookies
def _group_by_domain(self, contents: Dict) -> Dict[str, List[Dict]]:
"""
按域名分组cookies
Args:
contents: 原始cookie内容
Returns:
Dict[str, List[Dict]]: {domain: [cookie_data]}
"""
domain_groups = {}
for site, cookies in contents.items():
for cookie in cookies:
domain = cookie.get("domain", "")
if not domain:
continue
domain_key = self._extract_domain(domain)
if not domain_key:
continue
if domain_key not in domain_groups:
domain_groups[domain_key] = []
domain_groups[domain_key].append(cookie)
return domain_groups
def _extract_domain(self, domain: str) -> Optional[str]:
"""
提取主域名
Args:
domain: 原始域名
Returns:
Optional[str]: 主域名
"""
if not domain:
return None
domain = domain.lstrip('.')
parts = domain.split('.')
if len(parts) < 2:
return domain
if len(parts) == 2:
return domain
return '.'.join(parts[-2:])
def _is_cloudflare_only(self, cookie_list: List[Dict]) -> bool:
"""
检查是否仅包含Cloudflare验证cookie
Args:
cookie_list: cookie列表
Returns:
bool: 是否仅包含cf_clearance
"""
for cookie in cookie_list:
if cookie.get("name") != "cf_clearance":
return False
return True
@property
def last_download_time(self) -> Optional[float]:
"""获取最后下载时间"""
return self._last_download_time
@property
def download_count(self) -> int:
"""获取下载次数"""
return self._download_count
cookiecloud_client/examples.py
+259
View File
@@ -0,0 +1,259 @@
"""
CookieCloud客户端使用示例
演示如何使用cookiecloud_client模块
"""
import sys
import os
# 添加父目录到路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 示例1: 基本使用
def example_basic_usage():
"""基本使用示例"""
from cookiecloud_client import CookieCloudClient, CookieConfig
print("=" * 60)
print("示例1: 基本使用")
print("=" * 60)
# 创建配置
config = CookieConfig(
server="https://movie-pilot.org/cookiecloud",
username="hu6n2vcUmzpu7mqUN2rVCg",
password="jtDM5dV9AyqXkZdQVeA9f6"
)
# 创建客户端
client = CookieCloudClient(config)
# 测试连接
success, message = client.test_connection()
print(f"连接测试: {message}")
if success:
# 下载所有cookies
result = client.download()
if result.success:
print(f"\n✓ 下载成功！")
print(f" 总域名数: {result.total_domains}")
print(f" 总Cookie数: {result.total_cookies}")
print(f" 下载耗时: {result.download_time:.2f}秒")
# 显示前5个域名
domains = result.get_domains()[:5]
print(f"\n前5个域名:")
for idx, domain in enumerate(domains, 1):
cookie_str = result.get_cookie_string(domain)
preview = cookie_str[:50] + "..." if len(cookie_str) > 50 else cookie_str
print(f" {idx}. {domain}: {preview}")
else:
print(f"\n✗ 下载失败: {result.error_message}")
# 示例2: 下载指定域名的Cookie
def example_download_specific_domain():
"""下载指定域名示例"""
from cookiecloud_client import CookieCloudClient, CookieConfig
print("\n" + "=" * 60)
print("示例2: 下载指定域名的Cookie")
print("=" * 60)
config = CookieConfig(
server="https://movie-pilot.org/cookiecloud",
username="hu6n2vcUmzpu7mqUN2rVCg",
password="jtDM5dV9AyqXkZdQVeA9f6"
)
client = CookieCloudClient(config)
# 下载单个域名
domain = "baidu.com"
cookie_str = client.download_for_domain(domain)
if cookie_str:
print(f"✓ {domain}的Cookie:")
print(f" {cookie_str[:100]}...")
else:
print(f"✗ 未找到{domain}的Cookie")
# 示例3: 批量下载多个域名
def example_download_multiple_domains():
"""批量下载示例"""
from cookiecloud_client import CookieCloudClient, CookieConfig
print("\n" + "=" * 60)
print("示例3: 批量下载多个域名")
print("=" * 60)
config = CookieConfig(
server="https://movie-pilot.org/cookiecloud",
username="hu6n2vcUmzpu7mqUN2rVCg",
password="jtDM5dV9AyqXkZdQVeA9f6"
)
client = CookieCloudClient(config)
# 批量下载
domains = ["baidu.com", "github.com", "google.com", "bing.com"]
cookies = client.download_for_domains(domains)
print("批量下载结果:")
for domain, cookie_str in cookies.items():
if cookie_str:
preview = cookie_str[:50] + "..."
print(f" ✓ {domain}: {preview}")
else:
print(f" ✗ {domain}: 未找到Cookie")
# 示例4: 异常处理
def example_error_handling():
"""异常处理示例"""
from cookiecloud_client import (
CookieCloudClient,
CookieConfig,
CookieCloudError,
ConnectionError,
AuthenticationError,
NetworkError
)
print("\n" + "=" * 60)
print("示例4: 异常处理")
print("=" * 60)
try:
# 使用错误的凭据
config = CookieConfig(
server="https://movie-pilot.org/cookiecloud",
username="wrong_user",
password="wrong_pass"
)
client = CookieCloudClient(config)
result = client.download()
except AuthenticationError as e:
print(f"✗ 认证失败: {e.message}")
print(f" 用户名: {e.details.get('username')}")
except ConnectionError as e:
print(f"✗ 连接失败: {e.message}")
print(f" 服务器: {e.details.get('server')}")
print(f" 状态码: {e.details.get('status_code')}")
except NetworkError as e:
print(f"✗ 网络错误: {e.message}")
print(f" 原始错误: {e.details.get('original_error')}")
except CookieCloudError as e:
print(f"✗ CookieCloud错误: {e.message}")
# 示例5: 自定义配置
def example_custom_config():
"""自定义配置示例"""
from cookiecloud_client import CookieCloudClient, CookieConfig
print("\n" + "=" * 60)
print("示例5: 自定义配置")
print("=" * 60)
# 自定义配置
config = CookieConfig(
server="https://movie-pilot.org/cookiecloud",
username="hu6n2vcUmzpu7mqUN2rVCg",
password="jtDM5dV9AyqXkZdQVeA9f6",
timeout=60, # 60秒超时
verify_ssl=False, # 不验证SSL证书
ignore_cookies=[ # 忽略的cookie
"CookieAutoDeleteBrowsingDataCleanup",
"CookieAutoDeleteCleaningDiscarded",
"_ga", # 忽略Google Analytics
]
)
client = CookieCloudClient(config)
print(f"配置信息:")
print(f" 服务器: {config.server}")
print(f" 用户名: {config.username}")
print(f" 超时时间: {config.timeout}秒")
print(f" 验证SSL: {config.verify_ssl}")
print(f" 忽略Cookie: {len(config.ignore_cookies)}个")
result = client.download()
if result.success:
print(f"\n✓ 下载成功")
print(f" 总域名数: {result.total_domains}")
print(f" 总Cookie数: {result.total_cookies}")
# 示例6: 使用Cookie数据对象
def example_cookie_data_objects():
"""使用Cookie数据对象示例"""
from cookiecloud_client import CookieCloudClient, CookieConfig
print("\n" + "=" * 60)
print("示例6: 使用Cookie数据对象")
print("=" * 60)
config = CookieConfig(
server="https://movie-pilot.org/cookiecloud",
username="hu6n2vcUmzpu7mqUN2rVCg",
password="jtDM5dV9AyqXkZdQVeA9f6"
)
client = CookieCloudClient(config)
result = client.download()
if result.success:
# 获取第一个域名的详细信息
first_domain = result.get_domains()[0]
collection = result.cookies.get(first_domain)
if collection:
print(f"域名: {collection.domain}")
print(f"Cookie数量: {len(collection.cookies)}")
print(f"\nCookie详情:")
for idx, cookie in enumerate(collection.cookies[:3], 1):
print(f" {idx}. {cookie.name}")
print(f" 值: {cookie.value[:30]}...")
print(f" 路径: {cookie.path}")
print(f" 安全: {cookie.secure}")
print(f" HttpOnly: {cookie.http_only}")
print()
# 主函数
def main():
"""运行所有示例"""
print("\n")
print("╔" + "=" * 58 + "╗")
print("║" + " " * 15 + "CookieCloud客户端使用示例" + " " * 17 + "║")
print("╚" + "=" * 58 + "╝")
try:
example_basic_usage()
example_download_specific_domain()
example_download_multiple_domains()
example_error_handling()
example_custom_config()
example_cookie_data_objects()
print("\n" + "=" * 60)
print("所有示例执行完成！")
print("=" * 60)
except Exception as e:
print(f"\n示例执行出错: {str(e)}")
if __name__ == "__main__":
main()
cookiecloud_client/exceptions.py
+65
View File
@@ -0,0 +1,65 @@
"""
CookieCloud客户端异常类
"""
class CookieCloudError(Exception):
"""CookieCloud基础异常类"""
def __init__(self, message: str, details: dict = None):
self.message = message
self.details = details or {}
super().__init__(self.message)
def __str__(self):
if self.details:
return f"{self.message} - 详情: {self.details}"
return self.message
class ConfigurationError(CookieCloudError):
"""配置错误异常"""
def __init__(self, message: str, config_key: str = None):
details = {'config_key': config_key} if config_key else {}
super().__init__(message, details)
class ConnectionError(CookieCloudError):
"""连接错误异常"""
def __init__(self, message: str, server: str = None, status_code: int = None):
details = {}
if server:
details['server'] = server
if status_code:
details['status_code'] = status_code
super().__init__(message, details)
class AuthenticationError(CookieCloudError):
"""认证错误异常"""
def __init__(self, message: str, username: str = None):
details = {'username': username} if username else {}
super().__init__(message, details)
class DataParseError(CookieCloudError):
"""数据解析错误异常"""
def __init__(self, message: str, raw_data: str = None):
details = {}
if raw_data:
details['raw_data_length'] = len(raw_data)
super().__init__(message, details)
class NetworkError(CookieCloudError):
"""网络错误异常"""
def __init__(self, message: str, original_error: Exception = None):
details = {}
if original_error:
details['original_error'] = str(original_error)
super().__init__(message, details)
cookiecloud_client/models.py
+126
View File
@@ -0,0 +1,126 @@
"""
CookieCloud数据模型
"""
from dataclasses import dataclass, field
from typing import Optional, Dict, List
from datetime import datetime
@dataclass
class CookieConfig:
"""CookieCloud配置"""
server: str
username: str
password: str
timeout: int = 30
verify_ssl: bool = True
ignore_cookies: List[str] = field(default_factory=lambda: [
"CookieAutoDeleteBrowsingDataCleanup",
"CookieAutoDeleteCleaningDiscarded"
])
def __post_init__(self):
"""验证配置参数"""
if not self.server:
raise ValueError("服务器地址不能为空")
if not self.username:
raise ValueError("用户名不能为空")
if not self.password:
raise ValueError("密码不能为空")
if not self.server.startswith(('http://', 'https://')):
self.server = f"https://{self.server}"
@dataclass
class CookieData:
"""Cookie数据"""
domain: str
name: str
value: str
path: str = "/"
secure: bool = False
http_only: bool = False
expiry: Optional[datetime] = None
def to_dict(self) -> Dict:
"""转换为字典"""
return {
'domain': self.domain,
'name': self.name,
'value': self.value,
'path': self.path,
'secure': self.secure,
'httpOnly': self.http_only,
'expiry': self.expiry.isoformat() if self.expiry else None
}
@classmethod
def from_dict(cls, data: Dict) -> 'CookieData':
"""从字典创建"""
expiry = None
if data.get('expiry'):
try:
expiry = datetime.fromisoformat(data['expiry'])
except (ValueError, TypeError):
pass
return cls(
domain=data.get('domain', ''),
name=data.get('name', ''),
value=data.get('value', ''),
path=data.get('path', '/'),
secure=data.get('secure', False),
http_only=data.get('httpOnly', False),
expiry=expiry
)
@dataclass
class CookieCollection:
"""Cookie集合"""
domain: str
cookies: List[CookieData] = field(default_factory=list)
def to_cookie_string(self, ignore_list: List[str] = None) -> str:
"""转换为cookie字符串"""
ignore_list = ignore_list or []
cookie_parts = []
for cookie in self.cookies:
if cookie.name not in ignore_list:
cookie_parts.append(f"{cookie.name}={cookie.value}")
return ";".join(cookie_parts)
def add_cookie(self, cookie: CookieData):
"""添加cookie"""
self.cookies.append(cookie)
def get_cookie_by_name(self, name: str) -> Optional[CookieData]:
"""根据名称获取cookie"""
for cookie in self.cookies:
if cookie.name == name:
return cookie
return None
@dataclass
class DownloadResult:
"""下载结果"""
success: bool
cookies: Dict[str, CookieCollection] = field(default_factory=dict)
error_message: str = ""
total_domains: int = 0
total_cookies: int = 0
download_time: float = 0.0
def get_cookie_string(self, domain: str) -> Optional[str]:
"""获取指定域名的cookie字符串"""
collection = self.cookies.get(domain)
if collection:
return collection.to_cookie_string()
return None
def get_domains(self) -> List[str]:
"""获取所有域名列表"""
return list(self.cookies.keys())
cookiecloud_client/requirements.txt
+14
View File
@@ -0,0 +1,14 @@
# CookieCloud客户端依赖项
# 此模块完全独立，仅使用Python标准库
# Python版本要求
Python>=3.7
# 无外部依赖
# 仅使用Python标准库模块:
# - json
# - urllib
# - dataclasses
# - typing
# - datetime
# - unittest (用于测试)
cookiecloud_client/setup.py
+54
View File
@@ -0,0 +1,54 @@
"""
CookieCloud客户端安装脚本
"""
from setuptools import setup, find_packages
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setup(
name="cookiecloud-client",
version="1.0.0",
author="CookieManager Team",
author_email="support@example.com",
description="一个独立的、可复用的CookieCloud服务器客户端",
long_description=long_description,
long_description_content_type="text/markdown",
url="https://github.com/example/cookiecloud-client",
packages=find_packages(),
classifiers=[
"Development Status :: 5 - Production/Stable",
"Intended Audience :: Developers",
"Topic :: Software Development :: Libraries :: Python Modules",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
],
python_requires=">=3.7",
install_requires=[
# 无外部依赖，仅使用Python标准库
],
extras_require={
"dev": [
"pytest>=6.0",
"pytest-cov>=2.0",
"black>=21.0",
"flake8>=3.9",
],
},
entry_points={
"console_scripts": [
"cookiecloud-client=cookiecloud_client.cli:main",
],
},
project_urls={
"Bug Reports": "https://github.com/example/cookiecloud-client/issues",
"Source": "https://github.com/example/cookiecloud-client",
"Documentation": "https://github.com/example/cookiecloud-client#readme",
},
)
cookiecloud_client/tests.py
+295
View File
@@ -0,0 +1,295 @@
"""
CookieCloud客户端单元测试
"""
import unittest
from unittest.mock import Mock, patch, MagicMock
import json
from cookiecloud_client import (
CookieCloudClient,
CookieConfig,
CookieCloudError,
ConfigurationError,
ConnectionError,
AuthenticationError,
DataParseError,
NetworkError
)
class TestCookieConfig(unittest.TestCase):
"""测试CookieConfig配置类"""
def test_valid_config(self):
"""测试有效配置"""
config = CookieConfig(
server="https://example.com",
username="user",
password="pass"
)
self.assertEqual(config.server, "https://example.com")
self.assertEqual(config.username, "user")
self.assertEqual(config.password, "pass")
self.assertEqual(config.timeout, 30)
self.assertTrue(config.verify_ssl)
def test_config_with_custom_params(self):
"""测试自定义参数配置"""
config = CookieConfig(
server="https://example.com",
username="user",
password="pass",
timeout=60,
verify_ssl=False,
ignore_cookies=["test_cookie"]
)
self.assertEqual(config.timeout, 60)
self.assertFalse(config.verify_ssl)
self.assertEqual(config.ignore_cookies, ["test_cookie"])
def test_config_auto_add_protocol(self):
"""测试自动添加协议"""
config = CookieConfig(
server="example.com",
username="user",
password="pass"
)
self.assertTrue(config.server.startswith("https://"))
def test_config_empty_server(self):
"""测试空服务器地址"""
with self.assertRaises(ValueError):
CookieConfig(server="", username="user", password="pass")
def test_config_empty_username(self):
"""测试空用户名"""
with self.assertRaises(ValueError):
CookieConfig(server="https://example.com", username="", password="pass")
def test_config_empty_password(self):
"""测试空密码"""
with self.assertRaises(ValueError):
CookieConfig(server="https://example.com", username="user", password="")
class TestCookieCloudClient(unittest.TestCase):
"""测试CookieCloudClient客户端类"""
def setUp(self):
"""测试前准备"""
self.config = CookieConfig(
server="https://test.example.com",
username="testuser",
password="testpass"
)
self.client = CookieCloudClient(self.config)
def test_client_initialization(self):
"""测试客户端初始化"""
self.assertIsInstance(self.client.config, CookieConfig)
self.assertIsNone(self.client.last_download_time)
self.assertEqual(self.client.download_count, 0)
def test_client_invalid_config(self):
"""测试无效配置"""
with self.assertRaises(ConfigurationError):
CookieCloudClient("invalid_config")
@patch('urllib.request.urlopen')
def test_download_success(self, mock_urlopen):
"""测试成功下载"""
mock_response = Mock()
mock_response.status = 200
mock_response.read.return_value = json.dumps({
"cookie_data": {
"test.com": [
{
"domain": "test.com",
"name": "session",
"value": "test_value",
"path": "/",
"secure": False,
"httpOnly": False
}
]
}
}).encode('utf-8')
mock_urlopen.return_value = mock_response
result = self.client.download()
self.assertTrue(result.success)
self.assertEqual(result.total_domains, 1)
self.assertEqual(result.total_cookies, 1)
self.assertIsNotNone(self.client.last_download_time)
self.assertEqual(self.client.download_count, 1)
@patch('urllib.request.urlopen')
def test_download_authentication_error(self, mock_urlopen):
"""测试认证失败"""
import urllib.error
mock_urlopen.side_effect = urllib.error.HTTPError(
url="https://test.example.com/get/testuser",
code=401,
msg="Unauthorized",
hdrs={},
fp=None
)
with self.assertRaises(AuthenticationError):
self.client.download()
@patch('urllib.request.urlopen')
def test_download_connection_error(self, mock_urlopen):
"""测试连接错误"""
import urllib.error
mock_urlopen.side_effect = urllib.error.HTTPError(
url="https://test.example.com/get/testuser",
code=404,
msg="Not Found",
hdrs={},
fp=None
)
with self.assertRaises(ConnectionError):
self.client.download()
@patch('urllib.request.urlopen')
def test_download_network_error(self, mock_urlopen):
"""测试网络错误"""
import urllib.error
mock_urlopen.side_effect = urllib.error.URLError("Connection refused")
with self.assertRaises(NetworkError):
self.client.download()
@patch('urllib.request.urlopen')
def test_download_empty_data(self, mock_urlopen):
"""测试空数据"""
mock_response = Mock()
mock_response.status = 200
mock_response.read.return_value = json.dumps({}).encode('utf-8')
mock_urlopen.return_value = mock_response
with self.assertRaises(DataParseError):
self.client.download()
@patch('urllib.request.urlopen')
def test_test_connection_success(self, mock_urlopen):
"""测试连接测试成功"""
mock_response = Mock()
mock_response.status = 200
mock_response.read.return_value = json.dumps({
"cookie_data": {}
}).encode('utf-8')
mock_urlopen.return_value = mock_response
success, message = self.client.test_connection()
self.assertTrue(success)
self.assertEqual(message, "连接成功")
@patch('urllib.request.urlopen')
def test_test_connection_failure(self, mock_urlopen):
"""测试连接测试失败"""
import urllib.error
mock_urlopen.side_effect = urllib.error.HTTPError(
url="https://test.example.com/get/testuser",
code=401,
msg="Unauthorized",
hdrs={},
fp=None
)
success, message = self.client.test_connection()
self.assertFalse(success)
self.assertIn("认证失败", message)
@patch('urllib.request.urlopen')
def test_download_for_domain(self, mock_urlopen):
"""测试下载指定域名"""
mock_response = Mock()
mock_response.status = 200
mock_response.read.return_value = json.dumps({
"cookie_data": {
"test.com": [
{
"domain": "test.com",
"name": "session",
"value": "test_value",
"path": "/",
"secure": False,
"httpOnly": False
}
]
}
}).encode('utf-8')
mock_urlopen.return_value = mock_response
cookie_str = self.client.download_for_domain("test.com")
self.assertIsNotNone(cookie_str)
self.assertIn("session=test_value", cookie_str)
@patch('urllib.request.urlopen')
def test_download_for_domains(self, mock_urlopen):
"""测试批量下载多个域名"""
mock_response = Mock()
mock_response.status = 200
mock_response.read.return_value = json.dumps({
"cookie_data": {
"test1.com": [
{"domain": "test1.com", "name": "cookie1", "value": "value1", "path": "/"}
],
"test2.com": [
{"domain": "test2.com", "name": "cookie2", "value": "value2", "path": "/"}
]
}
}).encode('utf-8')
mock_urlopen.return_value = mock_response
domains = ["test1.com", "test2.com", "test3.com"]
cookies = self.client.download_for_domains(domains)
self.assertEqual(len(cookies), 3)
self.assertIn("cookie1=value1", cookies["test1.com"])
self.assertIn("cookie2=value2", cookies["test2.com"])
self.assertIsNone(cookies["test3.com"])
class TestExceptions(unittest.TestCase):
"""测试异常类"""
def test_cookie_cloud_error(self):
"""测试基础异常"""
error = CookieCloudError("测试错误", {"key": "value"})
self.assertEqual(error.message, "测试错误")
self.assertEqual(error.details, {"key": "value"})
self.assertIn("测试错误", str(error))
def test_configuration_error(self):
"""测试配置错误"""
error = ConfigurationError("配置错误", config_key="server")
self.assertEqual(error.message, "配置错误")
self.assertEqual(error.details["config_key"], "server")
def test_connection_error(self):
"""测试连接错误"""
error = ConnectionError("连接失败", server="example.com", status_code=404)
self.assertEqual(error.message, "连接失败")
self.assertEqual(error.details["server"], "example.com")
self.assertEqual(error.details["status_code"], 404)
def test_authentication_error(self):
"""测试认证错误"""
error = AuthenticationError("认证失败", username="testuser")
self.assertEqual(error.message, "认证失败")
self.assertEqual(error.details["username"], "testuser")
if __name__ == '__main__':
unittest.main()
cookiecloud_client/version.py
+8
View File
@@ -0,0 +1,8 @@
"""
版本信息
"""
__version__ = '1.0.0'
__author__ = 'CookieManager Team'
__email__ = 'support@example.com'
__license__ = 'MIT'