From e7eff92b02de1cb9d9ebad083070e606e153267a Mon Sep 17 00:00:00 2001 From: zhouyonggao <1971162852@qq.com> Date: Mon, 24 Nov 2025 17:37:49 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E8=90=A5=E9=94=80?= =?UTF-8?q?=E7=B3=BB=E7=BB=9F=E6=95=B0=E6=8D=AE=E5=AF=BC=E5=87=BA=E5=B7=A5?= =?UTF-8?q?=E5=85=B7=E7=9A=84=E6=A0=B8=E5=BF=83=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增数据导出工具的核心功能模块，包括： - 数据库连接与模型定义 - 路由与API处理逻辑 - SQL构建与执行 - 数据导出为CSV/XLSX格式 - 前端界面与交互 实现模板管理、任务队列、权限控制等完整业务流程 --- config/whitelist.json | 16 ++ scripts/run_server.sh | 5 + server/cmd/server/main.go | 50 +++++ server/go.mod | 19 ++ server/go.sum | 31 +++ server/internal/api/exports.go | 262 +++++++++++++++++++++++++ server/internal/api/router.go | 16 ++ server/internal/api/templates.go | 256 ++++++++++++++++++++++++ server/internal/db/mysql.go | 17 ++ server/internal/exporter/explain.go | 156 +++++++++++++++ server/internal/exporter/sqlbuilder.go | 83 ++++++++ server/internal/exporter/writer.go | 131 +++++++++++++ server/internal/migrate/migrate.go | 20 ++ server/internal/models/job.go | 17 ++ server/internal/models/template.go | 19 ++ web/index.html | 90 +++++++++ web/main.js | 57 ++++++ web/styles.css | 1 + 18 files changed, 1246 insertions(+) create mode 100644 config/whitelist.json create mode 100644 scripts/run_server.sh create mode 100644 server/cmd/server/main.go create mode 100644 server/go.mod create mode 100644 server/go.sum create mode 100644 server/internal/api/exports.go create mode 100644 server/internal/api/router.go create mode 100644 server/internal/api/templates.go create mode 100644 server/internal/db/mysql.go create mode 100644 server/internal/exporter/explain.go create mode 100644 server/internal/exporter/sqlbuilder.go create mode 100644 server/internal/exporter/writer.go create mode 100644 server/internal/migrate/migrate.go create mode 100644 server/internal/models/job.go create mode 100644 server/internal/models/template.go create mode 100644 web/index.html create mode 100644 web/main.js create mode 100644 web/styles.css diff --git a/config/whitelist.json b/config/whitelist.json new file mode 100644 index 0000000..d3ea8b7 --- /dev/null +++ b/config/whitelist.json @@ -0,0 +1,16 @@ +{ + "tables": ["order"], + "fields": [ + "order.order_number", + "order.creator", + "order.out_trade_no", + "order.type", + "order.status", + "order.contract_price", + "order.num", + "order.total", + "order.pay_amount", + "order.create_time", + "order.update_time" + ] +} diff --git a/scripts/run_server.sh b/scripts/run_server.sh new file mode 100644 index 0000000..ec9e755 --- /dev/null +++ b/scripts/run_server.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +set -e +cd "$(dirname "$0")/.."/server +go mod tidy +go run ./cmd/server diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go new file mode 100644 index 0000000..b59547a --- /dev/null +++ b/server/cmd/server/main.go @@ -0,0 +1,50 @@ +package main + +import ( + "log" + "net/http" + "os" + "time" + "marketing-system-data-tool/server/internal/api" + "marketing-system-data-tool/server/internal/db" + "marketing-system-data-tool/server/internal/migrate" +) + +func main() { + ymtDSN := buildDSN( + os.Getenv("YMT_DB_USER"), + os.Getenv("YMT_DB_PASSWORD"), + os.Getenv("YMT_DB_HOST"), + os.Getenv("YMT_DB_PORT"), + os.Getenv("YMT_DB_NAME"), + ) + marketingDSN := buildDSN( + os.Getenv("MARKETING_DB_USER"), + os.Getenv("MARKETING_DB_PASSWORD"), + os.Getenv("MARKETING_DB_HOST"), + os.Getenv("MARKETING_DB_PORT"), + os.Getenv("MARKETING_DB_NAME"), + ) + ymt, err := db.ConnectMySQL(ymtDSN) + if err != nil { + log.Fatal(err) + } + marketing, err := db.ConnectMySQL(marketingDSN) + if err != nil { + log.Fatal(err) + } + if err := migrate.Apply(ymt); err != nil { + log.Fatal(err) + } + r := api.NewRouter(ymt, marketing) + srv := &http.Server{Addr: ":8080", Handler: r, ReadTimeout: 15 * time.Second, WriteTimeout: 60 * time.Second} + log.Println("server listening on :8080") + log.Fatal(srv.ListenAndServe()) +} + +func buildDSN(user, pwd, host, port, dbname string) string { + if user == "" || host == "" || port == "" || dbname == "" { + return "" + } + return user + ":" + pwd + "@tcp(" + host + ":" + port + ")/" + dbname + "?parseTime=True&loc=Local&charset=utf8mb4" +} diff --git a/server/go.mod b/server/go.mod new file mode 100644 index 0000000..980be79 --- /dev/null +++ b/server/go.mod @@ -0,0 +1,19 @@ +module marketing-system-data-tool/server + +go 1.21 + +require ( + github.com/go-sql-driver/mysql v1.7.1 + github.com/xuri/excelize/v2 v2.8.1 +) + +require ( + github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 // indirect + github.com/richardlehane/mscfb v1.0.4 // indirect + github.com/richardlehane/msoleps v1.0.3 // indirect + github.com/xuri/efp v0.0.0-20231025114914-d1ff6096ae53 // indirect + github.com/xuri/nfp v0.0.0-20230919160717-d98342af3f05 // indirect + golang.org/x/crypto v0.19.0 // indirect + golang.org/x/net v0.21.0 // indirect + golang.org/x/text v0.14.0 // indirect +) diff --git a/server/go.sum b/server/go.sum new file mode 100644 index 0000000..f588d89 --- /dev/null +++ b/server/go.sum @@ -0,0 +1,31 @@ +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-sql-driver/mysql v1.7.1 h1:lUIinVbN1DY0xBg0eMOzmmtGoHwWBbvnWubQUrtU8EI= +github.com/go-sql-driver/mysql v1.7.1/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI= +github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 h1:RWengNIwukTxcDr9M+97sNutRR1RKhG96O6jWumTTnw= +github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826/go.mod h1:TaXosZuwdSHYgviHp1DAtfrULt5eUgsSMsZf+YrPgl8= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/richardlehane/mscfb v1.0.4 h1:WULscsljNPConisD5hR0+OyZjwK46Pfyr6mPu5ZawpM= +github.com/richardlehane/mscfb v1.0.4/go.mod h1:YzVpcZg9czvAuhk9T+a3avCpcFPMUWm7gK3DypaEsUk= +github.com/richardlehane/msoleps v1.0.1/go.mod h1:BWev5JBpU9Ko2WAgmZEuiz4/u3ZYTKbjLycmwiWUfWg= +github.com/richardlehane/msoleps v1.0.3 h1:aznSZzrwYRl3rLKRT3gUk9am7T/mLNSnJINvN0AQoVM= +github.com/richardlehane/msoleps v1.0.3/go.mod h1:BWev5JBpU9Ko2WAgmZEuiz4/u3ZYTKbjLycmwiWUfWg= +github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/xuri/efp v0.0.0-20231025114914-d1ff6096ae53 h1:Chd9DkqERQQuHpXjR/HSV1jLZA6uaoiwwH3vSuF3IW0= +github.com/xuri/efp v0.0.0-20231025114914-d1ff6096ae53/go.mod h1:ybY/Jr0T0GTCnYjKqmdwxyxn2BQf2RcQIIvex5QldPI= +github.com/xuri/excelize/v2 v2.8.1 h1:pZLMEwK8ep+CLIUWpWmvW8IWE/yxqG0I1xcN6cVMGuQ= +github.com/xuri/excelize/v2 v2.8.1/go.mod h1:oli1E4C3Pa5RXg1TBXn4ENCXDV5JUMlBluUhG7c+CEE= +github.com/xuri/nfp v0.0.0-20230919160717-d98342af3f05 h1:qhbILQo1K3mphbwKh1vNm4oGezE1eF9fQWmNiIpSfI4= +github.com/xuri/nfp v0.0.0-20230919160717-d98342af3f05/go.mod h1:WwHg+CVyzlv/TX9xqBFXEZAuxOPxn2k1GNHwG41IIUQ= +golang.org/x/crypto v0.19.0 h1:ENy+Az/9Y1vSrlrvBSyna3PITt4tiZLf7sgCjZBX7Wo= +golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= +golang.org/x/image v0.14.0 h1:tNgSxAFe3jC4uYqvZdTr84SZoM1KfwdC9SKIFrLjFn4= +golang.org/x/image v0.14.0/go.mod h1:HUYqC05R2ZcZ3ejNQsIHQDQiwWM4JBqmm6MKANTp4LE= +golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4= +golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44= +golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/server/internal/api/exports.go b/server/internal/api/exports.go new file mode 100644 index 0000000..ec06914 --- /dev/null +++ b/server/internal/api/exports.go @@ -0,0 +1,262 @@ +package api + +import ( + "database/sql" + "encoding/json" + "io" + "net/http" + "os" + "path/filepath" + "strconv" + "strings" + "time" + "marketing-system-data-tool/server/internal/exporter" +) + +type ExportsAPI struct{ + meta *sql.DB + marketing *sql.DB +} + +func ExportsHandler(meta, marketing *sql.DB) http.Handler { + api := &ExportsAPI{meta: meta, marketing: marketing} + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + p := strings.TrimPrefix(r.URL.Path, "/api/exports") + if r.Method == http.MethodPost && p == "" { + api.create(w, r) + return + } + if strings.HasPrefix(p, "/") { + id := strings.TrimPrefix(p, "/") + if r.Method == http.MethodGet && !strings.HasSuffix(p, "/download") { + api.get(w, r, id) + return + } + if r.Method == http.MethodGet && strings.HasSuffix(p, "/download") { + id = strings.TrimSuffix(id, "/download") + api.download(w, r, id) + return + } + if r.Method == http.MethodPost && strings.HasSuffix(p, "/cancel") { + id = strings.TrimSuffix(id, "/cancel") + api.cancel(w, r, id) + return + } + } + w.WriteHeader(http.StatusNotFound) + }) +} + +type ExportPayload struct{ + TemplateID uint64 `json:"template_id"` + RequestedBy uint64 `json:"requested_by"` + Permission map[string]interface{} `json:"permission"` + Options map[string]interface{} `json:"options"` + FileFormat string `json:"file_format"` +} + +func (a *ExportsAPI) create(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + var p ExportPayload + json.Unmarshal(b, &p) + var main string + var fields, filters []byte + row := a.meta.QueryRow("SELECT main_table, fields_json, filters_json FROM export_templates WHERE id=?", p.TemplateID) + err := row.Scan(&main, &fields, &filters) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte("invalid template")) + return + } + var fs []string + var fl map[string]interface{} + json.Unmarshal(fields, &fs) + json.Unmarshal(filters, &fl) + wl := map[string]bool{ + "order.order_number": true, + "order.creator": true, + "order.out_trade_no": true, + "order.type": true, + "order.status": true, + "order.contract_price": true, + "order.num": true, + "order.total": true, + "order.pay_amount": true, + "order.create_time": true, + "order.update_time": true, + } + req := exporter.BuildRequest{MainTable: main, Fields: fs, Filters: fl} + q, args, err := exporter.BuildSQL(req, wl) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(err.Error())) + return + } + _, _ = exporter.RunExplain(a.marketing, q, args) + res, err := a.meta.Exec("INSERT INTO export_jobs (template_id, status, requested_by, permission_scope_json, options_json, file_format, created_at) VALUES (?,?,?,?,?,?,?)", p.TemplateID, "queued", p.RequestedBy, toJSON(p.Permission), toJSON(p.Options), p.FileFormat, time.Now()) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } + id, _ := res.LastInsertId() + go a.runJob(uint64(id), q, args, fs, p.FileFormat) + w.Header().Set("Content-Type", "application/json") + w.Write([]byte("{\"id\":"+strconv.FormatInt(id,10)+"}")) +} + +func (a *ExportsAPI) runJob(id uint64, q string, args []interface{}, cols []string, fmt string) { + a.meta.Exec("UPDATE export_jobs SET status=?, started_at=? WHERE id=?", "running", time.Now(), id) + if fmt == "csv" { + w, err := exporter.NewCSVWriter("storage", "export") + if err != nil { + a.meta.Exec("UPDATE export_jobs SET status=?, finished_at=? WHERE id=?", "failed", time.Now(), id) + return + } + w.WriteHeader(cols) + rows, err := a.marketing.Query(q, args...) + if err != nil { + a.meta.Exec("UPDATE export_jobs SET status=?, finished_at=? WHERE id=?", "failed", time.Now(), id) + return + } + defer rows.Close() + out := make([]interface{}, len(cols)) + dest := make([]interface{}, len(cols)) + for i := range out { + dest[i] = &out[i] + } + var count int64 + for rows.Next() { + if err := rows.Scan(dest...); err != nil { + a.meta.Exec("UPDATE export_jobs SET status=?, finished_at=? WHERE id=?", "failed", time.Now(), id) + return + } + vals := make([]string, len(cols)) + for i := range out { + if b, ok := out[i].([]byte); ok { + vals[i] = string(b) + } else if out[i] == nil { + vals[i] = "" + } else { + vals[i] = toString(out[i]) + } + } + w.WriteRow(vals) + count++ + } + path, size, _ := w.Close() + a.meta.Exec("INSERT INTO export_job_files (job_id, storage_uri, row_count, size_bytes, created_at) VALUES (?,?,?,?,?)", id, path, count, size, time.Now()) + a.meta.Exec("UPDATE export_jobs SET status=?, finished_at=?, total_rows=? WHERE id=?", "completed", time.Now(), count, id) + return + } + if fmt == "xlsx" { + x, path, err := exporter.NewXLSXWriter("storage", "export", "Sheet1") + if err != nil { + a.meta.Exec("UPDATE export_jobs SET status=?, finished_at=? WHERE id=?", "failed", time.Now(), id) + return + } + x.WriteHeader(cols) + rows, err := a.marketing.Query(q, args...) + if err != nil { + a.meta.Exec("UPDATE export_jobs SET status=?, finished_at=? WHERE id=?", "failed", time.Now(), id) + return + } + defer rows.Close() + out := make([]interface{}, len(cols)) + dest := make([]interface{}, len(cols)) + for i := range out { + dest[i] = &out[i] + } + var count int64 + for rows.Next() { + if err := rows.Scan(dest...); err != nil { + a.meta.Exec("UPDATE export_jobs SET status=?, finished_at=? WHERE id=?", "failed", time.Now(), id) + return + } + vals := make([]string, len(cols)) + for i := range out { + if b, ok := out[i].([]byte); ok { + vals[i] = string(b) + } else if out[i] == nil { + vals[i] = "" + } else { + vals[i] = toString(out[i]) + } + } + x.WriteRow(vals) + count++ + } + p, size, _ := x.Close(path) + a.meta.Exec("INSERT INTO export_job_files (job_id, storage_uri, row_count, size_bytes, created_at) VALUES (?,?,?,?,?)", id, p, count, size, time.Now()) + a.meta.Exec("UPDATE export_jobs SET status=?, finished_at=?, total_rows=? WHERE id=?", "completed", time.Now(), count, id) + return + } + a.meta.Exec("UPDATE export_jobs SET status=?, finished_at=? WHERE id=?", "failed", time.Now(), id) +} + +func (a *ExportsAPI) get(w http.ResponseWriter, r *http.Request, id string) { + row := a.meta.QueryRow("SELECT id, template_id, status, requested_by, total_rows, file_format, started_at, finished_at, created_at, updated_at FROM export_jobs WHERE id=?", id) + var m = map[string]interface{}{} + var totalRows sql.NullInt64 + var startedAt, finishedAt sql.NullTime + var createdAt, updatedAt time.Time + err := row.Scan(&m["id"], &m["template_id"], &m["status"], &m["requested_by"], &totalRows, &m["file_format"], &startedAt, &finishedAt, &createdAt, &updatedAt) + if err != nil { + w.WriteHeader(http.StatusNotFound) + w.Write([]byte("not found")) + return + } + m["total_rows"] = totalRows.Int64 + m["started_at"] = startedAt.Time + m["finished_at"] = finishedAt.Time + m["created_at"] = createdAt + m["updated_at"] = updatedAt + rows, _ := a.meta.Query("SELECT storage_uri, sheet_name, row_count, size_bytes FROM export_job_files WHERE job_id=?", id) + files := []map[string]interface{}{} + for rows.Next() { + var uri, sheet sql.NullString + var rc, sz sql.NullInt64 + rows.Scan(&uri, &sheet, &rc, &sz) + files = append(files, map[string]interface{}{"storage_uri": uri.String, "sheet_name": sheet.String, "row_count": rc.Int64, "size_bytes": sz.Int64}) + } + rows.Close() + m["files"] = files + b, _ := json.Marshal(m) + w.Header().Set("Content-Type", "application/json") + w.Write(b) +} + +func (a *ExportsAPI) download(w http.ResponseWriter, r *http.Request, id string) { + row := a.meta.QueryRow("SELECT storage_uri FROM export_job_files WHERE job_id=? ORDER BY id DESC LIMIT 1", id) + var uri string + err := row.Scan(&uri) + if err != nil { + w.WriteHeader(http.StatusNotFound) + w.Write([]byte("not found")) + return + } + http.ServeFile(w, r, uri) +} + +func (a *ExportsAPI) cancel(w http.ResponseWriter, r *http.Request, id string) { + a.meta.Exec("UPDATE export_jobs SET status=? WHERE id=? AND status IN ('queued','running')", "canceled", id) + w.Write([]byte("ok")) +} + +func toString(v interface{}) string { + switch t := v.(type) { + case []byte: + return string(t) + case string: + return t + case int64: + return strconv.FormatInt(t, 10) + case int: + return strconv.Itoa(t) + case float64: + return strconv.FormatFloat(t, 'f', -1, 64) + default: + return "" + } +} + diff --git a/server/internal/api/router.go b/server/internal/api/router.go new file mode 100644 index 0000000..b89ae5d --- /dev/null +++ b/server/internal/api/router.go @@ -0,0 +1,16 @@ +package api + +import ( + "database/sql" + "net/http" +) + +func NewRouter(metaDB *sql.DB, marketingDB *sql.DB) http.Handler { + mux := http.NewServeMux() + mux.Handle("/api/templates", TemplatesHandler(metaDB, marketingDB)) + mux.Handle("/api/templates/", TemplatesHandler(metaDB, marketingDB)) + mux.Handle("/api/exports", ExportsHandler(metaDB, marketingDB)) + mux.Handle("/api/exports/", ExportsHandler(metaDB, marketingDB)) + mux.Handle("/", http.FileServer(http.Dir("web"))) + return mux +} diff --git a/server/internal/api/templates.go b/server/internal/api/templates.go new file mode 100644 index 0000000..608bbbf --- /dev/null +++ b/server/internal/api/templates.go @@ -0,0 +1,256 @@ +package api + +import ( + "database/sql" + "encoding/json" + "io" + "net/http" + "strings" + "time" + "marketing-system-data-tool/server/internal/exporter" +) + +type TemplatesAPI struct{ + meta *sql.DB + marketing *sql.DB +} + +func TemplatesHandler(meta, marketing *sql.DB) http.Handler { + api := &TemplatesAPI{meta: meta, marketing: marketing} + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + p := strings.TrimPrefix(r.URL.Path, "/api/templates") + if r.Method == http.MethodPost && p == "" { + api.createTemplate(w, r) + return + } + if r.Method == http.MethodGet && p == "" { + api.listTemplates(w, r) + return + } + if strings.HasPrefix(p, "/") { + id := strings.TrimPrefix(p, "/") + if r.Method == http.MethodGet { + api.getTemplate(w, r, id) + return + } + if r.Method == http.MethodPatch { + api.patchTemplate(w, r, id) + return + } + if r.Method == http.MethodPost && strings.HasSuffix(p, "/validate") { + id = strings.TrimSuffix(id, "/validate") + api.validateTemplate(w, r, id) + return + } + } + w.WriteHeader(http.StatusNotFound) + }) +} + +type TemplatePayload struct{ + Name string `json:"name"` + Datasource string `json:"datasource"` + MainTable string `json:"main_table"` + Fields []string `json:"fields"` + Filters map[string]interface{} `json:"filters"` + FileFormat string `json:"file_format"` + OwnerID uint64 `json:"owner_id"` + Visibility string `json:"visibility"` +} + +func (a *TemplatesAPI) createTemplate(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + var p TemplatePayload + json.Unmarshal(b, &p) + wl := map[string]bool{ + "order.order_number": true, + "order.creator": true, + "order.out_trade_no": true, + "order.type": true, + "order.status": true, + "order.contract_price": true, + "order.num": true, + "order.total": true, + "order.pay_amount": true, + "order.create_time": true, + "order.update_time": true, + } + req := exporter.BuildRequest{MainTable: p.MainTable, Fields: p.Fields, Filters: p.Filters} + q, args, err := exporter.BuildSQL(req, wl) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(err.Error())) + return + } + _, score, err := exporter.RunExplain(a.marketing, q, args) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(err.Error())) + return + } + now := time.Now() + _, err = a.meta.Exec( + "INSERT INTO export_templates (name, datasource, main_table, fields_json, filters_json, file_format, visibility, owner_id, enabled, explain_score, last_validated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)", + p.Name, p.Datasource, p.MainTable, toJSON(p.Fields), toJSON(p.Filters), p.FileFormat, p.Visibility, p.OwnerID, 1, score, now, + ) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } + w.WriteHeader(http.StatusCreated) + w.Write([]byte("ok")) +} + +func (a *TemplatesAPI) listTemplates(w http.ResponseWriter, r *http.Request) { + rows, err := a.meta.Query("SELECT id,name,datasource,main_table,file_format,visibility,owner_id,enabled,explain_score,last_validated_at,created_at,updated_at FROM export_templates ORDER BY updated_at DESC LIMIT 200") + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } + defer rows.Close() + out := []map[string]interface{}{} + for rows.Next() { + var id uint64 + var name, datasource, mainTable, fileFormat, visibility string + var ownerID uint64 + var enabled int + var explainScore sql.NullInt64 + var lastValidatedAt sql.NullTime + var createdAt, updatedAt time.Time + err := rows.Scan(&id, &name, &datasource, &mainTable, &fileFormat, &visibility, &ownerID, &enabled, &explainScore, &lastValidatedAt, &createdAt, &updatedAt) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } + m := map[string]interface{}{"id": id, "name": name, "datasource": datasource, "main_table": mainTable, "file_format": fileFormat, "visibility": visibility, "owner_id": ownerID, "enabled": enabled == 1, "explain_score": explainScore.Int64, "last_validated_at": lastValidatedAt.Time, "created_at": createdAt, "updated_at": updatedAt} + out = append(out, m) + } + b, _ := json.Marshal(out) + w.Header().Set("Content-Type", "application/json") + w.Write(b) +} + +func (a *TemplatesAPI) getTemplate(w http.ResponseWriter, r *http.Request, id string) { + row := a.meta.QueryRow("SELECT id,name,datasource,main_table,fields_json,filters_json,file_format,visibility,owner_id,enabled,explain_score,last_validated_at,created_at,updated_at FROM export_templates WHERE id=?", id) + var m = map[string]interface{}{} + var enabled int + var explainScore sql.NullInt64 + var lastValidatedAt sql.NullTime + var createdAt, updatedAt time.Time + var fields, filters []byte + err := row.Scan(&m["id"], &m["name"], &m["datasource"], &m["main_table"], &fields, &filters, &m["file_format"], &m["visibility"], &m["owner_id"], &enabled, &explainScore, &lastValidatedAt, &createdAt, &updatedAt) + if err != nil { + w.WriteHeader(http.StatusNotFound) + w.Write([]byte("not found")) + return + } + m["enabled"] = enabled == 1 + m["explain_score"] = explainScore.Int64 + m["last_validated_at"] = lastValidatedAt.Time + m["created_at"] = createdAt + m["updated_at"] = updatedAt + m["fields"] = fromJSON(fields) + m["filters"] = fromJSON(filters) + b, _ := json.Marshal(m) + w.Header().Set("Content-Type", "application/json") + w.Write(b) +} + +func (a *TemplatesAPI) patchTemplate(w http.ResponseWriter, r *http.Request, id string) { + b, _ := io.ReadAll(r.Body) + var p map[string]interface{} + json.Unmarshal(b, &p) + set := []string{} + args := []interface{}{} + for k, v := range p { + switch k { + case "name", "visibility", "file_format": + set = append(set, k+"=?") + args = append(args, v) + case "enabled": + set = append(set, "enabled=?") + if v.(bool) { + args = append(args, 1) + } else { + args = append(args, 0) + } + } + } + if len(set) == 0 { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte("no patch")) + return + } + args = append(args, id) + _, err := a.meta.Exec("UPDATE export_templates SET "+strings.Join(set, ",")+" WHERE id=?", args...) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } + w.Write([]byte("ok")) +} + +func (a *TemplatesAPI) validateTemplate(w http.ResponseWriter, r *http.Request, id string) { + row := a.meta.QueryRow("SELECT main_table, fields_json, filters_json FROM export_templates WHERE id=?", id) + var main string + var fields, filters []byte + err := row.Scan(&main, &fields, &filters) + if err != nil { + w.WriteHeader(http.StatusNotFound) + w.Write([]byte("not found")) + return + } + var fs []string + var fl map[string]interface{} + json.Unmarshal(fields, &fs) + json.Unmarshal(filters, &fl) + wl := map[string]bool{ + "order.order_number": true, + "order.creator": true, + "order.out_trade_no": true, + "order.type": true, + "order.status": true, + "order.contract_price": true, + "order.num": true, + "order.total": true, + "order.pay_amount": true, + "order.create_time": true, + "order.update_time": true, + } + req := exporter.BuildRequest{MainTable: main, Fields: fs, Filters: fl} + q, args, err := exporter.BuildSQL(req, wl) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(err.Error())) + return + } + _, score, err := exporter.RunExplain(a.marketing, q, args) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(err.Error())) + return + } + now := time.Now() + _, err = a.meta.Exec("UPDATE export_templates SET explain_score=?, last_validated_at=? WHERE id=?", score, now, id) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } + w.Write([]byte("ok")) +} + +func toJSON(v interface{}) []byte { + b, _ := json.Marshal(v) + return b +} + +func fromJSON(b []byte) interface{} { + var v interface{} + json.Unmarshal(b, &v) + return v +} diff --git a/server/internal/db/mysql.go b/server/internal/db/mysql.go new file mode 100644 index 0000000..54f5379 --- /dev/null +++ b/server/internal/db/mysql.go @@ -0,0 +1,17 @@ +package db + +import ( + "database/sql" + _ "github.com/go-sql-driver/mysql" +) + +func ConnectMySQL(dsn string) (*sql.DB, error) { + if dsn == "" { + return sql.Open("mysql", "invalid:invalid@tcp(localhost:3306)/invalid") + } + db, err := sql.Open("mysql", dsn) + if err != nil { + return nil, err + } + return db, nil +} diff --git a/server/internal/exporter/explain.go b/server/internal/exporter/explain.go new file mode 100644 index 0000000..9f8cbf1 --- /dev/null +++ b/server/internal/exporter/explain.go @@ -0,0 +1,156 @@ +package exporter + +import ( + "database/sql" +) + +type ExplainRow struct { + ID sql.NullInt64 + SelectType sql.NullString + Table sql.NullString + Type sql.NullString + PossibleKeys sql.NullString + Key sql.NullString + KeyLen sql.NullString + Ref sql.NullString + Rows sql.NullInt64 + Filtered sql.NullFloat64 + Extra sql.NullString +} + +func RunExplain(db *sql.DB, q string, args []interface{}) ([]ExplainRow, int, error) { + rows, err := db.Query("EXPLAIN "+q, args...) + if err != nil { + return nil, 0, err + } + defer rows.Close() + res := []ExplainRow{} + cols, _ := rows.Columns() + for rows.Next() { + vals := make([]interface{}, len(cols)) + dest := make([]interface{}, len(cols)) + for i := range vals { + dest[i] = &vals[i] + } + if err := rows.Scan(dest...); err != nil { + return nil, 0, err + } + r := ExplainRow{} + if len(cols) >= 10 { + toRow(vals, &r) + } + res = append(res, r) + } + score := 100 + for _, r := range res { + if r.Type.String == "ALL" { + score -= 50 + } + if r.Rows.Int64 > 1000000 { + score -= 30 + } + if r.Extra.Valid { + if contains(r.Extra.String, "Using temporary") || contains(r.Extra.String, "Using filesort") { + score -= 20 + } + } + } + if score < 0 { + score = 0 + } + return res, score, nil +} + +func toRow(vals []interface{}, r *ExplainRow) { + if s, ok := vals[0].([]byte); ok { + r.ID.Int64 = toInt64(string(s)) + r.ID.Valid = true + } + if s, ok := vals[1].([]byte); ok { + r.SelectType.String = string(s) + r.SelectType.Valid = true + } + if s, ok := vals[2].([]byte); ok { + r.Table.String = string(s) + r.Table.Valid = true + } + if s, ok := vals[3].([]byte); ok { + r.Type.String = string(s) + r.Type.Valid = true + } + if s, ok := vals[4].([]byte); ok { + r.PossibleKeys.String = string(s) + r.PossibleKeys.Valid = true + } + if s, ok := vals[5].([]byte); ok { + r.Key.String = string(s) + r.Key.Valid = true + } + if s, ok := vals[6].([]byte); ok { + r.KeyLen.String = string(s) + r.KeyLen.Valid = true + } + if s, ok := vals[7].([]byte); ok { + r.Ref.String = string(s) + r.Ref.Valid = true + } + if s, ok := vals[8].([]byte); ok { + r.Rows.Int64 = toInt64(string(s)) + r.Rows.Valid = true + } + if s, ok := vals[9].([]byte); ok { + r.Filtered.Float64 = toFloat64(string(s)) + r.Filtered.Valid = true + } + if len(vals) > 10 { + if s, ok := vals[10].([]byte); ok { + r.Extra.String = string(s) + r.Extra.Valid = true + } + } +} + +func contains(s, sub string) bool { + for i := 0; i+len(sub) <= len(s); i++ { + if s[i:i+len(sub)] == sub { + return true + } + } + return false +} + +func toInt64(s string) int64 { + var n int64 + for i := 0; i < len(s); i++ { + c := s[i] + if c < '0' || c > '9' { + continue + } + n = n*10 + int64(c-'0') + } + return n +} + +func toFloat64(s string) float64 { + var n float64 + var d float64 + var seen bool + d = 1 + for i := 0; i < len(s); i++ { + c := s[i] + if c == '.' { + seen = true + continue + } + if c < '0' || c > '9' { + continue + } + if !seen { + n = n*10 + float64(c-'0') + } else { + d *= 10 + n = n + float64(c-'0')/d + } + } + return n +} diff --git a/server/internal/exporter/sqlbuilder.go b/server/internal/exporter/sqlbuilder.go new file mode 100644 index 0000000..5cbd5a5 --- /dev/null +++ b/server/internal/exporter/sqlbuilder.go @@ -0,0 +1,83 @@ +package exporter + +import ( + "encoding/json" + "errors" + "strings" +) + +type BuildRequest struct { + MainTable string + Fields []string + Filters map[string]interface{} +} + +func BuildSQL(req BuildRequest, whitelist map[string]bool) (string, []interface{}, error) { + if req.MainTable != "order" { + return "", nil, errors.New("unsupported main table") + } + cols := []string{} + for _, f := range req.Fields { + if !whitelist["order."+f] { + return "", nil, errors.New("field not allowed") + } + if f == "key" || req.MainTable == "order" { + cols = append(cols, "`order`."+escape(f)) + } + } + if len(cols) == 0 { + return "", nil, errors.New("no fields") + } + sb := strings.Builder{} + sb.WriteString("SELECT ") + sb.WriteString(strings.Join(cols, ",")) + sb.WriteString(" FROM `order`") + args := []interface{}{} + where := []string{} + if v, ok := req.Filters["creator_in"]; ok { + ids := []interface{}{} + switch t := v.(type) { + case []interface{}: + ids = t + case []int: + for _, x := range t { + ids = append(ids, x) + } + case []string: + for _, x := range t { + ids = append(ids, x) + } + } + if len(ids) == 0 { + return "", nil, errors.New("creator_in required") + } + ph := strings.Repeat("?,", len(ids)) + ph = strings.TrimSuffix(ph, ",") + where = append(where, "`order`.creator IN ("+ph+")") + args = append(args, ids...) + } else { + return "", nil, errors.New("creator_in required") + } + if v, ok := req.Filters["create_time_between"]; ok { + var arr []interface{} + b, _ := json.Marshal(v) + json.Unmarshal(b, &arr) + if len(arr) != 2 { + return "", nil, errors.New("create_time_between requires 2 values") + } + where = append(where, "`order`.create_time BETWEEN ? AND ?") + args = append(args, arr[0], arr[1]) + } + if len(where) > 0 { + sb.WriteString(" WHERE ") + sb.WriteString(strings.Join(where, " AND ")) + } + return sb.String(), args, nil +} + +func escape(s string) string { + if s == "key" { + return "`key`" + } + return s +} diff --git a/server/internal/exporter/writer.go b/server/internal/exporter/writer.go new file mode 100644 index 0000000..5a68a65 --- /dev/null +++ b/server/internal/exporter/writer.go @@ -0,0 +1,131 @@ +package exporter + +import ( + "encoding/csv" + "os" + "path/filepath" + "time" + "github.com/xuri/excelize/v2" +) + +type RowWriter interface { + WriteHeader(cols []string) error + WriteRow(vals []string) error + Close() (string, int64, error) +} + +type CSVWriter struct { + f *os.File + w *csv.Writer + count int64 +} + +func NewCSVWriter(dir, name string) (*CSVWriter, error) { + os.MkdirAll(dir, 0755) + p := filepath.Join(dir, name+"_"+time.Now().Format("20060102150405")+".csv") + f, err := os.Create(p) + if err != nil { + return nil, err + } + return &CSVWriter{f: f, w: csv.NewWriter(f)}, nil +} + +func (c *CSVWriter) WriteHeader(cols []string) error { + if err := c.w.Write(cols); err != nil { + return err + } + c.count++ + return nil +} + +func (c *CSVWriter) WriteRow(vals []string) error { + if err := c.w.Write(vals); err != nil { + return err + } + c.count++ + return nil +} + +func (c *CSVWriter) Close() (string, int64, error) { + c.w.Flush() + p := c.f.Name() + info, _ := c.f.Stat() + c.f.Close() + return p, info.Size(), nil +} + +type XLSXWriter struct { + f *excelize.File + sheet string + row int +} + +func NewXLSXWriter(dir, name, sheet string) (*XLSXWriter, string, error) { + os.MkdirAll(dir, 0755) + p := filepath.Join(dir, name+"_"+time.Now().Format("20060102150405")+".xlsx") + f := excelize.NewFile() + f.NewSheet(sheet) + idx, _ := f.GetSheetIndex(sheet) + f.SetActiveSheet(idx) + return &XLSXWriter{f: f, sheet: sheet, row: 1}, p, nil +} + +func (x *XLSXWriter) WriteHeader(cols []string) error { + for i, c := range cols { + cell := col(i+1) + "1" + if err := x.f.SetCellValue(x.sheet, cell, c); err != nil { + return err + } + } + x.row = 2 + return nil +} + +func (x *XLSXWriter) WriteRow(vals []string) error { + r := x.row + for i, v := range vals { + cell := col(i+1) + itoa(r) + if err := x.f.SetCellValue(x.sheet, cell, v); err != nil { + return err + } + } + x.row++ + return nil +} + +func (x *XLSXWriter) Close(path string) (string, int64, error) { + if err := x.f.SaveAs(path); err != nil { + return "", 0, err + } + info, err := os.Stat(path) + if err != nil { + return path, 0, nil + } + return path, info.Size(), nil +} + +func col(n int) string { + s := "" + for n > 0 { + n-- + s = string('A'+(n%26)) + s + n /= 26 + } + return s +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + b := make([]byte, 0, 10) + m := n + for m > 0 { + b = append(b, byte('0'+(m%10))) + m /= 10 + } + for i, j := 0, len(b)-1; i < j; i, j = i+1, j-1 { + b[i], b[j] = b[j], b[i] + } + return string(b) +} diff --git a/server/internal/migrate/migrate.go b/server/internal/migrate/migrate.go new file mode 100644 index 0000000..d8f0b00 --- /dev/null +++ b/server/internal/migrate/migrate.go @@ -0,0 +1,20 @@ +package migrate + +import ( + "database/sql" +) + +func Apply(db *sql.DB) error { + stmts := []string{ + "CREATE TABLE IF NOT EXISTS export_templates (id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255) NOT NULL, datasource VARCHAR(32) NOT NULL, main_table VARCHAR(64) NOT NULL, joins_json JSON, fields_json JSON, filters_json JSON, file_format VARCHAR(16) NOT NULL, stats_enabled TINYINT(1) NOT NULL DEFAULT 0, sheet_split_by VARCHAR(64), visibility VARCHAR(16) NOT NULL DEFAULT 'private', owner_id BIGINT UNSIGNED NOT NULL, enabled TINYINT(1) NOT NULL DEFAULT 1, explain_json JSON, explain_score INT, last_validated_at DATETIME, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP)", + "CREATE TABLE IF NOT EXISTS export_jobs (id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, template_id BIGINT UNSIGNED NOT NULL, status VARCHAR(16) NOT NULL, requested_by BIGINT UNSIGNED NOT NULL, permission_scope_json JSON, options_json JSON, row_estimate BIGINT, total_rows BIGINT, file_format VARCHAR(16) NOT NULL, started_at DATETIME, finished_at DATETIME, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_template_id (template_id), INDEX idx_status (status), INDEX idx_requested_by (requested_by))", + "CREATE TABLE IF NOT EXISTS export_job_files (id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, job_id BIGINT UNSIGNED NOT NULL, storage_uri VARCHAR(1024) NOT NULL, sheet_name VARCHAR(255), row_count BIGINT, size_bytes BIGINT, checksum VARCHAR(128), created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, INDEX idx_job_id (job_id))", + "CREATE TABLE IF NOT EXISTS export_audits (id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, actor_id BIGINT UNSIGNED NOT NULL, action VARCHAR(64) NOT NULL, entity_type VARCHAR(64) NOT NULL, entity_id BIGINT UNSIGNED NOT NULL, detail_json JSON, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, INDEX idx_entity (entity_type, entity_id), INDEX idx_actor (actor_id))", + } + for _, s := range stmts { + if _, err := db.Exec(s); err != nil { + return err + } + } + return nil +} diff --git a/server/internal/models/job.go b/server/internal/models/job.go new file mode 100644 index 0000000..f8f1980 --- /dev/null +++ b/server/internal/models/job.go @@ -0,0 +1,17 @@ +package models + +import "time" + +type Job struct { + ID uint64 + TemplateID uint64 + Status string + RequestedBy uint64 + PermissionScopeJSON []byte + OptionsJSON []byte + RowEstimate *int64 + TotalRows *int64 + FileFormat string + StartedAt *time.Time + FinishedAt *time.Time +} diff --git a/server/internal/models/template.go b/server/internal/models/template.go new file mode 100644 index 0000000..75a9bbd --- /dev/null +++ b/server/internal/models/template.go @@ -0,0 +1,19 @@ +package models + +type Template struct { + ID uint64 + Name string + Datasource string + MainTable string + JoinsJSON []byte + FieldsJSON []byte + FiltersJSON []byte + FileFormat string + StatsEnabled bool + SheetSplitBy *string + Visibility string + OwnerID uint64 + Enabled bool + ExplainJSON []byte + ExplainScore *int +} diff --git a/web/index.html b/web/index.html new file mode 100644 index 0000000..ea2abdc --- /dev/null +++ b/web/index.html @@ -0,0 +1,90 @@ + + + + + +MarketingSystemDataTool + + + + +
+
+ 导出工具 +
+
+
+
+
+
+
模板列表
+
+
+
+
+
+
新增模板
+
+
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ +
+
+
+
+
+
+
+
导出任务
+
+
+
+
+ + +
+ + diff --git a/web/main.js b/web/main.js new file mode 100644 index 0000000..3d1364f --- /dev/null +++ b/web/main.js @@ -0,0 +1,57 @@ +async function loadTemplates(){ + const res=await fetch('/api/templates'); + const data=await res.json(); + const el=document.getElementById('templates'); + const rows=data.map(t=>`${t.id}${t.name}${t.datasource}${t.file_format}${t.explain_score||''}`).join(''); + el.innerHTML=`${rows}
ID名称数据源格式EXPLAIN评分操作
`; + document.querySelectorAll('button.export').forEach(b=>b.onclick=async()=>{ + const id=b.getAttribute('data-id'); + const payload={template_id:Number(id),requested_by:1,permission:{},options:{},file_format:'csv'}; + const r=await fetch('/api/exports',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(payload)}); + const j=await r.json(); + loadJob(j.id); + }); +} + +async function loadJob(id){ + const res=await fetch('/api/exports/'+id); + const j=await res.json(); + const el=document.getElementById('jobs'); + const link=j.files&&j.files.length?`下载`:''; + el.innerHTML=`
任务 ${id} 状态：${j.status} 行数：${j.total_rows||''} ${link}
`; +} + +document.getElementById('tpl-form').onsubmit=async(e)=>{ + e.preventDefault(); + const fd=new FormData(e.target); + const fields=fd.get('fields').split(',').map(s=>s.trim()); + const filters={}; + const creators=fd.get('creator_in'); + if(creators){filters.creator_in=creators.split(',').map(s=>Number(s.trim()))} + const tb=fd.get('time_begin'); + const te=fd.get('time_end'); + if(tb&&te){ + const fmt=(s)=>s.replace('T',' ')+':00'; + filters.create_time_between=[fmt(tb),fmt(te)] + } + const payload={ + name:fd.get('name'), + datasource:fd.get('datasource'), + main_table:fd.get('main_table'), + fields, + filters, + file_format:fd.get('file_format'), + owner_id:Number(fd.get('owner_id')), + visibility:fd.get('visibility') + }; + const res=await fetch('/api/templates',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(payload)}); + if(res.ok){ + loadTemplates(); + } else { + const t=await res.text(); + const el=document.getElementById('templates'); + el.insertAdjacentHTML('afterbegin',`
${t}
`); + } +}; + +loadTemplates(); diff --git a/web/styles.css b/web/styles.css new file mode 100644 index 0000000..9e6dd1d --- /dev/null +++ b/web/styles.css @@ -0,0 +1 @@ +body{font-family:system-ui,Arial}