git-sync/crawlab
1
0
Fork 0
mirror of https://github.com/crawlab-team/crawlab.git synced 2026-01-23 17:31:11 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into master

Browse Source
This commit is contained in:
Jeremy
2019-09-01 00:13:19 +08:00
committed by GitHub
parent d5dfe648af f755b57a12
commit 0d6bd8fb0a
31 changed files with 469 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
backend/constants/context.go Normal file
View File

@@ -0,0 +1,5 @@
package constants
const (
ContextUser = "currentUser"
)

8
backend/constants/errors.go Normal file
View File

@@ -0,0 +1,8 @@
package constants
import "crawlab/errors"
var (
//users
ErrorUserNotFound = errors.NewBusinessError(10001, "user not found.")
)

43
backend/errors/errors.go Normal file
View File

@@ -0,0 +1,43 @@
package errors
import "fmt"
type Scope int
const (
ScopeSystem Scope = 1
ScopeBusiness Scope = 2
)
type OPError struct {
Message string
Code int
Scope Scope
}
func (O OPError) Error() string {
var scope string
switch O.Scope {
case ScopeSystem:
scope = "system"
break
case ScopeBusiness:
scope = "business"
}
return fmt.Sprintf("%s : %d -> %s.", scope, O.Code, O.Message)
}
func NewSystemOPError(code int, message string) *OPError {
return &OPError{
Message: message,
Code: code,
Scope: ScopeSystem,
}
}
func NewBusinessError(code int, message string) *OPError {
return &OPError{
Message: message,
Code: code,
Scope: ScopeBusiness,
}
}

99
backend/main.go
View File

