"""
参数验证工具
提供统一的参数验证功能。
"""
from datetime import datetime
from typing import List, Optional
import os
import yaml
from .errors import InvalidParameterError
from .date_parser import DateParser
def get_supported_platforms() -> List[str]:
"""
从 config.yaml 动态获取支持的平台列表
Returns:
平台ID列表
Note:
- 读取失败时返回空列表，允许所有平台通过（降级策略）
- 平台列表来自 config/config.yaml 中的 platforms 配置
"""
try:
# 获取 config.yaml 路径（相对于当前文件）
current_dir = os.path.dirname(os.path.abspath(__file__))
config_path = os.path.join(current_dir, "..", "..", "config", "config.yaml")
config_path = os.path.normpath(config_path)
with open(config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
platforms = config.get('platforms', [])
return [p['id'] for p in platforms if 'id' in p]
except Exception as e:
# 降级方案：返回空列表，允许所有平台
print(f"警告：无法加载平台配置 ({config_path}): {e}")
return []
def validate_platforms(platforms: Optional[List[str]]) -> List[str]:
"""
验证平台列表
Args:
platforms: 平台ID列表，None表示使用 config.yaml 中配置的所有平台
Returns:
验证后的平台列表
Raises:
InvalidParameterError: 平台不支持
Note:
- platforms=None 时，返回 config.yaml 中配置的平台列表
- 会验证平台ID是否在 config.yaml 的 platforms 配置中
- 配置加载失败时，允许所有平台通过（降级策略）
"""
supported_platforms = get_supported_platforms()
if platforms is None:
# 返回配置文件中的平台列表（用户的默认配置）
return supported_platforms if supported_platforms else []
if not isinstance(platforms, list):
raise InvalidParameterError("platforms 参数必须是列表类型")
if not platforms:
# 空列表时，返回配置文件中的平台列表
return supported_platforms if supported_platforms else []
# 如果配置加载失败（supported_platforms为空），允许所有平台通过
if not supported_platforms:
print("警告：平台配置未加载，跳过平台验证")
return platforms
# 验证每个平台是否在配置中
invalid_platforms = [p for p in platforms if p not in supported_platforms]
if invalid_platforms:
raise InvalidParameterError(
f"不支持的平台: {', '.join(invalid_platforms)}",
suggestion=f"支持的平台（来自config.yaml）: {', '.join(supported_platforms)}"
)
return platforms
def validate_limit(limit: Optional[int], default: int = 20, max_limit: int = 1000) -> int:
"""
验证数量限制参数
Args:
limit: 限制数量
default: 默认值
max_limit: 最大限制
Returns:
验证后的限制值
Raises:
InvalidParameterError: 参数无效
"""
if limit is None:
return default
if not isinstance(limit, int):
raise InvalidParameterError("limit 参数必须是整数类型")
if limit <= 0:
raise InvalidParameterError("limit 必须大于0")
if limit > max_limit:
raise InvalidParameterError(
f"limit 不能超过 {max_limit}",
suggestion=f"请使用分页或降低limit值"
)
return limit
def validate_date(date_str: str) -> datetime:
"""
验证日期格式
Args:
date_str: 日期字符串 (YYYY-MM-DD)
Returns:
datetime对象
Raises:
InvalidParameterError: 日期格式错误
"""
try:
return datetime.strptime(date_str, "%Y-%m-%d")
except ValueError:
raise InvalidParameterError(
f"日期格式错误: {date_str}",
suggestion="请使用 YYYY-MM-DD 格式，例如: 2025-10-11"
)
def validate_date_range(date_range: Optional[dict]) -> Optional[tuple]:
"""
验证日期范围
Args:
date_range: 日期范围字典 {"start": "YYYY-MM-DD", "end": "YYYY-MM-DD"}
Returns:
(start_date, end_date) 元组，或 None
Raises:
InvalidParameterError: 日期范围无效
"""
if date_range is None:
return None
if not isinstance(date_range, dict):
raise InvalidParameterError("date_range 必须是字典类型")
start_str = date_range.get("start")
end_str = date_range.get("end")
if not start_str or not end_str:
raise InvalidParameterError(
"date_range 必须包含 start 和 end 字段",
suggestion='例如: {"start": "2025-10-01", "end": "2025-10-11"}'
)
start_date = validate_date(start_str)
end_date = validate_date(end_str)
if start_date > end_date:
raise InvalidParameterError(
"开始日期不能晚于结束日期",
suggestion=f"start: {start_str}, end: {end_str}"
)
# 检查日期是否在未来
today = datetime.now().date()
if start_date.date() > today or end_date.date() > today:
# 获取可用日期范围提示
try:
from ..services.data_service import DataService
data_service = DataService()
earliest, latest = data_service.get_available_date_range()
if earliest and latest:
available_range = f"{earliest.strftime('%Y-%m-%d')} 至 {latest.strftime('%Y-%m-%d')}"
else:
available_range = "无可用数据"
except Exception:
available_range = "未知（请检查 output 目录）"
future_dates = []
if start_date.date() > today:
future_dates.append(start_str)
if end_date.date() > today and end_str != start_str:
future_dates.append(end_str)
raise InvalidParameterError(
f"不允许查询未来日期: {', '.join(future_dates)}（当前日期: {today.strftime('%Y-%m-%d')}）",
suggestion=f"当前可用数据范围: {available_range}"
)
return (start_date, end_date)
def validate_keyword(keyword: str) -> str:
"""
验证关键词
Args:
keyword: 搜索关键词
Returns:
处理后的关键词
Raises:
InvalidParameterError: 关键词无效
"""
if not keyword:
raise InvalidParameterError("keyword 不能为空")
if not isinstance(keyword, str):
raise InvalidParameterError("keyword 必须是字符串类型")
keyword = keyword.strip()
if not keyword:
raise InvalidParameterError("keyword 不能为空白字符")
if len(keyword) > 100:
raise InvalidParameterError(
"keyword 长度不能超过100个字符",
suggestion="请使用更简洁的关键词"
)
return keyword
def validate_top_n(top_n: Optional[int], default: int = 10) -> int:
"""
验证TOP N参数
Args:
top_n: TOP N数量
default: 默认值
Returns:
验证后的值
Raises:
InvalidParameterError: 参数无效
"""
return validate_limit(top_n, default=default, max_limit=100)
def validate_mode(mode: Optional[str], valid_modes: List[str], default: str) -> str:
"""
验证模式参数
Args:
mode: 模式字符串
valid_modes: 有效模式列表
default: 默认模式
Returns:
验证后的模式
Raises:
InvalidParameterError: 模式无效
"""
if mode is None:
return default
if not isinstance(mode, str):
raise InvalidParameterError("mode 必须是字符串类型")
if mode not in valid_modes:
raise InvalidParameterError(
f"无效的模式: {mode}",
suggestion=f"支持的模式: {', '.join(valid_modes)}"
)
return mode
def validate_config_section(section: Optional[str]) -> str:
"""
验证配置节参数
Args:
section: 配置节名称
Returns:
验证后的配置节
Raises:
InvalidParameterError: 配置节无效
"""
valid_sections = ["all", "crawler", "push", "keywords", "weights"]
return validate_mode(section, valid_sections, "all")
def validate_date_query(
date_query: str,
allow_future: bool = False,
max_days_ago: int = 365
) -> datetime:
"""
验证并解析日期查询字符串
Args:
date_query: 日期查询字符串
allow_future: 是否允许未来日期
max_days_ago: 允许查询的最大天数
Returns:
解析后的datetime对象
Raises:
InvalidParameterError: 日期查询无效
Examples:
>>> validate_date_query("昨天")
datetime(2025, 10, 10)
>>> validate_date_query("2025-10-10")
datetime(2025, 10, 10)
"""
if not date_query:
raise InvalidParameterError(
"日期查询字符串不能为空",
suggestion="请提供日期查询，如：今天、昨天、2025-10-10"
)
# 使用DateParser解析日期
parsed_date = DateParser.parse_date_query(date_query)
# 验证日期不在未来
if not allow_future:
DateParser.validate_date_not_future(parsed_date)
# 验证日期不太久远
DateParser.validate_date_not_too_old(parsed_date, max_days=max_days_ago)
return parsed_date
