104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Dict, Optional, Tuple
|
|
|
|
|
|
FALLBACK_MODELS = {"qwen3-vl-plus", "kimi-k2.5", "minimax-m2.5"}
|
|
STATE_PATH = Path(__file__).resolve().parents[1] / "data" / "aliyun_fallback_state.json"
|
|
|
|
|
|
def _read_state() -> Dict:
|
|
if not STATE_PATH.exists():
|
|
return {"models": {}}
|
|
try:
|
|
data = json.loads(STATE_PATH.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {"models": {}}
|
|
if not isinstance(data, dict):
|
|
return {"models": {}}
|
|
if "models" not in data or not isinstance(data["models"], dict):
|
|
data["models"] = {}
|
|
return data
|
|
|
|
|
|
def _write_state(data: Dict) -> None:
|
|
STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
STATE_PATH.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def get_disabled_until(model_key: str) -> Optional[float]:
|
|
data = _read_state()
|
|
entry = (data.get("models") or {}).get(model_key) or {}
|
|
ts = entry.get("disabled_until")
|
|
try:
|
|
return float(ts) if ts is not None else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def is_fallback_active(model_key: str, now_ts: Optional[float] = None) -> bool:
|
|
if model_key not in FALLBACK_MODELS:
|
|
return False
|
|
now_ts = float(now_ts) if now_ts is not None else datetime.now(tz=timezone.utc).timestamp()
|
|
disabled_until = get_disabled_until(model_key)
|
|
return bool(disabled_until and disabled_until > now_ts)
|
|
|
|
|
|
def set_disabled_until(model_key: str, disabled_until_ts: float, reason: str = "") -> None:
|
|
if model_key not in FALLBACK_MODELS:
|
|
return
|
|
data = _read_state()
|
|
models = data.setdefault("models", {})
|
|
models[model_key] = {
|
|
"disabled_until": float(disabled_until_ts),
|
|
"reason": reason,
|
|
"updated_at": datetime.now(tz=timezone.utc).timestamp(),
|
|
}
|
|
_write_state(data)
|
|
|
|
|
|
def _next_monday_utc8(now: datetime) -> datetime:
|
|
# Monday = 0
|
|
weekday = now.weekday()
|
|
days_ahead = (7 - weekday) % 7
|
|
if days_ahead == 0:
|
|
days_ahead = 7
|
|
target = (now + timedelta(days=days_ahead)).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
return target
|
|
|
|
|
|
def _next_month_same_day_utc8(now: datetime) -> datetime:
|
|
year = now.year
|
|
month = now.month + 1
|
|
if month > 12:
|
|
month = 1
|
|
year += 1
|
|
# clamp day to last day of next month
|
|
if month == 12:
|
|
next_month = datetime(year + 1, 1, 1, tzinfo=now.tzinfo)
|
|
else:
|
|
next_month = datetime(year, month + 1, 1, tzinfo=now.tzinfo)
|
|
last_day = (next_month - timedelta(days=1)).day
|
|
day = min(now.day, last_day)
|
|
return datetime(year, month, day, 0, 0, 0, tzinfo=now.tzinfo)
|
|
|
|
|
|
def compute_disabled_until(error_text: str) -> Tuple[Optional[float], Optional[str]]:
|
|
if not error_text:
|
|
return None, None
|
|
text = str(error_text).lower()
|
|
tz8 = timezone(timedelta(hours=8))
|
|
now = datetime.now(tz=tz8)
|
|
|
|
if "hour allocated quota exceeded" in text or "每 5 小时请求额度已用完" in text:
|
|
until = now + timedelta(hours=5)
|
|
return until.astimezone(timezone.utc).timestamp(), "hour_quota"
|
|
if "week allocated quota exceeded" in text or "每周请求额度已用完" in text:
|
|
until = _next_monday_utc8(now)
|
|
return until.astimezone(timezone.utc).timestamp(), "week_quota"
|
|
if "month allocated quota exceeded" in text or "每月请求额度已用完" in text:
|
|
until = _next_month_same_day_utc8(now)
|
|
return until.astimezone(timezone.utc).timestamp(), "month_quota"
|
|
return None, None
|