@@ -99,53 +99,60 @@ func main() {
if services.IsMaster() {
// 中间件
app.Use(middlewares.CORSMiddleware())
app.Use(middlewares.AuthorizationMiddleware())
//app.Use(middlewares.AuthorizationMiddleware())
anonymousGroup := app.Group("/")
{
anonymousGroup.POST("/login", routes.Login) // 用户登录
anonymousGroup.PUT("/users", routes.PutUser) // 添加用户
}
authGroup := app.Group("/", middlewares.AuthorizationMiddleware())
{
// 路由
// 节点
authGroup.GET("/nodes", routes.GetNodeList) // 节点列表
authGroup.GET("/nodes/:id", routes.GetNode) // 节点详情
authGroup.POST("/nodes/:id", routes.PostNode) // 修改节点
authGroup.GET("/nodes/:id/tasks", routes.GetNodeTaskList) // 节点任务列表
authGroup.GET("/nodes/:id/system", routes.GetSystemInfo) // 节点任务列表
authGroup.DELETE("/nodes/:id", routes.DeleteNode) // 删除节点
// 爬虫
authGroup.GET("/spiders", routes.GetSpiderList) // 爬虫列表
authGroup.GET("/spiders/:id", routes.GetSpider) // 爬虫详情
authGroup.POST("/spiders", routes.PutSpider) // 上传爬虫
authGroup.POST("/spiders/:id", routes.PostSpider) // 修改爬虫
authGroup.POST("/spiders/:id/publish", routes.PublishSpider) // 发布爬虫
authGroup.DELETE("/spiders/:id", routes.DeleteSpider) // 删除爬虫
authGroup.GET("/spiders/:id/tasks", routes.GetSpiderTasks) // 爬虫任务列表
authGroup.GET("/spiders/:id/file", routes.GetSpiderFile) // 爬虫文件读取
authGroup.POST("/spiders/:id/file", routes.PostSpiderFile) // 爬虫目录写入
authGroup.GET("/spiders/:id/dir", routes.GetSpiderDir) // 爬虫目录
authGroup.GET("/spiders/:id/stats", routes.GetSpiderStats) // 爬虫统计数据
// 任务
authGroup.GET("/tasks", routes.GetTaskList) // 任务列表
authGroup.GET("/tasks/:id", routes.GetTask) // 任务详情
authGroup.PUT("/tasks", routes.PutTask) // 派发任务
authGroup.DELETE("/tasks/:id", routes.DeleteTask) // 删除任务
authGroup.POST("/tasks/:id/cancel", routes.CancelTask) // 取消任务
authGroup.GET("/tasks/:id/log", routes.GetTaskLog) // 任务日志
authGroup.GET("/tasks/:id/results", routes.GetTaskResults) // 任务结果
authGroup.GET("/tasks/:id/results/download", routes.DownloadTaskResultsCsv) // 下载任务结果
// 定时任务
authGroup.GET("/schedules", routes.GetScheduleList) // 定时任务列表
authGroup.GET("/schedules/:id", routes.GetSchedule) // 定时任务详情
authGroup.PUT("/schedules", routes.PutSchedule) // 创建定时任务
authGroup.POST("/schedules/:id", routes.PostSchedule) // 修改定时任务
authGroup.DELETE("/schedules/:id", routes.DeleteSchedule) // 删除定时任务
// 统计数据
authGroup.GET("/stats/home", routes.GetHomeStats) // 首页统计数据
// 用户
authGroup.GET("/users", routes.GetUserList) // 用户列表
authGroup.GET("/users/:id", routes.GetUser) // 用户详情
authGroup.POST("/users/:id", routes.PostUser) // 更改用户
authGroup.DELETE("/users/:id", routes.DeleteUser) // 删除用户
authGroup.GET("/me", routes.GetMe) // 获取自己账户
}
// 路由
// 节点
app.GET("/nodes", routes.GetNodeList) // 节点列表
app.GET("/nodes/:id", routes.GetNode) // 节点详情
app.POST("/nodes/:id", routes.PostNode) // 修改节点
app.GET("/nodes/:id/tasks", routes.GetNodeTaskList) // 节点任务列表
app.GET("/nodes/:id/system", routes.GetSystemInfo) // 节点任务列表
app.DELETE("/nodes/:id", routes.DeleteNode) // 删除节点
// 爬虫
app.GET("/spiders", routes.GetSpiderList) // 爬虫列表
app.GET("/spiders/:id", routes.GetSpider) // 爬虫详情
app.POST("/spiders", routes.PutSpider) // 上传爬虫
app.POST("/spiders/:id", routes.PostSpider) // 修改爬虫
app.POST("/spiders/:id/publish", routes.PublishSpider) // 发布爬虫
app.DELETE("/spiders/:id", routes.DeleteSpider) // 删除爬虫
app.GET("/spiders/:id/tasks", routes.GetSpiderTasks) // 爬虫任务列表
app.GET("/spiders/:id/file", routes.GetSpiderFile) // 爬虫文件读取
app.POST("/spiders/:id/file", routes.PostSpiderFile) // 爬虫目录写入
app.GET("/spiders/:id/dir", routes.GetSpiderDir) // 爬虫目录
app.GET("/spiders/:id/stats", routes.GetSpiderStats) // 爬虫统计数据
// 任务
app.GET("/tasks", routes.GetTaskList) // 任务列表
app.GET("/tasks/:id", routes.GetTask) // 任务详情
app.PUT("/tasks", routes.PutTask) // 派发任务
app.DELETE("/tasks/:id", routes.DeleteTask) // 删除任务
app.POST("/tasks/:id/cancel", routes.CancelTask) // 取消任务
app.GET("/tasks/:id/log", routes.GetTaskLog) // 任务日志
app.GET("/tasks/:id/results", routes.GetTaskResults) // 任务结果
app.GET("/tasks/:id/results/download", routes.DownloadTaskResultsCsv) // 下载任务结果
// 定时任务
app.GET("/schedules", routes.GetScheduleList) // 定时任务列表
app.GET("/schedules/:id", routes.GetSchedule) // 定时任务详情
app.PUT("/schedules", routes.PutSchedule) // 创建定时任务
app.POST("/schedules/:id", routes.PostSchedule) // 修改定时任务
app.DELETE("/schedules/:id", routes.DeleteSchedule) // 删除定时任务
// 统计数据
app.GET("/stats/home", routes.GetHomeStats) // 首页统计数据
// 用户
app.GET("/users", routes.GetUserList) // 用户列表
app.GET("/users/:id", routes.GetUser) // 用户详情
app.PUT("/users", routes.PutUser) // 添加用户
app.POST("/users/:id", routes.PostUser) // 更改用户
app.DELETE("/users/:id", routes.DeleteUser) // 删除用户
app.POST("/login", routes.Login) // 用户登录
app.GET("/me", routes.GetMe) // 获取自己账户
}
// 路由ping

13
backend/middlewares/auth.go
View File

@@ -12,12 +12,12 @@ import (
func AuthorizationMiddleware() gin.HandlerFunc {
return func(c *gin.Context) {
// 如果为登录或注册，不用校验
if c.Request.URL.Path == "/login" ||
(c.Request.URL.Path == "/users" && c.Request.Method == "PUT") ||
strings.HasSuffix(c.Request.URL.Path, "download") {
c.Next()
return
}
//if c.Request.URL.Path == "/login" ||
// (c.Request.URL.Path == "/users" && c.Request.Method == "PUT") ||
// strings.HasSuffix(c.Request.URL.Path, "download") {
// c.Next()
// return
//}
// 获取token string
tokenStr := c.GetHeader("Authorization")
@@ -46,6 +46,7 @@ func AuthorizationMiddleware() gin.HandlerFunc {
return
}
}
c.Set(constants.ContextUser, &user)
// 校验成功
c.Next()

2
backend/mock/schedule.go
View File

@@ -113,7 +113,7 @@ func PutSchedule(c *gin.Context) {
func DeleteSchedule(c *gin.Context) {
id := bson.ObjectIdHex("5d429e6c19f7abede924fee2")
for _, sch := range scheduleList {
if sch.Id == bson.ObjectId(id) {
if sch.Id == id {
fmt.Println("delete a schedule")
}
}

47
backend/model/node.go
View File

@@ -1,7 +1,9 @@
package model
import (
"crawlab/constants"
"crawlab/database"
"crawlab/services/register"
"github.com/apex/log"
"github.com/globalsign/mgo"
"github.com/globalsign/mgo/bson"
@@ -79,6 +81,7 @@ func GetNodeList(filter interface{}) ([]Node, error) {
var results []Node
if err := c.Find(filter).All(&results); err != nil {
log.Error("get node list error: " + err.Error())
debug.PrintStack()
return results, err
}
@@ -153,3 +156,47 @@ func GetNodeCount(query interface{}) (int, error) {
return count, nil
}
// 节点基本信息
func GetNodeBaseInfo() (ip string, mac string, key string, error error) {
ip, err := register.GetRegister().GetIp()
if err != nil {
debug.PrintStack()
return "", "", "", err
}
mac, err = register.GetRegister().GetMac()
if err != nil {
debug.PrintStack()
return "", "", "", err
}
key, err = register.GetRegister().GetKey()
if err != nil {
debug.PrintStack()
return "", "", "", err
}
return ip, mac, key, nil
}
// 根据redis的key值，重置node节点为offline
func ResetNodeStatusToOffline(list []string) {
nodes, _ := GetNodeList(nil)
for _, node := range nodes {
hasNode := false
for _, key := range list {
if key == node.Key {
hasNode = true
break
}
}
if !hasNode || node.Status == "" {
node.Status = constants.StatusOffline
if err := node.Save(); err != nil {
log.Errorf(err.Error())
return
}
continue
}
}
}

50
backend/model/node_test.go Normal file
View File

@@ -0,0 +1,50 @@
package model
import (
"crawlab/config"
"crawlab/constants"
"crawlab/database"
"github.com/apex/log"
. "github.com/smartystreets/goconvey/convey"
"runtime/debug"
"testing"
)
func TestAddNode(t *testing.T) {
Convey("Test AddNode", t, func() {
if err := config.InitConfig("../conf/config.yml"); err != nil {
log.Error("init config error:" + err.Error())
panic(err)
}
log.Info("初始化配置成功")
// 初始化Mongodb数据库
if err := database.InitMongo(); err != nil {
log.Error("init mongodb error:" + err.Error())
debug.PrintStack()
panic(err)
}
log.Info("初始化Mongodb数据库成功")
// 初始化Redis数据库
if err := database.InitRedis(); err != nil {
log.Error("init redis error:" + err.Error())
debug.PrintStack()
panic(err)
}
var node = Node{
Key: "c4:b3:01:bd:b5:e7",
Name: "10.27.238.101",
Ip: "10.27.238.101",
Port: "8000",
Mac: "c4:b3:01:bd:b5:e7",
Status: constants.StatusOnline,
IsMaster: true,
}
if err := node.Add(); err != nil {
log.Error("add node error:" + err.Error())
panic(err)
}
})
}

19
backend/model/spider.go
View File

@@ -23,13 +23,14 @@ type Spider struct {
Col string `json:"col"` // 结果储存位置
Site string `json:"site"` // 爬虫网站
Envs []Env `json:"envs" bson:"envs"` // 环境变量
Remark string `json:"remark"` // 备注
// 自定义爬虫
Src string `json:"src" bson:"src"` // 源码位置
Cmd string `json:"cmd" bson:"cmd"` // 执行命令
// 前端展示
LastRunTs time.Time `json:"last_run_ts"` // 最后一次执行时间
LastRunTs time.Time `json:"last_run_ts"` // 最后一次执行时间
LastStatus string `json:"last_status"` // 最后执行状态
// TODO: 可配置爬虫
//Fields []interface{} `json:"fields"`
@@ -92,15 +93,13 @@ func (spider *Spider) GetLastTask() (Task, error) {
return tasks[0], nil
}
func GetSpiderList(filter interface{}, skip int, limit int) ([]Spider, error) {
s, c := database.GetCol("spiders")
defer s.Close()
// 获取爬虫列表
spiders := []Spider{}
if err := c.Find(filter).Skip(skip).Limit(limit).All(&spiders); err != nil {
if err := c.Find(filter).Skip(skip).Limit(limit).Sort("name asc").All(&spiders); err != nil {
debug.PrintStack()
return spiders, err
}
@@ -117,6 +116,7 @@ func GetSpiderList(filter interface{}, skip int, limit int) ([]Spider, error) {
// 赋值
spiders[i].LastRunTs = task.CreateTs
spiders[i].LastStatus = task.Status
}
return spiders, nil
@@ -165,6 +165,15 @@ func RemoveSpider(id bson.ObjectId) error {
return err
}
// gf上的文件
s, gf := database.GetGridFs("files")
defer s.Close()
if err := gf.RemoveId(result.FileId); err != nil {
log.Error("remove file error, id:" + result.FileId.Hex())
return err
}
return nil
}

27
backend/routes/user.go
View File

@@ -4,6 +4,7 @@ import (
"crawlab/constants"
"crawlab/model"
"crawlab/services"
"crawlab/services/context"
"crawlab/utils"
"github.com/gin-gonic/gin"
"github.com/globalsign/mgo/bson"
@@ -171,7 +172,7 @@ func Login(c *gin.Context) {
}
// 获取token
tokenStr, err := services.GetToken(user.Username)
tokenStr, err := services.MakeToken(&user)
if err != nil {
HandleError(http.StatusUnauthorized, c, errors.New("not authorized"))
return
@@ -185,20 +186,16 @@ func Login(c *gin.Context) {
}
func GetMe(c *gin.Context) {
// 获取token string
tokenStr := c.GetHeader("Authorization")
// 校验token
user, err := services.CheckToken(tokenStr)
if err != nil {
HandleError(http.StatusUnauthorized, c, errors.New("not authorized"))
ctx := context.WithGinContext(c)
user := ctx.User()
if user == nil {
ctx.FailedWithError(constants.ErrorUserNotFound, http.StatusUnauthorized)
return
}
user.Password = ""
c.JSON(http.StatusOK, Response{
Status: "ok",
Message: "success",
Data: user,
})
ctx.Success(struct {
*model.User
Password string `json:"password,omitempty"`
}{
User: user,
}, nil)
}

6
backend/routes/utils.go
View File

@@ -1,13 +1,15 @@
package routes
import (
"github.com/apex/log"
"github.com/gin-gonic/gin"
"runtime/debug"
)
func HandleError(statusCode int, c *gin.Context, err error) {
log.Errorf("handle error:" + err.Error())
debug.PrintStack()
c.JSON(statusCode, Response{
c.AbortWithStatusJSON(statusCode, Response{
Status: "ok",
Message: "error",
Error: err.Error(),
@@ -16,7 +18,7 @@ func HandleError(statusCode int, c *gin.Context, err error) {
func HandleErrorF(statusCode int, c *gin.Context, err string) {
debug.PrintStack()
c.JSON(statusCode, Response{
c.AbortWithStatusJSON(statusCode, Response{
Status: "ok",
Message: "error",
Error: err,

73
backend/services/context/context.go Normal file
View File

@@ -0,0 +1,73 @@
package context
import (
"crawlab/constants"
"crawlab/errors"
"crawlab/model"
"fmt"
"github.com/apex/log"
"github.com/gin-gonic/gin"
errors2 "github.com/pkg/errors"
"net/http"
"runtime/debug"
)
type Context struct {
*gin.Context
}
func (c *Context) User() *model.User {
userIfe, exists := c.Get(constants.ContextUser)
if !exists {
return nil
}
user, ok := userIfe.(*model.User)
if !ok {
return nil
}
return user
}
func (c *Context) Success(data interface{}, meta interface{}) {
if meta == nil {
meta = gin.H{}
}
if data == nil {
data = gin.H{}
}
c.JSON(http.StatusOK, gin.H{
"status": "ok",
"message": "success",
"data": data,
"error": "",
})
}
func (c *Context) FailedWithError(err error, httpCode ...int) {
var code = 200
if len(httpCode) > 0 {
code = httpCode[0]
}
log.Errorf("handle error:" + err.Error())
debug.PrintStack()
switch errors2.Cause(err).(type) {
case errors.OPError:
c.AbortWithStatusJSON(code, gin.H{
"status": "ok",
"message": "error",
"error": err.Error(),
})
break
default:
fmt.Println("deprecated....")
c.AbortWithStatusJSON(code, gin.H{
"status": "ok",
"message": "error",
"error": err.Error(),
})
}
}
func WithGinContext(context *gin.Context) *Context {
return &Context{Context: context}
}

2
backend/services/log.go
View File

@@ -34,8 +34,10 @@ func GetLocalLog(logPath string) (fileBytes []byte, err error) {
return nil, err
}
defer f.Close()
const bufLen = 2048
logBuf := make([]byte, bufLen)
off := int64(0)
if fi.Size() > int64(len(logBuf)) {
off = fi.Size() - int64(len(logBuf))

126
backend/services/node.go
View File

@@ -73,23 +73,11 @@ func GetCurrentNode() (model.Node, error) {
if err != nil {
// 如果为主节点，表示为第一次注册，插入节点信息
if IsMaster() {
// 获取本机IP地址
ip, err := register.GetRegister().GetIp()
// 获取本机信息
ip, mac, key, err := model.GetNodeBaseInfo()
if err != nil {
debug.PrintStack()
return model.Node{}, err
}
mac, err := register.GetRegister().GetMac()
if err != nil {
debug.PrintStack()
return model.Node{}, err
}
key, err := register.GetRegister().GetKey()
if err != nil {
debug.PrintStack()
return model.Node{}, err
return node, err
}
// 生成节点
@@ -97,7 +85,7 @@ func GetCurrentNode() (model.Node, error) {
Key: key,
Id: bson.NewObjectId(),
Ip: ip,
Name: key,
Name: ip,
Mac: mac,
IsMaster: true,
}
@@ -124,6 +112,7 @@ func IsMaster() bool {
return viper.GetString("server.master") == Yes
}
// 所有调用IsMasterNode的方法，都永远会在master节点执行，所以GetCurrentNode方法返回永远是master节点
// 该ID的节点是否为主节点
func IsMasterNode(id string) bool {
curNode, _ := GetCurrentNode()
@@ -176,72 +165,54 @@ func UpdateNodeStatus() {
// 在Redis中删除该节点
if err := database.RedisClient.HDel("nodes", data.Key); err != nil {
log.Errorf(err.Error())
return
}
// 在MongoDB中该节点设置状态为离线
s, c := database.GetCol("nodes")
defer s.Close()
var node model.Node
if err := c.Find(bson.M{"key": key}).One(&node); err != nil {
log.Errorf(err.Error())
debug.PrintStack()
return
}
node.Status = constants.StatusOffline
if err := node.Save(); err != nil {
log.Errorf(err.Error())
return
}
continue
}
// 更新节点信息到数据库
s, c := database.GetCol("nodes")
defer s.Close()
var node model.Node
if err := c.Find(bson.M{"key": key}).One(&node); err != nil {
// 数据库不存在该节点
node = model.Node{
Key: key,
Name: key,
Ip: data.Ip,
Port: "8000",
Mac: data.Mac,
Status: constants.StatusOnline,
IsMaster: data.Master,
}
if err := node.Add(); err != nil {
log.Errorf(err.Error())
return
}
} else {
// 数据库存在该节点
node.Status = constants.StatusOnline
if err := node.Save(); err != nil {
log.Errorf(err.Error())
return
}
// 处理node信息
handleNodeInfo(key, data)
}
// 重置不在redis的key为offline
model.ResetNodeStatusToOffline(list)
}
func handleNodeInfo(key string, data Data) {
// 更新节点信息到数据库
s, c := database.GetCol("nodes")
defer s.Close()
// 同个key可能因为并发，被注册多次
var nodes []model.Node
_ = c.Find(bson.M{"key": key}).All(&nodes)
if nodes != nil && len(nodes) > 1 {
for _, node := range nodes {
_ = c.RemoveId(node.Id)
}
}
// 遍历数据库中的节点列表
nodes, err := model.GetNodeList(nil)
for _, node := range nodes {
hasNode := false
for _, key := range list {
if key == node.Key {
hasNode = true
break
}
var node model.Node
if err := c.Find(bson.M{"key": key}).One(&node); err != nil {
// 数据库不存在该节点
node = model.Node{
Key: key,
Name: data.Ip,
Ip: data.Ip,
Port: "8000",
Mac: data.Mac,
Status: constants.StatusOnline,
IsMaster: data.Master,
}
if !hasNode {
node.Status = constants.StatusOffline
if err := node.Save(); err != nil {
log.Errorf(err.Error())
return
}
continue
if err := node.Add(); err != nil {
log.Errorf(err.Error())
return
}
} else {
// 数据库存在该节点
node.Status = constants.StatusOnline
if err := node.Save(); err != nil {
log.Errorf(err.Error())
return
}
}
}
@@ -333,12 +304,15 @@ func WorkerNodeCallback(channel string, msgStr string) {
// 获取本地日志
logStr, err := GetLocalLog(msg.LogPath)
log.Info(string(logStr))
if err != nil {
log.Errorf(err.Error())
debug.PrintStack()
msgSd.Error = err.Error()
msgSd.Log = err.Error()
} else {
msgSd.Log = string(logStr)
}
msgSd.Log = string(logStr)
// 序列化
msgSdBytes, err := json.Marshal(&msgSd)
@@ -349,7 +323,7 @@ func WorkerNodeCallback(channel string, msgStr string) {
}
// 发布消息给主节点
fmt.Println(msgSd)
log.Info("publish get log msg to master")
if err := database.Publish("nodes:master", string(msgSdBytes)); err != nil {
log.Errorf(err.Error())
return

15
backend/services/spider.go
View File

@@ -20,6 +20,7 @@ import (
"path/filepath"
"runtime/debug"
"strings"
"syscall"
)
type SpiderFileData struct {
@@ -30,6 +31,7 @@ type SpiderFileData struct {
type SpiderUploadMessage struct {
FileId string
FileName string
SpiderId string
}
// 从项目目录中获取爬虫列表
@@ -39,9 +41,9 @@ func GetSpidersFromDir() ([]model.Spider, error) {
// 如果爬虫项目目录不存在，则创建一个
if !utils.Exists(srcPath) {
mask := syscall.Umask(0) // 改为 0000 八进制
mask := syscall.Umask(0) // 改为 0000 八进制
defer syscall.Umask(mask) // 改为原来的 umask
if err := os.MkdirAll(srcPath, 0666); err != nil {
if err := os.MkdirAll(srcPath, 0766); err != nil {
debug.PrintStack()
return []model.Spider{}, err
}
@@ -133,6 +135,8 @@ func ZipSpider(spider model.Spider) (filePath string, err error) {
// 如果源文件夹不存在，抛错
if !utils.Exists(spider.Src) {
debug.PrintStack()
// 删除该爬虫，否则会一直报错
_ = model.RemoveSpider(spider.Id)
return "", errors.New("source path does not exist")
}
@@ -173,6 +177,7 @@ func UploadToGridFs(spider model.Spider, fileName string, filePath string) (fid
// 如果存在FileId删除GridFS上的老文件
if !utils.IsObjectIdNull(spider.FileId) {
if err = gf.RemoveId(spider.FileId); err != nil {
log.Error("remove gf file:" + err.Error())
debug.PrintStack()
}
}
@@ -225,7 +230,7 @@ func ReadFileByStep(filePath string, handle func([]byte, *mgo.GridFile), fileCre
for {
switch nr, err := f.Read(s[:]); true {
case nr < 0:
fmt.Fprintf(os.Stderr, "cat: error reading: %s\n", err.Error())
_, _ = fmt.Fprintf(os.Stderr, "cat: error reading: %s\n", err.Error())
debug.PrintStack()
case nr == 0: // EOF
return nil
@@ -233,7 +238,6 @@ func ReadFileByStep(filePath string, handle func([]byte, *mgo.GridFile), fileCre
handle(s[0:nr], fileCreate)
}
}
return nil
}
// 发布所有爬虫
@@ -291,6 +295,7 @@ func PublishSpider(spider model.Spider) (err error) {
msg := SpiderUploadMessage{
FileId: fid.Hex(),
FileName: fileName,
SpiderId: spider.Id.Hex(),
}
msgStr, err := json.Marshal(msg)
if err != nil {
@@ -322,7 +327,7 @@ func OnFileUpload(channel string, msgStr string) {
// 从GridFS获取该文件
f, err := gf.OpenId(bson.ObjectIdHex(msg.FileId))
if err != nil {
log.Errorf(err.Error())
log.Errorf("open file id: " + msg.FileId + ", spider id:" + msg.SpiderId + ", error: " + err.Error())
debug.PrintStack()
return
}

8
backend/services/task.go
View File

@@ -408,9 +408,12 @@ func GetTaskLog(id string) (logStr string, err error) {
logStr = string(logBytes)
if err != nil {
log.Errorf(err.Error())
return "", err
logStr = string(err.Error())
// return "", err
} else {
logStr = string(logBytes)
}
logStr = string(logBytes)
} else {
// 若不为主节点，获取远端日志
logStr, err = GetRemoteLog(task)
@@ -472,6 +475,7 @@ func CancelTask(id string) (err error) {
}
func HandleTaskError(t model.Task, err error) {
log.Error("handle task error:" + err.Error())
t.Status = constants.StatusError
t.Error = err.Error()
t.FinishTs = time.Now()

40
backend/services/user.go
View File

@@ -5,11 +5,9 @@ import (
"crawlab/model"
"crawlab/utils"
"errors"
"github.com/apex/log"
"github.com/dgrijalva/jwt-go"
"github.com/globalsign/mgo/bson"
"github.com/spf13/viper"
"runtime/debug"
"time"
)
@@ -24,28 +22,38 @@ func InitUserService() error {
}
return nil
}
func GetToken(username string) (tokenStr string, err error) {
user, err := model.GetUserByUsername(username)
if err != nil {
log.Errorf(err.Error())
debug.PrintStack()
return
}
func MakeToken(user *model.User) (tokenStr string, err error) {
token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
"id": user.Id,
"username": user.Username,
"nbf": time.Now().Unix(),
})
tokenStr, err = token.SignedString([]byte(viper.GetString("server.secret")))
if err != nil {
return
}
return
return token.SignedString([]byte(viper.GetString("server.secret")))
}
//func GetToken(username string) (tokenStr string, err error) {
// user, err := model.GetUserByUsername(username)
// if err != nil {
// log.Errorf(err.Error())
// debug.PrintStack()
// return
// }
//
// token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
// "id": user.Id,
// "username": user.Username,
// "nbf": time.Now().Unix(),
// })
//
// tokenStr, err = token.SignedString([]byte(viper.GetString("server.secret")))
// if err != nil {
// return
// }
// return
//}
func SecretFunc() jwt.Keyfunc {
return func(token *jwt.Token) (interface{}, error) {
return []byte(viper.GetString("server.secret")), nil

2
docker/Dockerfile.master.alpine
View File

@@ -75,7 +75,7 @@ RUN sed -i 's/#rc_sys=""/rc_sys="lxc"/g' /etc/rc.conf && \
# working directory
WORKDIR /app/backend
ENV PYTHONIOENCODING utf-8
# frontend port
EXPOSE 8080

2
docker/Dockerfile.worker.alpine
View File

@@ -35,7 +35,7 @@ RUN apk del .build-deps
# working directory
WORKDIR /app/backend
ENV PYTHONIOENCODING utf-8
# backend port
EXPOSE 8000

1
frontend/package.json
View File

@@ -6,6 +6,7 @@
"serve": "vue-cli-service serve --ip=0.0.0.0",
"serve:prod": "vue-cli-service serve --mode=production --ip=0.0.0.0",
"config": "vue ui",
"build:dev": "vue-cli-service build --mode development",
"build:prod": "vue-cli-service build --mode production",
"lint": "vue-cli-service lint",
"test:unit": "vue-cli-service test:unit"

5
frontend/src/api/request.js
View File

@@ -3,7 +3,7 @@ import router from '../router'
let baseUrl = process.env.VUE_APP_BASE_URL ? process.env.VUE_APP_BASE_URL : 'http://localhost:8000'
const request = (method, path, params, data) => {
const request = (method, path, params, data, others = {}) => {
return new Promise((resolve, reject) => {
const url = baseUrl + path
const headers = {
@@ -14,7 +14,8 @@ const request = (method, path, params, data) => {
url,
params,
data,
headers
headers,
...others
})
.then(resolve)
.catch(error => {

2
frontend/src/components/InfoView/NodeInfoView.vue
View File

@@ -22,7 +22,7 @@
</el-form>
</el-row>
<el-row class="button-container" v-if="!isView">
<el-button type="success" @click="onSave">{{$t('Save')}}</el-button>
<el-button size="small" type="success" @click="onSave">{{$t('Save')}}</el-button>
</el-row>
</div>
</template>

3
frontend/src/components/InfoView/SpiderInfoView.vue
View File

@@ -44,6 +44,9 @@
<el-option value="customized" :label="$t('Customized')"></el-option>
</el-select>
</el-form-item>
<el-form-item :label="$t('Remark')">
<el-input v-model="spiderForm.remark"/>
</el-form-item>
</el-form>
</el-row>
<el-row class="button-container" v-if="!isView">

6
frontend/src/components/InfoView/TaskInfoView.vue
View File

@@ -86,15 +86,15 @@ export default {
return dayjs(str).format('YYYY-MM-DD HH:mm:ss')
},
getWaitDuration (row) {
if (row.start_ts.match('^0001')) return 'NA'
if (!row.start_ts || row.start_ts.match('^0001')) return 'NA'
return dayjs(row.start_ts).diff(row.create_ts, 'second')
},
getRuntimeDuration (row) {
if (row.finish_ts.match('^0001')) return 'NA'
if (!row.finish_ts || row.finish_ts.match('^0001')) return 'NA'
return dayjs(row.finish_ts).diff(row.start_ts, 'second')
},
getTotalDuration (row) {
if (row.finish_ts.match('^0001')) return 'NA'
if (!row.finish_ts || row.finish_ts.match('^0001')) return 'NA'
return dayjs(row.finish_ts).diff(row.create_ts, 'second')
}
}

2
frontend/src/i18n/zh.js
View File

@@ -154,6 +154,8 @@ export default {
'Last Run': '上次运行',
'Action': '操作',
'No command line': '没有执行命令',
'Last Status': '上次运行状态',
'Remark': '备注',
// 任务
'Task Info': '任务信息',

18
frontend/src/store/modules/node.js
View File

@@ -25,15 +25,7 @@ const mutations = {
const { id, systemInfo } = payload
for (let i = 0; i < state.nodeList.length; i++) {
if (state.nodeList[i]._id === id) {
// Vue.set(state.nodeList[i], 'systemInfo', {})
state.nodeList[i].systemInfo = systemInfo
// for (const key in systemInfo) {
// if (systemInfo.hasOwnProperty(key)) {
// console.log(key)
// state.nodeList[i].systemInfo[key] = systemInfo[key]
// // Vue.set(state.nodeList[i].systemInfo, key, systemInfo[key])
// }
// }
break
}
}
@@ -76,10 +68,12 @@ const actions = {
getTaskList ({ state, commit }, id) {
return request.get(`/nodes/${id}/tasks`)
.then(response => {
commit('task/SET_TASK_LIST',
response.data.data.map(d => d)
.sort((a, b) => a.create_ts < b.create_ts ? 1 : -1),
{ root: true })
if (response.data.data) {
commit('task/SET_TASK_LIST',
response.data.data.map(d => d)
.sort((a, b) => a.create_ts < b.create_ts ? 1 : -1),
{ root: true })
}
})
},
getNodeSystemInfo ({ state, commit }, id) {

16
frontend/src/store/modules/task.js
View File

@@ -120,6 +120,22 @@ const actions = {
commit('SET_TASK_RESULTS_TOTAL_COUNT', response.data.total)
})
},
async getTaskResultExcel ({ state, commit }, id) {
const { data } = await request.request('GET', '/tasks/' + id + '/results/download', {}, {
responseType: 'blob' // important
})
const downloadUrl = window.URL.createObjectURL(new Blob([data]))
const link = document.createElement('a')
link.href = downloadUrl
link.setAttribute('download', 'data.csv') // any other extension
document.body.appendChild(link)
link.click()
link.remove()
},
cancelTask ({ state, dispatch }, id) {
return request.post(`/tasks/${id}/cancel`)
.then(() => {

1
frontend/src/views/layout/components/Navbar.vue
View File

@@ -32,6 +32,7 @@
<font-awesome-icon :icon="['far', 'question-circle']"/>
<span style="margin-left: 5px;">{{$t('Documentation')}}</span>
</a>
<el-dropdown-menu slot="dropdown"></el-dropdown-menu>
</el-dropdown>
</div>
</template>

19
frontend/src/views/spider/SpiderList.vue
View File

@@ -190,6 +190,14 @@
{{getTime(scope.row[col.name])}}
</template>
</el-table-column>
<el-table-column v-else-if="col.name === 'last_status'"
:key="col.name"
:label="$t(col.label)"
align="left" :width="col.width">
<template slot-scope="scope">
<status-tag :status="scope.row.last_status"/>
</template>
</el-table-column>
<el-table-column v-else
:key="col.name"
:property="col.name"
@@ -239,10 +247,14 @@ import {
} from 'vuex'
import dayjs from 'dayjs'
import CrawlConfirmDialog from '../../components/Common/CrawlConfirmDialog'
import StatusTag from '../../components/Status/StatusTag'
export default {
name: 'SpiderList',
components: { CrawlConfirmDialog },
components: {
CrawlConfirmDialog,
StatusTag
},
data () {
return {
pagination: {
@@ -267,10 +279,11 @@ export default {
// { name: 'site_name', label: 'Site', width: '140', align: 'left' },
{ name: 'type', label: 'Spider Type', width: '120' },
// { name: 'cmd', label: 'Command Line', width: '200' },
// { name: 'lang', label: 'Language', width: '120', sortable: true },
{ name: 'last_status', label: 'Last Status', width: '120' },
{ name: 'last_run_ts', label: 'Last Run', width: '160' },
{ name: 'create_ts', label: 'Create Time', width: '160' },
{ name: 'update_ts', label: 'Update Time', width: '160' }
{ name: 'update_ts', label: 'Update Time', width: '160' },
{ name: 'remark', label: 'Remark', width: '160' }
// { name: 'last_7d_tasks', label: 'Last 7-Day Tasks', width: '80' },
// { name: 'last_5_errors', label: 'Last 5-Run Errors', width: '80' }
],

8
frontend/src/views/task/TaskDetail.vue
View File

@@ -95,16 +95,16 @@ export default {
this.$store.dispatch('task/getTaskResults', this.$route.params.id)
},
downloadCSV () {
window.location.href = this.$request.baseUrl + '/tasks/' + this.$route.params.id + '/results/download'
this.$store.dispatch('task/getTaskResultExcel', this.$route.params.id)
this.$st.sendEv('任务详情-结果', '下载CSV')
}
},
created () {
this.$store.dispatch('task/getTaskData', this.$route.params.id)
async created () {
await this.$store.dispatch('task/getTaskData', this.$route.params.id)
this.$store.dispatch('task/getTaskLog', this.$route.params.id)
this.$store.dispatch('task/getTaskResults', this.$route.params.id)
if (['running'].includes(this.taskForm.status)) {
if (this.taskForm && ['running'].includes(this.taskForm.status)) {
this.handle = setInterval(() => {
this.$store.dispatch('task/getTaskLog', this.$route.params.id)
}, 5000)

4
frontend/src/views/task/TaskList.vue
View File

@@ -45,7 +45,7 @@
:label="$t(col.label)"
:sortable="col.sortable"
:align="col.align"
:width="col.width">
>
<template slot-scope="scope">
<a href="javascript:" class="a-tag" @click="onClickSpider(scope.row)">{{scope.row[col.name]}}</a>
</template>
@@ -119,7 +119,7 @@
:width="col.width">
</el-table-column>
</template>
<el-table-column :label="$t('Action')" align="left" width="120" fixed="right">
<el-table-column :label="$t('Action')" align="left" fixed="right">
<template slot-scope="scope">
<el-tooltip :content="$t('View')" placement="top">
<el-button type="primary" icon="el-icon-search" size="mini" @click="onView(scope.row)"></el-button>