sync_lottery_data
Syncs recent lottery draw data for specified game types to a local database for analysis and querying.
Instructions
同步指定彩票类型的最新数据到本地数据库
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| lottery_type | Yes | 彩票类型 | |
| periods | No | 同步期数 |
Implementation Reference
- src/swlc_mcp/server.py:1429-1448 (handler)MCP tool handler implementation for 'sync_lottery_data' in the call_tool function. Extracts input parameters, invokes get_historical_data to perform synchronization, logs the operation, and returns formatted text response.
elif name == "sync_lottery_data": lottery_type = arguments.get("lottery_type") periods = arguments.get("periods", 10) try: # 从网络获取数据并保存到数据库 results = await lottery_service.get_historical_data(lottery_type, periods) if results: # 记录同步日志 lottery_service.db.log_sync(lottery_type, len(results)) return [types.TextContent( type="text", text=f"成功同步{lottery_type}数据{len(results)}期到本地数据库" )] else: lottery_service.db.log_sync(lottery_type, 0, 'failed', '获取数据失败') return [types.TextContent(type="text", text=f"同步{lottery_type}数据失败")] except Exception as e: lottery_service.db.log_sync(lottery_type, 0, 'failed', str(e)) return [types.TextContent(type="text", text=f"同步{lottery_type}数据失败：{str(e)}")] - src/swlc_mcp/server.py:1164-1184 (registration)Registration of the 'sync_lottery_data' tool in the list_tools method, defining the tool name, description, and input schema for MCP protocol.
name="sync_lottery_data", description="同步指定彩票类型的最新数据到本地数据库", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "periods": { "type": "integer", "minimum": 1, "maximum": 50, "default": 10, "description": "同步期数" } }, "required": ["lottery_type"] } ), - src/swlc_mcp/server.py:1164-1184 (schema)Input schema definition for the 'sync_lottery_data' tool, specifying parameters lottery_type (required, enum) and periods (optional, 1-50, default 10).
name="sync_lottery_data", description="同步指定彩票类型的最新数据到本地数据库", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "periods": { "type": "integer", "minimum": 1, "maximum": 50, "default": 10, "description": "同步期数" } }, "required": ["lottery_type"] } ), - src/swlc_mcp/server.py:475-654 (helper)Core helper method SWLCService.get_historical_data that implements the data synchronization logic: checks database, determines if network sync needed based on freshness and continuity, fetches from official API if required, parses and saves to local DB.
async def get_historical_data(self, lottery_type: str, periods: int = 10) -> List[LotteryResult]: """获取历史开奖数据""" try: # 首先尝试从数据库获取 db_results = self.db.get_historical_data(lottery_type, periods) # 检查是否需要从网络更新数据 should_update = False if not db_results: # 数据库没有数据，需要从网络获取 should_update = True elif len(db_results) < periods: # 数据库数据不足，需要从网络获取 should_update = True else: # 检查最新数据的新鲜度 latest_result = db_results[0] # 最新的数据 if not self._is_data_fresh(latest_result.get('draw_date', ''), lottery_type): should_update = True # 检查期号连续性 elif not self._check_period_continuity(db_results, lottery_type): logger.warning(f"{lottery_type}数据库中期号不连续，需要从网络更新") should_update = True if not should_update: logger.info(f"从本地数据库获取{lottery_type}历史数据") results = [] for item in db_results: if lottery_type == "双色球": result = LotteryResult( lottery_type="双色球", period=item['period'], draw_date=item['draw_date'], numbers=item['red_balls'], special_numbers=[item['blue_ball']], prize_pool=item['prize_pool'], sales_amount=item['sales_amount'] ) elif lottery_type == "福彩3D": result = LotteryResult( lottery_type="福彩3D", period=item['period'], draw_date=item['draw_date'], numbers=item['numbers'], sales_amount=item['sales_amount'] ) elif lottery_type == "七乐彩": result = LotteryResult( lottery_type="七乐彩", period=item['period'], draw_date=item['draw_date'], numbers=item['basic_numbers'], special_numbers=[item['special_number']], prize_pool=item['prize_pool'], sales_amount=item['sales_amount'] ) elif lottery_type == "快乐8": result = LotteryResult( lottery_type="快乐8", period=item['period'], draw_date=item['draw_date'], numbers=item['numbers'], prize_pool=item['prize_pool'], sales_amount=item['sales_amount'] ) else: continue results.append(result) return results # 从网络获取并保存数据 logger.info(f"从网络获取{lottery_type}历史数据") data = await self._fetch_lottery_data(lottery_type, periods) if not data or not data['result']: # 如果网络获取失败，尝试返回数据库中的可用数据 if db_results: logger.warning(f"网络获取{lottery_type}数据失败，返回数据库中的可用数据") return self._convert_db_results_to_lottery_results(db_results, lottery_type) return [] results = [] for item in data['result']: if lottery_type == "双色球": # 解析红球和蓝球 red_balls = item['red'].split(',') blue_ball = item['blue'] # 保存到数据库 self.db.save_ssq_result( period=item['code'], draw_date=item['date'], red_balls=red_balls, blue_ball=blue_ball ) result = LotteryResult( lottery_type="双色球", period=item['code'], draw_date=item['date'], numbers=red_balls, special_numbers=[blue_ball] ) elif lottery_type == "福彩3D": # 解析3D号码 numbers = item['red'].split(',') # 保存到数据库 self.db.save_3d_result( period=item['code'], draw_date=item['date'], numbers=numbers ) result = LotteryResult( lottery_type="福彩3D", period=item['code'], draw_date=item['date'], numbers=numbers ) elif lottery_type == "七乐彩": # 解析基本号码和特别号码 basic_numbers = item['red'].split(',') special_number = item['blue'] # 保存到数据库 self.db.save_qlc_result( period=item['code'], draw_date=item['date'], basic_numbers=basic_numbers, special_number=special_number ) result = LotteryResult( lottery_type="七乐彩", period=item['code'], draw_date=item['date'], numbers=basic_numbers, special_numbers=[special_number] ) elif lottery_type == "快乐8": # 解析快乐8号码 numbers = item['red'].split(',') # 保存到数据库 self.db.save_kl8_result( period=item['code'], draw_date=item['date'], numbers=numbers ) result = LotteryResult( lottery_type="快乐8", period=item['code'], draw_date=item['date'], numbers=numbers ) else: continue results.append(result) return results except Exception as e: logger.error(f"获取{lottery_type}历史数据失败: {e}") # 如果出错，尝试返回数据库中的可用数据 try: db_results = self.db.get_historical_data(lottery_type, periods) if db_results: logger.warning(f"返回数据库中的{lottery_type}数据作为备选") return self._convert_db_results_to_lottery_results(db_results, lottery_type) except: pass return []