# coding=utf-8
"""
配置工具模块 - 多账号配置解析和验证
提供多账号推送配置的解析、验证和限制功能
"""
from typing import Dict, List, Optional, Tuple
def parse_multi_account_config(config_value: str, separator: str = ";") -> List[str]:
"""
解析多账号配置,返回账号列表
Args:
config_value: 配置值字符串,多个账号用分隔符分隔
separator: 分隔符,默认为 ;
Returns:
账号列表,空字符串会被保留(用于占位)
Examples:
>>> parse_multi_account_config("url1;url2;url3")
['url1', 'url2', 'url3']
>>> parse_multi_account_config(";token2") # 第一个账号无token
['', 'token2']
>>> parse_multi_account_config("")
[]
"""
if not config_value:
return []
# 保留空字符串用于占位(如 ";token2" 表示第一个账号无token)
accounts = [acc.strip() for acc in config_value.split(separator)]
# 过滤掉全部为空的情况
if all(not acc for acc in accounts):
return []
return accounts
def validate_paired_configs(
configs: Dict[str, List[str]],
channel_name: str,
required_keys: Optional[List[str]] = None
) -> Tuple[bool, int]:
"""
验证配对配置的数量是否一致
对于需要多个配置项配对的渠道(如 Telegram 的 token 和 chat_id),
验证所有配置项的账号数量是否一致。
Args:
configs: 配置字典,key 为配置名,value 为账号列表
channel_name: 渠道名称,用于日志输出
required_keys: 必须有值的配置项列表
Returns:
(是否验证通过, 账号数量)
Examples:
>>> validate_paired_configs({
... "token": ["t1", "t2"],
... "chat_id": ["c1", "c2"]
... }, "Telegram", ["token", "chat_id"])
(True, 2)
>>> validate_paired_configs({
... "token": ["t1", "t2"],
... "chat_id": ["c1"] # 数量不匹配
... }, "Telegram", ["token", "chat_id"])
(False, 0)
"""
# 过滤掉空列表
non_empty_configs = {k: v for k, v in configs.items() if v}
if not non_empty_configs:
return True, 0
# 检查必须项
if required_keys:
for key in required_keys:
if key not in non_empty_configs or not non_empty_configs[key]:
return True, 0 # 必须项为空,视为未配置
# 获取所有非空配置的长度
lengths = {k: len(v) for k, v in non_empty_configs.items()}
unique_lengths = set(lengths.values())
if len(unique_lengths) > 1:
print(f"❌ {channel_name} 配置错误:配对配置数量不一致,将跳过该渠道推送")
for key, length in lengths.items():
print(f" - {key}: {length} 个")
return False, 0
return True, list(unique_lengths)[0] if unique_lengths else 0
def limit_accounts(
accounts: List[str],
max_count: int,
channel_name: str
) -> List[str]:
"""
限制账号数量
当配置的账号数量超过最大限制时,只使用前 N 个账号,
并输出警告信息。
Args:
accounts: 账号列表
max_count: 最大账号数量
channel_name: 渠道名称,用于日志输出
Returns:
限制后的账号列表
Examples:
>>> limit_accounts(["a1", "a2", "a3"], 2, "飞书")
⚠️ 飞书 配置了 3 个账号,超过最大限制 2,只使用前 2 个
['a1', 'a2']
"""
if len(accounts) > max_count:
print(f"⚠️ {channel_name} 配置了 {len(accounts)} 个账号,超过最大限制 {max_count},只使用前 {max_count} 个")
print(f" ⚠️ 警告:如果你是 fork 用户,过多账号可能导致 GitHub Actions 运行时间过长,存在账号风险")
return accounts[:max_count]
return accounts
def get_account_at_index(accounts: List[str], index: int, default: str = "") -> str:
"""
安全获取指定索引的账号值
当索引超出范围或账号值为空时,返回默认值。
Args:
accounts: 账号列表
index: 索引
default: 默认值
Returns:
账号值或默认值
Examples:
>>> get_account_at_index(["a", "b", "c"], 1)
'b'
>>> get_account_at_index(["a", "", "c"], 1, "default")
'default'
>>> get_account_at_index(["a"], 5, "default")
'default'
"""
if index < len(accounts):
return accounts[index] if accounts[index] else default
return default