不使用本地库，使用mariadb完成存储

message_report/write_db.py

@@ -1,27 +1,24 @@
 import schedule
 import time
-import sqlite3
 from datetime import datetime, timedelta
 import redis
+import pymysql
 
 # 连接到Redis
 r = redis.Redis(host='192.168.2.32', port=6379, db=0)
 
+# 配置数据库连接
+db_config = {
+    'host': '192.168.2.32',  # 替换为你的MariaDB服务器地址
+    'user': 'root',  # 替换为你的MariaDB用户名
+    'password': 'lw123456',  # 替换为你的MariaDB密码
+    'database': 'message_archive'
+}
+
 
 def write_to_db():
-    # 连接到SQLite数据库
+    # 连接到数据库
-    conn = sqlite3.connect('message_stats.db')
+    connection = pymysql.connect(**db_config)
-    c = conn.cursor()
-
-    # 创建表（如果不存在）
-    c.execute('''CREATE TABLE IF NOT EXISTS speech_counts (
-                    group_id TEXT,
-                    wx_id TEXT,
-                    date TEXT,
-                    count INTEGER,
-                    PRIMARY KEY (group_id, wx_id, date)
-                )''')
-
     # 获取当前日期的前一天
     yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
 
@@ -30,19 +27,28 @@ def write_to_db():
         parts = key.decode('utf-8').split(':')
         group_id, wx_id, _date = parts[0], parts[1], parts[2]  # _date应该是yesterday，但这里为了完整性还是保留
         count = int(r.hget(key, 'count')) if isinstance(r.hget(key, 'count'), bytes) else 0  # 处理可能的None或空值情况
+        # 插入消息信息
+        try:
+            with connection.cursor() as cursor:
+                # 插入消息信息
+                sql = """
+                    INSERT OR REPLACE INTO speech_counts (group_id, wx_id, date, count) VALUES (%s, %s, %s, %s)
+                """
+                cursor.execute(sql, (group_id, wx_id, yesterday, count))
 
-        # 插入或更新数据库记录
+            # 提交事务
-        c.execute("INSERT OR REPLACE INTO speech_counts (group_id, wx_id, date, count) VALUES (?, ?, ?, ?)",
+            connection.commit()
-                  (group_id, wx_id, yesterday, count))
+            print("write_to_db   successfully.")
 
-    conn.commit()
+        except Exception as e:
-    conn.close()
+            print(f"write_to_db   message: {e}")
+            connection.rollback()
 
 
 def generate_and_send_ranking(groupId, allContacts: dict):
     # 连接到SQLite数据库（假设数据库文件名为database.db）
-    conn = sqlite3.connect('message_stats.db')
+    connection = pymysql.connect(**db_config)
-    cursor = conn.cursor()
+    cursor = connection.cursor()
 
     yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
     # 编写SQL查询来获取发言数量前20的用户
@@ -50,7 +56,7 @@ def generate_and_send_ranking(groupId, allContacts: dict):
       SELECT wx_id,  count AS speech_count
         FROM speech_counts
         WHERE DATE(date) = DATE('now', '-1 day')
-        AND group_id = ?
+        AND group_id = %s
         GROUP BY wx_id
         ORDER BY count DESC
         LIMIT 20
@@ -63,13 +69,14 @@ def generate_and_send_ranking(groupId, allContacts: dict):
     # 格式化输出字符串
     ranking_str = yesterday + "发言数量前20的用户排名:\n"
     for rank, (username, speech_count) in enumerate(results, start=1):
-            ranking_str += f"{rank}. {allContacts[username]}: {speech_count} 次发言\n"
+        ranking_str += f"{rank}. {allContacts[username]}: {speech_count} 次发言\n"
     # 关闭数据库连接
-    conn.close()
+    connection.close()
     print(ranking_str)
     # 这里我们没有实际“发送”排名，只是返回字符串
     # 如果需要发送，可以在此添加发送逻辑
     return ranking_str
 
+
 if __name__ == '__main__':
     write_to_db()