config.py
3.76 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
"""
配置管理模块
"""
import json
import os
from datetime import datetime
from typing import Dict, List, Any
class Config:
"""配置管理类"""
CONFIG_FILE = "config.json"
DEFAULT_CONFIG = {
"git": {
"repo_url": "http://gitlab.zb100.com:10080/chaijin/EtsyCustomerNotify.git",
"ssh_key": "XKc2v_hs8-qkougieWvx"
},
"etsy": {
"messages_url": "https://www.etsy.com/messages?ref=seller-platform-mcnav",
"api_endpoints": {
"message_list": "https://www.etsy.com/api/v3/ajax/bespoke/member/conversations/message-list-data",
"tag_counts": "https://www.etsy.com/api/v3/ajax/member/conversations/unread-tag-counts"
}
},
"limits": {
"daily_limit": 100,
"batch_limit": 50,
"batch_pause_minutes": 5,
"message_delay_seconds": [10, 20]
},
"marketing_messages": [],
"excluded_tags": [],
"last_reset_date": None
}
def __init__(self):
self.config = self._load_config()
def _load_config(self) -> Dict[str, Any]:
"""加载配置文件"""
if os.path.exists(self.CONFIG_FILE):
try:
with open(self.CONFIG_FILE, 'r', encoding='utf-8') as f:
config = json.load(f)
# 合并默认配置（确保新字段存在）
return self._merge_config(self.DEFAULT_CONFIG, config)
except Exception as e:
print(f"配置文件加载失败: {e}")
return self.DEFAULT_CONFIG.copy()
def _merge_config(self, default: Dict, user: Dict) -> Dict:
"""递归合并配置"""
result = default.copy()
for key, value in user.items():
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = self._merge_config(result[key], value)
else:
result[key] = value
return result
def save(self):
"""保存配置文件"""
try:
with open(self.CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(self.config, f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"配置文件保存失败: {e}")
def get(self, key: str, default=None):
"""获取配置值（支持点号分隔的路径）"""
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
def set(self, key: str, value: Any):
"""设置配置值（支持点号分隔的路径）"""
keys = key.split('.')
config = self.config
for k in keys[:-1]:
if k not in config:
config[k] = {}
config = config[k]
config[keys[-1]] = value
def add_marketing_message(self, message: str):
"""添加营销消息"""
if "marketing_messages" not in self.config:
self.config["marketing_messages"] = []
self.config["marketing_messages"].append(message)
def get_marketing_messages(self) -> List[str]:
"""获取营销消息列表"""
return self.config.get("marketing_messages", [])
def set_excluded_tags(self, tags: List[str]):
"""设置排除标签"""
self.config["excluded_tags"] = tags
def get_excluded_tags(self) -> List[str]:
"""获取排除标签"""
return self.config.get("excluded_tags", [])
def reset_marketing_data(self):
"""重置营销数据"""
self.config["last_reset_date"] = datetime.now().isoformat()
# 这里会触发数据库清理，在database模块中实现