requests.py•4.69 kB
"""
请求和响应模型定义
"""
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from enum import Enum
from .article import WeChatArticle, ArticleContent, ArticleSummary, SearchResult, ResponseMetadata, ArticleMetadata
class ContentFormat(Enum):
"""内容格式枚举"""
MARKDOWN = "markdown"
TEXT = "text"
HTML = "html"
class SortOrder(Enum):
"""排序顺序枚举"""
RELEVANCE = "relevance"
TIME = "time"
POPULARITY = "popularity"
@dataclass
class BrowserConfig:
"""浏览器配置"""
headless: bool = True
viewport_width: int = 1280
viewport_height: int = 720
timeout: int = 30
user_agent: Optional[str] = None
proxy: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
result = {
"headless": self.headless,
"viewport_width": self.viewport_width,
"viewport_height": self.viewport_height,
"timeout": self.timeout
}
if self.user_agent:
result["user_agent"] = self.user_agent
if self.proxy:
result["proxy"] = self.proxy
return result
@dataclass
class RequestConfig:
"""请求配置"""
timeout: int = 30
max_retries: int = 3
retry_delay: float = 1.0
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return {
"timeout": self.timeout,
"max_retries": self.max_retries,
"retry_delay": self.retry_delay
}
@dataclass
class ServerConfig:
"""服务器配置"""
name: str
version: str
description: str
author: str
browser_config: BrowserConfig
request_config: RequestConfig
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return {
"name": self.name,
"version": self.version,
"description": self.description,
"author": self.author,
"browser_config": self.browser_config.to_dict(),
"request_config": self.request_config.to_dict()
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ServerConfig":
"""从字典创建配置"""
browser_config = BrowserConfig(**data.get("browser_config", {}))
request_config = RequestConfig(**data.get("request_config", {}))
return cls(
name=data["name"],
version=data["version"],
description=data["description"],
author=data["author"],
browser_config=browser_config,
request_config=request_config
)
@dataclass
class SearchQuery:
"""搜索查询"""
query: str
account: Optional[str] = None
sort_by: SortOrder = SortOrder.RELEVANCE
limit: int = 10
request_id: Optional[str] = None
@dataclass
class FetchArticleRequest:
"""获取文章请求"""
url: str
content_formats: List[ContentFormat] = None
request_id: Optional[str] = None
def __post_init__(self):
"""初始化后处理"""
if self.content_formats is None:
self.content_formats = [ContentFormat.MARKDOWN, ContentFormat.TEXT]
@dataclass
class FetchArticleResponse:
"""获取文章响应"""
metadata: ResponseMetadata
article: Optional[WeChatArticle] = None
content: Optional[Dict[str, str]] = None
error: Optional[Dict[str, Any]] = None
@dataclass
class SearchArticlesRequest:
"""搜索文章请求"""
query: SearchQuery
@dataclass
class SearchArticlesResponse:
"""搜索文章响应"""
metadata: ResponseMetadata
results: List[SearchResult]
total: int
error: Optional[Dict[str, Any]] = None
@dataclass
class ExtractArticleSummaryRequest:
"""提取文章摘要请求"""
url: str
max_length: int = 200
request_id: Optional[str] = None
@dataclass
class ExtractArticleSummaryResponse:
"""提取文章摘要响应"""
metadata: ResponseMetadata
summary: Optional[ArticleSummary] = None
error: Optional[Dict[str, Any]] = None
def generate_request_id() -> str:
"""生成请求ID"""
import uuid
return str(uuid.uuid4())
def create_error_response(error_code: str, message: str, details: Optional[Dict[str, Any]] = None, request_id: Optional[str] = None) -> Dict[str, Any]:
"""创建错误响应"""
response = {
"error_code": error_code,
"error_type": error_code,
"message": message
}
if details:
response["details"] = details
if request_id:
response["request_id"] = request_id
return response