87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
import requests
|
||
from requests import Response
|
||
from typing import List, TypeVar, Dict, Any, Optional
|
||
import logging
|
||
|
||
# 设置日志
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
T = TypeVar('T') # 泛型类型声明
|
||
|
||
REQUEST_URL = "http://aijinan.biaofun.com.cn/app/query/shandong" # 替换为实际API地址
|
||
|
||
def main():
|
||
cities = ["济南市", "青岛市", "临沂市", "菏泽市", "威海市", "东营市"]
|
||
|
||
# 修正：每次请求创建新的参数对象，避免累积
|
||
for city in cities:
|
||
params = {"city": city}
|
||
|
||
# 发起RPC请求
|
||
response_list = rpc_get(REQUEST_URL, params, False)
|
||
|
||
# 处理响应
|
||
if not response_list:
|
||
logger.warning(f"Empty response for city: {city}")
|
||
continue
|
||
|
||
# 设置城市名称并处理后续逻辑 (根据实际Response类调整)
|
||
response_list[0]['name'] = city # 这里假定response_list是字典列表
|
||
after(response_list)
|
||
|
||
def rpc_get(url: str, params: Dict[str, Any], right_or_wrong: bool) -> Optional[List[T]]:
|
||
"""
|
||
RPC请求封装
|
||
:param url: 请求URL
|
||
:param params: 请求参数
|
||
:param right_or_wrong: 未使用的标志（保留原Java参数）
|
||
:return: 响应对象列表
|
||
"""
|
||
logger.info(f"Request URL: {url}")
|
||
|
||
try:
|
||
# 发起POST请求
|
||
response: Response = requests.post(url, data=params)
|
||
|
||
# 检查响应状态
|
||
if response.status_code != 200:
|
||
logger.error(f"Request failed, Status Code: {response.status_code}, URL: {url}")
|
||
return None
|
||
|
||
response_text = response.text
|
||
logger.debug(f"Response: {response_text}")
|
||
|
||
# 解析响应（需要根据实际API响应格式实现）
|
||
return parse_response(response_text)
|
||
|
||
except Exception as e:
|
||
logger.exception(f"RPC request failed: {str(e)}")
|
||
return None
|
||
|
||
def parse_response(response_text: str) -> List[T]:
|
||
"""
|
||
解析API响应 (需要根据实际API响应格式实现)
|
||
示例实现 - 替换为实际解析逻辑
|
||
"""
|
||
# 这里需要根据实际API返回格式实现解析
|
||
# 示例：解析为JSON对象
|
||
import json
|
||
try:
|
||
data = json.loads(response_text)
|
||
# 根据实际数据结构返回列表
|
||
return [data] if isinstance(data, dict) else data
|
||
except json.JSONDecodeError:
|
||
logger.error("Invalid JSON response")
|
||
return []
|
||
|
||
def after(response_list: List[Any]):
|
||
"""
|
||
后续处理逻辑 (根据实际需求实现)
|
||
"""
|
||
# 示例实现
|
||
print(f"Processing response for: {response_list[0].get('name')}")
|
||
# 实际业务逻辑...
|
||
|
||
if __name__ == "__main__":
|
||
main() |