feat: 改造配置系统,使用API获取配置,支持IMAP和临时邮箱,移除本地env依赖
Some checks failed
Remove old artifacts / remove-old-artifacts (push) Has been cancelled
Some checks failed
Remove old artifacts / remove-old-artifacts (push) Has been cancelled
This commit is contained in:
182
config.py
182
config.py
@@ -1,58 +1,84 @@
|
||||
from dotenv import load_dotenv
|
||||
import os
|
||||
import sys
|
||||
import requests
|
||||
from logger import logging
|
||||
|
||||
|
||||
class Config:
|
||||
def __init__(self):
|
||||
# 获取应用程序的根目录路径
|
||||
if getattr(sys, "frozen", False):
|
||||
# 如果是打包后的可执行文件
|
||||
application_path = os.path.dirname(sys.executable)
|
||||
else:
|
||||
# 如果是开发环境
|
||||
application_path = os.path.dirname(os.path.abspath(__file__))
|
||||
self.api_base_url = "https://cursorapi.nosqli.com/admin"
|
||||
self.load_config()
|
||||
|
||||
# 指定 .env 文件的路径
|
||||
dotenv_path = os.path.join(application_path, ".env")
|
||||
|
||||
if not os.path.exists(dotenv_path):
|
||||
raise FileNotFoundError(f"文件 {dotenv_path} 不存在")
|
||||
|
||||
# 加载 .env 文件
|
||||
load_dotenv(dotenv_path)
|
||||
|
||||
self.imap = False
|
||||
self.temp_mail = os.getenv("TEMP_MAIL", "").strip().split("@")[0]
|
||||
self.temp_mail_epin = os.getenv("TEMP_MAIL_EPIN", "").strip()
|
||||
self.temp_mail_ext = os.getenv("TEMP_MAIL_EXT", "").strip()
|
||||
self.domain = os.getenv("DOMAIN", "").strip()
|
||||
|
||||
# 如果临时邮箱为null则加载IMAP
|
||||
if self.temp_mail == "null":
|
||||
self.imap = True
|
||||
self.imap_server = os.getenv("IMAP_SERVER", "").strip()
|
||||
self.imap_port = os.getenv("IMAP_PORT", "").strip()
|
||||
self.imap_user = os.getenv("IMAP_USER", "").strip()
|
||||
self.imap_pass = os.getenv("IMAP_PASS", "").strip()
|
||||
self.imap_dir = os.getenv("IMAP_DIR", "inbox").strip()
|
||||
|
||||
self.check_config()
|
||||
def load_config(self):
|
||||
"""从API加载配置"""
|
||||
try:
|
||||
endpoint = f"{self.api_base_url}/api.mail/getRandom"
|
||||
headers = {
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
response = requests.get(
|
||||
endpoint,
|
||||
headers=headers,
|
||||
timeout=10
|
||||
)
|
||||
|
||||
response_data = response.json()
|
||||
if response_data["code"] == 0:
|
||||
logging.info(f"成功获取配置: {response_data['data']['name']}")
|
||||
env_config = response_data["data"]["env"]
|
||||
|
||||
# 设置配置项
|
||||
self.domain = env_config.get("DOMAIN", "").strip()
|
||||
self.temp_mail = env_config.get("TEMP_MAIL", "").strip()
|
||||
self.temp_mail_ext = env_config.get("TEMP_MAIL_EXT", "").strip()
|
||||
self.temp_mail_epin = env_config.get("TEMP_MAIL_EPIN", "").strip()
|
||||
self.browser_user_agent = env_config.get("BROWSER_USER_AGENT", "").strip()
|
||||
self.mail_server = env_config.get("MAIL_SERVER", "").strip()
|
||||
|
||||
# IMAP相关配置
|
||||
self.imap = env_config.get("TEMP_MAIL", "") == "null"
|
||||
if self.imap:
|
||||
self.imap_server = env_config.get("IMAP_SERVER", "").strip()
|
||||
self.imap_port = env_config.get("IMAP_PORT", "").strip()
|
||||
self.imap_user = env_config.get("IMAP_USER", "").strip()
|
||||
self.imap_pass = env_config.get("IMAP_PASS", "").strip()
|
||||
self.imap_dir = env_config.get("IMAP_DIR", "inbox").strip()
|
||||
else:
|
||||
self.imap_server = ""
|
||||
self.imap_port = ""
|
||||
self.imap_user = ""
|
||||
self.imap_pass = ""
|
||||
self.imap_dir = "inbox"
|
||||
|
||||
self.check_config()
|
||||
else:
|
||||
raise ValueError(f"获取配置失败: {response_data['msg']}")
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"加载配置失败: {str(e)}")
|
||||
raise
|
||||
|
||||
def get_temp_mail(self):
|
||||
|
||||
return self.temp_mail
|
||||
|
||||
def get_temp_mail_epin(self):
|
||||
|
||||
return self.temp_mail_epin
|
||||
|
||||
def get_temp_mail_ext(self):
|
||||
|
||||
return self.temp_mail_ext
|
||||
|
||||
def get_temp_mail_epin(self):
|
||||
return self.temp_mail_epin
|
||||
|
||||
def get_domain(self):
|
||||
return self.domain
|
||||
|
||||
def get_mail_server(self):
|
||||
return self.mail_server
|
||||
|
||||
def get_browser_user_agent(self):
|
||||
return self.browser_user_agent
|
||||
|
||||
def get_imap(self):
|
||||
"""获取IMAP配置"""
|
||||
if not self.imap:
|
||||
return False
|
||||
return {
|
||||
@@ -63,84 +89,64 @@ class Config:
|
||||
"imap_dir": self.imap_dir,
|
||||
}
|
||||
|
||||
def get_domain(self):
|
||||
return self.domain
|
||||
|
||||
def check_config(self):
|
||||
"""检查配置项是否有效
|
||||
|
||||
检查规则:
|
||||
1. 如果使用 tempmail.plus,需要配置 TEMP_MAIL 和 DOMAIN
|
||||
2. 如果使用 IMAP,需要配置 IMAP_SERVER、IMAP_PORT、IMAP_USER、IMAP_PASS
|
||||
3. IMAP_DIR 是可选的
|
||||
"""
|
||||
"""检查配置项是否有效"""
|
||||
# 基础配置检查
|
||||
required_configs = {
|
||||
"domain": "域名",
|
||||
"browser_user_agent": "浏览器UA",
|
||||
"mail_server": "邮件服务器"
|
||||
}
|
||||
|
||||
# 检查基础配置
|
||||
for key, name in required_configs.items():
|
||||
if not self.check_is_valid(getattr(self, key)):
|
||||
raise ValueError(f"{name}未配置,请在 .env 文件中设置 {key.upper()}")
|
||||
raise ValueError(f"{name}配置无效")
|
||||
|
||||
# 检查邮箱配置
|
||||
if self.temp_mail != "null":
|
||||
if not self.imap:
|
||||
# tempmail.plus 模式
|
||||
if not self.check_is_valid(self.temp_mail):
|
||||
raise ValueError("临时邮箱未配置,请在 .env 文件中设置 TEMP_MAIL")
|
||||
if not self.check_is_valid(self.temp_mail) or not self.check_is_valid(self.temp_mail_ext):
|
||||
raise ValueError("临时邮箱配置无效")
|
||||
else:
|
||||
# IMAP 模式
|
||||
imap_configs = {
|
||||
"imap_server": "IMAP服务器",
|
||||
"imap_port": "IMAP端口",
|
||||
"imap_user": "IMAP用户名",
|
||||
"imap_pass": "IMAP密码",
|
||||
"imap_pass": "IMAP密码"
|
||||
}
|
||||
|
||||
for key, name in imap_configs.items():
|
||||
value = getattr(self, key)
|
||||
if value == "null" or not self.check_is_valid(value):
|
||||
raise ValueError(
|
||||
f"{name}未配置,请在 .env 文件中设置 {key.upper()}"
|
||||
)
|
||||
|
||||
# IMAP_DIR 是可选的,如果设置了就检查其有效性
|
||||
if self.imap_dir != "null" and not self.check_is_valid(self.imap_dir):
|
||||
raise ValueError(
|
||||
"IMAP收件箱目录配置无效,请在 .env 文件中正确设置 IMAP_DIR"
|
||||
)
|
||||
if not self.check_is_valid(getattr(self, key)):
|
||||
raise ValueError(f"{name}配置无效")
|
||||
|
||||
def check_is_valid(self, value):
|
||||
"""检查配置项是否有效
|
||||
|
||||
Args:
|
||||
value: 配置项的值
|
||||
|
||||
Returns:
|
||||
bool: 配置项是否有效
|
||||
"""
|
||||
"""检查配置项是否有效"""
|
||||
return isinstance(value, str) and len(str(value).strip()) > 0
|
||||
|
||||
def print_config(self):
|
||||
"""打印当前配置"""
|
||||
logging.info("\n=== 当前配置 ===")
|
||||
logging.info(f"域名: {self.domain}")
|
||||
if self.imap:
|
||||
logging.info(f"\033[32mIMAP服务器: {self.imap_server}\033[0m")
|
||||
logging.info(f"\033[32mIMAP端口: {self.imap_port}\033[0m")
|
||||
logging.info(f"\033[32mIMAP用户名: {self.imap_user}\033[0m")
|
||||
logging.info(f"\033[32mIMAP密码: {'*' * len(self.imap_pass)}\033[0m")
|
||||
logging.info(f"\033[32mIMAP收件箱目录: {self.imap_dir}\033[0m")
|
||||
if self.temp_mail != "null":
|
||||
logging.info(
|
||||
f"\033[32m临时邮箱: {self.temp_mail}{self.temp_mail_ext}\033[0m"
|
||||
)
|
||||
logging.info(f"\033[32m域名: {self.domain}\033[0m")
|
||||
logging.info("=== IMAP配置 ===")
|
||||
logging.info(f"IMAP服务器: {self.imap_server}")
|
||||
logging.info(f"IMAP端口: {self.imap_port}")
|
||||
logging.info(f"IMAP用户名: {self.imap_user}")
|
||||
logging.info(f"IMAP密码: {'*' * len(self.imap_pass)}")
|
||||
logging.info(f"IMAP目录: {self.imap_dir}")
|
||||
else:
|
||||
logging.info(f"临时邮箱: {self.temp_mail}{self.temp_mail_ext}")
|
||||
if self.temp_mail_epin:
|
||||
logging.info(f"邮箱PIN码: {self.temp_mail_epin}")
|
||||
logging.info(f"邮件服务器: {self.mail_server}")
|
||||
logging.info(f"浏览器UA: {self.browser_user_agent}")
|
||||
|
||||
|
||||
# 使用示例
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
config = Config()
|
||||
print("环境变量加载成功!")
|
||||
print("配置加载成功!")
|
||||
config.print_config()
|
||||
except ValueError as e:
|
||||
print(f"错误: {e}")
|
||||
except Exception as e:
|
||||
print(f"错误: {str(e)}")
|
||||
|
||||
Reference in New Issue
Block a user