# -*- coding: utf-8 -*- """ 威科先行（Wolters Kluwer China）API 自动化检索模块 =============================================== 占位符版本 - 需填充实际 API 信息后方可运行 模块功能： - 自动化法律检索（法规、案例、司法解释等） - 与 legal-issue-research skill 流程整合 - 结果解析与格式化输出 """ import json import requests from typing import Dict, List, Optional, Any from datetime import datetime # ============================================================ # 第一部分：API 配置（需填充） # ============================================================ class WoltersKluwerConfig: """威科先行 API 配置""" # ---------- 需填充 ---------- API_BASE_URL = "https://api.wkinfo.com.cn/v1" # 实际 API 端点 API_KEY = "YOUR_API_KEY_HERE" # API Key API_SECRET = "YOUR_API_SECRET_HERE" # API Secret（如需要） # --------------------------- # 认证方式 AUTH_TYPE = "Bearer" # 或 "API-Key" / "OAuth2" # 请求头 HEADERS = { "Content-Type": "application/json", "Accept": "application/json", # "Authorization": "Bearer {token}", # 动态填充 } # API 限流配置 RATE_LIMIT_REQUESTS = 100 # 每分钟请求数 RATE_LIMIT_SECONDS = 60 # 时间窗口（秒） # 超时配置 REQUEST_TIMEOUT = 30 # 秒 # ============================================================ # 第二部分：API 认证 # ============================================================ class WoltersKluwerAuth: """认证管理""" def __init__(self, config: WoltersKluwerConfig): self.config = config self._token = None self._token_expires = None def get_token(self) -> str: """ 获取认证 Token Returns: str: 认证令牌 """ # ---------- 需填充 ---------- # 根据 AUTH_TYPE 实现对应认证逻辑 # # 方式1: API-Key 直传 # if self.config.AUTH_TYPE == "API-Key": # return self.config.API_KEY # # 方式2: Bearer Token # if self.config.AUTH_TYPE == "Bearer": # if self._is_token_valid(): # return self._token # self._token = self._fetch_token() # return self._token # # 方式3: OAuth2 # if self.config.AUTH_TYPE == "OAuth2": # return self._get_oauth_token() # raise NotImplementedError("请实现认证逻辑或提供 API 文档") # --------------------------- def _is_token_valid(self) -> bool: """检查 Token 是否有效""" if not self._token or not self._token_expires: return False return datetime.now() < self._token_expires def _fetch_token(self) -> str: """从 API 获取 Token""" # ---------- 需填充 ---------- # 实现 Token 申请逻辑 # # 示例（实际需根据威科 API 文档调整）： # response = requests.post( # f"{self.config.API_BASE_URL}/oauth/token", # data={ # "grant_type": "client_credentials", # "client_id": self.config.API_KEY, # "client_secret": self.config.API_SECRET, # }, # timeout=self.config.REQUEST_TIMEOUT # ) # result = response.json() # self._token = result["access_token"] # self._token_expires = datetime.now() + timedelta(seconds=result["expires_in"]) # return self._token # --------------------------- raise NotImplementedError("请实现 Token 获取逻辑") def get_headers(self) -> Dict[str, str]: """获取带认证信息的请求头""" token = self.get_token() headers = self.config.HEADERS.copy() # ---------- 需填充 ---------- if self.config.AUTH_TYPE == "Bearer": headers["Authorization"] = f"Bearer {token}" elif self.config.AUTH_TYPE == "API-Key": headers["X-API-Key"] = token # --------------------------- return headers # ============================================================ # 第三部分：检索请求构建 # ============================================================ class SearchRequestBuilder: """检索请求构建器""" # 检索类型枚举 SEARCH_TYPE_LAW = "law" # 法律法规 SEARCH_TYPE_CASE = "case" # 案例 SEARCH_TYPE_JUDGMENT = "judgment" # 裁判文书 SEARCH_TYPE_INTERPRETATION = "interpretation" # 司法解释 SEARCH_TYPE_GUIDANCE = "guidance" # 指导性案例 SEARCH_TYPE_ALL = "all" # 全部 def __init__(self, config: WoltersKluwerConfig): self.config = config def build_request( self, keyword: str, search_type: str = SEARCH_TYPE_ALL, region: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, page: int = 1, page_size: int = 20, **kwargs ) -> Dict[str, Any]: """ 构建检索请求 Args: keyword: 检索关键词 search_type: 检索类型 region: 地域限定（如 "上海"、"广东"） date_from: 开始日期（YYYY-MM-DD） date_to: 结束日期（YYYY-MM-DD） page: 页码 page_size: 每页条数 **kwargs: 其他可选参数 Returns: Dict: 请求参数字典 """ # ---------- 需填充 ---------- # 根据威科 API 文档构建请求参数 # # 示例（实际需根据 API 调整）： # request_body = { # "keyword": keyword, # "searchType": search_type, # "filters": { # "region": region, # "dateRange": { # "from": date_from, # "to": date_to # } if date_from or date_to else None, # # 添加其他筛选条件 # }, # "pagination": { # "page": page, # "pageSize": page_size # }, # "sort": "relevance", # 或 "date", "citation_count" # # 扩展参数 # **kwargs # } # return request_body # --------------------------- # 临时返回结构示例 return { "keyword": keyword, "searchType": search_type, "region": region, "dateFrom": date_from, "dateTo": date_to, "page": page, "pageSize": page_size, "extra": kwargs } # ============================================================ # 第四部分：API 响应解析 # ============================================================ class ResponseParser: """响应解析器 - 将 API 返回解析为统一格式""" @staticmethod def parse_law_result(raw: Dict) -> Dict: """ 解析法规检索结果 Args: raw: API 返回的原始数据 Returns: Dict: 标准化格式 """ # ---------- 需填充 ---------- # 根据威科 API 返回的实际字段名进行映射 # # 示例映射： # return { # "title": raw.get("lawTitle") or raw.get("name"), # "document_number": raw.get("documentNumber") or raw.get("lawNo"), # "issue_date": raw.get("issueDate"), # "effective_date": raw.get("effectiveDate"), # "category": raw.get("category") or raw.get("lawType"), # "issuer": raw.get("issuer") or raw.get("promulgatingAgency"), # "content": raw.get("content") or raw.get("fullText"), # "status": raw.get("status") or raw.get("effectiveStatus"), # "url": raw.get("url") or raw.get("detailLink"), # # 扩展字段 # } # --------------------------- # 临时返回 return raw @staticmethod def parse_case_result(raw: Dict) -> Dict: """ 解析案例检索结果 """ # ---------- 需填充 ---------- # return { # "case_name": raw.get("caseName"), # "case_number": raw.get("caseNumber"), # "case_type": raw.get("caseType"), # "court": raw.get("court"), # "judge_date": raw.get("judgeDate"), # "judgment_summary": raw.get("judgmentSummary"), # "full_text_url": raw.get("fullTextUrl"), # # 扩展字段 # } # --------------------------- return raw @staticmethod def parse_response( raw_response: Dict, search_type: str ) -> Dict[str, Any]: """ 统一响应解析入口 Args: raw_response: API 原始响应 search_type: 检索类型 Returns: Dict: 解析后的标准化结果 """ # ---------- 需填充 ---------- # 统一解析逻辑 # # 通常威科 API 返回格式： # { # "success": bool, # "data": { ... }, # "total": int, # "page": int, # "pageSize": int, # "error": { "code": str, "message": str } # 如有错误 # } # # return { # "success": raw_response.get("success", True), # "total": raw_response.get("total", 0), # "page": raw_response.get("page", 1), # "pageSize": raw_response.get("pageSize", 20), # "results": [...], # 解析后的结果列表 # "error": raw_response.get("error") # } # --------------------------- return raw_response # ============================================================ # 第五部分：核心检索器 # ============================================================ class WoltersKluwerSearcher: """威科先行检索器 - 主类""" def __init__(self, config: Optional[WoltersKluwerConfig] = None): self.config = config or WoltersKluwerConfig() self.auth = WoltersKluwerAuth(self.config) self.request_builder = SearchRequestBuilder(self.config) self.parser = ResponseParser() def search( self, keyword: str, search_type: str = "all", region: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, page: int = 1, page_size: int = 20, **kwargs ) -> Dict[str, Any]: """ 执行检索 Args: keyword: 检索关键词 search_type: 检索类型 (law/case/judgment/interpretation/guidance/all) region: 地域限定 date_from: 开始日期 date_to: 结束日期 page: 页码 page_size: 每页条数 **kwargs: 其他参数 Returns: Dict: 检索结果 """ # 1. 构建请求 request_body = self.request_builder.build_request( keyword=keyword, search_type=search_type, region=region, date_from=date_from, date_to=date_to, page=page, page_size=page_size, **kwargs ) # 2. 发送请求 # ---------- 需填充 ---------- # endpoint = f"{self.config.API_BASE_URL}/search" # headers = self.auth.get_headers() # # try: # response = requests.post( # endpoint, # headers=headers, # json=request_body, # timeout=self.config.REQUEST_TIMEOUT # ) # response.raise_for_status() # raw_result = response.json() # except requests.exceptions.RequestException as e: # return { # "success": False, # "error": f"请求失败: {str(e)}", # "results": [] # } # --------------------------- # 3. 解析响应 # ---------- 需填充 ---------- # parsed = self.parser.parse_response(raw_result, search_type) # return parsed # --------------------------- return { "success": False, "error": "API 调用代码待填充", "results": [], "request": request_body } def search_laws( self, keyword: str, region: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """检索法律法规""" return self.search( keyword=keyword, search_type=SearchRequestBuilder.SEARCH_TYPE_LAW, region=region, date_from=date_from, date_to=date_to, page=page, page_size=page_size ) def search_cases( self, keyword: str, region: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """检索案例""" return self.search( keyword=keyword, search_type=SearchRequestBuilder.SEARCH_TYPE_CASE, region=region, date_from=date_from, date_to=date_to, page=page, page_size=page_size ) def search_judgments( self, keyword: str, region: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """检索裁判文书""" return self.search( keyword=keyword, search_type=SearchRequestBuilder.SEARCH_TYPE_JUDGMENT, region=region, date_from=date_from, date_to=date_to, page=page, page_size=page_size ) def search_interpretations( self, keyword: str, date_from: Optional[str] = None, date_to: Optional[str] = None, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """检索司法解释""" return self.search( keyword=keyword, search_type=SearchRequestBuilder.SEARCH_TYPE_INTERPRETATION, region=None, date_from=date_from, date_to=date_to, page=page, page_size=page_size ) def search_guidance_cases( self, keyword: str, date_from: Optional[str] = None, date_to: Optional[str] = None, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """检索指导性案例""" return self.search( keyword=keyword, search_type=SearchRequestBuilder.SEARCH_TYPE_GUIDANCE, region=None, date_from=date_from, date_to=date_to, page=page, page_size=page_size ) def comprehensive_search( self, keyword: str, region: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """ 综合检索 - 同时检索所有类型 Returns: { "laws": [...], "cases": [...], "judgments": [...], "interpretations": [...], "guidance_cases": [...] } """ results = {} # 并行或串行调用各类型检索 # ---------- 需填充 ---------- # 可选1: 串行调用 # results["laws"] = self.search_laws(keyword, region, date_from, date_to, page, page_size) # results["cases"] = self.search_cases(keyword, region, date_from, date_to, page, page_size) # ... # 可选2: 并行调用（使用 threading 或 asyncio） # import concurrent.futures # with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: # future_laws = executor.submit(self.search_laws, keyword, region, date_from, date_to, page, page_size) # ... # results["laws"] = future_laws.result() # ... # --------------------------- return results # ============================================================ # 第六部分：Skill 集成接口 # ============================================================ class LegalSearchIntegration: """ 与 legal-issue-research skill 集成的接口 此模块负责： 1. 接收 skill 传来的检索需求 2. 调用威科 API 执行检索 3. 将结果格式化为 skill 要求的格式 4. 返回给 skill 处理流程 """ def __init__(self, config: Optional[WoltersKluwerConfig] = None): self.searcher = WoltersKluwerSearcher(config) def search_by_legal_issue( self, legal_issue: str, legal_type: Optional[str] = None, region: Optional[str] = None, date_range: Optional[Dict[str, str]] = None ) -> Dict[str, Any]: """ 根据法律问题执行检索 Args: legal_issue: 法律问题描述（如 "违法解除劳动合同赔偿金"） legal_type: 法律类型（labor/civil/criminal/administrative/company） region: 地域 date_range: {"from": "2020-01-01", "to": "2024-12-31"} Returns: Dict: 检索结果，格式兼容 legal-issue-research skill """ # 1. 关键词处理 keyword = legal_issue # 2. 根据法律类型映射检索类型 # ---------- 需填充 ---------- # type_mapping = { # "labor": "law+case", # 劳动法规+案例 # "civil": "law+case", # 民事法规+案例 # "criminal": "law+case", # 刑事法规+案例 # "administrative": "law", # 行政法规 # "company": "law+case", # 公司法规模 # } # search_type = type_mapping.get(legal_type, "all") # --------------------------- # 3. 日期范围 date_from = date_range.get("from") if date_range else None date_to = date_range.get("to") if date_range else None # 4. 执行检索 # ---------- 需填充 ---------- # results = self.searcher.comprehensive_search( # keyword=keyword, # region=region, # date_from=date_from, # date_to=date_to # ) # --------------------------- # 5. 格式化为 skill 要求的格式 # ---------- 需填充 ---------- # formatted = { # "source": "威科先行", # "keyword": keyword, # "search_time": datetime.now().isoformat(), # "laws": self._format_laws(results.get("laws", [])), # "cases": self._format_cases(results.get("cases", [])), # "judgments": self._format_judgments(results.get("judgments", [])), # "interpretations": self._format_interpretations(results.get("interpretations", [])), # "guidance_cases": self._format_guidance_cases(results.get("guidance_cases", [])), # "total_count": sum([ # len(results.get("laws", [])), # len(results.get("cases", [])), # ... # ]) # } # return formatted # --------------------------- return { "source": "威科先行", "keyword": keyword, "status": "pending_api_config", "message": "API 配置待填充" } def _format_laws(self, laws: List[Dict]) -> List[Dict]: """格式化法规结果""" formatted = [] for law in laws: formatted.append({ "title": law.get("title", ""), "document_number": law.get("document_number", ""), "effective_date": law.get("effective_date", ""), "category": law.get("category", ""), "issuer": law.get("issuer", ""), "content_summary": law.get("content", "")[:500] if law.get("content") else "", "url": law.get("url", "") }) return formatted def _format_cases(self, cases: List[Dict]) -> List[Dict]: """格式化案例结果""" formatted = [] for case in cases: formatted.append({ "case_name": case.get("case_name", ""), "case_number": case.get("case_number", ""), "case_type": case.get("case_type", ""), "court": case.get("court", ""), "judge_date": case.get("judge_date", ""), "summary": case.get("judgment_summary", "")[:500] if case.get("judgment_summary") else "", "url": case.get("full_text_url", "") }) return formatted def _format_judgments(self, judgments: List[Dict]) -> List[Dict]: """格式化裁判文书结果""" return self._format_cases(judgments) def _format_interpretations(self, interpretations: List[Dict]) -> List[Dict]: """格式化司法解释结果""" return self._format_laws(interpretations) def _format_guidance_cases(self, guidance_cases: List[Dict]) -> List[Dict]: """格式化指导性案例结果""" return self._format_cases(guidance_cases) # ============================================================ # 第七部分：使用示例 # ============================================================ def example_usage(): """使用示例""" # 初始化 integration = LegalSearchIntegration() # 执行检索 result = integration.search_by_legal_issue( legal_issue="违法解除劳动合同赔偿金", legal_type="labor", region="全国", date_range={"from": "2020-01-01", "to": "2024-12-31"} ) print(json.dumps(result, ensure_ascii=False, indent=2)) def direct_api_example(): """直接调用 API 示例""" config = WoltersKluwerConfig() searcher = WoltersKluwerSearcher(config) # 检索法律法规 laws = searcher.search_laws( keyword="劳动合同法第四十七条", date_from="2020-01-01", date_to="2024-12-31" ) # 检索案例 cases = searcher.search_cases( keyword="违法解除劳动合同", region="上海", page=1, page_size=10 ) # 检索指导性案例 guidance = searcher.search_guidance_cases( keyword="劳动合同", date_from="2020-01-01" ) print("法规结果:", json.dumps(laws, ensure_ascii=False, indent=2)) print("案例结果:", json.dumps(cases, ensure_ascii=False, indent=2)) print("指导性案例:", json.dumps(guidance, ensure_ascii=False, indent=2)) # ============================================================ # 入口点 # ============================================================ if __name__ == "__main__": print("=" * 60) print("威科先行 API 自动化检索模块") print("=" * 60) print() print("⚠️ 当前为占位符版本，API 配置待填充") print() print("下一步：") print("1. 获取威科先行 API 文档") print("2. 填充 WoltersKluwerConfig 类中的配置") print("3. 实现 WoltersKluwerAuth 类中的认证逻辑") print("4. 完善 ResponseParser 类中的解析逻辑") print("5. 取消注释 searcher.search() 中的实际调用代码") print() print("=" * 60)