""" 配置管理模块 支持三层配置来源（优先级从低到高）： 1. .env 文件（默认值） 2. ~/.wellfound/config.json（界面保存的配置，持久化） 3. 运行时内存（界面实时修改） 导入方式： import src.config as cfg # 推荐：每次访问当前值 cfg.get_llm_api_key() # 总是返回最新值 """ import json import os import logging from pathlib import Path from typing import Optional from dotenv import load_dotenv logger = logging.getLogger(__name__) # ---- 持久化文件路径 ---- CONFIG_DIR = Path.home() / ".wellfound" CONFIG_FILE = CONFIG_DIR / "config.json" # ---- 模块级变量（可被运行时更新）---- _llm_provider = "deepseek" _openai_api_key = "" _openai_model = "gpt-4o-mini" _deepseek_api_key = "" _deepseek_model = "deepseek-chat" _google_sheet_id = "" _google_sheet_name = "West" # 所有区域工作表名称（按优先级顺序查询） _google_sheet_names = ["West", "South", "Northeast", "Midwest"] _google_service_account_json = "" _sheet_company_col = "C" _sheet_email_type_col = "AA" _sheet_email_body_col = "AB" _sheet_email_date_col = "AC" _sheet_timestamp_col = "AD" _fuzzy_match_threshold = 85 _fuzzy_ambiguous_gap = 3.0 _max_concurrent = 5 _email_body_max_chars = 500 # 列索引缓存（配置变更时重建） _indices: dict = {} _config_loaded = False def _col_letter_to_index(col: str) -> int: col = col.upper().strip() result = 0 for ch in col: result = result * 26 + (ord(ch) - ord("A") + 1) return result - 1 def _rebuild_indices(): global _indices _indices = { "company": _col_letter_to_index(_sheet_company_col), "email_type": _col_letter_to_index(_sheet_email_type_col), "email_body": _col_letter_to_index(_sheet_email_body_col), "email_date": _col_letter_to_index(_sheet_email_date_col), "timestamp": _col_letter_to_index(_sheet_timestamp_col), } # ---- 初始化：加载配置 ---- def _load_env(): """用 .env 文件值填充模块变量（仅在首次调用时作为默认值）""" load_dotenv() # 优先从环境变量读取（HF Spaces Secrets 或本地 .env） # 如果环境变量不存在，则保持当前值（可能来自 config.json） env_provider = os.getenv("LLM_PROVIDER") if env_provider: global _llm_provider _llm_provider = env_provider env_openai_key = os.getenv("OPENAI_API_KEY") if env_openai_key: global _openai_api_key _openai_api_key = env_openai_key env_openai_model = os.getenv("OPENAI_MODEL") if env_openai_model: global _openai_model _openai_model = env_openai_model env_deepseek_key = os.getenv("DEEPSEEK_API_KEY") if env_deepseek_key: global _deepseek_api_key _deepseek_api_key = env_deepseek_key env_deepseek_model = os.getenv("DEEPSEEK_MODEL") if env_deepseek_model: global _deepseek_model _deepseek_model = env_deepseek_model env_sheet_id = os.getenv("GOOGLE_SHEET_ID") if env_sheet_id: global _google_sheet_id _google_sheet_id = env_sheet_id env_sheet_name = os.getenv("GOOGLE_SHEET_NAME") if env_sheet_name: global _google_sheet_name _google_sheet_name = env_sheet_name env_sa_json = os.getenv("GOOGLE_SERVICE_ACCOUNT_JSON") if env_sa_json: global _google_service_account_json _google_service_account_json = env_sa_json # 可选配置 env_company_col = os.getenv("SHEET_COMPANY_COL") if env_company_col: global _sheet_company_col _sheet_company_col = env_company_col env_type_col = os.getenv("SHEET_EMAIL_TYPE_COL") if env_type_col: global _sheet_email_type_col _sheet_email_type_col = env_type_col env_body_col = os.getenv("SHEET_EMAIL_BODY_COL") if env_body_col: global _sheet_email_body_col _sheet_email_body_col = env_body_col env_date_col = os.getenv("SHEET_EMAIL_DATE_COL") if env_date_col: global _sheet_email_date_col _sheet_email_date_col = env_date_col env_ts_col = os.getenv("SHEET_TIMESTAMP_COL") if env_ts_col: global _sheet_timestamp_col _sheet_timestamp_col = env_ts_col env_threshold = os.getenv("FUZZY_MATCH_THRESHOLD") if env_threshold: global _fuzzy_match_threshold _fuzzy_match_threshold = int(env_threshold) env_gap = os.getenv("FUZZY_AMBIGUOUS_GAP") if env_gap: global _fuzzy_ambiguous_gap _fuzzy_ambiguous_gap = float(env_gap) _rebuild_indices() def _apply_dict(d: dict): """用字典更新模块变量""" global _llm_provider, _openai_api_key, _openai_model global _deepseek_api_key, _deepseek_model global _google_sheet_id, _google_sheet_name, _google_service_account_json global _sheet_company_col, _sheet_email_type_col, _sheet_email_body_col global _sheet_email_date_col, _sheet_timestamp_col global _fuzzy_match_threshold, _fuzzy_ambiguous_gap global _max_concurrent, _email_body_max_chars _llm_provider = d.get("llm_provider", _llm_provider) _openai_api_key = d.get("openai_api_key", _openai_api_key) _openai_model = d.get("openai_model", _openai_model) _deepseek_api_key = d.get("deepseek_api_key", _deepseek_api_key) _deepseek_model = d.get("deepseek_model", _deepseek_model) _google_sheet_id = d.get("google_sheet_id", _google_sheet_id) _google_sheet_name = d.get("google_sheet_name", _google_sheet_name) _google_service_account_json = d.get("google_service_account_json", _google_service_account_json) _sheet_company_col = d.get("sheet_company_col", _sheet_company_col) _sheet_email_type_col = d.get("sheet_email_type_col", _sheet_email_type_col) _sheet_email_body_col = d.get("sheet_email_body_col", _sheet_email_body_col) _sheet_email_date_col = d.get("sheet_email_date_col", _sheet_email_date_col) _sheet_timestamp_col = d.get("sheet_timestamp_col", _sheet_timestamp_col) _fuzzy_match_threshold = d.get("fuzzy_match_threshold", _fuzzy_match_threshold) _fuzzy_ambiguous_gap = d.get("fuzzy_ambiguous_gap", _fuzzy_ambiguous_gap) _max_concurrent = d.get("max_concurrent", _max_concurrent) _email_body_max_chars = d.get("email_body_max_chars", _email_body_max_chars) _rebuild_indices() def load_config() -> dict: """ 加载配置（优先级：config.json > 环境变量 > 内置默认值） 在应用启动时调用一次即可。 Returns: 当前配置字典 """ global _config_loaded # 先加载环境变量（作为基础值） _load_env() # 如果在 HF Spaces 环境中，不加载 config.json（优先使用 Secrets） # 但可以提供一个选项：如果 HF_SPACES_LOAD_CONFIG=true，则仍然加载 if os.getenv("SPACE_ID") or os.getenv("HF_SPACE"): if not os.getenv("HF_SPACES_LOAD_CONFIG"): logger.info("🌐 检测到 HF Spaces 环境，优先使用环境变量（Secrets）") _config_loaded = True return get_config_dict() # 本地环境或显式要求：尝试从 config.json 加载 if CONFIG_FILE.exists(): try: with open(CONFIG_FILE, "r", encoding="utf-8") as f: file_cfg = json.load(f) _apply_dict(file_cfg) logger.info(f"✅ 已从 {CONFIG_FILE} 加载持久化配置") except Exception as e: logger.warning(f"读取配置文件失败: {e}，使用环境变量 / 默认值") _config_loaded = True return get_config_dict() def save_config(cfg: dict) -> tuple[bool, str]: """ 将配置保存到 ~/.wellfound/config.json Args: cfg: 完整配置字典（来自界面输入） Returns: (success, message) """ try: CONFIG_DIR.mkdir(parents=True, exist_ok=True) # 写入文件 with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(cfg, f, indent=2, ensure_ascii=False) # 同步到内存 _apply_dict(cfg) logger.info(f"✅ 配置已保存到 {CONFIG_FILE}") return True, f"✅ 配置已保存至 {CONFIG_FILE.name}" except Exception as e: logger.error(f"保存配置失败: {e}") return False, f"❌ 保存失败: {str(e)}" def apply_config(cfg: dict): """仅更新内存中的配置（不写文件），用于界面实时预览""" _apply_dict(cfg) def get_config_dict() -> dict: """返回当前所有配置的字典（用于界面回填）""" return { "llm_provider": _llm_provider, "openai_api_key": _openai_api_key, "openai_model": _openai_model, "deepseek_api_key": _deepseek_api_key, "deepseek_model": _deepseek_model, "google_sheet_id": _google_sheet_id, "google_sheet_name": _google_sheet_name, "google_service_account_json": _google_service_account_json, "sheet_company_col": _sheet_company_col, "sheet_email_type_col": _sheet_email_type_col, "sheet_email_body_col": _sheet_email_body_col, "sheet_email_date_col": _sheet_email_date_col, "sheet_timestamp_col": _sheet_timestamp_col, "fuzzy_match_threshold": _fuzzy_match_threshold, "fuzzy_ambiguous_gap": _fuzzy_ambiguous_gap, "max_concurrent": _max_concurrent, "email_body_max_chars": _email_body_max_chars, } # ---- 便捷读取函数（各模块调用这些函数，总是拿到最新值）---- def get_llm_provider() -> str: return _llm_provider def get_llm_api_key() -> str: return _deepseek_api_key if _llm_provider == "deepseek" else _openai_api_key def get_llm_model() -> str: return _deepseek_model if _llm_provider == "deepseek" else _openai_model def get_llm_base_url() -> Optional[str]: if _llm_provider == "deepseek": return "https://api.deepseek.com" return None def get_google_sheet_id() -> str: return _google_sheet_id def get_google_sheet_name() -> str: return _google_sheet_name def get_google_sheet_names() -> list[str]: """返回所有需要查询的区域工作表名称列表""" return list(_google_sheet_names) def get_service_account_info() -> Optional[dict]: """解析 Service Account JSON（支持 str 或 dict）""" raw = _google_service_account_json if not raw: return None if isinstance(raw, dict): return raw if isinstance(raw, str): raw = raw.strip() if not raw: return None # 尝试解析 JSON 字符串 try: return json.loads(raw) except json.JSONDecodeError: # 尝试作为文件路径 if os.path.exists(raw): with open(raw, "r", encoding="utf-8") as f: return json.load(f) return None return None def get_sheet_col_indices() -> dict: """返回所有列 0-based 索引""" return _indices def get_fuzzy_threshold() -> int: return int(_fuzzy_match_threshold) def get_fuzzy_ambiguous_gap() -> float: return float(_fuzzy_ambiguous_gap) def get_max_concurrent() -> int: return int(_max_concurrent) def get_email_body_max_chars() -> int: return int(_email_body_max_chars) # ---- 常量 ---- HIGH_PRIORITY_TYPES = {"Forward", "not fit", "considered", "connect more", "career page", "not hiring"} LOW_PRIORITY_TYPES = {"Auto reply", "do not exist"} ALL_CATEGORIES = [ "Auto reply", "career page", "connect more", "considered", "not hiring", "do not exist", "not fit", "Forward", ] # ---- 启动时自动加载 ---- load_config()