zeroleo12345 · zeroleo12345 · Feb 12, 2026 · Feb 12, 2026 · Feb 12, 2026 · Feb 12, 2026
diff --git a/.env.x b/.env.x
diff --git a/src/acct/accounting_flow.py b/src/acct/accounting_flow.py
@@ -37,7 +37,7 @@ def accounting_handler(cls, request: AcctRequest, acct_user_profile: AcctUserPro
             current_session = AccountingSession.put(acct_user_profile.packet.outer_username, acct_user_profile.packet.user_mac)
             if current_session > 1 and account.role != Account.Role.PLATFORM_OWNER.value:
                 text = f'{acct_user_profile.packet.outer_username} 账号多拨!'
-                Feishu.send_groud_msg(receiver_id=Feishu.FEISHU_SESSION_CHAT_ID, text=text)
+                Feishu.send_group_msg(receiver_id=Feishu.FEISHU_SESSION_CHAT_ID, text=text)
                 # cls.disconnect(user_name=acct_user_profile.packet.outer_username, user_mac=acct_user_profile.packet.user_mac)
 
         # cls.push_metric(username=account.username, request=request)

diff --git a/src/auth/mac_flow.py b/src/auth/mac_flow.py
@@ -35,7 +35,7 @@ def mac_auth(cls, request: AuthRequest, session: BaseSession):
             # notify
             notify_url = f'{API_URL}/mac-account?username={session.auth_user_profile.packet.outer_username}&ssid={request.ssid}&ap_mac={request.ap_mac}'
             text = f'设备首次请求放通:\nMAC: {session.auth_user_profile.packet.user_mac}\nSSID: {request.ssid}\n若允许访问, 请点击: {notify_url}'
-            Feishu.send_groud_msg(receiver_id=Feishu.FEISHU_MAC_CHAT_ID, text=text)
+            Feishu.send_group_msg(receiver_id=Feishu.FEISHU_MAC_CHAT_ID, text=text)
 
         # mac Flow: 用户不存在则创建
         account = MacAccount.get_(username=session.auth_user_profile.packet.outer_username)
@@ -53,7 +53,7 @@ def mac_auth(cls, request: AuthRequest, session: BaseSession):
                 expired_at=expired_at, created_at=created_at,
             )
             text = f'新增放通 MAC 设备, MAC: {session.auth_user_profile.packet.user_mac}, SSID: {request.ssid}'
-            Feishu.send_groud_msg(receiver_id=Feishu.FEISHU_MAC_CHAT_ID, text=text)
+            Feishu.send_group_msg(receiver_id=Feishu.FEISHU_MAC_CHAT_ID, text=text)
             redis.delete(enable_flag_key)
         if not account.is_enable:
             log.warning(f'account is disabled')

diff --git a/src/library/crypto.py b/src/library/crypto.py
@@ -6,15 +6,15 @@
 import ctypes
 from loguru import logger as log
 # 项目库
-from utils.config import config
+from utils.config import settings
 
 # HOSTAPD 动态库
-HOSTAPD_LIBRARY = config('HOSTAPD_LIBRARY')
-CA_CERT = config('CA_CERT')
-CLIENT_CERT = config('CLIENT_CERT')
-PRIVATE_KEY = config('PRIVATE_KEY')
-PRIVATE_KEY_PASSWORD = str(config('PRIVATE_KEY_PASSWORD'))
-DH_FILE = config('DH_FILE')
+HOSTAPD_LIBRARY = settings.get('HOSTAPD_LIBRARY')
+CA_CERT = settings.get('CA_CERT')
+CLIENT_CERT = settings.get('CLIENT_CERT')
+PRIVATE_KEY = settings.get('PRIVATE_KEY')
+PRIVATE_KEY_PASSWORD = str(settings.get('PRIVATE_KEY_PASSWORD'))
+DH_FILE = settings.get('DH_FILE')
 
 
 class EapCryptoError(Exception):

diff --git a/src/processor/auth_processor.py b/src/processor/auth_processor.py
@@ -21,11 +21,11 @@
 from settings import RADIUS_DICTIONARY_DIR, RADIUS_SECRET, RADIUS_LISTEN_IP, RADIUS_LISTEN_PORT
 from loguru import logger as log
 from controls.user import AuthUserProfile
