数据迁移OK

class/core/db.py

@@ -281,7 +281,16 @@ class Sql():
         except Exception as ex:
             return "error: " + str(ex)
 
-    def execute(self, sql, param):
+    def originExecute(self, sql, param=()):
+        self.__GetConn()
+        try:
+            result = self.__DB_CONN.execute(sql, param)
+            self.__DB_CONN.commit()
+            return result
+        except Exception as ex:
+            return "error: " + str(ex)
+
+    def execute(self, sql, param=()):
         # 执行SQL语句返回受影响行
         self.__GetConn()
         # print sql, param

plugins/webstats/tool_migrate.py

@@ -32,19 +32,166 @@ def getTaskConf():
     return conf
 
 
+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.text_factory = lambda x: str(x, encoding="utf-8", errors='ignore')
+    # conn.text_factory = lambda x: unicode(x, "utf-8", "ignore")
+    return conn
+
+
 def migrateSiteHotLogs(site_name, query_date):
     print(site_name, query_date)
-    return mw.returnMsg(True, "{} 日志合并成功！".format(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
+    # 1. copy to tmp file
+    try:
+        import shutil
+        print("coping {} to {} ...".format(hot_db, hot_db_tmp))
+        mw.writeFile(migrating_flag, "yes")
+        shutil.copy(hot_db, hot_db_tmp)
+        if not os.path.exists(hot_db_tmp):
+            return mw.returnMsg(False, "migrating fail, copy tmp file!")
+    except:
+        return mw.returnMsg(False, "{} migrating fail.".format(site_name))
+    finally:
+        if os.path.exists(migrating_flag):
+            os.remove(migrating_flag)
+
+     # 2. 从临时备份中迁移热日志数据到历史日志
+    print("begin tmp to hot log data ...")
+    try:
+        print("history file: {}".format(history_logs_db))
+        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 = ",".join([c[1] for c in hot_db_columns if c[1] != "id"])
+        query_start = 0
+        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 = "select {} from web_logs where time<{}".format(
+            _columns, todayUt)
+        selector = logs_conn.originExecute(logs_sql)
+        log = selector.fetchone()
+        while log:
+            params = ""
+            for field in log:
+                if params:
+                    params += ","
+                if field is None:
+                    field = "\'\'"
+                elif type(field) == str:
+                    field = "\'" + field.replace("\'", "\”") + "\'"
+                params += str(field)
+            insert_sql = "insert into web_logs(" + \
+                _columns + ") values(" + params + ")"
+            history_logs_conn.execute(insert_sql)
+            log = selector.fetchone()
+
+        print("sorting historical data, this action takes a long time...")
+        history_logs_conn.execute("VACUUM;")
+
+        gcfg = getGlobalConf()
+        save_day = gcfg['global']["save_day"]
+        print("delete historical data {} days ago...".format(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 = "delete from site_logs where time <= {}".format(
+            save_timestamp)
+        history_logs_conn.execute(delete_sql)
+        history_logs_conn.commit()
+
+        # 3. delete merged data and clean up statistics
+        print("delete merged thermal data...")
+        mw.writeFile(migrating_flag, "yes")
+
+        hot_db_conn = pSqliteDb('web_logs', site_name)
+        del_hot_log = "delete from web_logs where time<{}".format(todayUt)
+        print(del_hot_log)
+        r = hot_db_conn.execute(del_hot_log)
+        print("delete:", r)
+        print("deleting statistics over 180 days...")
+        save_time_key = time.strftime(
+            '%Y%m%d00', time.localtime(time.time() - 180 * 86400))
+
+        del_request_stat_sql = "delete from request_stat where time<={}".format(
+            save_time_key)
+        hot_db_conn.execute(del_request_stat_sql)
+
+        hot_db_conn.execute(
+            "delete from spider_stat where time<={}".format(save_time_key))
+        hot_db_conn.execute(
+            "delete from client_stat where time<={}".format(save_time_key))
+        hot_db_conn.execute(
+            "delete from referer_stat where time<={}".format(save_time_key))
+        hot_db_conn.commit()
+        print("clean up the hot database...")
+        hot_db_conn.execute("VACUUM;")
+        hot_db_conn.commit()
+
+        if os.path.exists(migrating_flag):
+            os.remove(migrating_flag)
+    except Exception as e:
+        if site_name:
+            print("{} logs to history error:{}".format(site_name, e))
+        else:
+            print("logs to history error:{}".format(e))
+    finally:
+        if os.path.exists(hot_db_tmp):
+            os.remove(hot_db_tmp)
+
+    print("{} logs migrate ok.".format(site_name))
+    return mw.returnMsg(True, "{} logs migrate ok".format(site_name))
 
 
 def migrateHotLogs(query_date="today"):
     print("begin migrate hot logs")
     sites = mw.M('sites').field('name').order("addtime").select()
-    # print(sites)
 
     unset_site = {"name": "unset"}
     sites.append(unset_site)
 
+    # migrateSiteHotLogs('t1.cn', query_date)
+
     for site_info in sites:
         # print(site_info['name'])
         site_name = site_info["name"]