From 728a38c55ce7295262ec0668f0b36f97b5760dbc Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 18:35:05 +0800 Subject: [PATCH 01/17] Update webstats_log.lua --- plugins/webstats/lua/webstats_log.lua | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/plugins/webstats/lua/webstats_log.lua b/plugins/webstats/lua/webstats_log.lua index 54af3199d..67e2501ed 100644 --- a/plugins/webstats/lua/webstats_log.lua +++ b/plugins/webstats/lua/webstats_log.lua @@ -276,7 +276,11 @@ local function run_app() log_kv = kv, } - cache:rpush(total_key, json.encode(data)) + local len, err, forcible = cache:rpush(total_key, json.encode(data)) + if not len then + C:D("webstats rpush failed: " .. tostring(err or "unknown") + .. ", forcible=" .. tostring(forcible) .. ", queue=" .. total_key) + end end -- C:D("webstats_log run_app start, server_name=" .. tostring(server_name)) From 468cc54cf53e7b4ca17620534fd8006bffd09c75 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 18:55:25 +0800 Subject: [PATCH 02/17] Update webstats_common.lua --- plugins/webstats/lua/webstats_common.lua | 1 + 1 file changed, 1 insertion(+) diff --git a/plugins/webstats/lua/webstats_common.lua b/plugins/webstats/lua/webstats_common.lua index 741dc6709..ac70163b8 100644 --- a/plugins/webstats/lua/webstats_common.lua +++ b/plugins/webstats/lua/webstats_common.lua @@ -589,6 +589,7 @@ function _M.cron(self) local insert_ok = self:store_logs_line(db, stmt, input_sn, info) if not insert_ok then + self:D("store_logs_line failed for " .. input_sn .. ": " .. tostring(insert_ok)) ngx.shared.mw_total:rpush(total_key, data) rollback_sites[input_sn] = true return From 31106f2b13a2ef7d114697cf8f4f1a5936ae1f24 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 19:03:27 +0800 Subject: [PATCH 03/17] Update webstats_common.lua --- plugins/webstats/lua/webstats_common.lua | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/plugins/webstats/lua/webstats_common.lua b/plugins/webstats/lua/webstats_common.lua index ac70163b8..3a4bbfa08 100644 --- a/plugins/webstats/lua/webstats_common.lua +++ b/plugins/webstats/lua/webstats_common.lua @@ -961,10 +961,16 @@ function _M.D(self,msg) end function _M.is_migrating(self, input_sn) - local file = io.open(self.app_dir.."/migrating", "rb") - if file then return true end - local file = io.open(self.app_dir.."/logs/"..input_sn.."/migrating", "rb") - if file then return true end + local fp = io.open(self.app_dir.."/migrating", "rb") + if fp then + fp:close() + return true + end + local fp = io.open(self.app_dir.."/logs/"..input_sn.."/migrating", "rb") + if fp then + fp:close() + return true + end return false end From f34e93192d931887cc345d43a01b0e96b0d846c7 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 19:10:17 +0800 Subject: [PATCH 04/17] Update webstats_common.lua --- plugins/webstats/lua/webstats_common.lua | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/plugins/webstats/lua/webstats_common.lua b/plugins/webstats/lua/webstats_common.lua index 3a4bbfa08..bf74d4371 100644 --- a/plugins/webstats/lua/webstats_common.lua +++ b/plugins/webstats/lua/webstats_common.lua @@ -117,15 +117,11 @@ end - journal_size_limit = 20GB: WAL 文件大小限制 ]] function _M.initDB(self, input_sn) - log_dir = self.app_dir .. "/logs" - if log_dir == "" then - return nil - end - + local log_dir = self.app_dir .. "/logs" local path = log_dir .. '/' .. input_sn .. "/logs.db" local db, err = sqlite3.open(path) - if err then + C:D("initDB failed for " .. input_sn .. ": " .. tostring(err)) return nil end From 58ab48ac15478e33e4f57e13103c8c90b1792fd7 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 19:11:48 +0800 Subject: [PATCH 05/17] Update webstats_common.lua --- plugins/webstats/lua/webstats_common.lua | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/webstats/lua/webstats_common.lua b/plugins/webstats/lua/webstats_common.lua index bf74d4371..e67f4e855 100644 --- a/plugins/webstats/lua/webstats_common.lua +++ b/plugins/webstats/lua/webstats_common.lua @@ -121,7 +121,7 @@ function _M.initDB(self, input_sn) local path = log_dir .. '/' .. input_sn .. "/logs.db" local db, err = sqlite3.open(path) if err then - C:D("initDB failed for " .. input_sn .. ": " .. tostring(err)) + self:D("initDB failed for " .. input_sn .. ": " .. tostring(err)) return nil end From f9be534101e72734f69087f1281ab7168e0060fe Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 19:24:36 +0800 Subject: [PATCH 06/17] update --- plugins/webstats/lua/webstats_common.lua | 265 +++++++++++++++---------------- plugins/webstats/tool_migrate.py | 29 ++-- 2 files changed, 145 insertions(+), 149 deletions(-) diff --git a/plugins/webstats/lua/webstats_common.lua b/plugins/webstats/lua/webstats_common.lua index e67f4e855..bd91f8b15 100644 --- a/plugins/webstats/lua/webstats_common.lua +++ b/plugins/webstats/lua/webstats_common.lua @@ -397,41 +397,43 @@ function _M.cronPre(self) for _, site_v in ipairs(sites) do local input_sn = site_v["name"] - -- 安全地初始化数据库连接 - local db_ok, db = pcall(function() return self:initDB(input_sn) end) - if not db_ok or not db then - self:D("cronPre initDB failed for " .. tostring(input_sn) .. ": " .. tostring(db)) - return false - end - - -- 开启事务 - db:exec([[BEGIN TRANSACTION]]) - - local success = true - -- 预创建当前小时和下一小时的统计记录 - for _, ws_v in ipairs(wc_stat) do - if not self:_update_stat_pre(db, ws_v, time_key) then - success = false - break + if not self:is_migrating(input_sn) then + -- 安全地初始化数据库连接 + local db_ok, db = pcall(function() return self:initDB(input_sn) end) + if not db_ok or not db then + self:D("cronPre initDB failed for " .. tostring(input_sn) .. ": " .. tostring(db)) + return false end - if not self:_update_stat_pre(db, ws_v, time_key_next) then - success = false - break + + -- 开启事务 + db:exec([[BEGIN TRANSACTION]]) + + local success = true + -- 预创建当前小时和下一小时的统计记录 + for _, ws_v in ipairs(wc_stat) do + if not self:_update_stat_pre(db, ws_v, time_key) then + success = false + break + end + if not self:_update_stat_pre(db, ws_v, time_key_next) then + success = false + break + end end - end - -- 提交或回滚事务 - if success then - pcall(function() db:execute([[COMMIT]]) end) - else - pcall(function() db:exec([[ROLLBACK]]) end) - end + -- 提交或回滚事务 + if success then + pcall(function() db:execute([[COMMIT]]) end) + else + pcall(function() db:exec([[ROLLBACK]]) end) + end - -- 安全关闭数据库 - pcall(function() db:close() end) + -- 安全关闭数据库 + pcall(function() db:close() end) - if not success then - return false + if not success then + return false + end end end @@ -515,45 +517,45 @@ function _M.cron(self) end local ok, err = pcall(function() + if self:is_working('cron_init_stat') then + return + end + for _, site_v in ipairs(sites) do local input_sn = site_v["name"] if self:is_migrating(input_sn) then - return - end - - if self:is_working('cron_init_stat') then - return - end - - local db_ok, db = pcall(function() return self:initDB(input_sn) end) - if not db_ok or not db then - self:D("initDB failed for " .. input_sn .. ": " .. tostring(db)) - return - end - - stat_fields[input_sn] = {} - - dbs[input_sn] = db - self:clean_stats(db, input_sn) - - local stmt_ok, stmt = pcall(function() - return db:prepare[[INSERT INTO web_logs( - time, ip, scheme, domain, server_name, method, status_code, uri, body_length, - referer, user_agent, protocol, request_time, is_spider, request_headers, ip_list, client_port) - VALUES(:time, :ip, :scheme, :domain, :server_name, :method, :status_code, :uri, - :body_length, :referer, :user_agent, :protocol, :request_time, :is_spider, - :request_headers, :ip_list, :client_port)]] - end) - - if stmt_ok and stmt then - stmts[input_sn] = stmt - db:exec([[BEGIN TRANSACTION]]) + -- 该站点迁移中，跳过；其它站点继续写入 else - if db and db:isopen() then - db:close() + local db_ok, db = pcall(function() return self:initDB(input_sn) end) + if not db_ok or not db then + self:D("initDB failed for " .. input_sn .. ": " .. tostring(db)) + return + end + + stat_fields[input_sn] = {} + + dbs[input_sn] = db + self:clean_stats(db, input_sn) + + local stmt_ok, stmt = pcall(function() + return db:prepare[[INSERT INTO web_logs( + time, ip, scheme, domain, server_name, method, status_code, uri, body_length, + referer, user_agent, protocol, request_time, is_spider, request_headers, ip_list, client_port) + VALUES(:time, :ip, :scheme, :domain, :server_name, :method, :status_code, :uri, + :body_length, :referer, :user_agent, :protocol, :request_time, :is_spider, + :request_headers, :ip_list, :client_port)]] + end) + + if stmt_ok and stmt then + stmts[input_sn] = stmt + db:exec([[BEGIN TRANSACTION]]) + else + if db and db:isopen() then + db:close() + end + dbs[input_sn] = nil + return end - dbs[input_sn] = nil - return end end @@ -571,82 +573,79 @@ function _M.cron(self) end local input_sn = info['server_name'] - local db = dbs[input_sn] - if not db then + if self:is_migrating(input_sn) then ngx.shared.mw_total:rpush(total_key, data) - return - end + else + local db = dbs[input_sn] + local stmt = stmts[input_sn] + if not db or not stmt then + ngx.shared.mw_total:rpush(total_key, data) + else + local insert_ok = self:store_logs_line(db, stmt, input_sn, info) + if not insert_ok then + self:D("store_logs_line failed for " .. input_sn .. ": " .. tostring(insert_ok)) + ngx.shared.mw_total:rpush(total_key, data) + rollback_sites[input_sn] = true + else + local log_kv = info["log_kv"] + local excluded = log_kv['excluded'] + local stat_tmp_fields = info['stat_fields'] - local stmt = stmts[input_sn] - if not stmt then - ngx.shared.mw_total:rpush(total_key, data) - return - end + local stat_fields_is = stat_fields[input_sn] + for stf_k, stf_v in pairs(stat_tmp_fields) do + if excluded then + if stf_k == "spider_stat_fields" or stf_k == "client_stat_fields" then + break + end + end - local insert_ok = self:store_logs_line(db, stmt, input_sn, info) - if not insert_ok then - self:D("store_logs_line failed for " .. input_sn .. ": " .. tostring(insert_ok)) - ngx.shared.mw_total:rpush(total_key, data) - rollback_sites[input_sn] = true - return - end + local stf_is = stat_fields_is[stf_k] + if not stf_is then + stf_is = {} + stat_fields_is[stf_k] = stf_is + end - local log_kv = info["log_kv"] - local excluded = log_kv['excluded'] - local stat_tmp_fields = info['stat_fields'] + for sv_k, sv_v in pairs(stf_v) do + stf_is[sv_k] = (stf_is[sv_k] or 0) + sv_v + end + end - local stat_fields_is = stat_fields[input_sn] - for stf_k, stf_v in pairs(stat_tmp_fields) do - if excluded then - if stf_k == "spider_stat_fields" or stf_k == "client_stat_fields" then - break + if not excluded then + local ip = log_kv['ip'] + local body_length = log_kv["body_length"] + + local ip_stats_sn = ip_stats[input_sn] + if not ip_stats_sn then + ip_stats_sn = {} + ip_stats[input_sn] = ip_stats_sn + end + + local ip_stat = ip_stats_sn[ip] + if not ip_stat then + ip_stats_sn[ip] = {ip_num = 1, body_length = body_length} + else + ip_stat.ip_num = ip_stat.ip_num + 1 + ip_stat.body_length = ip_stat.body_length + body_length + end + + local url_stats_sn = url_stats[input_sn] + if not url_stats_sn then + url_stats_sn = {} + url_stats[input_sn] = url_stats_sn + end + + local request_uri = log_kv["request_uri"] + local request_uri_md5 = ngx.md5(request_uri) + local url_stat = url_stats_sn[request_uri_md5] + if not url_stat then + url_stats_sn[request_uri_md5] = {url_num = 1, uri = request_uri, body_length = body_length} + else + url_stat.url_num = url_stat.url_num + 1 + url_stat.body_length = url_stat.body_length + body_length + end + end end end - - local stf_is = stat_fields_is[stf_k] - if not stf_is then - stf_is = {} - stat_fields_is[stf_k] = stf_is - end - - for sv_k, sv_v in pairs(stf_v) do - stf_is[sv_k] = (stf_is[sv_k] or 0) + sv_v - end - end - - if not excluded then - local ip = log_kv['ip'] - local body_length = log_kv["body_length"] - - local ip_stats_sn = ip_stats[input_sn] - if not ip_stats_sn then - ip_stats_sn = {} - ip_stats[input_sn] = ip_stats_sn - end - - local ip_stat = ip_stats_sn[ip] - if not ip_stat then - ip_stats_sn[ip] = {ip_num = 1, body_length = body_length} - else - ip_stat.ip_num = ip_stat.ip_num + 1 - ip_stat.body_length = ip_stat.body_length + body_length - end - - local url_stats_sn = url_stats[input_sn] - if not url_stats_sn then - url_stats_sn = {} - url_stats[input_sn] = url_stats_sn - end - - local request_uri = log_kv["request_uri"] - local request_uri_md5 = ngx.md5(request_uri) - local url_stat = url_stats_sn[request_uri_md5] - if not url_stat then - url_stats_sn[request_uri_md5] = {url_num = 1, uri = request_uri, body_length = body_length} - else - url_stat.url_num = url_stat.url_num + 1 - url_stat.body_length = url_stat.body_length + body_length - end end end diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index cb84306d3..f563fe2cc 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -88,7 +88,11 @@ def migrateSiteHotLogs(site_name, query_date): print(f"[{site_name}] 正在迁移中，跳过") return mw.returnMsg(True, f"{site_name} is migrating, skip") - # 1. 备份到临时文件（使用文件直接拷贝） + todayTime = time.strftime('%Y-%m-%d 00:00:00', time.localtime()) + todayUt = int(time.mktime(time.strptime( + todayTime, "%Y-%m-%d %H:%M:%S"))) + + # 1. 备份到临时文件（copy 期间短暂互斥，完成后立即解除） try: import shutil print(f"[{site_name}] 备份 {hot_db} -> {hot_db_tmp} ...") @@ -98,11 +102,12 @@ def migrateSiteHotLogs(site_name, query_date): if not os.path.exists(hot_db_tmp): return mw.returnMsg(False, f"{site_name} migrating fail, copy tmp file!") except Exception as e: + return mw.returnMsg(False, f"{site_name} migrating fail: {e}") + finally: if os.path.exists(migrating_flag): os.remove(migrating_flag) - return mw.returnMsg(False, f"{site_name} migrating fail: {e}") - # 2. 从临时备份中迁移热日志数据到历史日志（批量插入） + # 2. 从临时备份中迁移热日志数据到历史日志（读 logs_tmp，不阻塞 live 写入） try: logs_conn = pSqliteDb('web_log', site_name, 'logs_tmp') history_logs_conn = pSqliteDb('web_log', site_name, 'history_logs') @@ -113,10 +118,6 @@ def migrateSiteHotLogs(site_name, query_date): columns_str = ",".join(_columns) placeholders = ",".join(["?"] * len(_columns)) - todayTime = time.strftime('%Y-%m-%d 00:00:00', time.localtime()) - todayUt = int(time.mktime(time.strptime( - todayTime, "%Y-%m-%d %H:%M:%S"))) - logs_sql = f"select {columns_str} from web_logs where time<{todayUt}" selector = logs_conn.originExecute(logs_sql) @@ -158,18 +159,15 @@ def migrateSiteHotLogs(site_name, query_date): history_logs_conn.commit() except Exception as e: - if site_name: - print(f"[{site_name}] logs to history error: {e}") - else: - print(f"logs to history error: {e}") + if os.path.exists(hot_db_tmp): + os.remove(hot_db_tmp) + print(f"[{site_name}] logs to history error: {e}") return mw.returnMsg(False, f"{site_name} logs migrate error: {e}") - # 3. 删除已迁移的数据并清理统计（批量删除） + # 3. 删除已迁移的数据并清理统计（仅 DELETE/VACUUM 期间互斥） try: - if os.path.exists(migrating_flag): - os.remove(migrating_flag) - mw.writeFile(migrating_flag, "yes") + time.sleep(0.5) hot_db_conn = pSqliteDb('web_logs', site_name) @@ -242,7 +240,6 @@ def migrateHotLogs(query_date="today"): print(f"\n迁移完成! 成功: {success_count}, 失败: {fail_count}") - mw.opWeb('restart') return mw.returnMsg(True, f"logs migrate ok, success: {success_count}, fail: {fail_count}") except BlockingIOError: From 9c83695187c6aae5b6878879c5c066bf2ae75fe3 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 19:44:14 +0800 Subject: [PATCH 07/17] Create tool_migrate.py --- plugins/webstats/bak/tool_migrate.py | 259 +++++++++++++++++++++++++++++++++++ 1 file changed, 259 insertions(+) create mode 100644 plugins/webstats/bak/tool_migrate.py diff --git a/plugins/webstats/bak/tool_migrate.py b/plugins/webstats/bak/tool_migrate.py new file mode 100644 index 000000000..f563fe2cc --- /dev/null +++ b/plugins/webstats/bak/tool_migrate.py @@ -0,0 +1,259 @@ +# coding:utf-8 + +import sys +import io +import os +import time +import json +import fcntl + +web_dir = os.getcwd() + "/web" +if os.path.exists(web_dir): + sys.path.append(web_dir) + os.chdir(web_dir) + +import core.mw as mw +from utils.crontab import crontab as MwCrontab + +app_debug = False +if mw.isAppleSystem(): + app_debug = True + + +def getPluginName(): + return 'webstats' + + +def getPluginDir(): + return mw.getPluginDir() + '/' + getPluginName() + + +def getServerDir(): + return mw.getServerDir() + '/' + getPluginName() + + +def getConf(): + conf = getServerDir() + "/lua/config.json" + return conf + + +def getGlobalConf(): + conf = getConf() + content = mw.readFile(conf) + result = json.loads(content) + return result + + +def pSqliteDb(dbname='web_logs', site_name='unset', fn="logs"): + db_dir = getServerDir() + '/logs/' + site_name + if not os.path.exists(db_dir): + mw.execShell('mkdir -p ' + db_dir) + + name = fn + file = db_dir + '/' + name + '.db' + + if not os.path.exists(file): + conn = mw.M(dbname).dbPos(db_dir, name) + sql = mw.readFile(getPluginDir() + '/conf/init.sql') + sql_list = sql.split(';') + for index in range(len(sql_list)): + conn.execute(sql_list[index], ()) + else: + conn = mw.M(dbname).dbPos(db_dir, name) + + conn.execute("PRAGMA synchronous = 0", ()) + conn.execute("PRAGMA page_size = 4096", ()) + conn.execute("PRAGMA journal_mode = wal", ()) + + conn.autoTextFactory() + + return conn + + +def migrateSiteHotLogs(site_name, query_date): + start_time = time.time() + print(f"[{site_name}] 开始迁移日志...") + + migrating_flag = getServerDir() + "/logs/%s/migrating" % site_name + hot_db = getServerDir() + "/logs/%s/logs.db" % site_name + hot_db_tmp = getServerDir() + "/logs/%s/logs_tmp.db" % site_name + history_logs_db = getServerDir() + "/logs/%s/history_logs.db" % site_name + + if not os.path.exists(hot_db): + print(f"[{site_name}] 热日志数据库不存在，跳过") + return mw.returnMsg(True, f"{site_name} logs not exist, skip") + + # 检查是否正在迁移 + if os.path.exists(migrating_flag): + print(f"[{site_name}] 正在迁移中，跳过") + return mw.returnMsg(True, f"{site_name} is migrating, skip") + + todayTime = time.strftime('%Y-%m-%d 00:00:00', time.localtime()) + todayUt = int(time.mktime(time.strptime( + todayTime, "%Y-%m-%d %H:%M:%S"))) + + # 1. 备份到临时文件（copy 期间短暂互斥，完成后立即解除） + try: + import shutil + print(f"[{site_name}] 备份 {hot_db} -> {hot_db_tmp} ...") + mw.writeFile(migrating_flag, "yes") + time.sleep(0.5) + shutil.copy(hot_db, hot_db_tmp) + if not os.path.exists(hot_db_tmp): + return mw.returnMsg(False, f"{site_name} migrating fail, copy tmp file!") + except Exception as e: + return mw.returnMsg(False, f"{site_name} migrating fail: {e}") + finally: + if os.path.exists(migrating_flag): + os.remove(migrating_flag) + + # 2. 从临时备份中迁移热日志数据到历史日志（读 logs_tmp，不阻塞 live 写入） + try: + logs_conn = pSqliteDb('web_log', site_name, 'logs_tmp') + history_logs_conn = pSqliteDb('web_log', site_name, 'history_logs') + + hot_db_columns = logs_conn.originExecute( + "PRAGMA table_info([web_logs])") + _columns = [c[1] for c in hot_db_columns if c[1] != "id"] + columns_str = ",".join(_columns) + placeholders = ",".join(["?"] * len(_columns)) + + logs_sql = f"select {columns_str} from web_logs where time<{todayUt}" + selector = logs_conn.originExecute(logs_sql) + + # 批量插入配置 + batch_size = 10000 + batch_count = 0 + total_count = 0 + insert_sql = f"insert into web_logs({columns_str}) values({placeholders})" + + while True: + logs = selector.fetchmany(batch_size) + if not logs: + break + + batch_count += 1 + total_count += len(logs) + # 使用原生 executemany 批量插入 + history_logs_conn.executemany(insert_sql, logs) + history_logs_conn.commit() + + if batch_count % 10 == 0: + elapsed = time.time() - start_time + print(f"[{site_name}] 已迁移 {total_count} 条记录, 耗时 {elapsed:.2f}s") + + print(f"[{site_name}] 共迁移 {total_count} 条记录") + + # 清理历史数据 + gcfg = getGlobalConf() + save_day = gcfg['global']["save_day"] + print(f"[{site_name}] 删除 {save_day} 天前的历史数据...") + + time_now = time.localtime() + save_timestamp = time.mktime((time_now.tm_year, time_now.tm_mon, time_now.tm_mday - save_day, 0, 0, 0, 0, 0, 0)) + delete_sql = f"delete from web_logs where time <= {save_timestamp}" + history_logs_conn.execute(delete_sql) + history_logs_conn.commit() + + history_logs_conn.execute("VACUUM;") + history_logs_conn.commit() + + except Exception as e: + if os.path.exists(hot_db_tmp): + os.remove(hot_db_tmp) + print(f"[{site_name}] logs to history error: {e}") + return mw.returnMsg(False, f"{site_name} logs migrate error: {e}") + + # 3. 删除已迁移的数据并清理统计（仅 DELETE/VACUUM 期间互斥） + try: + mw.writeFile(migrating_flag, "yes") + time.sleep(0.5) + + hot_db_conn = pSqliteDb('web_logs', site_name) + + # 分批删除热日志 + del_hot_log = f"delete from web_logs where time<{todayUt}" + print(f"[{site_name}] 删除已迁移的热日志...") + hot_db_conn.execute(del_hot_log) + + # 删除过期统计数据 + print(f"[{site_name}] 删除180天前的统计数据...") + save_time_key = time.strftime( + '%Y%m%d00', time.localtime(time.time() - 180 * 86400)) + + del_request_stat_sql = f"delete from request_stat where time<={save_time_key}" + hot_db_conn.execute(del_request_stat_sql) + hot_db_conn.execute(f"delete from spider_stat where time<={save_time_key}") + hot_db_conn.execute(f"delete from client_stat where time<={save_time_key}") + hot_db_conn.execute(f"delete from referer_stat where time<={save_time_key}") + + hot_db_conn.commit() + print(f"[{site_name}] 压缩热数据库...") + hot_db_conn.execute("VACUUM;") + hot_db_conn.commit() + + except Exception as e: + print(f"[{site_name}] delete hot logs error: {e}") + finally: + if os.path.exists(migrating_flag): + os.remove(migrating_flag) + if os.path.exists(hot_db_tmp): + os.remove(hot_db_tmp) + + elapsed = time.time() - start_time + print(f"[{site_name}] 日志迁移完成，耗时 {elapsed:.2f}s") + + if not mw.isAppleSystem(): + mw.execShell("chown -R www:www " + getServerDir()) + + return mw.returnMsg(True, f"{site_name} logs migrate ok") + + +def migrateHotLogs(query_date="today"): + # 使用文件锁防止并发迁移 + lock_file = getServerDir() + "/logs/migrate_hot_logs.lock" + lock_fd = None + + try: + lock_fd = open(lock_file, 'w') + fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + + print("开始迁移热日志...") + sites = mw.M('sites').field('name').order("add_time").select() + + unset_site = {"name": "unset"} + sites.append(unset_site) + + total_sites = len(sites) + success_count = 0 + fail_count = 0 + + for i, site_info in enumerate(sites): + site_name = site_info["name"] + print(f"\n[{i+1}/{total_sites}] 处理站点: {site_name}") + migrate_res = migrateSiteHotLogs(site_name, query_date) + if migrate_res["status"]: + success_count += 1 + else: + fail_count += 1 + print(f"[{site_name}] 迁移失败: {migrate_res['msg']}") + + print(f"\n迁移完成! 成功: {success_count}, 失败: {fail_count}") + + return mw.returnMsg(True, f"logs migrate ok, success: {success_count}, fail: {fail_count}") + + except BlockingIOError: + print("正在迁移中，跳过") + return mw.returnMsg(True, "正在迁移中，跳过") + except Exception as e: + print(f"migrateHotLogs error: {e}") + return mw.returnMsg(False, f"migrateHotLogs error: {e}") + finally: + if lock_fd: + try: + fcntl.flock(lock_fd.fileno(), fcntl.LOCK_UN) + lock_fd.close() + if os.path.exists(lock_file): + os.remove(lock_file) + except: + pass \ No newline at end of file From 4f38a9f3ff8556618b0861705458b6596f87f56a Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 19:49:58 +0800 Subject: [PATCH 08/17] Update tool_migrate.py --- plugins/webstats/tool_migrate.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index f563fe2cc..96ba605c0 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -97,7 +97,7 @@ def migrateSiteHotLogs(site_name, query_date): import shutil print(f"[{site_name}] 备份 {hot_db} -> {hot_db_tmp} ...") mw.writeFile(migrating_flag, "yes") - time.sleep(0.5) + time.sleep(1) shutil.copy(hot_db, hot_db_tmp) if not os.path.exists(hot_db_tmp): return mw.returnMsg(False, f"{site_name} migrating fail, copy tmp file!") From 5cdc5af8cd5b65e8112d27d7d9413cc76377b2ee Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 19:50:13 +0800 Subject: [PATCH 09/17] Update tool_migrate.py --- plugins/webstats/tool_migrate.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index 96ba605c0..2e0d1981a 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -167,7 +167,7 @@ def migrateSiteHotLogs(site_name, query_date): # 3. 删除已迁移的数据并清理统计（仅 DELETE/VACUUM 期间互斥） try: mw.writeFile(migrating_flag, "yes") - time.sleep(0.5) + time.sleep(1) hot_db_conn = pSqliteDb('web_logs', site_name) From b46f45569ee5b1b91c61f61129c19ff4544918a1 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 20:01:23 +0800 Subject: [PATCH 10/17] Update tool_migrate.py --- plugins/webstats/tool_migrate.py | 51 +++++++++++++++++++++++----------------- 1 file changed, 29 insertions(+), 22 deletions(-) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index 2e0d1981a..b960fd363 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -70,6 +70,23 @@ def pSqliteDb(dbname='web_logs', site_name='unset', fn="logs"): return conn +def batch_delete(conn, table, where_clause, batch_size=5000, pause_sec=0.05): + """分批删除，缩短单次持锁时间，WAL 模式下可与 OpenResty 并发写入。""" + total = 0 + while True: + sql = ( + f"DELETE FROM {table} WHERE rowid IN " + f"(SELECT rowid FROM {table} WHERE {where_clause} LIMIT {batch_size})" + ) + affected = conn.execute(sql) + if not isinstance(affected, int) or affected == 0: + break + total += affected + if pause_sec > 0: + time.sleep(pause_sec) + return total + + def migrateSiteHotLogs(site_name, query_date): start_time = time.time() print(f"[{site_name}] 开始迁移日志...") @@ -164,39 +181,29 @@ def migrateSiteHotLogs(site_name, query_date): print(f"[{site_name}] logs to history error: {e}") return mw.returnMsg(False, f"{site_name} logs migrate error: {e}") - # 3. 删除已迁移的数据并清理统计（仅 DELETE/VACUUM 期间互斥） + # 3. 分批删除已迁移的热日志并清理统计（不设 migrating 锁，不阻塞 OpenResty 写入） try: - mw.writeFile(migrating_flag, "yes") - time.sleep(1) - hot_db_conn = pSqliteDb('web_logs', site_name) + hot_db_conn.execute("PRAGMA busy_timeout = 30000") - # 分批删除热日志 - del_hot_log = f"delete from web_logs where time<{todayUt}" - print(f"[{site_name}] 删除已迁移的热日志...") - hot_db_conn.execute(del_hot_log) + print(f"[{site_name}] 分批删除已迁移的热日志...") + deleted = batch_delete(hot_db_conn, 'web_logs', f"time<{todayUt}") + print(f"[{site_name}] 已删除 {deleted} 条热日志") - # 删除过期统计数据 - print(f"[{site_name}] 删除180天前的统计数据...") + print(f"[{site_name}] 分批删除180天前的统计数据...") save_time_key = time.strftime( '%Y%m%d00', time.localtime(time.time() - 180 * 86400)) + stat_where = f"time<='{save_time_key}'" + for stat_table in ('request_stat', 'spider_stat', 'client_stat', 'referer_stat'): + stat_deleted = batch_delete(hot_db_conn, stat_table, stat_where) + print(f"[{site_name}] {stat_table} 已删除 {stat_deleted} 条") - del_request_stat_sql = f"delete from request_stat where time<={save_time_key}" - hot_db_conn.execute(del_request_stat_sql) - hot_db_conn.execute(f"delete from spider_stat where time<={save_time_key}") - hot_db_conn.execute(f"delete from client_stat where time<={save_time_key}") - hot_db_conn.execute(f"delete from referer_stat where time<={save_time_key}") - - hot_db_conn.commit() - print(f"[{site_name}] 压缩热数据库...") - hot_db_conn.execute("VACUUM;") - hot_db_conn.commit() + print(f"[{site_name}] 执行 WAL checkpoint...") + hot_db_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") except Exception as e: print(f"[{site_name}] delete hot logs error: {e}") finally: - if os.path.exists(migrating_flag): - os.remove(migrating_flag) if os.path.exists(hot_db_tmp): os.remove(hot_db_tmp) From 5839dbcd6b5ab0d2f56e83a0ee5aef7e8e9e048b Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 20:05:10 +0800 Subject: [PATCH 11/17] Update tool_migrate.py --- plugins/webstats/tool_migrate.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index b960fd363..ee8203b66 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -115,7 +115,9 @@ def migrateSiteHotLogs(site_name, query_date): print(f"[{site_name}] 备份 {hot_db} -> {hot_db_tmp} ...") mw.writeFile(migrating_flag, "yes") time.sleep(1) + copy_start = time.time() shutil.copy(hot_db, hot_db_tmp) + print(f"[{site_name}] 备份完成，耗时 {time.time() - copy_start:.2f}s") if not os.path.exists(hot_db_tmp): return mw.returnMsg(False, f"{site_name} migrating fail, copy tmp file!") except Exception as e: From 6a6f6e6062b9d75e66dc2d4eec08b2a8ec977c04 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 20:27:17 +0800 Subject: [PATCH 12/17] Update tool_migrate.py --- plugins/webstats/tool_migrate.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index ee8203b66..745df97c5 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -6,6 +6,7 @@ import os import time import json import fcntl +import shutil web_dir = os.getcwd() + "/web" if os.path.exists(web_dir): @@ -43,6 +44,10 @@ def getGlobalConf(): result = json.loads(content) return result +def fastCopy(src, dst, buffer_size=128 * 1024 * 1024): # 128MB 缓冲区 + with open(src, 'rb') as fsrc: + with open(dst, 'wb') as fdst: + shutil.copyfileobj(fsrc, fdst, length=buffer_size) def pSqliteDb(dbname='web_logs', site_name='unset', fn="logs"): db_dir = getServerDir() + '/logs/' + site_name @@ -111,12 +116,12 @@ def migrateSiteHotLogs(site_name, query_date): # 1. 备份到临时文件（copy 期间短暂互斥，完成后立即解除） try: - import shutil print(f"[{site_name}] 备份 {hot_db} -> {hot_db_tmp} ...") mw.writeFile(migrating_flag, "yes") time.sleep(1) copy_start = time.time() - shutil.copy(hot_db, hot_db_tmp) + # shutil.copy(hot_db, hot_db_tmp) + fastCopy(hot_db, hot_db_tmp, 512 * 1024 * 1024) print(f"[{site_name}] 备份完成，耗时 {time.time() - copy_start:.2f}s") if not os.path.exists(hot_db_tmp): return mw.returnMsg(False, f"{site_name} migrating fail, copy tmp file!") From b88ea7aa3ccd9c54edf535dcf480c9f6f6d0fc38 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 20:32:36 +0800 Subject: [PATCH 13/17] update --- plugins/webstats/tool_migrate.py | 7 +------ web/core/mw.py | 6 ++++++ 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index 745df97c5..2b74bb3da 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -44,11 +44,6 @@ def getGlobalConf(): result = json.loads(content) return result -def fastCopy(src, dst, buffer_size=128 * 1024 * 1024): # 128MB 缓冲区 - with open(src, 'rb') as fsrc: - with open(dst, 'wb') as fdst: - shutil.copyfileobj(fsrc, fdst, length=buffer_size) - def pSqliteDb(dbname='web_logs', site_name='unset', fn="logs"): db_dir = getServerDir() + '/logs/' + site_name if not os.path.exists(db_dir): @@ -121,7 +116,7 @@ def migrateSiteHotLogs(site_name, query_date): time.sleep(1) copy_start = time.time() # shutil.copy(hot_db, hot_db_tmp) - fastCopy(hot_db, hot_db_tmp, 512 * 1024 * 1024) + mw.fastCopy(hot_db, hot_db_tmp, 512 * 1024 * 1024) print(f"[{site_name}] 备份完成，耗时 {time.time() - copy_start:.2f}s") if not os.path.exists(hot_db_tmp): return mw.returnMsg(False, f"{site_name} migrating fail, copy tmp file!") diff --git a/web/core/mw.py b/web/core/mw.py index 6fbb975e2..5a0315aac 100644 --- a/web/core/mw.py +++ b/web/core/mw.py @@ -25,6 +25,7 @@ import subprocess import glob import base64 import re +import shutil from random import Random @@ -318,6 +319,11 @@ def toSize(size, middle='') -> str: s = u return str(round(size, 2)) + middle + u +def fastCopy(src, dst, buffer_size=128 * 1024 * 1024): # 128MB 缓冲区 + with open(src, 'rb') as fsrc: + with open(dst, 'wb') as fdst: + shutil.copyfileobj(fsrc, fdst, length=buffer_size) + def returnData(status, msg, data=None): if data is None: return {'status': status, 'msg': msg} From 18443aab608463eba855f2970b389f7502cdb726 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 20:49:20 +0800 Subject: [PATCH 14/17] Update tool_migrate.py --- plugins/webstats/tool_migrate.py | 1 + 1 file changed, 1 insertion(+) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index 2b74bb3da..14139803b 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -115,6 +115,7 @@ def migrateSiteHotLogs(site_name, query_date): mw.writeFile(migrating_flag, "yes") time.sleep(1) copy_start = time.time() + # import shutil # shutil.copy(hot_db, hot_db_tmp) mw.fastCopy(hot_db, hot_db_tmp, 512 * 1024 * 1024) print(f"[{site_name}] 备份完成，耗时 {time.time() - copy_start:.2f}s") From 39bd87e868a797b27c708e28391f6f69010383b0 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 20:54:00 +0800 Subject: [PATCH 15/17] Update tool_migrate.py --- plugins/webstats/tool_migrate.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index 14139803b..ec58c453f 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -113,7 +113,7 @@ def migrateSiteHotLogs(site_name, query_date): try: print(f"[{site_name}] 备份 {hot_db} -> {hot_db_tmp} ...") mw.writeFile(migrating_flag, "yes") - time.sleep(1) + time.sleep(0.5) copy_start = time.time() # import shutil # shutil.copy(hot_db, hot_db_tmp) From 6f51eaa29856d3d21d4cdc0a4fa5dd63067e089d Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 20:59:32 +0800 Subject: [PATCH 16/17] update --- plugins/webstats/tool_migrate.py | 2 +- web/core/mw.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index ec58c453f..60d24506b 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -117,7 +117,7 @@ def migrateSiteHotLogs(site_name, query_date): copy_start = time.time() # import shutil # shutil.copy(hot_db, hot_db_tmp) - mw.fastCopy(hot_db, hot_db_tmp, 512 * 1024 * 1024) + mw.fastCopy(hot_db, hot_db_tmp, 256 * 1024) print(f"[{site_name}] 备份完成，耗时 {time.time() - copy_start:.2f}s") if not os.path.exists(hot_db_tmp): return mw.returnMsg(False, f"{site_name} migrating fail, copy tmp file!") diff --git a/web/core/mw.py b/web/core/mw.py index 5a0315aac..e6100a285 100644 --- a/web/core/mw.py +++ b/web/core/mw.py @@ -319,7 +319,7 @@ def toSize(size, middle='') -> str: s = u return str(round(size, 2)) + middle + u -def fastCopy(src, dst, buffer_size=128 * 1024 * 1024): # 128MB 缓冲区 +def fastCopy(src, dst, buffer_size=256 * 1024): # 128MB 缓冲区 with open(src, 'rb') as fsrc: with open(dst, 'wb') as fdst: shutil.copyfileobj(fsrc, fdst, length=buffer_size) From 4e3e7864be07022b299135864d39d9051448b0e9 Mon Sep 17 00:00:00 2001 From: dami Date: Mon, 6 Jul 2026 21:02:13 +0800 Subject: [PATCH 17/17] Update tool_migrate.py --- plugins/webstats/tool_migrate.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/plugins/webstats/tool_migrate.py b/plugins/webstats/tool_migrate.py index 60d24506b..e372d14fc 100644 --- a/plugins/webstats/tool_migrate.py +++ b/plugins/webstats/tool_migrate.py @@ -209,6 +209,10 @@ def migrateSiteHotLogs(site_name, query_date): finally: if os.path.exists(hot_db_tmp): os.remove(hot_db_tmp) + if os.path.exists(hot_db_tmp+"-shm"): + os.remove(hot_db_tmp+"-shm") + if os.path.exists(hot_db_tmp+"-wal"): + os.remove(hot_db_tmp+"-wal") elapsed = time.time() - start_time print(f"[{site_name}] 日志迁移完成，耗时 {elapsed:.2f}s")