-from utils.config import config
+from utils.config import settings
 from library.crypto import libhostapd
 
 
-if config('USE_GTC', default=False, cast='@bool'):
+if settings.get('USE_GTC', default=False, cast='@bool'):
     log.info('## PEAP-GTC mode ##')
     USE_GTC = True
 else:

diff --git a/src/processor/dae_processor.py b/src/processor/dae_processor.py
@@ -96,7 +96,7 @@ def run(self):
         # 发送报文
         try:
             self.socket.sendto(request.RequestPacket(), request.address)
-            res_data, from_address = self.socket.recvfrom(__bufsize=1024)
+            res_data, from_address = self.socket.recvfrom(1024)   # buf size: 1024
         except Exception as e:
             log.error(traceback.format_exc())
             return False

diff --git a/src/settings.py b/src/settings.py
@@ -3,37 +3,37 @@
 # 第三方库
 import sentry_sdk
 # 项目库
-from utils.config import config
+from utils.config import settings
 from loguru import logger as log
 
 
-SENTRY_DSN = config('SENTRY_DSN', mandatory=False)
-SENTRY_PROXY = config('SENTRY_PROXY', default='')
+SENTRY_DSN = settings.get('SENTRY_DSN', mandatory=False)
+SENTRY_PROXY = settings.get('SENTRY_PROXY', default='')
 sentry_sdk.init(
     dsn=SENTRY_DSN,
     debug=False,
     http_proxy=SENTRY_PROXY,
     https_proxy=SENTRY_PROXY,
 )
 
-RADIUS_DICTIONARY_DIR = config('RADIUS_DICTIONARY_DIR')
-RADIUS_SECRET: bytes = str.encode(config('RADIUS_SECRET'))
-RADIUS_LISTEN_IP = config('RADIUS_LISTEN_IP', default='')
-RADIUS_LISTEN_PORT = config('RADIUS_LISTEN_PORT', default='')
-ACCOUNTING_INTERVAL = config('ACCOUNTING_INTERVAL', default=60, cast='@int')
-API_URL = config('API_URL')
+RADIUS_DICTIONARY_DIR = settings.get('RADIUS_DICTIONARY_DIR')
+RADIUS_SECRET: bytes = str.encode(settings.get('RADIUS_SECRET'))
+RADIUS_LISTEN_IP = settings.get('RADIUS_LISTEN_IP', default='')
+RADIUS_LISTEN_PORT = settings.get('RADIUS_LISTEN_PORT', default='')
+ACCOUNTING_INTERVAL = settings.get('ACCOUNTING_INTERVAL', default=60, cast='@int')
+API_URL = settings.get('API_URL')
 
 # DB
-DATABASE_URI = config('DATABASE_URI')
+DATABASE_URI = settings.get('DATABASE_URI')
 
 # Redis
-REDIS_HOST = config('REDIS_HOST')
-REDIS_PORT = config('REDIS_PORT')
-REDIS_PASSWORD = config('REDIS_PASSWORD')
-REDIS_DB = config('REDIS_DB')
+REDIS_HOST = settings.get('REDIS_HOST')
+REDIS_PORT = settings.get('REDIS_PORT')
+REDIS_PASSWORD = settings.get('REDIS_PASSWORD')
+REDIS_DB = settings.get('REDIS_DB')
 
 # Log
-LOG_LEVEL = config('LOG_LEVEL')
+LOG_LEVEL = settings.get('LOG_LEVEL')
 # 初始化日志
 log.remove()    # workaround: https://github.com/Delgan/loguru/issues/208
 # log_console_format = "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <level>{message}</level>"

diff --git a/src/utils/config.py b/src/utils/config.py
@@ -10,11 +10,12 @@ class Config(object):
         dotenv_override=False,  # 设置.env配置是否覆盖环境变量
     )
 
-    def __call__(self, key, default=None, cast=None, mandatory=True, fresh=False, dotted_lookup=True, parent=None):
-        value = self._settings.get(key, default=default, cast=cast, fresh=fresh, dotted_lookup=dotted_lookup, parent=None)
+    @classmethod
+    def get(cls, key, default=None, cast=None, mandatory=True, fresh=False, dotted_lookup=True, parent=None):
+        value = cls._settings.get(key, default=default, cast=cast, fresh=fresh, dotted_lookup=dotted_lookup, parent=None)
         if mandatory and value is None:
             raise Exception(f'config key: {key} is missing')
         return value
 
 
