报错、成功发送钉钉消息

update_data.py

@@ -6,6 +6,8 @@ import akshare as ak
 from index_downloader import get_all_stock_index
 import schedule
 import time
+import traceback
+from dingtalk import DingTalkBot
 
 
 def get_latest_index_kline_date():
@@ -60,7 +62,13 @@ def get_latest_index_kline_date():
     return df
 
 
+
+
 def main():
+    webhook = "https://oapi.dingtalk.com/robot/send?access_token=fb70c1561d8beba94b4f11568f4bb15e3ae07ccbdc8ac19676434a9d1cd17546"  # 填写你的webhook
+    secret = "SEC1ae7cd2f1a6f9da3611af37da3e7d954c1e8533fc073c6c8cc5e5af3b6e5926b"   # 填写你的加签token（如果有），否则留空
+    dingtalk = DingTalkBot(webhook, secret)
+    try:
         db_config = DatabaseConfig()
         logger.info(f"数据库连接: {db_config.connection_string}")
 
@@ -70,6 +78,7 @@ def main():
             logger.info("✅ 数据库连接测试成功")
         else:
             logger.error("❌ 数据库连接测试失败")
+            raise Exception("数据库连接测试失败")
 
         df = get_latest_index_kline_date()
         logger.info(df.head())
@@ -82,6 +91,9 @@ def main():
         if len(res) == 0:
             res = db_manager.insert_dataframe(df, "index_kline")
             logger.info(res)
+    except Exception as e:
+        logger.error(f"main 函数执行异常: {e}\n{traceback.format_exc()}")
+        dingtalk.send_text(f"main 函数执行异常: {e}\n{traceback.format_exc()}")
 
 
 if __name__ == "__main__":