| from __future__ import annotations |
|
|
| import copy |
| from dataclasses import dataclass |
| from datetime import datetime, timedelta, timezone |
| import json |
| import os |
| import sys |
| from pathlib import Path |
| from threading import Lock |
| import time |
| from urllib.parse import urlparse |
|
|
| try: |
| from zoneinfo import ZoneInfo |
| except ImportError: |
| ZoneInfo = None |
|
|
| from services.storage.base import StorageBackend |
|
|
| BASE_DIR = Path(__file__).resolve().parents[1] |
| PERSIST_DIR = Path(os.getenv("CHATGPT2API_PERSIST_DIR", "").strip() or "/app/persist") |
|
|
|
|
| def _env_path(name: str, fallback: Path) -> Path: |
| value = str(os.getenv(name) or "").strip() |
| return Path(value) if value else fallback |
|
|
|
|
| DATA_DIR = _env_path("CHATGPT2API_DATA_DIR", PERSIST_DIR if PERSIST_DIR.exists() else BASE_DIR / "data") |
| CONFIG_FILE = _env_path("CHATGPT2API_CONFIG_FILE", PERSIST_DIR / "config" / "config.json" if PERSIST_DIR.exists() else BASE_DIR / "config.json") |
| VERSION_FILE = BASE_DIR / "VERSION" |
| BACKUP_STATE_FILE = DATA_DIR / "backup_state.json" |
| LOG_FILE = _env_path("CHATGPT2API_LOG_FILE", PERSIST_DIR / "logs" / "logs.jsonl" if PERSIST_DIR.exists() else DATA_DIR / "logs.jsonl") |
|
|
| DEFAULT_BACKUP_INCLUDE = { |
| "config": True, |
| "register": True, |
| "cpa": True, |
| "sub2api": True, |
| "logs": True, |
| "image_tasks": True, |
| "accounts_snapshot": True, |
| "auth_keys_snapshot": True, |
| "images": False, |
| } |
|
|
| DEFAULT_IMAGE_STORAGE = { |
| "enabled": False, |
| "mode": "local", |
| "webdav_url": "", |
| "webdav_username": "", |
| "webdav_password": "", |
| "webdav_root_path": "chatgpt2api/images", |
| "public_base_url": "", |
| } |
|
|
| DEFAULT_CHAT_COMPLETION_CACHE = { |
| "enabled": True, |
| "ttl_seconds": 60, |
| "max_entries": 256, |
| "dedupe_inflight": True, |
| "stream_cache": True, |
| "normalize_messages": True, |
| "drop_adjacent_duplicates": True, |
| "drop_assistant_history": False, |
| } |
|
|
| |
| |
| |
| DEFAULT_CHECKIN_CONTENT = { |
| "notice": "奖励积分一部分来源公众号广告费，大家有时间可以帮忙点广告，每天闲了点几下，赚腾讯给钱给大家增加签到奖励的积分。", |
| "dialog_text": "打扰啦，辛苦有空逛逛公众号，留言点广告腾讯会给广告费", |
| "dialog_footer": ( |
| "仅手机点广告评论才有收益，当前赠送积分已超广告收入\n" |
| "基金亏麻了，天才交易员陨落，暂不能自费给大家送太多积分" |
| ), |
| "qrcode_url": "https://wx.z-l.top/qrcode.png", |
| } |
|
|
| DEFAULT_PROXY_RUNTIME_USER_AGENT = ( |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " |
| "AppleWebKit/537.36 (KHTML, like Gecko) " |
| "Chrome/145.0.0.0 Safari/537.36" |
| ) |
|
|
| DEFAULT_PROXY_RUNTIME = { |
| "enabled": False, |
| "egress_mode": "direct", |
| "proxy_url": "", |
| "resource_proxy_url": "", |
| "skip_ssl_verify": False, |
| "reset_session_status_codes": [403], |
| "clearance": { |
| "enabled": False, |
| "mode": "none", |
| "cf_cookies": "", |
| "cf_clearance": "", |
| "user_agent": DEFAULT_PROXY_RUNTIME_USER_AGENT, |
| "browser": "chrome", |
| "flaresolverr_url": "", |
| "timeout_sec": 60, |
| "refresh_interval": 3600, |
| "warm_up_on_start": False, |
| }, |
| } |
|
|
| DEFAULT_FREE_ACCOUNT_CLEANUP = { |
| "enabled": False, |
| "interval_minutes": 10, |
| "failure_threshold": 3, |
| "register_precheck_enabled": True, |
| "action": "mark_abnormal", |
| } |
|
|
| DEFAULT_REDEEM_CODE_CLEANUP = { |
| "enabled": False, |
| "interval_minutes": 1440, |
| "delete_expired": True, |
| "delete_disabled": False, |
| "archive_redeemed": False, |
| "redeemed_older_than_days": 90, |
| "limit": 1000, |
| } |
|
|
| DEFAULT_NORMAL_USER_NAVIGATION: list[dict[str, object]] = [] |
| CHAT_NAVIGATION_MIGRATION_KEY = "_chat_navigation_v1" |
| CHAT_ROOM_NAVIGATION_MIGRATION_KEY = "_chat_room_navigation_v1" |
| DEFAULT_NORMAL_USER_POOL_PRICING: dict[str, object] = { |
| "enabled": False, |
| "timezone": "Asia/Shanghai", |
| "base_unit_price": 1, |
| "calendar_refresh_hours": 24, |
| "rules": [], |
| } |
| DEFAULT_NORMAL_USER_THIRD_PARTY_IMAGE_PRICING: dict[str, object] = copy.deepcopy(DEFAULT_NORMAL_USER_POOL_PRICING) |
|
|
| BRANDING_KEYS = ( |
| "site_title", |
| "site_logo_url", |
| "site_header_title", |
| "site_external_link_label", |
| "site_external_link_url", |
| "site_external_link_icon_url", |
| "site_external_link_icon_svg", |
| ) |
| DEFAULT_BRANDING = { |
| "site_title": "ChatGPT 号池管理", |
| "site_logo_url": "", |
| "site_header_title": "chatgpt2api", |
| "site_external_link_label": "GitHub", |
| "site_external_link_url": "https://github.com/basketikun/chatgpt2api", |
| "site_external_link_icon_url": "", |
| "site_external_link_icon_svg": "", |
| } |
|
|
| ANNOUNCEMENT_TYPES = {"notice", "timeline"} |
| ANNOUNCEMENT_TARGETS = {"internal", "normal", "all"} |
| SHARED_ANNOUNCEMENTS_KEY = "announcements" |
| SHARED_SETTINGS_KEY = "settings" |
| DEFAULT_PROMPT_OPTIMIZER_SYSTEM_PROMPT = ( |
| "You are an expert prompt engineer. Improve the user's prompt to be clearer, more detailed, and effective for an LLM. " |
| "Respond ONLY with the improved prompt text, in the same language as the user. " |
| "Do not add introductions like \"Here is the improved prompt:\"." |
| ) |
| DEFAULT_REVERSE_PROMPT_SYSTEM_PROMPT = """请根据参考图片反推一段适合用于 AI 生图的提示词。 |
| |
| 要求： |
| 1. 只输出提示词正文，不要解释。 |
| 2. 覆盖主体、构图、风格、光线、色彩、材质、镜头和氛围。 |
| 3. 尽量写成可直接用于生图模型的完整提示词。""" |
| DEFAULT_PRODUCT_VISUAL_PLANNER_SYSTEM_PROMPT = ( |
| "你是资深电商商品视觉策划专家，擅长为具体商品量身定制商品素材包与连续详情页，而不是套模板。\n" |
| "用户会提供商品图片（多角度）与商品资料（描述、卖点、规格）以及目标平台。\n" |
| "你必须完成四步思考，最终输出一套为该商品「量身定制」的图片任务方案。\n\n" |
| "第一步——深度分析商品（结合图片与资料，禁止编造事实）：\n" |
| "- 品类与定位：这是什么品类？中高低哪个价位段？面向谁？\n" |
| "- 目标人群：核心购买人群是谁？他们在什么场景下会买？最在意什么？\n" |
| "- 购买决策点：消费者下单前最想确认哪 3-5 件事（如性能、材质、效果、安全、售后、口碑）？\n" |
| "- 差异化卖点：和同类竞品相比，这个商品真正独特、值得讲的是什么？\n" |
| "- 潜在顾虑：消费者最可能犹豫/怀疑什么？哪些点需要用图打消？\n\n" |
| "第二步——提炼全局视觉基调（后续每一张图都必须严格遵循，保证整套是同一套设计）：\n" |
| "- product_appearance：商品主体外观特征（形状、颜色、材质、包装样式、logo 位置），所有图片必须呈现同一商品主体；\n" |
| "- background_style：统一的背景/场景风格（如高级灰渐变、简约米白、深色科技光效等），整套只允许 1 种主背景风格；\n" |
| "- color_palette：统一色彩体系（主色/辅助色/强调色），整套图片共用同一配色；\n" |
| "- lighting：统一光线方向与质感（如左侧柔光棚拍、顶部主光+冷色边缘光）；\n" |
| "- composition_language：统一构图语言（商品占比、主体位置、留白比例、信息区位置）；\n" |
| "- typography_style：统一的画面文字风格（字体气质、文字大小层级、文案位置），保证整套像同一设计师出品。\n\n" |
| "第三步——自主设计每一屏/每一张（这是核心，禁止机械套模板）：\n" |
| "- 【商品素材包】不要只会 主图/场景/卖点/细节 四件套，而要围绕该商品真实卖点与人群，设计真正有用的图集；\n" |
| "- 【商品详情页】不要固定按 首屏→氛围→细节→开盒→卖点→场景→清单→收尾 顺序走，而是：\n" |
| " ① 先想清楚这套详情页的「叙事主线」（例如：问题痛点切入→解决方案→眼见为实→信任背书→消除顾虑→临门一脚）；\n" |
| " ② 每一屏围绕一个明确的「信息任务」设计（首屏抓注意力、中屏讲证据、末屏促转化等），形式可自由发挥：\n" |
| " 产品包装图、场景代入图、工艺细节图、对比图、信息图、数据可视化、使用前后、拆解图、生活方式图、\n" |
| " 品牌故事图、资质认证图、用户评价图、FAQ 图、套装内容图等——只要适合这个商品就用，不要硬凑；\n" |
| " ③ 整体要有节奏：吸引 → 建立信任 → 打消顾虑 → 促进购买，避免每屏都在平铺卖点。\n\n" |
| "硬性要求：\n" |
| "1. 每一屏/每一张必须承担明确且互不重复的内容职责（role 字段写清该张的信息任务），禁止两张图职责重叠或空洞凑数。\n" |
| "2. 商品主体、包装识别、色彩、质感、光线、构图语言在全部图片中保持一致；整套图片必须呈现为同一套设计。\n" |
| "3. 每张图写清：画面主体、构图、场景/背景、光线、风格、画面中的文字内容（如卖点短句），文字要口语化、有说服力。\n" |
| "4. 只使用用户提供的商品事实（名称、卖点、规格、材质、认证、数据），没有依据的信息不得编造成商品事实；\n" |
| " 竞品对比、数据、认证等无依据内容不得虚构。\n" |
| "5. 每张图片任务必须是独立的、可直接交给生图模型的完整中文提示词，并显式复述视觉基调要求。\n" |
| "必须仅返回一个 JSON 对象，禁止返回 markdown 代码块或额外解释：\n" |
| '{"analysis":{"category":"品类定位","audience":"目标人群与购买场景","decision_points":["关键决策点"],"differentiation":"差异化卖点","concerns":["潜在顾虑"]},"visual_style":{"product_appearance":"...","background_style":"...","color_palette":"...","lighting":"...","composition_language":"...","typography_style":"..."},"items":[{"title":"用途名","role":"该张的信息任务与理由","prompt":"完整中文提示词"},...]}' |
| ) |
| SHARED_SETTINGS_KEYS = ( |
| "normal_user_default_credits", |
| "normal_user_daily_credits", |
| "checkin_daily_credits", |
| "checkin_random_min", |
| "checkin_random_max", |
| "checkin_invite_reward", |
| "checkin_credit_valid_days", |
| "checkin_ads", |
| "internal_image_model_daily_limits", |
| "wechat_login_enabled", |
| "wechat_auto_register_enabled", |
| "internal_user_login_session_duration_hours", |
| "normal_user_login_session_duration_hours", |
| "prompt_optimizer_system_prompt", |
| "prompt_optimizer_provider", |
| "prompt_optimizer_base_url", |
| "prompt_optimizer_api_key", |
| "prompt_optimizer_model", |
| "reverse_prompt_system_prompt", |
| "reverse_prompt_provider", |
| "reverse_prompt_base_url", |
| "reverse_prompt_api_key", |
| "reverse_prompt_model", |
| "product_visual_planner_system_prompt", |
| "product_visual_agent_provider", |
| "product_visual_agent_base_url", |
| "product_visual_agent_api_key", |
| "product_visual_agent_model", |
| "normal_user_pool_pricing", |
| "normal_user_third_party_image_pricing", |
| "error_message_rules", |
| "error_message_fallback_enabled", |
| "error_message_ai_model", |
| ) |
| SHARED_SETTINGS_CACHE_SECONDS = 10.0 |
|
|
|
|
| def _normalize_bool(value: object, default: bool = False) -> bool: |
| if isinstance(value, str): |
| lowered = value.strip().lower() |
| if lowered in {"1", "true", "yes", "on"}: |
| return True |
| if lowered in {"0", "false", "no", "off"}: |
| return False |
| return default |
| if value is None: |
| return default |
| return bool(value) |
|
|
|
|
| def _default_normal_user_pool_pricing_rules() -> list[dict[str, object]]: |
| """Starter templates are intentionally disabled so upgrades never change billing.""" |
| return [ |
| { |
| "id": "starter-workday-offpeak", |
| "name": "工作日低谷时段", |
| "target_type": "pool", |
| "third_party_group_id": "", |
| "enabled": False, |
| "pay_percentage": 50, |
| "date_type": "official_workday", |
| "holiday_names": [], |
| "weekend_basis": "natural", |
| "dates": [], |
| "start_date": "", |
| "end_date": "", |
| "all_day": False, |
| "start_time": "00:00", |
| "end_time": "08:00", |
| }, |
| { |
| "id": "starter-natural-weekend", |
| "name": "自然周末优惠", |
| "target_type": "pool", |
| "third_party_group_id": "", |
| "enabled": False, |
| "pay_percentage": 80, |
| "date_type": "weekend", |
| "holiday_names": [], |
| "weekend_basis": "natural", |
| "dates": [], |
| "start_date": "", |
| "end_date": "", |
| "all_day": True, |
| "start_time": "00:00", |
| "end_time": "00:00", |
| }, |
| { |
| "id": "starter-official-holiday", |
| "name": "法定节假日半价", |
| "target_type": "pool", |
| "third_party_group_id": "", |
| "enabled": False, |
| "pay_percentage": 50, |
| "date_type": "official_holiday", |
| "holiday_names": [], |
| "weekend_basis": "official", |
| "dates": [], |
| "start_date": "", |
| "end_date": "", |
| "all_day": True, |
| "start_time": "00:00", |
| "end_time": "00:00", |
| }, |
| { |
| "id": "starter-specific-date", |
| "name": "指定日期活动", |
| "target_type": "pool", |
| "third_party_group_id": "", |
| "enabled": False, |
| "pay_percentage": 70, |
| "date_type": "specific_dates", |
| "holiday_names": [], |
| "weekend_basis": "natural", |
| "dates": ["2026-11-11"], |
| "start_date": "", |
| "end_date": "", |
| "all_day": True, |
| "start_time": "00:00", |
| "end_time": "00:00", |
| }, |
| ] |
|
|
|
|
| def _normalize_normal_user_pool_pricing(value: object) -> dict[str, object]: |
| source = value if isinstance(value, dict) else {} |
| raw_rules = ( |
| source.get("rules") |
| if isinstance(source.get("rules"), list) |
| else _default_normal_user_pool_pricing_rules() |
| ) |
| rules = [copy.deepcopy(item) for item in raw_rules if isinstance(item, dict)] |
| try: |
| refresh_hours = max(1, min(168, int(source.get("calendar_refresh_hours") or 24))) |
| except (TypeError, ValueError): |
| refresh_hours = 24 |
| return { |
| "enabled": _normalize_bool(source.get("enabled"), False), |
| "timezone": "Asia/Shanghai", |
| "base_unit_price": 1, |
| "calendar_refresh_hours": refresh_hours, |
| "rules": rules, |
| } |
|
|
|
|
| def _skip_user_storage_migration() -> bool: |
| return os.getenv("CHATGPT2API_SKIP_USER_STORAGE_MIGRATION", "").strip().lower() in {"1", "true", "yes", "on"} |
|
|
|
|
| def _normalize_positive_int(value: object, default: int, minimum: int = 0) -> int: |
| try: |
| normalized = int(value) |
| except (OverflowError, TypeError, ValueError): |
| normalized = default |
| return max(minimum, normalized) |
|
|
|
|
| def _today_key() -> str: |
| tz = ZoneInfo("Asia/Shanghai") if ZoneInfo else timezone(timedelta(hours=8)) |
| return datetime.now(tz).strftime("%Y-%m-%d") |
|
|
|
|
| def _now_iso() -> str: |
| return datetime.now(timezone.utc).isoformat() |
|
|
|
|
| def _normalize_internal_image_model_limit_scope(value: object) -> str: |
| normalized = str(value or "").strip().lower() |
| if normalized in {"total", "global", "shared", "all"}: |
| return "total" |
| return "per_user" |
|
|
|
|
| def _normalize_internal_image_model_daily_limits(value: object) -> list[dict[str, object]]: |
| if isinstance(value, dict): |
| source = [ |
| {"model_id": model_id, **entry} if isinstance(entry, dict) else {"model_id": model_id, "daily_limit": entry} |
| for model_id, entry in value.items() |
| ] |
| else: |
| source = value if isinstance(value, list) else [] |
| limits_by_model: dict[str, dict[str, object]] = {} |
| for raw in source: |
| if not isinstance(raw, dict): |
| continue |
| model_id = str(raw.get("model_id") or raw.get("model") or raw.get("id") or "").strip() |
| if not model_id: |
| continue |
| date_key = str(raw.get("date") or raw.get("daily_key") or "").strip() |
| previous = limits_by_model.get(model_id) |
| used = _normalize_positive_int(raw.get("used"), 0, 0) |
| reserved = _normalize_positive_int(raw.get("reserved"), 0, 0) |
| if previous is not None and str(previous.get("date") or "") == date_key: |
| used += _normalize_positive_int(previous.get("used"), 0, 0) |
| reserved += _normalize_positive_int(previous.get("reserved"), 0, 0) |
| limit_scope = _normalize_internal_image_model_limit_scope(raw.get("limit_scope") or raw.get("scope")) |
| limits_by_model[model_id] = { |
| "model_id": model_id, |
| "daily_limit": _normalize_positive_int(raw.get("daily_limit") if "daily_limit" in raw else raw.get("limit"), 0, 0), |
| "limit_scope": limit_scope, |
| "date": date_key, |
| "used": used if limit_scope == "total" else 0, |
| "reserved": reserved if limit_scope == "total" else 0, |
| "updated_at": str(raw.get("updated_at") or "").strip(), |
| } |
| limits = list(limits_by_model.values()) |
| limits.sort(key=lambda item: str(item.get("model_id") or "")) |
| return limits |
|
|
|
|
| def _reset_internal_image_model_daily_limit(entry: dict[str, object]) -> dict[str, object]: |
| next_entry = dict(entry) |
| if str(next_entry.get("date") or "").strip() != _today_key() and _normalize_positive_int(next_entry.get("reserved"), 0, 0) <= 0: |
| next_entry["date"] = _today_key() |
| next_entry["used"] = 0 |
| next_entry["reserved"] = 0 |
| return next_entry |
|
|
|
|
| def _internal_image_model_daily_limit_public(entry: dict[str, object]) -> dict[str, object]: |
| normalized = _reset_internal_image_model_daily_limit(entry) |
| daily_limit = _normalize_positive_int(normalized.get("daily_limit"), 0, 0) |
| used = _normalize_positive_int(normalized.get("used"), 0, 0) |
| reserved = _normalize_positive_int(normalized.get("reserved"), 0, 0) |
| limit_scope = _normalize_internal_image_model_limit_scope(normalized.get("limit_scope")) |
| return { |
| "model_id": str(normalized.get("model_id") or "").strip(), |
| "daily_limit": daily_limit, |
| "limit_scope": limit_scope, |
| "used": used if limit_scope == "total" else 0, |
| "reserved": reserved if limit_scope == "total" else 0, |
| "remaining": max(0, daily_limit - used - reserved) if limit_scope == "total" else daily_limit, |
| "date": str(normalized.get("date") or "").strip() or _today_key(), |
| "updated_at": normalized.get("updated_at"), |
| } |
|
|
|
|
| def _normalize_text(value: object, default: str = "", max_length: int = 500) -> str: |
| if value is None: |
| return default |
| normalized = str(value).strip() |
| return normalized[:max_length] |
|
|
|
|
| def _normalize_branding_settings(value: dict[str, object] | None) -> dict[str, str]: |
| source = value if isinstance(value, dict) else {} |
| return { |
| "site_title": _normalize_text(source.get("site_title"), DEFAULT_BRANDING["site_title"], 80) or DEFAULT_BRANDING["site_title"], |
| "site_logo_url": _normalize_text(source.get("site_logo_url"), DEFAULT_BRANDING["site_logo_url"], 500), |
| "site_header_title": _normalize_text(source.get("site_header_title"), DEFAULT_BRANDING["site_header_title"], 40) or DEFAULT_BRANDING["site_header_title"], |
| "site_external_link_label": _normalize_text(source.get("site_external_link_label"), DEFAULT_BRANDING["site_external_link_label"], 40), |
| "site_external_link_url": _normalize_text(source.get("site_external_link_url"), DEFAULT_BRANDING["site_external_link_url"], 500), |
| "site_external_link_icon_url": _normalize_text(source.get("site_external_link_icon_url"), DEFAULT_BRANDING["site_external_link_icon_url"], 500), |
| "site_external_link_icon_svg": _normalize_text(source.get("site_external_link_icon_svg"), DEFAULT_BRANDING["site_external_link_icon_svg"], 6000), |
| } |
|
|
|
|
| def _normalize_backup_include(value: object) -> dict[str, bool]: |
| source = value if isinstance(value, dict) else {} |
| normalized = dict(DEFAULT_BACKUP_INCLUDE) |
| for key in normalized: |
| normalized[key] = _normalize_bool(source.get(key), normalized[key]) |
| return normalized |
|
|
|
|
| def _normalize_backup_settings(value: object) -> dict[str, object]: |
| source = value if isinstance(value, dict) else {} |
| return { |
| "enabled": _normalize_bool(source.get("enabled"), False), |
| "provider": "cloudflare_r2", |
| "account_id": str(source.get("account_id") or "").strip(), |
| "access_key_id": str(source.get("access_key_id") or "").strip(), |
| "secret_access_key": str(source.get("secret_access_key") or "").strip(), |
| "bucket": str(source.get("bucket") or "").strip(), |
| "prefix": str(source.get("prefix") or "backups").strip().strip("/") or "backups", |
| "interval_minutes": _normalize_positive_int(source.get("interval_minutes"), 360, 1), |
| "rotation_keep": _normalize_positive_int(source.get("rotation_keep"), 10, 0), |
| "encrypt": _normalize_bool(source.get("encrypt"), False), |
| "passphrase": str(source.get("passphrase") or "").strip(), |
| "include": _normalize_backup_include(source.get("include")), |
| } |
|
|
|
|
| def _normalize_backup_state(value: object) -> dict[str, object]: |
| source = value if isinstance(value, dict) else {} |
| return { |
| "last_started_at": str(source.get("last_started_at") or "").strip() or None, |
| "last_finished_at": str(source.get("last_finished_at") or "").strip() or None, |
| "last_status": str(source.get("last_status") or "idle").strip() or "idle", |
| "last_error": str(source.get("last_error") or "").strip() or None, |
| "last_object_key": str(source.get("last_object_key") or "").strip() or None, |
| } |
|
|
|
|
| def _normalize_image_storage_settings(value: object) -> dict[str, object]: |
| source = value if isinstance(value, dict) else {} |
| mode = str(source.get("mode") or "local").strip().lower() |
| if mode not in {"local", "webdav", "both"}: |
| mode = "local" |
| enabled = _normalize_bool(source.get("enabled"), False) |
| if not enabled: |
| mode = "local" |
| root_path = str(source.get("webdav_root_path") or DEFAULT_IMAGE_STORAGE["webdav_root_path"]).strip().strip("/") |
| return { |
| "enabled": enabled, |
| "mode": mode, |
| "webdav_url": str(source.get("webdav_url") or "").strip().rstrip("/"), |
| "webdav_username": str(source.get("webdav_username") or "").strip(), |
| "webdav_password": str(source.get("webdav_password") or "").strip(), |
| "webdav_root_path": root_path or str(DEFAULT_IMAGE_STORAGE["webdav_root_path"]), |
| "public_base_url": str(source.get("public_base_url") or "").strip().rstrip("/"), |
| } |
|
|
|
|
| def _normalize_chat_completion_cache_settings(value: object) -> dict[str, object]: |
| source = value if isinstance(value, dict) else {} |
| return { |
| "enabled": _normalize_bool(source.get("enabled"), bool(DEFAULT_CHAT_COMPLETION_CACHE["enabled"])), |
| "ttl_seconds": _normalize_positive_int(source.get("ttl_seconds"), int(DEFAULT_CHAT_COMPLETION_CACHE["ttl_seconds"]), 0), |
| "max_entries": _normalize_positive_int(source.get("max_entries"), int(DEFAULT_CHAT_COMPLETION_CACHE["max_entries"]), 1), |
| "dedupe_inflight": _normalize_bool(source.get("dedupe_inflight"), bool(DEFAULT_CHAT_COMPLETION_CACHE["dedupe_inflight"])), |
| "stream_cache": _normalize_bool(source.get("stream_cache"), bool(DEFAULT_CHAT_COMPLETION_CACHE["stream_cache"])), |
| "normalize_messages": _normalize_bool(source.get("normalize_messages"), bool(DEFAULT_CHAT_COMPLETION_CACHE["normalize_messages"])), |
| "drop_adjacent_duplicates": _normalize_bool(source.get("drop_adjacent_duplicates"), bool(DEFAULT_CHAT_COMPLETION_CACHE["drop_adjacent_duplicates"])), |
| "drop_assistant_history": _normalize_bool(source.get("drop_assistant_history"), bool(DEFAULT_CHAT_COMPLETION_CACHE["drop_assistant_history"])), |
| } |
|
|
|
|
| MAX_CHAT_MODEL_ENTRIES = 300 |
| MAX_CHAT_MODEL_ID_LENGTH = 140 |
|
|
|
|
| def _normalize_model_id_list(value: object) -> list[str]: |
| items = value if isinstance(value, list) else [] |
| normalized: list[str] = [] |
| seen: set[str] = set() |
| for item in items: |
| model_id = str(item or "").strip()[:MAX_CHAT_MODEL_ID_LENGTH] |
| if not model_id or model_id in seen: |
| continue |
| seen.add(model_id) |
| normalized.append(model_id) |
| if len(normalized) >= MAX_CHAT_MODEL_ENTRIES: |
| break |
| return normalized |
|
|
|
|
| def _normalize_model_overrides(value: object) -> dict[str, list[str]]: |
| """模型下拉的人工覆盖项（对话与生图共用同一套结构）。 |
| |
| ``include`` 强制显示某个模型，``exclude`` 强制隐藏某个模型； |
| 都没命中的模型交给自动规则决定（services/chat_model_service.py / |
| services/image_model_service.py）。两个名单冲突时以 include 为准。 |
| """ |
|
|
| source = value if isinstance(value, dict) else {} |
| include = _normalize_model_id_list(source.get("include")) |
| include_set = set(include) |
| exclude = [item for item in _normalize_model_id_list(source.get("exclude")) if item not in include_set] |
| return {"include": include, "exclude": exclude} |
|
|
|
|
| def _normalize_chat_model_overrides(value: object) -> dict[str, list[str]]: |
| """对话模型下拉的人工覆盖项。""" |
|
|
| return _normalize_model_overrides(value) |
|
|
|
|
| def _normalize_image_model_overrides(value: object) -> dict[str, list[str]]: |
| """生图模型下拉的人工覆盖项。""" |
|
|
| return _normalize_model_overrides(value) |
|
|
|
|
| def _normalize_system_model_overrides(value: object) -> dict[str, list[str]]: |
| """「系统可用模型」列表的人工覆盖项。""" |
|
|
| return _normalize_model_overrides(value) |
|
|
|
|
| def _normalize_status_codes(value: object) -> list[int]: |
| items = value if isinstance(value, list) else DEFAULT_PROXY_RUNTIME["reset_session_status_codes"] |
| normalized: list[int] = [] |
| for item in items: |
| if isinstance(item, bool): |
| continue |
| try: |
| status = int(item) |
| except (OverflowError, TypeError, ValueError): |
| continue |
| if 100 <= status <= 599 and status not in normalized: |
| normalized.append(status) |
| return normalized or list(DEFAULT_PROXY_RUNTIME["reset_session_status_codes"]) |
|
|
|
|
| def _normalize_proxy_runtime_settings(value: object) -> dict[str, object]: |
| source = value if isinstance(value, dict) else {} |
| default_clearance = DEFAULT_PROXY_RUNTIME["clearance"] |
| clearance_source = source.get("clearance") if isinstance(source.get("clearance"), dict) else {} |
| egress_mode = str(source.get("egress_mode") or DEFAULT_PROXY_RUNTIME["egress_mode"]).strip().lower() |
| if egress_mode not in {"direct", "single_proxy"}: |
| egress_mode = str(DEFAULT_PROXY_RUNTIME["egress_mode"]) |
| clearance_mode = str(clearance_source.get("mode") or default_clearance["mode"]).strip().lower() |
| if clearance_mode not in {"none", "manual", "flaresolverr"}: |
| clearance_mode = str(default_clearance["mode"]) |
| existing_cookies = str(source.get("_existing_cf_cookies") or "").strip() |
| existing_clearance = str(source.get("_existing_cf_clearance") or "").strip() |
| cf_cookies = str(clearance_source.get("cf_cookies") or "").strip() |
| cf_clearance = str(clearance_source.get("cf_clearance") or "").strip() |
| if not cf_cookies and _normalize_bool(clearance_source.get("has_cf_cookies"), False): |
| cf_cookies = existing_cookies |
| if not cf_clearance and _normalize_bool(clearance_source.get("has_cf_clearance"), False): |
| cf_clearance = existing_clearance |
| return { |
| "enabled": _normalize_bool(source.get("enabled"), bool(DEFAULT_PROXY_RUNTIME["enabled"])), |
| "egress_mode": egress_mode, |
| "proxy_url": str(source.get("proxy_url") or "").strip(), |
| "resource_proxy_url": str(source.get("resource_proxy_url") or "").strip(), |
| "skip_ssl_verify": _normalize_bool(source.get("skip_ssl_verify"), bool(DEFAULT_PROXY_RUNTIME["skip_ssl_verify"])), |
| "reset_session_status_codes": _normalize_status_codes(source.get("reset_session_status_codes")), |
| "clearance": { |
| "enabled": _normalize_bool(clearance_source.get("enabled"), bool(default_clearance["enabled"])), |
| "mode": clearance_mode, |
| "cf_cookies": cf_cookies, |
| "cf_clearance": cf_clearance, |
| "user_agent": str(clearance_source.get("user_agent") or default_clearance["user_agent"]).strip(), |
| "browser": str(clearance_source.get("browser") or default_clearance["browser"]).strip(), |
| "flaresolverr_url": str(clearance_source.get("flaresolverr_url") or "").strip(), |
| "timeout_sec": _normalize_positive_int(clearance_source.get("timeout_sec"), int(default_clearance["timeout_sec"]), 1), |
| "refresh_interval": _normalize_positive_int(clearance_source.get("refresh_interval"), int(default_clearance["refresh_interval"]), 60), |
| "warm_up_on_start": _normalize_bool(clearance_source.get("warm_up_on_start"), bool(default_clearance["warm_up_on_start"])), |
| }, |
| } |
|
|
|
|
| def _normalize_free_account_cleanup_settings(value: object) -> dict[str, object]: |
| source = value if isinstance(value, dict) else {} |
| action = str(source.get("action") or DEFAULT_FREE_ACCOUNT_CLEANUP["action"]).strip().lower() |
| if action not in {"mark_abnormal", "delete"}: |
| action = str(DEFAULT_FREE_ACCOUNT_CLEANUP["action"]) |
| return { |
| "enabled": _normalize_bool(source.get("enabled"), bool(DEFAULT_FREE_ACCOUNT_CLEANUP["enabled"])), |
| "interval_minutes": _normalize_positive_int( |
| source.get("interval_minutes"), |
| int(DEFAULT_FREE_ACCOUNT_CLEANUP["interval_minutes"]), |
| 1, |
| ), |
| "failure_threshold": _normalize_positive_int( |
| source.get("failure_threshold"), |
| int(DEFAULT_FREE_ACCOUNT_CLEANUP["failure_threshold"]), |
| 1, |
| ), |
| "register_precheck_enabled": _normalize_bool( |
| source.get("register_precheck_enabled"), |
| bool(DEFAULT_FREE_ACCOUNT_CLEANUP["register_precheck_enabled"]), |
| ), |
| "action": action, |
| } |
|
|
|
|
| def _normalize_redeem_code_cleanup_settings(value: object) -> dict[str, object]: |
| source = value if isinstance(value, dict) else {} |
| return { |
| "enabled": _normalize_bool(source.get("enabled"), bool(DEFAULT_REDEEM_CODE_CLEANUP["enabled"])), |
| "interval_minutes": _normalize_positive_int( |
| source.get("interval_minutes"), |
| int(DEFAULT_REDEEM_CODE_CLEANUP["interval_minutes"]), |
| 1, |
| ), |
| "delete_expired": _normalize_bool( |
| source.get("delete_expired"), |
| bool(DEFAULT_REDEEM_CODE_CLEANUP["delete_expired"]), |
| ), |
| "delete_disabled": _normalize_bool( |
| source.get("delete_disabled"), |
| bool(DEFAULT_REDEEM_CODE_CLEANUP["delete_disabled"]), |
| ), |
| "archive_redeemed": _normalize_bool( |
| source.get("archive_redeemed"), |
| bool(DEFAULT_REDEEM_CODE_CLEANUP["archive_redeemed"]), |
| ), |
| "redeemed_older_than_days": _normalize_positive_int( |
| source.get("redeemed_older_than_days"), |
| int(DEFAULT_REDEEM_CODE_CLEANUP["redeemed_older_than_days"]), |
| 1, |
| ), |
| "limit": min( |
| 20000, |
| _normalize_positive_int(source.get("limit"), int(DEFAULT_REDEEM_CODE_CLEANUP["limit"]), 1), |
| ), |
| } |
|
|
|
|
| def _normalize_navigation_url(value: object) -> str: |
| text = str(value or "").strip()[:1000] |
| if not text: |
| return "" |
| if text.startswith("/") and not text.startswith("//"): |
| return text |
| parsed = urlparse(text) |
| return text if parsed.scheme in {"http", "https"} and parsed.netloc else "" |
|
|
|
|
| DEFAULT_ADMIN_USER_NAVIGATION: list[dict[str, object]] = [ |
| {"id": "admin-image", "title": "画图", "url": "/image"}, |
| {"id": "admin-batch", "title": "批量生图", "url": "/batch"}, |
| {"id": "admin-chat", "title": "对话", "url": "/chat"}, |
| {"id": "admin-chat-room", "title": "聊天室", "url": "/chat-room"}, |
| {"id": "admin-product-visual", "title": "商品视觉", "url": "/product-visual"}, |
| {"id": "admin-accounts", "title": "号池管理", "url": "/accounts"}, |
| {"id": "admin-register", "title": "注册机", "url": "/register"}, |
| {"id": "admin-image-manager", "title": "图片管理", "url": "/image-manager"}, |
| {"id": "admin-logs", "title": "日志管理", "url": "/logs"}, |
| {"id": "admin-dashboard", "title": "数据看板", "url": "/dashboard"}, |
| {"id": "admin-debug", "title": "调试", "url": "/debug"}, |
| {"id": "admin-settings", "title": "设置", "url": "/settings"}, |
| ] |
|
|
|
|
| def _normalize_navigation_items(value: object, *, allow_children: bool = True) -> list[dict[str, object]]: |
| source = value if isinstance(value, list) else [] |
| normalized: list[dict[str, object]] = [] |
| used_ids: set[str] = set() |
| for index, raw in enumerate(source[:30]): |
| if not isinstance(raw, dict): |
| continue |
| title = _normalize_text(raw.get("title") or raw.get("label"), "", 40) |
| url = _normalize_navigation_url(raw.get("url") or raw.get("href")) |
| children = _normalize_navigation_items(raw.get("children"), allow_children=False) if allow_children else [] |
| if not title or (not url and not children): |
| continue |
| raw_id = _normalize_text(raw.get("id"), "", 64) or f"navigation-{index + 1}" |
| item_id = raw_id |
| suffix = 2 |
| while item_id in used_ids: |
| item_id = f"{raw_id[:56]}-{suffix}" |
| suffix += 1 |
| used_ids.add(item_id) |
| normalized.append({ |
| "id": item_id, |
| "title": title, |
| "url": url, |
| "new_window": _normalize_bool(raw.get("new_window"), False), |
| "append_api_credentials": _normalize_bool(raw.get("append_api_credentials"), False), |
| "api_key_param": _normalize_text(raw.get("api_key_param"), "apiKey", 64), |
| "base_url_param": _normalize_text(raw.get("base_url_param"), "baseUrl", 64), |
| "enabled": _normalize_bool(raw.get("enabled"), True), |
| "children": children, |
| }) |
| return normalized |
|
|
|
|
| def _normalize_image_proxy_base_url(value: object) -> str: |
| text = str(value or "").strip().rstrip("/") |
| if not text: |
| return "" |
| parsed = urlparse(text) |
| if parsed.scheme not in {"http", "https"} or not parsed.netloc: |
| raise ValueError("图片加速地址必须是完整的 http/https 地址") |
| return text |
|
|
|
|
| def _validate_image_storage_settings(settings: dict[str, object]) -> None: |
| if not _normalize_bool(settings.get("enabled"), False): |
| return |
| if not str(settings.get("webdav_url") or "").strip(): |
| raise ValueError("启用 WebDAV 图片存储后必须填写 WebDAV URL") |
| if not str(settings.get("webdav_password") or "").strip(): |
| raise ValueError("启用 WebDAV 图片存储后必须填写 WebDAV 密码") |
|
|
|
|
| def _normalize_announcements(value: object) -> list[dict[str, object]]: |
| source = value if isinstance(value, list) else [] |
| timeline_items: list[dict[str, object]] = [] |
| notice_items: dict[str, dict[str, object]] = {} |
| for index, item in enumerate(source): |
| if not isinstance(item, dict): |
| continue |
| title = str(item.get("title") or "").strip() |
| content = str(item.get("content") or "").strip() |
| if not title and not content: |
| continue |
| item_type = str(item.get("type") or "notice").strip().lower() |
| if item_type not in ANNOUNCEMENT_TYPES: |
| item_type = "notice" |
| target = str(item.get("target") or "all").strip().lower() |
| if target not in ANNOUNCEMENT_TARGETS: |
| target = "all" |
| if item_type == "notice" and target not in {"internal", "normal"}: |
| target = "normal" |
| created_at = str(item.get("created_at") or "").strip() |
| updated_at = str(item.get("updated_at") or created_at).strip() |
| item_id = str(item.get("id") or "").strip() or f"announcement-{index + 1}" |
| normalized_item = { |
| "id": item_id[:64], |
| "type": item_type, |
| "target": target, |
| "title": title[:80], |
| "content": content[:2000], |
| "enabled": _normalize_bool(item.get("enabled"), True), |
| "created_at": created_at, |
| "updated_at": updated_at or created_at, |
| } |
| if item_type == "notice": |
| |
| notice_items[target] = normalized_item |
| else: |
| timeline_items.append(normalized_item) |
| return [*notice_items.values(), *timeline_items[-100:]] |
|
|
|
|
| def _normalize_checkin_content(value: object, fallback: object = None) -> dict[str, str]: |
| """签到页文案。 |
| |
| 缺键时先沿用已保存的值（``fallback``），再回落到默认文案 —— 配置是嵌套 dict， |
| ``update()`` 只做顶层合并，不这样兜底的话「只改一个字段」会把其余字段打回默认。 |
| 显式给出空串表示「不显示这一段」。 |
| """ |
| source = value if isinstance(value, dict) else {} |
| previous = fallback if isinstance(fallback, dict) else {} |
| normalized: dict[str, str] = {} |
| for key, default in DEFAULT_CHECKIN_CONTENT.items(): |
| raw = source[key] if key in source else previous.get(key) |
| if raw is None: |
| normalized[key] = default |
| continue |
| |
| if isinstance(raw, (list, tuple)): |
| normalized[key] = "\n".join(str(item).strip() for item in raw if str(item).strip()) |
| continue |
| normalized[key] = str(raw).strip() |
| return normalized |
|
|
|
|
| def _normalize_error_message_rules(value: object) -> list[dict[str, object]]: |
| """归一化前台报错文案映射规则。 |
| |
| 真正的校验逻辑在 error_message_service.normalize_rules；此处延迟导入， |
| 因为该模块会反向读取 config，导入期不能互相依赖。 |
| """ |
| from services.error_message_service import normalize_rules |
|
|
| return normalize_rules(value) |
|
|
|
|
| def _normalize_shared_settings(value: object, fallback: dict[str, object] | None = None) -> dict[str, object]: |
| source = value if isinstance(value, dict) else {} |
| fallback = fallback or {} |
| pool_pricing = _normalize_normal_user_pool_pricing( |
| source.get("normal_user_pool_pricing", fallback.get("normal_user_pool_pricing")) |
| ) |
| third_party_source = source.get("normal_user_third_party_image_pricing") |
| if not isinstance(third_party_source, dict): |
| legacy_third_party_rules = [ |
| rule for rule in pool_pricing.get("rules", []) |
| if isinstance(rule, dict) and rule.get("target_type") == "third_party_group" |
| ] |
| if legacy_third_party_rules: |
| pool_pricing["rules"] = [ |
| rule for rule in pool_pricing.get("rules", []) |
| if not (isinstance(rule, dict) and rule.get("target_type") == "third_party_group") |
| ] |
| third_party_source = { |
| **pool_pricing, |
| "enabled": bool(pool_pricing.get("enabled")), |
| "rules": legacy_third_party_rules, |
| } |
| return { |
| "normal_user_default_credits": _normalize_positive_int( |
| source.get("normal_user_default_credits"), |
| _normalize_positive_int(fallback.get("normal_user_default_credits"), 0, 0), |
| 0, |
| ), |
| "normal_user_daily_credits": _normalize_positive_int( |
| source.get("normal_user_daily_credits"), |
| _normalize_positive_int(fallback.get("normal_user_daily_credits"), 0, 0), |
| 0, |
| ), |
| "checkin_daily_credits": _normalize_positive_int( |
| source.get("checkin_daily_credits"), |
| _normalize_positive_int(fallback.get("checkin_daily_credits"), _normalize_positive_int(source.get("normal_user_daily_credits"), _normalize_positive_int(fallback.get("normal_user_daily_credits"), 0, 0), 0), 0), |
| 0, |
| ), |
| "checkin_random_min": _normalize_positive_int(source.get("checkin_random_min"), _normalize_positive_int(fallback.get("checkin_random_min"), 6, 1), 1), |
| "checkin_random_max": max( |
| _normalize_positive_int(source.get("checkin_random_max"), _normalize_positive_int(fallback.get("checkin_random_max"), 10, 1), 1), |
| _normalize_positive_int(source.get("checkin_random_min"), _normalize_positive_int(fallback.get("checkin_random_min"), 6, 1), 1), |
| ), |
| "checkin_invite_reward": _normalize_positive_int(source.get("checkin_invite_reward"), _normalize_positive_int(fallback.get("checkin_invite_reward"), 30, 0), 0), |
| |
| "checkin_credit_valid_days": min( |
| 3650, |
| _normalize_positive_int( |
| source.get("checkin_credit_valid_days"), |
| _normalize_positive_int(fallback.get("checkin_credit_valid_days"), 0, 0), |
| 0, |
| ), |
| ), |
| "internal_image_model_daily_limits": _normalize_internal_image_model_daily_limits( |
| source.get("internal_image_model_daily_limits", fallback.get("internal_image_model_daily_limits")) |
| ), |
| "wechat_login_enabled": _normalize_bool( |
| source.get("wechat_login_enabled"), |
| _normalize_bool(fallback.get("wechat_login_enabled"), True), |
| ), |
| "wechat_auto_register_enabled": _normalize_bool( |
| source.get("wechat_auto_register_enabled"), |
| _normalize_bool(fallback.get("wechat_auto_register_enabled"), True), |
| ), |
| "internal_user_login_session_duration_hours": _normalize_positive_int( |
| source.get("internal_user_login_session_duration_hours"), |
| _normalize_positive_int(fallback.get("internal_user_login_session_duration_hours"), 0, 0), |
| 0, |
| ), |
| "normal_user_login_session_duration_hours": _normalize_positive_int( |
| source.get("normal_user_login_session_duration_hours"), |
| _normalize_positive_int( |
| fallback.get("normal_user_login_session_duration_hours", fallback.get("login_session_duration_hours")), |
| 72, |
| 0, |
| ), |
| 0, |
| ), |
| "prompt_optimizer_system_prompt": str( |
| source.get( |
| "prompt_optimizer_system_prompt", |
| fallback.get("prompt_optimizer_system_prompt", DEFAULT_PROMPT_OPTIMIZER_SYSTEM_PROMPT), |
| ) |
| or DEFAULT_PROMPT_OPTIMIZER_SYSTEM_PROMPT |
| ).strip(), |
| "prompt_optimizer_provider": str(source.get("prompt_optimizer_provider", fallback.get("prompt_optimizer_provider", "internal")) or "internal").strip().lower(), |
| "prompt_optimizer_base_url": str(source.get("prompt_optimizer_base_url", fallback.get("prompt_optimizer_base_url", "")) or "").strip(), |
| "prompt_optimizer_api_key": str(source.get("prompt_optimizer_api_key", fallback.get("prompt_optimizer_api_key", "")) or "").strip(), |
| "prompt_optimizer_model": str(source.get("prompt_optimizer_model", fallback.get("prompt_optimizer_model", "gpt-5-5")) or "gpt-5-5").strip(), |
| "reverse_prompt_system_prompt": str( |
| source.get( |
| "reverse_prompt_system_prompt", |
| fallback.get("reverse_prompt_system_prompt", DEFAULT_REVERSE_PROMPT_SYSTEM_PROMPT), |
| ) |
| or DEFAULT_REVERSE_PROMPT_SYSTEM_PROMPT |
| ).strip(), |
| "reverse_prompt_provider": str(source.get("reverse_prompt_provider", fallback.get("reverse_prompt_provider", "internal")) or "internal").strip().lower(), |
| "reverse_prompt_base_url": str(source.get("reverse_prompt_base_url", fallback.get("reverse_prompt_base_url", "")) or "").strip(), |
| "reverse_prompt_api_key": str(source.get("reverse_prompt_api_key", fallback.get("reverse_prompt_api_key", "")) or "").strip(), |
| "reverse_prompt_model": str(source.get("reverse_prompt_model", fallback.get("reverse_prompt_model", "gpt-5-5")) or "gpt-5-5").strip(), |
| "product_visual_planner_system_prompt": str( |
| source.get( |
| "product_visual_planner_system_prompt", |
| fallback.get("product_visual_planner_system_prompt", DEFAULT_PRODUCT_VISUAL_PLANNER_SYSTEM_PROMPT), |
| ) |
| or DEFAULT_PRODUCT_VISUAL_PLANNER_SYSTEM_PROMPT |
| ).strip(), |
| "product_visual_agent_provider": str( |
| source.get( |
| "product_visual_agent_provider", |
| source.get( |
| "prompt_optimizer_provider", |
| fallback.get("product_visual_agent_provider", fallback.get("prompt_optimizer_provider", "internal")), |
| ), |
| ) |
| or "internal" |
| ).strip().lower(), |
| "product_visual_agent_base_url": str( |
| source.get( |
| "product_visual_agent_base_url", |
| source.get( |
| "prompt_optimizer_base_url", |
| fallback.get("product_visual_agent_base_url", fallback.get("prompt_optimizer_base_url", "")), |
| ), |
| ) |
| or "" |
| ).strip(), |
| "product_visual_agent_api_key": str( |
| source.get( |
| "product_visual_agent_api_key", |
| source.get( |
| "prompt_optimizer_api_key", |
| fallback.get("product_visual_agent_api_key", fallback.get("prompt_optimizer_api_key", "")), |
| ), |
| ) |
| or "" |
| ).strip(), |
| "product_visual_agent_model": str( |
| source.get( |
| "product_visual_agent_model", |
| source.get( |
| "prompt_optimizer_model", |
| fallback.get("product_visual_agent_model", fallback.get("prompt_optimizer_model", "gpt-5-5")), |
| ), |
| ) |
| or "gpt-5-5" |
| ).strip(), |
| "normal_user_pool_pricing": pool_pricing, |
| "normal_user_third_party_image_pricing": _normalize_normal_user_pool_pricing( |
| third_party_source |
| if isinstance(third_party_source, dict) |
| else fallback.get("normal_user_third_party_image_pricing", DEFAULT_NORMAL_USER_THIRD_PARTY_IMAGE_PRICING) |
| ), |
| |
| |
| |
| "error_message_rules": _normalize_error_message_rules( |
| source.get("error_message_rules", fallback.get("error_message_rules")) |
| ), |
| "error_message_fallback_enabled": _normalize_bool( |
| source.get("error_message_fallback_enabled"), |
| _normalize_bool(fallback.get("error_message_fallback_enabled"), False), |
| ), |
| |
| "error_message_ai_model": str( |
| source.get("error_message_ai_model", fallback.get("error_message_ai_model", "")) or "" |
| ).strip(), |
| } |
|
|
|
|
| @dataclass(frozen=True) |
| class LoadedSettings: |
| auth_key: str |
| refresh_account_interval_minute: int |
|
|
|
|
| def _normalize_auth_key(value: object) -> str: |
| return str(value or "").strip() |
|
|
|
|
| def _is_invalid_auth_key(value: object) -> bool: |
| return _normalize_auth_key(value) == "" |
|
|
|
|
| def _read_json_object(path: Path, *, name: str) -> dict[str, object]: |
| if not path.exists(): |
| return {} |
| if path.is_dir(): |
| print( |
| f"Warning: {name} at '{path}' is a directory, ignoring it and falling back to other configuration sources.", |
| file=sys.stderr, |
| ) |
| return {} |
| try: |
| data = json.loads(path.read_text(encoding="utf-8")) |
| except Exception: |
| return {} |
| return data if isinstance(data, dict) else {} |
|
|
|
|
| def _load_settings() -> LoadedSettings: |
| DATA_DIR.mkdir(parents=True, exist_ok=True) |
| raw_config = _read_json_object(CONFIG_FILE, name="config.json") |
| auth_key = _normalize_auth_key(os.getenv("CHATGPT2API_AUTH_KEY") or raw_config.get("auth-key")) |
| if _is_invalid_auth_key(auth_key): |
| raise ValueError( |
| "❌ auth-key 未设置！\n" |
| "请在环境变量 CHATGPT2API_AUTH_KEY 中设置，或者在 config.json 中填写 auth-key。" |
| ) |
|
|
| try: |
| refresh_interval = int(raw_config.get("refresh_account_interval_minute", 5)) |
| except (TypeError, ValueError): |
| refresh_interval = 5 |
|
|
| return LoadedSettings( |
| auth_key=auth_key, |
| refresh_account_interval_minute=refresh_interval, |
| ) |
|
|
|
|
| class ConfigStore: |
| def __init__(self, path: Path): |
| self.path = path |
| DATA_DIR.mkdir(parents=True, exist_ok=True) |
| self.data = self._load() |
| if self._migrate_chat_navigation(): |
| self._save() |
| if self._migrate_chat_room_navigation(): |
| self._save() |
| self._storage_backend: StorageBackend | None = None |
| self._user_storage_backend: StorageBackend | None = None |
| self._account_pool_storage_backend: StorageBackend | None = None |
| self._shared_settings_cache: dict[str, object] | None = None |
| self._shared_settings_cache_at = 0.0 |
| self._shared_settings_lock = Lock() |
| if _is_invalid_auth_key(self.auth_key): |
| raise ValueError( |
| "❌ auth-key 未设置！\n" |
| "请按以下任意一种方式解决：\n" |
| "1. 在 Render 的 Environment 变量中添加：\n" |
| " CHATGPT2API_AUTH_KEY = your_real_auth_key\n" |
| "2. 或者在 config.json 中填写：\n" |
| ' "auth-key": "your_real_auth_key"' |
| ) |
| self._prime_shared_announcements() |
| self._prime_shared_settings() |
|
|
| def _load(self) -> dict[str, object]: |
| return _read_json_object(self.path, name="config.json") |
|
|
| def _save(self) -> None: |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
| self.path.write_text(json.dumps(self.data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") |
|
|
| def _migrate_chat_navigation(self) -> bool: |
| """Add the new chat surface once without overriding later admin choices.""" |
| if self.data.get(CHAT_NAVIGATION_MIGRATION_KEY): |
| return False |
| for audience, key in ( |
| ("normal", "normal_user_navigation"), |
| ("internal", "internal_user_navigation"), |
| ("admin", "admin_user_navigation"), |
| ): |
| raw = self.data.get(key) |
| if not isinstance(raw, list) or any( |
| isinstance(item, dict) and str(item.get("url") or "").rstrip("/") == "/chat" |
| for item in raw |
| ): |
| continue |
| image_index = next(( |
| index |
| for index, item in enumerate(raw) |
| if isinstance(item, dict) and str(item.get("url") or "").rstrip("/") == "/image" |
| ), -1) |
| if image_index < 0: |
| continue |
| raw.insert(image_index + 1, { |
| "id": f"{audience}-chat", |
| "title": "对话", |
| "url": "/chat", |
| "new_window": False, |
| "append_api_credentials": False, |
| "api_key_param": "apiKey", |
| "base_url_param": "baseUrl", |
| "enabled": True, |
| "children": [], |
| }) |
| self.data[CHAT_NAVIGATION_MIGRATION_KEY] = True |
| return True |
|
|
| def _migrate_chat_room_navigation(self) -> bool: |
| """把「聊天室」入口补进三套导航，只补一次（不覆盖站长后来的增删）。 |
| |
| 与 `_migrate_chat_navigation` 同一套路：只在导航里确实没有 /chat-room 时插入， |
| 插到「对话」之后（找不到 /chat 就追加到末尾），并对已存在 /chat-room 的配置跳过。 |
| """ |
| if self.data.get(CHAT_ROOM_NAVIGATION_MIGRATION_KEY): |
| return False |
| for audience, key in ( |
| ("normal", "normal_user_navigation"), |
| ("internal", "internal_user_navigation"), |
| ("admin", "admin_user_navigation"), |
| ): |
| raw = self.data.get(key) |
| if not isinstance(raw, list) or any( |
| isinstance(item, dict) and str(item.get("url") or "").rstrip("/") == "/chat-room" |
| for item in raw |
| ): |
| continue |
| chat_index = next(( |
| index |
| for index, item in enumerate(raw) |
| if isinstance(item, dict) and str(item.get("url") or "").rstrip("/") == "/chat" |
| ), -1) |
| entry = { |
| "id": f"{audience}-chat-room", |
| "title": "聊天室", |
| "url": "/chat-room", |
| "new_window": False, |
| "append_api_credentials": False, |
| "api_key_param": "apiKey", |
| "base_url_param": "baseUrl", |
| "enabled": True, |
| "children": [], |
| } |
| if chat_index < 0: |
| raw.append(entry) |
| else: |
| raw.insert(chat_index + 1, entry) |
| self.data[CHAT_ROOM_NAVIGATION_MIGRATION_KEY] = True |
| return True |
|
|
| @property |
| def auth_key(self) -> str: |
| return _normalize_auth_key(os.getenv("CHATGPT2API_AUTH_KEY") or self.data.get("auth-key")) |
|
|
| @property |
| def accounts_file(self) -> Path: |
| return DATA_DIR / "accounts.json" |
|
|
| @property |
| def thread_pool_tokens(self) -> int: |
| """Capacity of the shared anyio worker-thread pool (anyio default: 40).""" |
| try: |
| return max(4, min(512, int(self.data.get("thread_pool_tokens", 40)))) |
| except (TypeError, ValueError): |
| return 40 |
|
|
| @property |
| def chat_models(self) -> dict[str, list[str]]: |
| """对话页模型下拉的人工覆盖项（include 强制显示 / exclude 强制隐藏）。 |
| |
| 为空时由 services/chat_model_service.py 的结构化规则自动决定， |
| 所以上游新增模型会自动出现在对话下拉里，不需要人工维护名单。 |
| """ |
|
|
| return _normalize_chat_model_overrides(self.data.get("chat_models")) |
|
|
| @property |
| def image_models(self) -> dict[str, list[str]]: |
| """生图模型下拉的人工覆盖项（include 强制显示 / exclude 强制隐藏）。 |
| |
| 默认全部启用（含第三方接入的生图模型）——这里的开关只做展示层过滤， |
| 不改变第三方生图配置本身（计费、路由、额度、对外 API 均不受影响）。 |
| """ |
|
|
| return _normalize_image_model_overrides(self.data.get("image_models")) |
|
|
| @property |
| def system_models(self) -> dict[str, list[str]]: |
| """「系统可用模型」列表的人工覆盖项（include 强制显示 / exclude 强制隐藏）。 |
| |
| 默认全部启用。这里的开关只做展示层过滤，不影响上游账号池、第三方配置、 |
| 计费、路由，也不影响对外 /v1/models（后者必须保持全量）。 |
| """ |
|
|
| return _normalize_system_model_overrides(self.data.get("system_models")) |
|
|
| @property |
| def refresh_account_interval_minute(self) -> int: |
| try: |
| return int(self.data.get("refresh_account_interval_minute", 5)) |
| except (TypeError, ValueError): |
| return 5 |
|
|
| @property |
| def image_retention_days(self) -> int: |
| try: |
| return max(1, int(self.data.get("image_retention_days", 30))) |
| except (TypeError, ValueError): |
| return 30 |
|
|
| @property |
| def image_poll_timeout_secs(self) -> int: |
| try: |
| return max(1, int(self.data.get("image_poll_timeout_secs", 120))) |
| except (TypeError, ValueError): |
| return 120 |
|
|
| @property |
| def image_poll_max_attempts(self) -> int: |
| """Maximum account attempts, including the initial account.""" |
| try: |
| return max(1, min(10, int(self.data.get("image_poll_max_attempts", 2)))) |
| except (TypeError, ValueError): |
| return 2 |
|
|
| @property |
| def image_poll_interval_secs(self) -> float: |
| try: |
| return max(0.5, float(self.data.get("image_poll_interval_secs", 10.0))) |
| except (TypeError, ValueError): |
| return 10.0 |
|
|
| @property |
| def image_poll_initial_wait_secs(self) -> float: |
| """Image generation upstream takes ~30s; polling immediately wastes requests |
| and trips a transient 429. Default 10s gives the conversation document time |
| to commit before the first poll.""" |
| try: |
| return max(0.0, float(self.data.get("image_poll_initial_wait_secs", 10.0))) |
| except (TypeError, ValueError): |
| return 10.0 |
|
|
| @property |
| def image_check_before_hit_enabled(self) -> bool: |
| return _normalize_bool(self.data.get("image_check_before_hit_enabled"), True) |
|
|
| @property |
| def image_settle_enabled(self) -> bool: |
| return _normalize_bool(self.data.get("image_settle_enabled"), True) |
|
|
| @property |
| def image_settle_secs(self) -> float: |
| try: |
| return max(0.0, float(self.data.get("image_settle_secs", 2.0))) |
| except (TypeError, ValueError): |
| return 2.0 |
|
|
| @property |
| def image_timeout_retry_secs(self) -> int: |
| try: |
| return max(5, min(120, int(self.data.get("image_timeout_retry_secs", 30)))) |
| except (TypeError, ValueError): |
| return 30 |
|
|
| @property |
| def image_account_concurrency(self) -> int: |
| try: |
| return max(1, int(self.data.get("image_account_concurrency", 3))) |
| except (TypeError, ValueError): |
| return 3 |
|
|
| @property |
| def image_parallel_generation(self) -> bool: |
| return _normalize_bool(self.data.get("image_parallel_generation"), True) |
|
|
| @property |
| def image_prompt_aspect_hint(self) -> bool: |
| """是否在图片提示词尾部追加比例指令（如“输出宽高比：16:9。若上文中的宽高比或构图指令与此输出比例冲突，请以此比例为准。”）。 |
| |
| 账号池走 ChatGPT 网页会话，没有 aspect-ratio 这样的接口参数，比例只能写进文本。 |
| 默认开启：画图页的比例下拉默认是「智能比例」（size 为空），而 size 为空时**不追加**， |
| 所以开启不会影响默认出图；只有用户主动选了比例才追加，这时候追加正是他想要的结果—— |
| 否则账号池路线上那个下拉等于摆设。仅当 size 是比例（16:9）或像素尺寸（1024x1024） |
| 写法时才追加，自然语言值直接忽略。 |
| """ |
| return _normalize_bool(self.data.get("image_prompt_aspect_hint"), True) |
|
|
| @property |
| def image_remove_conversation_after_result(self) -> bool: |
| return _normalize_bool(self.data.get("image_remove_conversation_after_result"), False) |
|
|
| @property |
| def image_remove_conversation_always(self) -> bool: |
| return _normalize_bool(self.data.get("image_remove_conversation_always"), False) |
|
|
| @property |
| def image_memory_button_enabled(self) -> bool: |
| return _normalize_bool(self.data.get("image_memory_button_enabled"), True) |
|
|
| @property |
| def cleanup_shared_conversations(self) -> bool: |
| return _normalize_bool(self.data.get("cleanup_shared_conversations"), False) |
|
|
| @property |
| def image_share_evidence_enabled(self) -> bool: |
| """是否允许把生成结果「取证」为 ChatGPT 官方分享链接。 |
| |
| 开启后，画图页每张结果图的右上角会出现入口，用户按需生成一条 |
| ``https://chatgpt.com/share/{share_id}`` 匿名链接来自证「这确实是官网生成的」。 |
| |
| 默认关闭：分享页会把用户这次的完整提示词公开在 chatgpt.com 上， |
| 属于站长需要显式知情并承担的暴露面，不能默认打开。 |
| """ |
| return _normalize_bool(self.data.get("image_share_evidence_enabled"), False) |
|
|
| @property |
| def image_share_retention_hours(self) -> int: |
| """分享链接保留时长（小时），默认 12，范围 1-720。 |
| |
| ⚠️ 这个值同时决定「多久后删除分享」和「多久后隐藏上游会话」，两者必须一起做： |
| 实测隐藏会话等价于官方删除，会让分享链接立刻 404 |
| （``shared_conversation_deleted`` / ``source_conversation_deleted``）， |
| 所以保留期内绝不能提前隐藏会话。 |
| """ |
| try: |
| hours = int(self.data.get("image_share_retention_hours", 12)) |
| except (TypeError, ValueError): |
| hours = 12 |
| return max(1, min(720, hours)) |
|
|
| @property |
| def normal_user_default_credits(self) -> int: |
| return int(self._load_shared_settings().get("normal_user_default_credits") or 0) |
|
|
| @property |
| def normal_user_daily_credits(self) -> int: |
| return int(self._load_shared_settings().get("normal_user_daily_credits") or 0) |
|
|
| @property |
| def checkin_daily_credits(self) -> int: |
| return int(self._load_shared_settings().get("checkin_daily_credits") or 0) |
|
|
| @property |
| def checkin_random_min(self) -> int: |
| return int(self._load_shared_settings().get("checkin_random_min") or 6) |
|
|
| @property |
| def checkin_random_max(self) -> int: |
| return int(self._load_shared_settings().get("checkin_random_max") or 10) |
|
|
| @property |
| def checkin_invite_reward(self) -> int: |
| return int(self._load_shared_settings().get("checkin_invite_reward") or 30) |
|
|
| @property |
| def checkin_credit_valid_days(self) -> int: |
| """签到赠送积分的可用天数；0 表示永久有效（发放进基础积分池）。""" |
| return min(3650, max(0, int(self._load_shared_settings().get("checkin_credit_valid_days") or 0))) |
|
|
| @property |
| def error_message_rules(self) -> list[dict[str, object]]: |
| """前台报错文案替换规则（关键词 -> 大白话）。""" |
| return _normalize_error_message_rules(self._load_shared_settings().get("error_message_rules")) |
|
|
| @property |
| def error_message_fallback_enabled(self) -> bool: |
| """未命中规则时是否强制隐藏原始报错（默认关闭，即照原样显示）。""" |
| return _normalize_bool(self._load_shared_settings().get("error_message_fallback_enabled"), False) |
|
|
| @property |
| def error_message_ai_model(self) -> str: |
| """「AI 归纳」使用的模型名；空串表示用池子默认文本模型。""" |
| return str(self._load_shared_settings().get("error_message_ai_model") or "").strip() |
|
|
| @property |
| def checkin_content(self) -> dict[str, str]: |
| return _normalize_checkin_content(self.data.get("checkin_content")) |
|
|
| @property |
| def internal_image_model_daily_limits(self) -> list[dict[str, object]]: |
| return [ |
| _internal_image_model_daily_limit_public(item) |
| for item in _normalize_internal_image_model_daily_limits( |
| self._load_shared_settings().get("internal_image_model_daily_limits") |
| ) |
| ] |
|
|
| @property |
| def wechat_login_enabled(self) -> bool: |
| return _normalize_bool(self._load_shared_settings().get("wechat_login_enabled"), True) |
|
|
| @property |
| def wechat_auto_register_enabled(self) -> bool: |
| return _normalize_bool(self._load_shared_settings().get("wechat_auto_register_enabled"), True) |
|
|
| @property |
| def internal_user_login_session_duration_hours(self) -> int: |
| return int(self._load_shared_settings().get("internal_user_login_session_duration_hours") or 0) |
|
|
| @property |
| def normal_user_login_session_duration_hours(self) -> int: |
| return int(self._load_shared_settings().get("normal_user_login_session_duration_hours") or 0) |
|
|
| @property |
| def prompt_optimizer_system_prompt(self) -> str: |
| return str( |
| self._load_shared_settings().get("prompt_optimizer_system_prompt") |
| or DEFAULT_PROMPT_OPTIMIZER_SYSTEM_PROMPT |
| ).strip() |
|
|
| @property |
| def prompt_optimizer_provider(self) -> str: |
| value = str(self._load_shared_settings().get("prompt_optimizer_provider") or "internal").strip().lower() |
| return value if value in {"internal", "external"} else "internal" |
|
|
| @property |
| def prompt_optimizer_base_url(self) -> str: |
| return str(self._load_shared_settings().get("prompt_optimizer_base_url") or "").strip() |
|
|
| @property |
| def prompt_optimizer_api_key(self) -> str: |
| return str(self._load_shared_settings().get("prompt_optimizer_api_key") or "").strip() |
|
|
| @property |
| def prompt_optimizer_model(self) -> str: |
| return str(self._load_shared_settings().get("prompt_optimizer_model") or "gpt-5-5").strip() |
|
|
| @property |
| def reverse_prompt_system_prompt(self) -> str: |
| return str( |
| self._load_shared_settings().get("reverse_prompt_system_prompt") |
| or DEFAULT_REVERSE_PROMPT_SYSTEM_PROMPT |
| ).strip() |
|
|
| @property |
| def reverse_prompt_provider(self) -> str: |
| value = str(self._load_shared_settings().get("reverse_prompt_provider") or "internal").strip().lower() |
| return value if value in {"internal", "external"} else "internal" |
|
|
| @property |
| def reverse_prompt_base_url(self) -> str: |
| return str(self._load_shared_settings().get("reverse_prompt_base_url") or "").strip() |
|
|
| @property |
| def reverse_prompt_api_key(self) -> str: |
| return str(self._load_shared_settings().get("reverse_prompt_api_key") or "").strip() |
|
|
| @property |
| def reverse_prompt_model(self) -> str: |
| return str(self._load_shared_settings().get("reverse_prompt_model") or "gpt-5-5").strip() |
|
|
| @property |
| def product_visual_planner_system_prompt(self) -> str: |
| return str( |
| self._load_shared_settings().get("product_visual_planner_system_prompt") |
| or DEFAULT_PRODUCT_VISUAL_PLANNER_SYSTEM_PROMPT |
| ).strip() |
|
|
| @property |
| def product_visual_agent_provider(self) -> str: |
| value = str(self._load_shared_settings().get("product_visual_agent_provider") or "internal").strip().lower() |
| return value if value in {"internal", "external"} else "internal" |
|
|
| @property |
| def product_visual_agent_base_url(self) -> str: |
| return str(self._load_shared_settings().get("product_visual_agent_base_url") or "").strip() |
|
|
| @property |
| def product_visual_agent_api_key(self) -> str: |
| return str(self._load_shared_settings().get("product_visual_agent_api_key") or "").strip() |
|
|
| @property |
| def product_visual_agent_model(self) -> str: |
| return str(self._load_shared_settings().get("product_visual_agent_model") or "gpt-5-5").strip() |
|
|
| @property |
| def normal_user_pool_pricing(self) -> dict[str, object]: |
| return _normalize_normal_user_pool_pricing( |
| self._load_shared_settings().get("normal_user_pool_pricing") |
| ) |
|
|
| @property |
| def normal_user_third_party_image_pricing(self) -> dict[str, object]: |
| return _normalize_normal_user_pool_pricing( |
| self._load_shared_settings().get("normal_user_third_party_image_pricing") |
| ) |
|
|
| def login_session_duration_hours_for_role(self, role: object) -> int: |
| return self.normal_user_login_session_duration_hours if str(role or "").strip() == "normal" else self.internal_user_login_session_duration_hours |
|
|
| def get_branding(self) -> dict[str, str]: |
| return _normalize_branding_settings(self.data) |
|
|
| @property |
| def shop_purchase_url(self) -> str: |
| return str(self.data.get("shop_purchase_url") or "").strip() |
|
|
| @property |
| def auto_remove_invalid_accounts(self) -> bool: |
| value = self.data.get("auto_remove_invalid_accounts", False) |
| if isinstance(value, str): |
| return value.strip().lower() in {"1", "true", "yes", "on"} |
| return bool(value) |
|
|
| @property |
| def auto_remove_rate_limited_accounts(self) -> bool: |
| value = self.data.get("auto_remove_rate_limited_accounts", False) |
| if isinstance(value, str): |
| return value.strip().lower() in {"1", "true", "yes", "on"} |
| return bool(value) |
|
|
| def get_free_account_cleanup_settings(self) -> dict[str, object]: |
| return _normalize_free_account_cleanup_settings(self.data.get("free_account_cleanup")) |
|
|
| def get_redeem_code_cleanup_settings(self) -> dict[str, object]: |
| return _normalize_redeem_code_cleanup_settings(self.data.get("redeem_code_cleanup")) |
|
|
| def get_normal_user_navigation(self, *, include_disabled: bool = False) -> list[dict[str, object]]: |
| items = _normalize_navigation_items(self.data.get("normal_user_navigation")) |
| return self._filter_user_navigation(items, include_disabled=include_disabled) |
|
|
| def get_internal_user_navigation(self, *, include_disabled: bool = False) -> list[dict[str, object]]: |
| items = _normalize_navigation_items(self.data.get("internal_user_navigation")) |
| return self._filter_user_navigation(items, include_disabled=include_disabled) |
|
|
| def get_admin_user_navigation(self, *, include_disabled: bool = False) -> list[dict[str, object]]: |
| raw = self.data.get("admin_user_navigation") |
| if raw is None: |
| |
| raw = [dict(item) for item in DEFAULT_ADMIN_USER_NAVIGATION] |
| items = _normalize_navigation_items(raw) |
| return self._filter_user_navigation(items, include_disabled=include_disabled) |
|
|
| @staticmethod |
| def _filter_user_navigation( |
| items: list[dict[str, object]], |
| *, |
| include_disabled: bool = False, |
| ) -> list[dict[str, object]]: |
| if include_disabled: |
| return items |
| visible: list[dict[str, object]] = [] |
| for item in items: |
| if not item.get("enabled"): |
| continue |
| next_item = dict(item) |
| next_item["children"] = [ |
| dict(child) |
| for child in item.get("children", []) |
| if isinstance(child, dict) and child.get("enabled") |
| ] |
| if next_item.get("url") or next_item["children"]: |
| visible.append(next_item) |
| return visible |
|
|
| @property |
| def log_levels(self) -> list[str]: |
| levels = self.data.get("log_levels") |
| if not isinstance(levels, list): |
| return [] |
| allowed = {"debug", "info", "warning", "error"} |
| return [level for item in levels if (level := str(item or "").strip().lower()) in allowed] |
|
|
| @property |
| def sensitive_words(self) -> list[str]: |
| words = self.data.get("sensitive_words") |
| return [word for item in words if (word := str(item or "").strip())] if isinstance(words, list) else [] |
|
|
| @property |
| def ai_review(self) -> dict[str, object]: |
| value = self.data.get("ai_review") |
| return value if isinstance(value, dict) else {} |
|
|
| @property |
| def local_review(self) -> dict[str, object]: |
| value = self.data.get("local_review") |
| return value if isinstance(value, dict) else {} |
|
|
| @property |
| def global_system_prompt(self) -> str: |
| return str(self.data.get("global_system_prompt") or "").strip() |
|
|
| @property |
| def default_upstream_model_name(self) -> str: |
| return str(self.data.get("default_upstream_model_name") or "gpt-5-5").strip() |
|
|
| @property |
| def default_thinking_effort(self) -> str: |
| value = str(self.data.get("default_thinking_effort") or "auto").strip().lower() |
| return value if value in {"auto", "standard", "extended", "max"} else "auto" |
|
|
| @property |
| def images_dir(self) -> Path: |
| path = _env_path("CHATGPT2API_IMAGES_DIR", PERSIST_DIR / "images" if PERSIST_DIR.exists() else DATA_DIR / "images") |
| path.mkdir(parents=True, exist_ok=True) |
| return path |
|
|
| @property |
| def image_thumbnails_dir(self) -> Path: |
| path = _env_path("CHATGPT2API_IMAGE_THUMBNAILS_DIR", PERSIST_DIR / "image_thumbnails" if PERSIST_DIR.exists() else DATA_DIR / "image_thumbnails") |
| path.mkdir(parents=True, exist_ok=True) |
| return path |
|
|
| def cleanup_old_images(self) -> int: |
| cutoff = time.time() - self.image_retention_days * 86400 |
| removed = 0 |
| for path in self.images_dir.rglob("*"): |
| try: |
| if path.is_file() and path.stat().st_mtime < cutoff: |
| path.unlink() |
| removed += 1 |
| except (FileNotFoundError, PermissionError, OSError): |
| continue |
| for path in sorted((p for p in self.images_dir.rglob("*") if p.is_dir()), key=lambda p: len(p.parts), reverse=True): |
| try: |
| path.rmdir() |
| except OSError: |
| pass |
| return removed |
|
|
| @property |
| def base_url(self) -> str: |
| return str( |
| os.getenv("CHATGPT2API_BASE_URL") |
| or self.data.get("base_url") |
| or "" |
| ).strip().rstrip("/") |
|
|
| @property |
| def sentinel_solver_url(self) -> str: |
| """sentinel 求解服务地址（sentinel_solver/），为空时退回本地 Turnstile 求解。""" |
| return str( |
| os.getenv("CHATGPT2API_SENTINEL_SOLVER_URL") |
| or self.data.get("sentinel_solver_url") |
| or "" |
| ).strip().rstrip("/") |
|
|
| @property |
| def sentinel_solver_key(self) -> str: |
| return str( |
| os.getenv("CHATGPT2API_SENTINEL_SOLVER_KEY") |
| or self.data.get("sentinel_solver_key") |
| or "" |
| ).strip() |
|
|
| @property |
| def image_proxy_base_url(self) -> str: |
| return _normalize_image_proxy_base_url( |
| os.getenv("CHATGPT2API_IMAGE_PROXY_BASE_URL") |
| or self.data.get("image_proxy_base_url") |
| or "" |
| ) |
|
|
| @property |
| def app_version(self) -> str: |
| try: |
| value = VERSION_FILE.read_text(encoding="utf-8").strip() |
| except FileNotFoundError: |
| return "0.0.0" |
| return value or "0.0.0" |
|
|
| def get(self) -> dict[str, object]: |
| data = dict(self.data) |
| data["thread_pool_tokens"] = self.thread_pool_tokens |
| data["chat_models"] = self.chat_models |
| data["image_models"] = self.image_models |
| data["system_models"] = self.system_models |
| data["refresh_account_interval_minute"] = self.refresh_account_interval_minute |
| data["image_retention_days"] = self.image_retention_days |
| data["image_poll_timeout_secs"] = self.image_poll_timeout_secs |
| data["image_poll_max_attempts"] = self.image_poll_max_attempts |
| data["image_poll_interval_secs"] = self.image_poll_interval_secs |
| data["image_poll_initial_wait_secs"] = self.image_poll_initial_wait_secs |
| data["image_check_before_hit_enabled"] = self.image_check_before_hit_enabled |
| data["image_settle_enabled"] = self.image_settle_enabled |
| data["image_settle_secs"] = self.image_settle_secs |
| data["image_timeout_retry_secs"] = self.image_timeout_retry_secs |
| data["image_account_concurrency"] = self.image_account_concurrency |
| data["image_parallel_generation"] = self.image_parallel_generation |
| data["image_prompt_aspect_hint"] = self.image_prompt_aspect_hint |
| data["image_remove_conversation_after_result"] = self.image_remove_conversation_after_result |
| data["image_remove_conversation_always"] = self.image_remove_conversation_always |
| data["image_memory_button_enabled"] = self.image_memory_button_enabled |
| data["cleanup_shared_conversations"] = self.cleanup_shared_conversations |
| data["image_share_evidence_enabled"] = self.image_share_evidence_enabled |
| data["image_share_retention_hours"] = self.image_share_retention_hours |
| data["normal_user_default_credits"] = self.normal_user_default_credits |
| data["normal_user_daily_credits"] = self.normal_user_daily_credits |
| data["normal_user_pool_pricing"] = self.normal_user_pool_pricing |
| data["normal_user_third_party_image_pricing"] = self.normal_user_third_party_image_pricing |
| data["checkin_daily_credits"] = self.checkin_daily_credits |
| data["checkin_random_min"] = self.checkin_random_min |
| data["checkin_random_max"] = self.checkin_random_max |
| data["checkin_invite_reward"] = self.checkin_invite_reward |
| data["checkin_credit_valid_days"] = self.checkin_credit_valid_days |
| data["checkin_content"] = self.checkin_content |
| data["error_message_rules"] = self.error_message_rules |
| data["error_message_fallback_enabled"] = self.error_message_fallback_enabled |
| data["error_message_ai_model"] = self.error_message_ai_model |
| data["internal_image_model_daily_limits"] = self.internal_image_model_daily_limits |
| data["wechat_login_enabled"] = self.wechat_login_enabled |
| data["wechat_auto_register_enabled"] = self.wechat_auto_register_enabled |
| data["internal_user_login_session_duration_hours"] = self.internal_user_login_session_duration_hours |
| data["normal_user_login_session_duration_hours"] = self.normal_user_login_session_duration_hours |
| data["prompt_optimizer_system_prompt"] = self.prompt_optimizer_system_prompt |
| data["prompt_optimizer_provider"] = self.prompt_optimizer_provider |
| data["prompt_optimizer_base_url"] = self.prompt_optimizer_base_url |
| data["prompt_optimizer_api_key"] = self.prompt_optimizer_api_key |
| data["prompt_optimizer_model"] = self.prompt_optimizer_model |
| data["reverse_prompt_system_prompt"] = self.reverse_prompt_system_prompt |
| data["reverse_prompt_provider"] = self.reverse_prompt_provider |
| data["reverse_prompt_base_url"] = self.reverse_prompt_base_url |
| data["reverse_prompt_api_key"] = self.reverse_prompt_api_key |
| data["reverse_prompt_model"] = self.reverse_prompt_model |
| data["product_visual_planner_system_prompt"] = self.product_visual_planner_system_prompt |
| data["product_visual_agent_provider"] = self.product_visual_agent_provider |
| data["product_visual_agent_base_url"] = self.product_visual_agent_base_url |
| data["product_visual_agent_api_key"] = self.product_visual_agent_api_key |
| data["product_visual_agent_model"] = self.product_visual_agent_model |
| data.update(self.get_branding()) |
| data["shop_purchase_url"] = self.shop_purchase_url |
| data["auto_remove_invalid_accounts"] = self.auto_remove_invalid_accounts |
| data["auto_remove_rate_limited_accounts"] = self.auto_remove_rate_limited_accounts |
| data["free_account_cleanup"] = self.get_free_account_cleanup_settings() |
| data["redeem_code_cleanup"] = self.get_redeem_code_cleanup_settings() |
| data["normal_user_navigation"] = self.get_normal_user_navigation(include_disabled=True) |
| data["internal_user_navigation"] = self.get_internal_user_navigation(include_disabled=True) |
| data["admin_user_navigation"] = self.get_admin_user_navigation(include_disabled=True) |
| data["log_levels"] = self.log_levels |
| data["sensitive_words"] = self.sensitive_words |
| data["image_proxy_base_url"] = self.image_proxy_base_url |
| data["local_review"] = self.local_review |
| data["ai_review"] = self.ai_review |
| data["global_system_prompt"] = self.global_system_prompt |
| data["default_upstream_model_name"] = self.default_upstream_model_name |
| data["default_thinking_effort"] = self.default_thinking_effort |
| data["backup"] = self.get_backup_settings() |
| data["image_storage"] = self.get_image_storage_settings() |
| data["chat_completion_cache"] = self.get_chat_completion_cache_settings() |
| data["proxy_runtime"] = self.get_public_proxy_runtime_settings() |
| data["announcements"] = self.get_announcements(include_disabled=True) |
| data["sentinel_solver_url"] = self.sentinel_solver_url |
| data.pop("auth-key", None) |
| data.pop("sentinel_solver_key", None) |
| data.pop(CHAT_NAVIGATION_MIGRATION_KEY, None) |
| data.pop(CHAT_ROOM_NAVIGATION_MIGRATION_KEY, None) |
| return data |
|
|
| def get_proxy_settings(self) -> str: |
| return str(self.data.get("proxy") or "").strip() |
|
|
| def update(self, data: dict[str, object]) -> dict[str, object]: |
| incoming_data = dict(data or {}) |
| next_data = dict(self.data) |
| next_data.update(incoming_data) |
| if "backup" in next_data: |
| next_data["backup"] = _normalize_backup_settings(next_data.get("backup")) |
| if "image_storage" in next_data: |
| next_data["image_storage"] = _normalize_image_storage_settings(next_data.get("image_storage")) |
| _validate_image_storage_settings(next_data["image_storage"]) |
| if "chat_completion_cache" in next_data: |
| next_data["chat_completion_cache"] = _normalize_chat_completion_cache_settings(next_data.get("chat_completion_cache")) |
| if "chat_models" in next_data: |
| next_data["chat_models"] = _normalize_chat_model_overrides(next_data.get("chat_models")) |
| if "image_models" in next_data: |
| next_data["image_models"] = _normalize_image_model_overrides(next_data.get("image_models")) |
| if "system_models" in next_data: |
| next_data["system_models"] = _normalize_system_model_overrides(next_data.get("system_models")) |
| if "proxy_runtime" in next_data: |
| incoming_runtime = next_data.get("proxy_runtime") |
| if isinstance(incoming_runtime, dict): |
| previous_clearance = self.get_proxy_runtime_settings().get("clearance") |
| if isinstance(previous_clearance, dict): |
| incoming_runtime = dict(incoming_runtime) |
| incoming_runtime["_existing_cf_cookies"] = previous_clearance.get("cf_cookies") |
| incoming_runtime["_existing_cf_clearance"] = previous_clearance.get("cf_clearance") |
| next_data["proxy_runtime"] = _normalize_proxy_runtime_settings(incoming_runtime) |
| if "free_account_cleanup" in next_data: |
| next_data["free_account_cleanup"] = _normalize_free_account_cleanup_settings( |
| next_data.get("free_account_cleanup") |
| ) |
| if "redeem_code_cleanup" in next_data: |
| next_data["redeem_code_cleanup"] = _normalize_redeem_code_cleanup_settings( |
| next_data.get("redeem_code_cleanup") |
| ) |
| if "normal_user_navigation" in next_data: |
| next_data["normal_user_navigation"] = _normalize_navigation_items( |
| next_data.get("normal_user_navigation") |
| ) |
| if "internal_user_navigation" in next_data: |
| next_data["internal_user_navigation"] = _normalize_navigation_items( |
| next_data.get("internal_user_navigation") |
| ) |
| if "admin_user_navigation" in next_data: |
| next_data["admin_user_navigation"] = _normalize_navigation_items( |
| next_data.get("admin_user_navigation") |
| ) |
| if "image_proxy_base_url" in next_data: |
| next_data["image_proxy_base_url"] = _normalize_image_proxy_base_url(next_data.get("image_proxy_base_url")) |
| if any(key in incoming_data for key in BRANDING_KEYS): |
| next_data.update(_normalize_branding_settings(next_data)) |
| if "checkin_content" in next_data: |
| |
| next_data["checkin_content"] = _normalize_checkin_content( |
| next_data.get("checkin_content"), self.data.get("checkin_content") |
| ) |
| if any(key in incoming_data for key in SHARED_SETTINGS_KEYS): |
| next_data.update(self._save_shared_settings(_normalize_shared_settings(next_data, self.data))) |
| if "announcements" in next_data: |
| next_data["announcements"] = self._save_shared_announcements( |
| _normalize_announcements(next_data.get("announcements")) |
| ) |
| next_data.pop("backup_state", None) |
| self.data = next_data |
| self._save() |
| return self.get() |
|
|
| def get_backup_settings(self) -> dict[str, object]: |
| return _normalize_backup_settings(self.data.get("backup")) |
|
|
| def get_image_storage_settings(self) -> dict[str, object]: |
| return _normalize_image_storage_settings(self.data.get("image_storage")) |
|
|
| def get_chat_completion_cache_settings(self) -> dict[str, object]: |
| return _normalize_chat_completion_cache_settings(self.data.get("chat_completion_cache")) |
|
|
| def get_proxy_runtime_settings(self) -> dict[str, object]: |
| return _normalize_proxy_runtime_settings(self.data.get("proxy_runtime")) |
|
|
| def get_public_proxy_runtime_settings(self) -> dict[str, object]: |
| runtime = copy.deepcopy(self.get_proxy_runtime_settings()) |
| clearance = runtime.get("clearance") if isinstance(runtime.get("clearance"), dict) else {} |
| if isinstance(clearance, dict): |
| has_cookies = bool(str(clearance.get("cf_cookies") or "").strip()) |
| has_clearance = bool(str(clearance.get("cf_clearance") or "").strip()) |
| clearance["cf_cookies"] = "" |
| clearance["cf_clearance"] = "" |
| clearance["has_cf_cookies"] = has_cookies |
| clearance["has_cf_clearance"] = has_clearance |
| return runtime |
|
|
| def get_announcements(self, *, include_disabled: bool = False, role: str = "") -> list[dict[str, object]]: |
| normalized = self._load_announcements() |
| target_role = str(role or "").strip().lower() |
| visible_targets = {"all"} |
| if target_role == "normal": |
| visible_targets.add("normal") |
| elif target_role in {"user", "admin"}: |
| visible_targets.add("internal") |
| items = [] |
| for item in normalized: |
| if not include_disabled and not item.get("enabled"): |
| continue |
| if target_role and item.get("target") not in visible_targets: |
| continue |
| items.append(dict(item)) |
| return sorted(items, key=lambda item: str(item.get("updated_at") or item.get("created_at") or ""), reverse=True) |
|
|
| def _shared_state_storage(self) -> StorageBackend | None: |
| if not os.getenv("SHARED_USER_DATABASE_URL", "").strip(): |
| return None |
| try: |
| storage = self.get_user_storage_backend() |
| except Exception as exc: |
| print(f"[shared-settings] shared storage unavailable: {exc}") |
| return None |
| if not callable(getattr(storage, "load_named_state", None)) or not callable(getattr(storage, "save_named_state", None)): |
| return None |
| return storage |
|
|
| def _shared_announcements_storage(self) -> StorageBackend | None: |
| return self._shared_state_storage() |
|
|
| def _local_shared_settings(self) -> dict[str, object]: |
| return _normalize_shared_settings(self.data) |
|
|
| def _load_shared_settings(self) -> dict[str, object]: |
| local_settings = self._local_shared_settings() |
| storage = self._shared_state_storage() |
| if storage is None: |
| return local_settings |
| now = time.time() |
| if self._shared_settings_cache is not None and now - self._shared_settings_cache_at < SHARED_SETTINGS_CACHE_SECONDS: |
| return dict(self._shared_settings_cache) |
| try: |
| load_named_state = getattr(storage, "load_named_state") |
| state = load_named_state(SHARED_SETTINGS_KEY) |
| except Exception as exc: |
| print(f"[shared-settings] load settings failed: {exc}") |
| return local_settings |
|
|
| if isinstance(state, dict) and any(key in state for key in SHARED_SETTINGS_KEYS): |
| settings = _normalize_shared_settings(state, local_settings) |
| else: |
| settings = local_settings |
| if any(key in self.data for key in SHARED_SETTINGS_KEYS) and not _skip_user_storage_migration(): |
| print( |
| "[shared-settings] Migrating local settings to shared storage: " |
| f"default={settings['normal_user_default_credits']}, " |
| f"daily={settings['normal_user_daily_credits']}" |
| ) |
| self._save_shared_settings(settings) |
| self._shared_settings_cache = dict(settings) |
| self._shared_settings_cache_at = now |
| return settings |
|
|
| def _prime_shared_settings(self) -> None: |
| if not os.getenv("SHARED_USER_DATABASE_URL", "").strip(): |
| return |
| try: |
| self._load_shared_settings() |
| except Exception as exc: |
| print(f"[shared-settings] prime settings failed: {exc}") |
|
|
| def _save_shared_settings(self, settings: dict[str, object]) -> dict[str, object]: |
| normalized = _normalize_shared_settings(settings, self.data) |
| self._shared_settings_cache = dict(normalized) |
| self._shared_settings_cache_at = time.time() |
| storage = self._shared_state_storage() |
| if storage is None: |
| return normalized |
| try: |
| save_named_state = getattr(storage, "save_named_state") |
| save_named_state(SHARED_SETTINGS_KEY, normalized) |
| except Exception as exc: |
| print(f"[shared-settings] save settings failed: {exc}") |
| return normalized |
|
|
| def get_internal_image_model_daily_limit(self, model_id: object) -> dict[str, object] | None: |
| normalized_model_id = str(model_id or "").strip() |
| if not normalized_model_id: |
| return None |
| for entry in self.internal_image_model_daily_limits: |
| if str(entry.get("model_id") or "").strip() == normalized_model_id: |
| return dict(entry) |
| return None |
|
|
| def reserve_internal_image_model_total_daily_limit(self, model_id: object, amount: int) -> dict[str, object]: |
| normalized_model_id = str(model_id or "").strip() |
| normalized_amount = max(0, int(amount or 0)) |
| if not normalized_model_id or normalized_amount <= 0: |
| return {"_internal_image_model_daily_limit_reservation": {"amount": 0}} |
| with self._shared_settings_lock: |
| settings = self._load_shared_settings() |
| limits = _normalize_internal_image_model_daily_limits(settings.get("internal_image_model_daily_limits")) |
| limit = next((entry for entry in limits if str(entry.get("model_id") or "").strip() == normalized_model_id), None) |
| if limit is None or _normalize_internal_image_model_limit_scope(limit.get("limit_scope")) != "total": |
| return {"_internal_image_model_daily_limit_reservation": {"amount": 0}} |
| limit = _reset_internal_image_model_daily_limit(limit) |
| daily_limit = _normalize_positive_int(limit.get("daily_limit"), 0, 0) |
| used = _normalize_positive_int(limit.get("used"), 0, 0) |
| reserved = _normalize_positive_int(limit.get("reserved"), 0, 0) |
| remaining = max(0, daily_limit - used - reserved) |
| if normalized_amount > remaining: |
| if remaining <= 0: |
| raise ValueError(f"{normalized_model_id} 今日内部用户总使用数量已达上限") |
| raise ValueError(f"{normalized_model_id} 今日内部用户总剩余 {remaining} 张，不足以生成 {normalized_amount} 张") |
| limit["reserved"] = reserved + normalized_amount |
| limit["date"] = str(limit.get("date") or "").strip() or _today_key() |
| limit["updated_at"] = _now_iso() |
| next_limits = [ |
| limit if str(entry.get("model_id") or "").strip() == normalized_model_id else entry |
| for entry in limits |
| ] |
| settings["internal_image_model_daily_limits"] = next_limits |
| saved = self._save_shared_settings(settings) |
| self.data.update(saved) |
| self._save() |
| public_limit = _internal_image_model_daily_limit_public(limit) |
| public_limit["_internal_image_model_daily_limit_reservation"] = { |
| "model_id": normalized_model_id, |
| "amount": normalized_amount, |
| "scope": "default_total", |
| "limit_scope": "total", |
| } |
| return public_limit |
|
|
| def consume_internal_image_model_total_daily_limit(self, model_id: object, amount: int) -> dict[str, object] | None: |
| return self._settle_internal_image_model_total_daily_limit(model_id, amount, consume=True) |
|
|
| def refund_internal_image_model_total_daily_limit(self, model_id: object, amount: int) -> dict[str, object] | None: |
| return self._settle_internal_image_model_total_daily_limit(model_id, amount, consume=False) |
|
|
| def _settle_internal_image_model_total_daily_limit(self, model_id: object, amount: int, *, consume: bool) -> dict[str, object] | None: |
| normalized_model_id = str(model_id or "").strip() |
| normalized_amount = max(0, int(amount or 0)) |
| if not normalized_model_id or normalized_amount <= 0: |
| return None |
| with self._shared_settings_lock: |
| settings = self._load_shared_settings() |
| limits = _normalize_internal_image_model_daily_limits(settings.get("internal_image_model_daily_limits")) |
| for index, entry in enumerate(limits): |
| if str(entry.get("model_id") or "").strip() != normalized_model_id: |
| continue |
| if _normalize_internal_image_model_limit_scope(entry.get("limit_scope")) != "total": |
| return None |
| limit = _reset_internal_image_model_daily_limit(entry) |
| reserved = _normalize_positive_int(limit.get("reserved"), 0, 0) |
| settled = min(reserved, normalized_amount) |
| limit["reserved"] = reserved - settled |
| if consume: |
| limit["used"] = _normalize_positive_int(limit.get("used"), 0, 0) + settled |
| limit["updated_at"] = _now_iso() |
| limits[index] = limit |
| settings["internal_image_model_daily_limits"] = limits |
| saved = self._save_shared_settings(settings) |
| self.data.update(saved) |
| self._save() |
| return _internal_image_model_daily_limit_public(limit) |
| return None |
|
|
| def _load_announcements(self) -> list[dict[str, object]]: |
| local_items = _normalize_announcements(self.data.get("announcements")) |
| storage = self._shared_announcements_storage() |
| if storage is None: |
| return local_items |
| try: |
| load_named_state = getattr(storage, "load_named_state") |
| state = load_named_state(SHARED_ANNOUNCEMENTS_KEY) |
| shared_items = _normalize_announcements(state.get("items") if isinstance(state, dict) else None) |
| except Exception as exc: |
| print(f"[announcements] load shared announcements failed: {exc}") |
| return local_items |
| if isinstance(state, dict) and "items" in state: |
| return shared_items |
| if local_items and not _skip_user_storage_migration(): |
| print(f"[announcements] Migrating local announcements to shared storage: {len(local_items)} items") |
| self._save_shared_announcements(local_items) |
| return local_items |
|
|
| def _prime_shared_announcements(self) -> None: |
| if not os.getenv("SHARED_USER_DATABASE_URL", "").strip(): |
| return |
| try: |
| self._load_announcements() |
| except Exception as exc: |
| print(f"[announcements] prime shared announcements failed: {exc}") |
|
|
| def _save_shared_announcements(self, items: list[dict[str, object]]) -> list[dict[str, object]]: |
| normalized = _normalize_announcements(items) |
| storage = self._shared_announcements_storage() |
| if storage is None: |
| return normalized |
| try: |
| save_named_state = getattr(storage, "save_named_state") |
| save_named_state(SHARED_ANNOUNCEMENTS_KEY, {"items": normalized}) |
| except Exception as exc: |
| print(f"[announcements] save shared announcements failed: {exc}") |
| return normalized |
|
|
| def get_storage_backend(self) -> StorageBackend: |
| """获取存储后端实例（单例）""" |
| if self._storage_backend is None: |
| from services.storage.factory import create_storage_backend |
| self._storage_backend = create_storage_backend(DATA_DIR) |
| return self._storage_backend |
|
|
| def get_user_storage_backend(self) -> StorageBackend: |
| """获取用户、额度、微信绑定与卡密使用的存储后端。""" |
| if self._user_storage_backend is None: |
| from services.storage.factory import create_user_storage_backend |
| self._user_storage_backend = create_user_storage_backend(DATA_DIR, self.get_storage_backend()) |
| return self._user_storage_backend |
|
|
| def get_account_pool_storage_backend(self) -> StorageBackend: |
| if self._account_pool_storage_backend is None: |
| from services.storage.factory import create_account_pool_storage_backend |
| self._account_pool_storage_backend = create_account_pool_storage_backend( |
| DATA_DIR, |
| self.get_storage_backend(), |
| self.get_user_storage_backend(), |
| ) |
| return self._account_pool_storage_backend |
|
|
|
|
| def load_backup_state() -> dict[str, object]: |
| return _normalize_backup_state(_read_json_object(BACKUP_STATE_FILE, name="backup_state.json")) |
|
|
|
|
| def save_backup_state(state: dict[str, object]) -> dict[str, object]: |
| normalized = _normalize_backup_state(state) |
| BACKUP_STATE_FILE.write_text(json.dumps(normalized, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") |
| return normalized |
|
|
|
|
| config = ConfigStore(CONFIG_FILE) |
|
|