-config = Config()
+settings = Config
diff --git a/src/utils/dingding.py b/src/utils/dingding.py
@@ -0,0 +1,93 @@
+import requests
+import time
+# 第三方库
+from utils.config import settings
+from loguru import logger as log
+
+
+class Dingding(object):
+    class Token(object):
+        def __init__(self, token='', ttl=-1):
+            self.token = token
+            self.expired_at = int(time.time()) + ttl
+
+    # Must:
+    DINGDING_APP_ID = settings.get('DINGDING_APP_ID')
+    DINGDING_APP_SECRET = settings.get('DINGDING_APP_SECRET')
+    DINGDING_ROBOT_CODE = settings.get('DINGDING_ROBOT_CODE', default='dingqnettcbcq4tpecq7')
+    # Optional:
+    DINGDING_MAC_CHAT_ID = settings.get('DINGDING_MAC_CHAT_ID', default='cidVhnIuNh9n5Q0MoN8ddMqNw==')            # MAC请求放通群
+    DINGDING_SESSION_CHAT_ID = settings.get('DINGDING_SESSION_CHAT_ID', default='cidVhnIuNh9n5Q0MoN8ddMqNw==')    # 多拨告警群
+    #
+    _ACCESS_TOKEN = Token()
+
+    """
+    获取access_token
+    https://open.dingtalk.com/document/development/obtain-the-access-token-of-an-internal-app
+
+    POST /v1.0/oauth2/accessToken HTTP/1.1
+    Host:api.dingtalk.com
+    Content-Type:application/json
+
+    {
+      "appKey" : "dingeqqpkv3xxxxxx",
+      "appSecret" : "GT-lsu-taDAxxxsTsxxxx"
+    }
+
+    :return:
+    {
+      "accessToken" : "fw8ef8we8f76e6f7s8dxxxx",
+      "expireIn" : 7200
+    }
+    """
+    @classmethod
+    def get_access_token(cls) -> str:
+        if int(time.time()) > cls._ACCESS_TOKEN.expired_at:
+            data = {
+                'appKey': cls.DINGDING_APP_ID,
+                'appSecret': cls.DINGDING_APP_SECRET,
+            }
+            response = requests.post('https://api.dingtalk.com/v1.0/oauth2/accessToken', json=data)
+            body = response.json()
+            log.debug(f'API get_access_token: {body}')
+            cls._ACCESS_TOKEN = cls.Token(token=body['accessToken'], ttl=body['expireIn'])
+        log.debug(f'fetched access token: {cls._ACCESS_TOKEN.token}')
+        return cls._ACCESS_TOKEN.token
+
+    """
+    发送应用消息
+    https://open.dingtalk.com/document/development/the-robot-sends-a-group-message
+
+    POST /v1.0/robot/groupMessages/send HTTP/1.1
+    Host:api.dingtalk.com
+    x-acs-dingtalk-access-token:nvosnghskaknz8xxxxxx
+    Content-Type:application/json
+
+    {
+      "msgParam" : "{\"content\":\"钉钉，让进步发生\"}",
+      "msgKey" : "sampleText",
+      "openConversationId" : "cid6KeBBLoveMJOGXoYKF5xxxxxxx==",
+      "robotCode" : "dingue4kfzdxbynxxxxxx",
+      "coolAppCode" : "COOLAPP-1-10182EEDD1AC0BA60xxxxxx"
+    }
+
+    :return:
+    {
+      "processQueryKey" : "jkasdfb8va9hnxxxxxx"
+    }
+    """
+    @classmethod
+    def send_group_msg(cls, receiver_id: str, text: str):
+        headers = {
+            'x-acs-dingtalk-access-token': cls.get_access_token(),
+        }
+        data = {
+            'msgParam': f'{{"content":"{text}"}}',
+            'msgKey': 'sampleText',
+            'openConversationId': receiver_id,
+            'robotCode': cls.DINGDING_ROBOT_CODE,
+        }
+        response = requests.post('https://api.dingtalk.com/v1.0/robot/groupMessages/send', json=data, headers=headers)
+        body = response.json()
+        log.debug(f'API send_group_msg: {body}')
+        assert response.ok