projectAIpopular/gateway/proxy/config.py

"""ProxyConfig：从 settings 的 proxy 段构建代理层配置（T-P0）。

锁定决策落地：
- D-P1 货币：账本毫元整数。配置里价格是浮点"元/1M tokens"（人类可读），
  **加载期一次性转为整数毫元/1M tokens**，运行期纯整数运算（毫元漂移只可能
  发生在这一次换算，黄金用例锁死）。
- 售价折扣采用差异化 {in: 0.5, out: 0.8}（设计文档 §2.5 结论：统一 5 折
  盈亏平衡 h0≈32% 结构性危险；执行版 §4 示例与之冲突，按设计文档取差异化，
  纯配置可热改）。
- D-P7：enabled=False 时不注册任何 /proxy 路由。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple

# 元 -> 毫元换算系数（D-P1：1 元 = 1000 毫元）
YUAN_TO_MILLI = 1000


@dataclass
class ModelPrice:
    """单一模型的整数价格表（毫元 / 1M tokens）。"""
    in_miss: int
    in_hit: int
    out: int


@dataclass
class BucketCfg:
    """一个课程桶的配置。"""
    name: str
    system_template: str = "你是校园学习助手。"
    doc_prefix_file: Optional[str] = None
    doc_version: int = 1
    ttl_hours: int = 72


@dataclass
class ProxyConfig:
    """代理层运行配置（全部为加载期定型值）。"""
    enabled: bool = False
    admin_key: str = ""
    db_path: str = "data/proxy.sqlite3"
    buckets: Dict[str, BucketCfg] = field(default_factory=dict)
    model_prices: Dict[str, ModelPrice] = field(default_factory=dict)
    peak_start: str = "08:30"
    peak_end: str = "23:59"
    offpeak_factor: float = 0.5
    sale_in: float = 0.5
    sale_out: float = 0.8
    rpm_per_key: int = 10
    day_req_cap: int = 200
    concurrent_per_key: int = 2
    max_body_chars: int = 60000
    semcache_enabled: bool = True
    sim_threshold: float = 0.92
    max_entries: int = 300000
    promote_frequency: int = 5

    def bucket(self, name: str) -> BucketCfg:
        """取桶配置；未知名回落 default（D-P2 缺省桶）。"""
        if name in self.buckets:
            return self.buckets[name]
        return self.buckets.get("default") or BucketCfg(name="default")

    def price(self, model: str) -> Optional[ModelPrice]:
        """取模型价格表；未配置返回 None（调用方应拒绝或用 default 档）。"""
        return self.model_prices.get(model)


def _to_milli_per_m(yuan_per_m: float) -> int:
    """元/1M -> 毫元/1M（四舍五入取整，加载期唯一换算点）。"""
    return int(round(float(yuan_per_m) * YUAN_TO_MILLI))


def _parse_hhmm(text: str, fallback: str) -> str:
    """校验 HH:MM 形态，非法回落默认。"""
    t = str(text or fallback).strip()
    parts = t.split(":")
    if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
        h, m = int(parts[0]), int(parts[1])
        if 0 <= h <= 23 and 0 <= m <= 59:
            return f"{h:02d}:{m:02d}"
    return fallback


def build_proxy_config(settings_dict: Dict[str, Any]) -> ProxyConfig:
    """从 settings.to_dict() 的 proxy 段构建 ProxyConfig（缺省值兜底）。

    settings_dict 为整个设置 dict（含 worker/architect/... 与 proxy 段）。
    """
    raw = settings_dict.get("proxy") or {}
    if not isinstance(raw, dict):
        raw = {}

    # ---- 桶 ----
    buckets: Dict[str, BucketCfg] = {}
    raw_buckets = raw.get("buckets") or {}
    if isinstance(raw_buckets, dict):
        for name, b in raw_buckets.items():
            if not isinstance(b, dict):
                continue
            buckets[str(name)] = BucketCfg(
                name=str(name),
                system_template=str(b.get("system_template") or "你是校园学习助手。"),
                doc_prefix_file=(str(b["doc_prefix_file"])
                                 if b.get("doc_prefix_file") else None),
                doc_version=int(b.get("doc_version", 1) or 1),
                ttl_hours=int(b.get("ttl_hours", 72) or 72),
            )
    if "default" not in buckets:
        buckets["default"] = BucketCfg(name="default")

    # ---- 价格表：元/1M 浮点 -> 毫元/1M 整数（D-P1 加载期换算） ----
    model_prices: Dict[str, ModelPrice] = {}
    raw_pricing = raw.get("pricing") or {}
    if isinstance(raw_pricing, dict):
        for model, p in raw_pricing.items():
            if not isinstance(p, dict) or "in_miss" not in p:
                continue  # peak_window/offpeak_factor/sale_discount 非模型键
            in_miss = _to_milli_per_m(p.get("in_miss", 0))
            in_hit = _to_milli_per_m(p.get("in_hit", 0))
            out = _to_milli_per_m(p.get("out", 0))
            if in_miss <= 0:
                continue
            if in_hit <= 0:
                in_hit = max(1, in_miss // 30)   # 缺省 = in_miss × 1/30（D-P3）
            model_prices[str(model)] = ModelPrice(
                in_miss=in_miss, in_hit=in_hit, out=out)

    # ---- 峰谷窗口 / 系数 ----
    pw = raw_pricing.get("peak_window") or {} if isinstance(raw_pricing, dict) else {}
    off = float(raw_pricing.get("offpeak_factor", 0.5)) if isinstance(raw_pricing, dict) else 0.5
    sale = raw_pricing.get("sale_discount") or {} if isinstance(raw_pricing, dict) else {}

    # ---- 限流 / 缓存 ----
    limits = raw.get("limits") or {}
    if not isinstance(limits, dict):
        limits = {}
    sem = raw.get("semcache") or {}
    if not isinstance(sem, dict):
        sem = {}

    return ProxyConfig(
        enabled=bool(raw.get("enabled", False)),
        admin_key=str(raw.get("admin_key") or ""),
        db_path=str(raw.get("db_path") or "data/proxy.sqlite3"),
        buckets=buckets,
        model_prices=model_prices,
        peak_start=_parse_hhmm(pw.get("start"), "08:30") if isinstance(pw, dict) else "08:30",
        peak_end=_parse_hhmm(pw.get("end"), "23:59") if isinstance(pw, dict) else "23:59",
        offpeak_factor=max(0.0, min(1.0, off)),
        sale_in=max(0.0, min(1.0, float(sale.get("in", 0.5)))),
        sale_out=max(0.0, min(1.0, float(sale.get("out", 0.8)))),
        rpm_per_key=int(limits.get("rpm_per_key", 10) or 10),
        day_req_cap=int(limits.get("day_req_cap", 200) or 200),
        concurrent_per_key=int(limits.get("concurrent_per_key", 2) or 2),
        max_body_chars=int(limits.get("max_body_chars", 60000) or 60000),
        semcache_enabled=bool(sem.get("enabled", True)),
        sim_threshold=float(sem.get("sim_threshold", 0.92) or 0.92),
        max_entries=int(sem.get("max_entries", 300000) or 300000),
        promote_frequency=int(sem.get("promote_frequency", 5) or 5),
    )