update

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"]
 
-            -- 安全地初始化数据库连接
+            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
+                local db_ok, db = pcall(function() return self:initDB(input_sn) end)
-                self:D("cronPre initDB failed for " .. tostring(input_sn) .. ": " .. tostring(db))
+                if not db_ok or not db then
-                return false
+                    self:D("cronPre initDB failed for " .. tostring(input_sn) .. ": " .. tostring(db))
-            end
+                    return false
-
-            -- 开启事务
-            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
+                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
+                if success then
-                pcall(function() db:execute([[COMMIT]]) end)
+                    pcall(function() db:execute([[COMMIT]]) end)
-            else
+                else
-                pcall(function() db:exec([[ROLLBACK]]) end)
+                    pcall(function() db:exec([[ROLLBACK]]) end)
-            end
+                end
 
-            -- 安全关闭数据库
+                -- 安全关闭数据库
-            pcall(function() db:close() end)
+                pcall(function() db:close() end)
 
-            if not success then
+                if not success then
-                return false
+                    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
+                    local db_ok, db = pcall(function() return self:initDB(input_sn) end)
-                        db:close()
+                    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 self:is_migrating(input_sn) then
-                if not db then
                     ngx.shared.mw_total:rpush(total_key, data)
-                    return
+                else
-                end
+                    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]
+                            local stat_fields_is = stat_fields[input_sn]
-                if not stmt then
+                            for stf_k, stf_v in pairs(stat_tmp_fields) do
-                    ngx.shared.mw_total:rpush(total_key, data)
+                                if excluded then
-                    return
+                                    if stf_k == "spider_stat_fields" or stf_k == "client_stat_fields" then
-                end
+                                        break
+                                    end
+                                end
 
-                local insert_ok = self:store_logs_line(db, stmt, input_sn, info)
+                                local stf_is = stat_fields_is[stf_k]
-                if not insert_ok then
+                                if not stf_is then
-                    self:D("store_logs_line failed for " .. input_sn .. ": " .. tostring(insert_ok))
+                                    stf_is = {}
-                    ngx.shared.mw_total:rpush(total_key, data)
+                                    stat_fields_is[stf_k] = stf_is
-                    rollback_sites[input_sn] = true
+                                end
-                    return
-                end
 
-                local log_kv = info["log_kv"]
+                                for sv_k, sv_v in pairs(stf_v) do
-                local excluded = log_kv['excluded']
+                                    stf_is[sv_k] = (stf_is[sv_k] or 0) + sv_v
-                local stat_tmp_fields = info['stat_fields']
+                                end
+                            end
 
-                local stat_fields_is = stat_fields[input_sn]
+                            if not excluded then
-                for stf_k, stf_v in pairs(stat_tmp_fields) do
+                                local ip = log_kv['ip']
-                    if excluded then
+                                local body_length = log_kv["body_length"]
-                        if stf_k == "spider_stat_fields" or stf_k == "client_stat_fields" then
+
-                            break
+                                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
 

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:
+        if os.path.exists(hot_db_tmp):
-            print(f"[{site_name}] logs to history error: {e}")
+            os.remove(hot_db_tmp)
-        else:
+        print(f"[{site_name}] logs to history error: {e}")
-            print(f"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: