123

service/agent.lua

@@ -50,6 +50,7 @@ function  M.status_task(id, msg)
         RECV_TASK.updatePlatformConfig()
         RECV_TASK.updateFqKeyList()
         RECV_TASK.updateBlackBooks()
+        RECV_TASK.getPullDataConfig()
     end)
 
 end
@@ -118,6 +119,19 @@ function RECV_TASK.startOneTask(msg_body)
     end
 end
 
+function RECV_TASK.updatePullConig(msg_body)
+    if AGENT_ID~=0 then
+        isRunTask = true --工人进入工作状态
+        websocket.write(AGENT_ID, cjson.encode({cmd="updatePullConig"}))
+    end
+end
+
+function RECV_TASK.getPullDataConfig(msg_body)
+    if AGENT_ID~=0 then
+        isRunTask = true --工人进入工作状态
+        websocket.write(AGENT_ID, cjson.encode({cmd="getPullDataConfig"}))
+    end
+end
 
 function RECV_TASK.updateAppConfig(msg_body)
     -- local body = cjson.decode(msg_body)

service/agent_manager.lua

@@ -114,9 +114,14 @@ function CMD.updateBlackBooks(_,msg_body)
     end
 end
 
---收到一个获取黑岩书的任务
-function CMD.task_get_heiyan_book() 
-    
+--更新拉取配置
+function CMD.updatePullConig(_,msg_body) 
+    local agent = getOneAgent()
+    if agent~=nil then
+        skynet.call(agent,"lua",nil,nil,"updatePullConig",msg_body)
+    else
+        skynet.error("把任务放到队列")
+    end
 end
 
 skynet.start(function()

service/backmgr/init.lua

@@ -19,6 +19,8 @@ local filter_task = require "filter_task"
 local fq_book = require "fq_book"
 local book_black_list = require "book_black_list"
 local video_titles = require "video_titles"
+local pull_data_config = require "pull_data_config"
+local pull_data_status = require "pull_data_status"
 local status_200 = 200
 local CMD = {
     
@@ -36,6 +38,8 @@ CMD["filter_task"] = filter_task;
 CMD["fq_book"] = fq_book;
 CMD["book_black_list"] = book_black_list;
 CMD["video_titles"] = video_titles;
+CMD["pull_data_config"] = pull_data_config;
+CMD["pull_data_status"] = pull_data_status;
 function run(target,fun,msg_body,fd)
     if target~=nil and fun~=nil and target[fun]~=nil then
         local isok,data,total = target[fun](msg_body)

service/backmgr/pull_data_config.lua

@@ -0,0 +1,81 @@
+--拉取数据配置
+local M = {}
+local mysqldbx = require "mysqldbx"
+local tools = require "tools"
+local skynet = require "skynet"
+local cjson = require "cjson"
+
+function M.get_all_pull_data_list(msg_body)
+    local sql = "SELECT * FROM pull_data_config "
+    local list = mysqldbx.query(sql)
+    return true,list
+end
+
+
+function M.modify_pull_data_config(msg_body)
+    local isok ,key =  tools.checkData({"id","start","name","interval_minute"},msg_body)
+    if not isok then
+        return false,string.format("缺少字段： %s.", key)
+    end
+    local sql = string.format("UPDATE `pull_data_config` SET  start = '%s' ,name ='%s' , interval_minute = %d  WHERE id = %d ",
+    msg_body.start,msg_body.name,msg_body.interval_minute,msg_body.id)
+    mysqldbx.query(sql)
+    return true, {}
+end
+
+function M.add_pull_data_list_config(msg_body)
+    local isok ,key =  tools.checkData({"config_list"},msg_body)
+    if not isok then
+        return false,string.format("缺少字段： %s.", key)
+    end
+    for i = 1, #msg_body.config_list, 1 do
+        local config_item = msg_body.config_list[i]
+        local sql = string.format("INSERT INTO `pull_data_config` (start,name,interval_minute)  VALUES ('%s','%s',%d)",config_item.start,config_item.name,config_item.interval_minute)
+        skynet.error(sql)
+        mysqldbx.query(sql)
+    end
+    return true, {}
+end
+
+function M.add_pull_data_config(msg_body)
+    local isok ,key =  tools.checkData({"start","name","interval_minute"},msg_body)
+    if not isok then
+        return false,string.format("缺少字段： %s.", key)
+    end
+    local sql = string.format("INSERT INTO `pull_data_config` (start,name,interval_minute)  VALUES ('%s','%s',%d)",msg_body.start,msg_body.name,msg_body.interval_minute)
+    mysqldbx.query(sql)
+    return true, {}
+end
+
+function M.delete_pull_data_config(msg_body)
+    local isok ,key =  tools.checkData({"id"},msg_body)
+    if not isok then
+        return false,string.format("缺少字段： %s.", key)
+    end
+    local sql = string.format("DELETE FROM pull_data_config WHERE id = %d ",msg_body.id)
+    mysqldbx.query(sql)
+    return true, {}
+end
+
+function M.pull_data_list(msg_body)
+    local isok ,key =  tools.checkData({"page_size","page_number"},msg_body)
+    if not isok then
+        return false,string.format("缺少字段： %s.", key)
+    end
+    local page_size = msg_body.page_size
+    local page_number = msg_body.page_number
+    local offset = (page_number - 1) * page_size
+    local sql = "SELECT * FROM pull_data_config "..string.format(" LIMIT %d OFFSET %d ",page_size, offset)
+    local list = mysqldbx.query(sql)
+    sql = "SELECT   COUNT(*) AS total  FROM pull_data_config "
+    local total = mysqldbx.query(sql)
+    return true,list,total[1].total
+end
+
+function M.sync_pull_data_config(msg_body)
+    local sql = "SELECT * FROM pull_data_config "
+    local list = mysqldbx.query(sql)
+    skynet.send("backmgr","lua","on_recv",nil,"ws_push_msg",cjson.encode({cmd="updatePullConig",data=list}))
+    return true,{}
+end
+return M

service/backmgr/pull_data_status.lua

@@ -0,0 +1,17 @@
+--拉取数据状态
+local M = {}
+local mysqldbx = require "mysqldbx"
+local tools = require "tools"
+local skynet = require "skynet"
+local cjson = require "cjson"
+function M.getRecord(msg_body)
+    local isok ,key =  tools.checkData({"record_date"},msg_body)
+    if not isok then
+        return false,string.format("缺少字段： %s.", key)
+    end
+    local sql = string.format("SELECT * FROM daily_records WHERE record_date = '%s' ",msg_body.record_date)
+    local list = mysqldbx.query(sql)
+    return true,list[1]
+end
+
+return M

service/doc/cmd_sql.sql

@@ -163,5 +163,13 @@ id int not null auto_increment COMMENT '唯一id',
 create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
 primary key (id));
 
+create table pull_data_config (  
+id int not null auto_increment COMMENT '唯一id',
+name  varchar(100)  COMMENT '名称',	
+start  varchar(100)  COMMENT '开始拉取数据的事件的时间',	
+interval_minute  int  COMMENT '从开始时间向前间隔多少分钟',	
+create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+primary key (id));
+