diff --git a/im/backend/internal/app/admin.go b/im/backend/internal/app/admin.go index 99f687a..c021fd8 100644 --- a/im/backend/internal/app/admin.go +++ b/im/backend/internal/app/admin.go @@ -3,6 +3,7 @@ package app import ( "database/sql" "encoding/json" + "fmt" "net/http" "net/url" "strconv" @@ -577,9 +578,33 @@ func (a *App) adminHandleReport(w http.ResponseWriter, r *http.Request) { func (a *App) adminRiskUsers(w http.ResponseWriter, r *http.Request) { page, size, offset := pagination(r) + // 风险分只有配上"这个人现在是什么状态"才能拿来做决定：已经封了的不用再看， + // 加过白的不该再排在最前面。 + where := " WHERE 1=1" + args := []any{} + if level := strings.TrimSpace(r.URL.Query().Get("level")); level != "" { + where += " AND rp.risk_level>=?" + args = append(args, level) + } + switch strings.TrimSpace(r.URL.Query().Get("state")) { + case "whitelisted": + where += " AND rp.whitelisted=1" + case "pending": + where += " AND rp.whitelisted=0" + } + if keyword := strings.TrimSpace(r.URL.Query().Get("keyword")); keyword != "" { + where += " AND (p.nickname LIKE ? OR u.public_id LIKE ?)" + like := "%" + keyword + "%" + args = append(args, like, like) + } + from := ` FROM user_risk_profiles rp JOIN user_profiles p ON p.user_id=rp.user_id + JOIN users u ON u.id=rp.user_id` + where var total int64 - _ = a.db.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM user_risk_profiles`).Scan(&total) - rows, err := a.db.QueryContext(r.Context(), `SELECT rp.user_id,p.nickname,p.avatar_url,rp.risk_score,rp.risk_level,rp.message_score,rp.device_score,rp.report_score,rp.behavior_score,rp.updated_at FROM user_risk_profiles rp JOIN user_profiles p ON p.user_id=rp.user_id ORDER BY rp.risk_score DESC LIMIT ? OFFSET ?`, size, offset) + _ = a.db.QueryRowContext(r.Context(), `SELECT COUNT(*)`+from, args...).Scan(&total) + rows, err := a.db.QueryContext(r.Context(), `SELECT rp.user_id,u.public_id,p.nickname,p.avatar_url,rp.risk_score,rp.risk_level, + rp.message_score,rp.device_score,rp.report_score,rp.behavior_score,rp.updated_at, + rp.whitelisted,rp.whitelist_note,u.status`+from+` ORDER BY rp.whitelisted ASC,rp.risk_score DESC LIMIT ? OFFSET ?`, + append(args, size, offset)...) if err != nil { fail(w, 500, 50001, "查询失败") return @@ -588,19 +613,40 @@ func (a *App) adminRiskUsers(w http.ResponseWriter, r *http.Request) { items := []map[string]any{} for rows.Next() { var id int64 - var nick, avatar string - var score, level, message, device, report, behavior int + var publicID, nick, avatar, note string + var score, level, message, device, report, behavior, userStatus int + var whitelisted bool var updated time.Time - _ = rows.Scan(&id, &nick, &avatar, &score, &level, &message, &device, &report, &behavior, &updated) - items = append(items, map[string]any{"userId": id, "nickname": nick, "avatar": avatar, "riskScore": score, "riskLevel": level, "messageScore": message, "deviceScore": device, "reportScore": report, "behaviorScore": behavior, "updatedAt": updated}) + _ = rows.Scan(&id, &publicID, &nick, &avatar, &score, &level, &message, &device, &report, &behavior, + &updated, &whitelisted, ¬e, &userStatus) + items = append(items, map[string]any{"userId": id, "publicId": publicID, "nickname": nick, "avatar": avatar, + "riskScore": score, "riskLevel": level, "messageScore": message, "deviceScore": device, + "reportScore": report, "behaviorScore": behavior, "updatedAt": updated, + "whitelisted": whitelisted, "whitelistNote": note, "userStatus": userStatus}) } reply(w, pageResult{Items: items, Total: total, Page: page, Size: size}) } func (a *App) adminRiskEvents(w http.ResponseWriter, r *http.Request) { page, size, offset := pagination(r) + // 事件列表是一份待办：默认先看还没跟进的那些。 + where := " WHERE 1=1" + args := []any{} + switch strings.TrimSpace(r.URL.Query().Get("state")) { + case "handled": + where += " AND e.handled_at IS NOT NULL" + case "pending": + where += " AND e.handled_at IS NULL" + } + if eventType := strings.TrimSpace(r.URL.Query().Get("eventType")); eventType != "" { + where += " AND e.event_type=?" + args = append(args, eventType) + } + from := ` FROM risk_events e JOIN user_profiles p ON p.user_id=e.user_id` + where var total int64 - _ = a.db.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM risk_events`).Scan(&total) - rows, err := a.db.QueryContext(r.Context(), `SELECT e.id,e.user_id,p.nickname,e.event_type,e.score_delta,e.device_id,e.ip,e.metadata,e.created_at FROM risk_events e JOIN user_profiles p ON p.user_id=e.user_id ORDER BY e.created_at DESC LIMIT ? OFFSET ?`, size, offset) + _ = a.db.QueryRowContext(r.Context(), `SELECT COUNT(*)`+from, args...).Scan(&total) + rows, err := a.db.QueryContext(r.Context(), `SELECT e.id,e.user_id,p.nickname,e.event_type,e.score_delta,e.device_id,e.ip, + e.metadata,e.created_at,e.handled_at,e.handled_note`+from+` ORDER BY e.handled_at IS NOT NULL,e.created_at DESC LIMIT ? OFFSET ?`, + append(args, size, offset)...) if err != nil { fail(w, 500, 50001, "查询失败") return @@ -609,14 +655,17 @@ func (a *App) adminRiskEvents(w http.ResponseWriter, r *http.Request) { items := []map[string]any{} for rows.Next() { var id, userID int64 - var nick, typ, device, ip string + var nick, typ, device, ip, note string var delta int var metadata []byte var created time.Time - _ = rows.Scan(&id, &userID, &nick, &typ, &delta, &device, &ip, &metadata, &created) + var handled any + _ = rows.Scan(&id, &userID, &nick, &typ, &delta, &device, &ip, &metadata, &created, &handled, ¬e) var meta any _ = json.Unmarshal(metadata, &meta) - items = append(items, map[string]any{"id": id, "userId": userID, "nickname": nick, "eventType": typ, "scoreDelta": delta, "deviceId": device, "ip": ip, "metadata": meta, "createdAt": created}) + items = append(items, map[string]any{"id": id, "userId": userID, "nickname": nick, "eventType": typ, + "scoreDelta": delta, "deviceId": device, "ip": ip, "metadata": meta, "createdAt": created, + "handledAt": handled, "handledNote": note}) } reply(w, pageResult{Items: items, Total: total, Page: page, Size: size}) } @@ -689,10 +738,18 @@ func (a *App) adminOrders(w http.ResponseWriter, r *http.Request) { like := "%" + keyword + "%" args = append(args, like, like, like) } + // 一张订单表里现在装着会员、代喂和保证金三种东西，不分开看就没法核对。 + if productType := strings.TrimSpace(r.URL.Query().Get("productType")); productType != "" { + where += ` AND o.product_type=?` + args = append(args, productType) + } var total int64 _ = a.db.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM orders o JOIN users u ON u.id=o.user_id JOIN user_profiles p ON p.user_id=o.user_id`+where, args...).Scan(&total) args = append(args, size, offset) - rows, err := a.db.QueryContext(r.Context(), `SELECT o.id,o.order_no,o.user_id,u.public_id,p.nickname,o.product_id,COALESCE(mp.name,''),o.amount_cent,o.currency,o.status,o.channel,o.paid_at,o.created_at FROM orders o JOIN users u ON u.id=o.user_id JOIN user_profiles p ON p.user_id=o.user_id LEFT JOIN membership_plans mp ON mp.id=o.product_id`+where+` ORDER BY o.created_at DESC LIMIT ? OFFSET ?`, args...) + rows, err := a.db.QueryContext(r.Context(), `SELECT o.id,o.order_no,o.user_id,u.public_id,p.nickname,o.product_type,o.product_id, + COALESCE(mp.name,''),o.amount_cent,COALESCE(o.refunded_cent,0),o.currency,o.status,o.channel,o.provider_order_no,o.paid_at,o.created_at + FROM orders o JOIN users u ON u.id=o.user_id JOIN user_profiles p ON p.user_id=o.user_id + LEFT JOIN membership_plans mp ON mp.id=o.product_id AND o.product_type='membership'`+where+` ORDER BY o.created_at DESC LIMIT ? OFFSET ?`, args...) if err != nil { fail(w, 500, 50001, "查询失败") return @@ -701,12 +758,27 @@ func (a *App) adminOrders(w http.ResponseWriter, r *http.Request) { items := []map[string]any{} for rows.Next() { var id, userID, productID int64 - var orderNo, publicID, nickname, product, currency, orderStatus, channel string - var amount int + var orderNo, publicID, nickname, product, currency, orderStatus, channel, productType, providerOrderNo string + var amount, refunded int var paid any var created time.Time - _ = rows.Scan(&id, &orderNo, &userID, &publicID, &nickname, &productID, &product, &amount, ¤cy, &orderStatus, &channel, &paid, &created) - items = append(items, map[string]any{"id": id, "orderNo": orderNo, "userId": userID, "publicId": publicID, "nickname": nickname, "productId": productID, "productName": product, "amountCent": amount, "currency": currency, "status": orderStatus, "channel": channel, "paidAt": paid, "createdAt": created}) + _ = rows.Scan(&id, &orderNo, &userID, &publicID, &nickname, &productType, &productID, &product, &amount, + &refunded, ¤cy, &orderStatus, &channel, &providerOrderNo, &paid, &created) + // 会员订单的名字来自套餐表；代喂和保证金没有套餐，名字要自己拼出来。 + switch productType { + case productPetFeed: + product = fmt.Sprintf("代喂任务 #%d", productID) + case productPetDeposit: + product = "喂养者保证金" + case productMembership: + if product == "" { + product = "会员套餐" + } + } + items = append(items, map[string]any{"id": id, "orderNo": orderNo, "userId": userID, "publicId": publicID, + "nickname": nickname, "productType": productType, "productId": productID, "productName": product, + "amountCent": amount, "refundedCent": refunded, "currency": currency, "status": orderStatus, + "channel": channel, "providerOrderNo": providerOrderNo, "paidAt": paid, "createdAt": created}) } reply(w, pageResult{Items: items, Total: total, Page: page, Size: size}) } diff --git a/im/backend/internal/app/admin_broadcast.go b/im/backend/internal/app/admin_broadcast.go new file mode 100644 index 0000000..6b50631 --- /dev/null +++ b/im/backend/internal/app/admin_broadcast.go @@ -0,0 +1,287 @@ +package app + +import ( + "context" + "fmt" + "net/http" + "strconv" + "strings" + "time" +) + +// Announcements. Until now operations had no way to say anything to users: +// an outage, a rule change, a campaign — all of it depended on people noticing +// by themselves. Sending is deliberately recorded and revocable, because the +// two questions afterwards are always "who got it" and "can we take it back". + +type broadcastRequest struct { + Title string `json:"title"` + Content string `json:"content"` + Audience string `json:"audience"` + UserIDs []int64 `json:"userIds"` + CityCode string `json:"cityCode"` + WithPush bool `json:"withPush"` +} + +// broadcastAudience turns the chosen audience into a recipient query. Keeping +// the SQL here (rather than building it from request fields) is what stops an +// audience picker from turning into an injection point. +func (a *App) broadcastAudience(req broadcastRequest) (string, []any, error) { + switch strings.TrimSpace(req.Audience) { + case "all": + return `SELECT id FROM users WHERE status=1 AND deleted_at IS NULL`, nil, nil + case "vip": + return `SELECT u.id FROM users u JOIN user_profiles p ON p.user_id=u.id + WHERE u.status=1 AND u.deleted_at IS NULL AND p.is_vip=1`, nil, nil + case "city": + code := strings.TrimSpace(req.CityCode) + if code == "" { + return "", nil, fmt.Errorf("请选择城市") + } + return `SELECT u.id FROM users u JOIN user_location_states l ON l.user_id=u.id + WHERE u.status=1 AND u.deleted_at IS NULL AND l.city_code=?`, []any{code}, nil + case "sitters": + return `SELECT u.id FROM users u JOIN pet_sitters s ON s.user_id=u.id + WHERE u.status=1 AND u.deleted_at IS NULL AND s.status=1`, nil, nil + case "owners": + return `SELECT DISTINCT u.id FROM users u JOIN pets p ON p.owner_user_id=u.id + WHERE u.status=1 AND u.deleted_at IS NULL AND p.deleted_at IS NULL`, nil, nil + case "users": + if len(req.UserIDs) == 0 { + return "", nil, fmt.Errorf("请指定接收的用户") + } + if len(req.UserIDs) > 500 { + return "", nil, fmt.Errorf("指定用户最多 500 个，再多请改用人群条件") + } + placeholders := make([]string, 0, len(req.UserIDs)) + args := make([]any, 0, len(req.UserIDs)) + for _, id := range req.UserIDs { + placeholders = append(placeholders, "?") + args = append(args, id) + } + return `SELECT id FROM users WHERE status=1 AND deleted_at IS NULL AND id IN (` + + strings.Join(placeholders, ",") + `)`, args, nil + } + return "", nil, fmt.Errorf("人群条件无效") +} + +func (a *App) adminSendBroadcast(w http.ResponseWriter, r *http.Request) { + var req broadcastRequest + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "格式错误") + return + } + req.Title = strings.TrimSpace(req.Title) + req.Content = strings.TrimSpace(req.Content) + if req.Title == "" || len([]rune(req.Title)) > 40 { + fail(w, http.StatusBadRequest, 20001, "标题为 1 到 40 个字") + return + } + if len([]rune(req.Content)) < 5 || len([]rune(req.Content)) > 500 { + fail(w, http.StatusBadRequest, 20001, "正文为 5 到 500 个字") + return + } + query, args, err := a.broadcastAudience(req) + if err != nil { + fail(w, http.StatusBadRequest, 20001, err.Error()) + return + } + + rows, err := a.db.QueryContext(r.Context(), query, args...) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "查询接收人失败") + return + } + recipients := []int64{} + for rows.Next() { + var id int64 + if rows.Scan(&id) == nil { + recipients = append(recipients, id) + } + } + rows.Close() + if len(recipients) == 0 { + fail(w, http.StatusBadRequest, 20001, "这个条件下没有匹配到用户") + return + } + limit := a.configInt(r.Context(), "notification.broadcast_max_recipients", 20000) + if len(recipients) > limit { + fail(w, http.StatusBadRequest, 20001, fmt.Sprintf("匹配到 %d 人，超过单次上限 %d 人", len(recipients), limit)) + return + } + + reference := strings.TrimSpace(req.CityCode) + if req.Audience == "users" { + ids := make([]string, 0, len(req.UserIDs)) + for _, id := range req.UserIDs { + ids = append(ids, strconv.FormatInt(id, 10)) + } + reference = strings.Join(ids, ",") + if len(reference) > 500 { + reference = reference[:500] + } + } + withPush := req.WithPush && a.configBool(r.Context(), "notification.broadcast_push_enabled", true) + result, err := a.db.ExecContext(r.Context(), `INSERT INTO notification_broadcasts + (title,content,audience,audience_ref,with_push,recipient_count,created_by) + VALUES(?,?,?,?,?,?,?)`, req.Title, req.Content, req.Audience, reference, withPush, len(recipients), current(r).ID) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "创建公告失败") + return + } + broadcastID, _ := result.LastInsertId() + + // 一条一条插会把几万人的公告拖成分钟级，所以按批写； + // 关掉了系统通知的人不发，这是他自己的设置。 + delivered := 0 + const batchSize = 500 + for start := 0; start < len(recipients); start += batchSize { + end := start + batchSize + if end > len(recipients) { + end = len(recipients) + } + values := make([]string, 0, end-start) + params := make([]any, 0, (end-start)*6) + for _, userID := range recipients[start:end] { + if !a.notificationAllowed(r.Context(), userID, "system") { + continue + } + values = append(values, "(?,?,?,?,?,?)") + params = append(params, userID, "system", req.Title, req.Content, "broadcast", broadcastID) + } + if len(values) == 0 { + continue + } + if _, err = a.db.ExecContext(r.Context(), `INSERT INTO notifications(user_id,type,title,content,biz_type,biz_id) VALUES`+ + strings.Join(values, ","), params...); err != nil { + fail(w, http.StatusInternalServerError, 50001, "写入通知失败") + return + } + delivered += len(values) + } + _, _ = a.db.ExecContext(r.Context(), `UPDATE notification_broadcasts SET recipient_count=? WHERE id=?`, delivered, broadcastID) + + if withPush { + // 推送是尽力而为：走不通不该让公告本身失败，所以放到后台并单独计数。 + go a.pushBroadcast(context.Background(), broadcastID, recipients, req.Title, req.Content) + } + a.audit(r, "broadcast", "notification", broadcastID, map[string]any{ + "audience": req.Audience, "recipients": delivered, "withPush": withPush, "title": req.Title, + }) + reply(w, map[string]any{"id": broadcastID, "recipientCount": delivered, "withPush": withPush}) +} + +func (a *App) pushBroadcast(ctx context.Context, broadcastID int64, recipients []int64, title, content string) { + credentials, ok := a.pushSettings(ctx) + if !ok { + return + } + token, err := a.pushToken(ctx, credentials) + if err != nil { + return + } + sent := 0 + for _, userID := range recipients { + targets := a.pushRecipients(ctx, 0, 0, []int64{userID}) + for _, target := range targets { + if a.sendPush(ctx, credentials, token, target, title, content, 0) == nil { + sent++ + } + } + } + _, _ = a.db.ExecContext(ctx, `UPDATE notification_broadcasts SET push_count=? WHERE id=?`, sent, broadcastID) +} + +func (a *App) adminBroadcasts(w http.ResponseWriter, r *http.Request) { + page, size, offset := pagination(r) + var total int64 + _ = a.db.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM notification_broadcasts`).Scan(&total) + rows, err := a.db.QueryContext(r.Context(), `SELECT b.id,b.title,b.content,b.audience,b.audience_ref,b.with_push, + b.recipient_count,b.push_count,b.status,b.revoked_count,b.created_at,COALESCE(admin.username,''), + (SELECT COUNT(*) FROM notifications n WHERE n.biz_type='broadcast' AND n.biz_id=b.id AND n.read_at IS NOT NULL) + FROM notification_broadcasts b LEFT JOIN admin_users admin ON admin.id=b.created_by + ORDER BY b.id DESC LIMIT ? OFFSET ?`, size, offset) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "查询公告失败") + return + } + defer rows.Close() + items := []map[string]any{} + for rows.Next() { + var id int64 + var title, content, audience, reference, status, operator string + var withPush bool + var recipients, pushes, revoked, read int + var created time.Time + if rows.Scan(&id, &title, &content, &audience, &reference, &withPush, &recipients, &pushes, + &status, &revoked, &created, &operator, &read) != nil { + continue + } + items = append(items, map[string]any{ + "id": id, "title": title, "content": content, "audience": audience, "audienceRef": reference, + "withPush": withPush, "recipientCount": recipients, "pushCount": pushes, "status": status, + "revokedCount": revoked, "readCount": read, "operator": operator, + "createdAt": created.Format(time.RFC3339), + }) + } + reply(w, map[string]any{"items": items, "total": total, "page": page, "size": size}) +} + +// adminRevokeBroadcast pulls back only what nobody has read yet. A notification +// someone already opened cannot be unsent, and pretending otherwise would make +// the counts lie. +func (a *App) adminRevokeBroadcast(w http.ResponseWriter, r *http.Request) { + id, err := pathID(r) + if err != nil { + fail(w, http.StatusBadRequest, 20001, "编号无效") + return + } + var status string + if a.db.QueryRowContext(r.Context(), `SELECT status FROM notification_broadcasts WHERE id=?`, id).Scan(&status) != nil { + fail(w, http.StatusNotFound, 30001, "公告不存在") + return + } + if status == "REVOKED" { + fail(w, http.StatusBadRequest, 20001, "这条公告已经撤回过了") + return + } + result, err := a.db.ExecContext(r.Context(), `DELETE FROM notifications WHERE biz_type='broadcast' AND biz_id=? AND read_at IS NULL`, id) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "撤回失败") + return + } + revoked, _ := result.RowsAffected() + if _, err = a.db.ExecContext(r.Context(), `UPDATE notification_broadcasts SET status='REVOKED',revoked_count=?,revoked_at=NOW(3) WHERE id=?`, revoked, id); err != nil { + fail(w, http.StatusInternalServerError, 50001, "撤回失败") + return + } + a.audit(r, "revoke_broadcast", "notification", id, map[string]any{"revoked": revoked}) + reply(w, map[string]any{"success": true, "revokedCount": revoked}) +} + +// adminBroadcastPreview answers "how many people would get this" before anyone +// presses send. +func (a *App) adminBroadcastPreview(w http.ResponseWriter, r *http.Request) { + req := broadcastRequest{ + Audience: strings.TrimSpace(r.URL.Query().Get("audience")), + CityCode: strings.TrimSpace(r.URL.Query().Get("cityCode")), + } + if raw := strings.TrimSpace(r.URL.Query().Get("userIds")); raw != "" { + for _, part := range strings.Split(raw, ",") { + if id, convErr := strconv.ParseInt(strings.TrimSpace(part), 10, 64); convErr == nil { + req.UserIDs = append(req.UserIDs, id) + } + } + } + query, args, err := a.broadcastAudience(req) + if err != nil { + fail(w, http.StatusBadRequest, 20001, err.Error()) + return + } + var count int + if a.db.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM (`+query+`) AS audience`, args...).Scan(&count) != nil { + fail(w, http.StatusInternalServerError, 50001, "预览失败") + return + } + reply(w, map[string]any{"count": count, "limit": a.configInt(r.Context(), "notification.broadcast_max_recipients", 20000)}) +} diff --git a/im/backend/internal/app/admin_finance.go b/im/backend/internal/app/admin_finance.go new file mode 100644 index 0000000..6362982 --- /dev/null +++ b/im/backend/internal/app/admin_finance.go @@ -0,0 +1,226 @@ +package app + +import ( + "fmt" + "net/http" + "strings" + "time" +) + +// The money view operations actually needs: what has been taken in, what is +// still held on someone's behalf, what has been paid out, and whether the two +// sides of the book still agree. Split across four pages it was impossible to +// answer "how much of our balance is other people's money" — that is the +// number this page exists for. + +func (a *App) adminFinanceOverview(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + days := 14 + if value := strings.TrimSpace(r.URL.Query().Get("days")); value != "" { + if parsed, err := fmt.Sscanf(value, "%d", &days); parsed != 1 || err != nil || days < 1 || days > 90 { + days = 14 + } + } + + money := map[string]int64{ + // 托管中 = 已经从主人那里收到、还没结算也没退的钱。这是平台账上"别人的钱"。 + "escrowedCent": a.countOne(ctx, `SELECT COALESCE(SUM(gross_cent-refunded_cent),0) FROM pet_feed_tasks + WHERE status IN ('ESCROWED','ASSIGNED','SERVING','COMPLETED','DISPUTED')`), + "settledCent": a.countOne(ctx, `SELECT COALESCE(SUM(payout_cent),0) FROM pet_feed_tasks WHERE status='SETTLED'`), + "platformFeeCent": a.countOne(ctx, `SELECT COALESCE(SUM(platform_fee_cent),0) FROM pet_feed_tasks WHERE status='SETTLED'`), + "tipCent": a.countOne(ctx, `SELECT COALESCE(SUM(tip_cent),0) FROM pet_feed_tasks WHERE status IN ('SETTLED','COMPLETED','SERVING','ASSIGNED','ESCROWED')`), + "walletBalanceCent": a.countOne(ctx, `SELECT COALESCE(SUM(available_cent),0) FROM wallet_accounts`), + "pendingPayoutCent": a.countOne(ctx, `SELECT COALESCE(SUM(amount_cent),0) FROM withdrawals WHERE status IN ('PENDING','APPROVED')`), + "paidPayoutCent": a.countOne(ctx, `SELECT COALESCE(SUM(payout_cent),0) FROM withdrawals WHERE status='PAID'`), + "orderPaidCent": a.countOne(ctx, `SELECT COALESCE(SUM(amount_cent),0) FROM orders WHERE status IN ('PAID','REFUNDING') AND deleted_at IS NULL`), + "orderRefundedCent": a.countOne(ctx, `SELECT COALESCE(SUM(refunded_cent),0) FROM orders WHERE deleted_at IS NULL`), + "feedRefundedCent": a.countOne(ctx, `SELECT COALESCE(SUM(refunded_cent),0) FROM pet_feed_tasks`), + "unpaidOrderCent": a.countOne(ctx, `SELECT COALESCE(SUM(amount_cent),0) FROM orders WHERE status='CREATED' AND deleted_at IS NULL`), + "platformOwesCent": a.countOne(ctx, `SELECT COALESCE(SUM(available_cent),0) FROM wallet_accounts`), + "disputeHeldCent": a.countOne(ctx, `SELECT COALESCE(SUM(gross_cent-refunded_cent),0) FROM pet_feed_tasks WHERE status='DISPUTED'`), + "ledgerMismatchUser": a.countOne(ctx, `SELECT COUNT(*) FROM (SELECT a.user_id FROM wallet_accounts a + LEFT JOIN wallet_transactions t ON t.user_id=a.user_id GROUP BY a.user_id,a.available_cent + HAVING a.available_cent<>COALESCE(SUM(t.amount_cent*t.direction),0)) AS drift`), + } + + // 按产品看收入：会员、代喂、保证金各自多少，退了多少。 + products := []map[string]any{} + rows, err := a.db.QueryContext(ctx, `SELECT product_type,COUNT(*),COALESCE(SUM(amount_cent),0),COALESCE(SUM(refunded_cent),0) + FROM orders WHERE status IN ('PAID','REFUNDING','REFUNDED') AND deleted_at IS NULL GROUP BY product_type`) + if err == nil { + defer rows.Close() + for rows.Next() { + var productType string + var count, amount, refunded int64 + if rows.Scan(&productType, &count, &amount, &refunded) == nil { + products = append(products, map[string]any{ + "productType": productType, "orderCount": count, + "amountCent": amount, "refundedCent": refunded, + }) + } + } + } + + // 按日趋势：收进来的和退出去的放在一起看，才能看出某天是不是在净流出。 + trend := []map[string]any{} + trendRows, err := a.db.QueryContext(ctx, `SELECT DATE(paid_at) AS day,COALESCE(SUM(amount_cent),0),COUNT(*) + FROM orders WHERE paid_at IS NOT NULL AND paid_at>=DATE_SUB(CURDATE(),INTERVAL ? DAY) AND deleted_at IS NULL + GROUP BY DATE(paid_at) ORDER BY day`, days) + if err == nil { + defer trendRows.Close() + for trendRows.Next() { + var day time.Time + var amount, count int64 + if trendRows.Scan(&day, &amount, &count) == nil { + trend = append(trend, map[string]any{ + "date": day.Format("2006-01-02"), "amountCent": amount, "orderCount": count, + }) + } + } + } + + reply(w, map[string]any{ + "money": money, "products": products, "trend": trend, "days": days, + "withdrawEnabled": a.configBool(ctx, "pet.withdraw_enabled", false), + "paymentMode": a.configPlain(ctx, "payment.mode", "sandbox"), + }) +} + +// adminOrderRefunds lists what has already been sent back on one order. +func (a *App) adminOrderRefunds(w http.ResponseWriter, r *http.Request) { + id, err := pathID(r) + if err != nil { + fail(w, http.StatusBadRequest, 20001, "编号无效") + return + } + rows, queryErr := a.db.QueryContext(r.Context(), `SELECT f.id,f.amount_cent,f.channel,f.reason,f.created_at, + COALESCE(admin.username,'') FROM order_refunds f LEFT JOIN admin_users admin ON admin.id=f.handled_by + WHERE f.order_id=? ORDER BY f.id DESC`, id) + if queryErr != nil { + fail(w, http.StatusInternalServerError, 50001, "查询退款记录失败") + return + } + defer rows.Close() + items := []map[string]any{} + for rows.Next() { + var refundID, amount int64 + var channel, reason, operator string + var created time.Time + if rows.Scan(&refundID, &amount, &channel, &reason, &created, &operator) != nil { + continue + } + items = append(items, map[string]any{ + "id": refundID, "amountCent": amount, "channel": channel, "reason": reason, + "operator": operator, "createdAt": created.Format(time.RFC3339), + }) + } + reply(w, map[string]any{"items": items}) +} + +// adminRefundOrder is the one refund entry point for every product. A partial +// refund leaves what it bought alone; only refunding the last cent revokes it, +// because half a membership is still a membership. +func (a *App) adminRefundOrder(w http.ResponseWriter, r *http.Request) { + id, err := pathID(r) + if err != nil { + fail(w, http.StatusBadRequest, 20001, "订单编号无效") + return + } + var req struct { + AmountCent int64 `json:"amountCent"` + Reason string `json:"reason"` + Offline bool `json:"offline"` + } + if r.ContentLength > 0 && decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "参数错误") + return + } + req.Reason = strings.TrimSpace(req.Reason) + if req.Reason == "" { + req.Reason = "客服操作退款" + } + + var userID, productID int64 + var amount, refunded int + var orderNo, providerOrderNo, status, productType string + if a.db.QueryRowContext(r.Context(), `SELECT user_id,product_type,product_id,amount_cent,COALESCE(refunded_cent,0), + order_no,provider_order_no,status FROM orders WHERE id=? AND deleted_at IS NULL`, id). + Scan(&userID, &productType, &productID, &amount, &refunded, &orderNo, &providerOrderNo, &status) != nil { + fail(w, http.StatusNotFound, 30001, "订单不存在") + return + } + if status != "PAID" && status != "REFUND_REQUESTED" && status != "REFUNDING" && status != "REFUNDED" { + fail(w, http.StatusBadRequest, 20001, "只有已支付的订单可以退款") + return + } + remaining := int64(amount - refunded) + if remaining <= 0 { + fail(w, http.StatusBadRequest, 20001, "这笔订单已经退完了") + return + } + refundAmount := req.AmountCent + if refundAmount <= 0 { + refundAmount = remaining + } + if refundAmount > remaining { + fail(w, http.StatusBadRequest, 20001, fmt.Sprintf("最多还能退 %.2f 元", float64(remaining)/100)) + return + } + + live := a.configPlain(r.Context(), "payment.mode", "sandbox") == "live" + channel := "gateway" + if req.Offline || providerOrderNo == "" || strings.HasPrefix(providerOrderNo, "sandbox:") || !live { + channel = "offline" + } + refundNo := fmt.Sprintf("order:%d:%d", id, time.Now().UnixMilli()) + if channel == "gateway" { + if err = a.createGatewayRefund(r.Context(), orderNo, providerOrderNo, refundNo, int(refundAmount)); err != nil { + a.audit(r, "refund_failed", "order", id, map[string]any{"amountCent": refundAmount, "error": err.Error()}) + fail(w, http.StatusBadGateway, 50003, "退款网关请求失败，未记账，可以重试或改为线下退款登记") + return + } + } + + tx, err := a.db.BeginTx(r.Context(), nil) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "退款记账失败") + return + } + defer func() { _ = tx.Rollback() }() + if _, err = tx.ExecContext(r.Context(), `INSERT INTO order_refunds(order_id,refund_no,user_id,amount_cent,channel,reason,handled_by) + VALUES(?,?,?,?,?,?,?)`, id, refundNo, userID, refundAmount, channel, req.Reason, current(r).ID); err != nil { + fail(w, http.StatusInternalServerError, 50001, "退款记账失败") + return + } + full := int64(refunded)+refundAmount >= int64(amount) + newStatus := status + if full { + newStatus = "REFUNDED" + } + if _, err = tx.ExecContext(r.Context(), `UPDATE orders SET refunded_cent=refunded_cent+?,status=? WHERE id=?`, + refundAmount, newStatus, id); err != nil { + fail(w, http.StatusInternalServerError, 50001, "退款记账失败") + return + } + // 退完才收回买到的东西：半份会员仍然是会员，代喂任务同理。 + if full { + if err = a.revokePaidOrderTx(r.Context(), tx, id, userID, productType, productID, int(refundAmount)); err != nil { + fail(w, http.StatusBadRequest, 20001, err.Error()) + return + } + } + if tx.Commit() != nil { + fail(w, http.StatusInternalServerError, 50001, "退款记账失败") + return + } + + a.audit(r, "refund", "order", id, map[string]any{ + "amountCent": refundAmount, "channel": channel, "full": full, "reason": req.Reason, "refundNo": refundNo, + }) + a.notifyUser(r.Context(), userID, "system", "订单已退款", + fmt.Sprintf("订单 %s 已退款 %.2f 元：%s", orderNo, float64(refundAmount)/100, req.Reason), "order", id) + reply(w, map[string]any{ + "success": true, "amountCent": refundAmount, "channel": channel, + "refundedCent": int64(refunded) + refundAmount, "remainingCent": remaining - refundAmount, "full": full, + }) +} diff --git a/im/backend/internal/app/admin_operations_integration_test.go b/im/backend/internal/app/admin_operations_integration_test.go new file mode 100644 index 0000000..85f7d30 --- /dev/null +++ b/im/backend/internal/app/admin_operations_integration_test.go @@ -0,0 +1,305 @@ +package app + +import ( + "context" + "encoding/json" + "fmt" + "net/http/httptest" + "strings" + "testing" +) + +// 这一批补的都是"管理端原来只能看"的地方，所以断言集中在两件事上： +// 动作真的落到了数据上，以及每个动作都留下了能复核的痕迹。 + +func TestAdminBroadcastMySQL(t *testing.T) { + db := isolatedIMDatabase(t) + a := &App{db: db, hub: NewHub(), config: Config{Environment: "development", JWTSecret: "isolated-im-regression-secret-only"}} + + t.Run("an announcement reaches the chosen audience and nobody else", func(t *testing.T) { + if _, err := db.Exec(`UPDATE user_profiles SET is_vip=1 WHERE user_id=2`); err != nil { + t.Fatal(err) + } + preview := httptest.NewRecorder() + a.adminBroadcastPreview(preview, imTestRequest(1, "GET", "/admin/v1/notifications/broadcasts/preview?audience=vip", "")) + if preview.Code != 200 || !strings.Contains(preview.Body.String(), `"count":1`) { + t.Fatalf("会员人群预览应当是 1 人: %s", preview.Body.String()) + } + send := httptest.NewRecorder() + a.adminSendBroadcast(send, imTestRequest(1, "POST", "/admin/v1/notifications/broadcasts", + `{"title":"会员权益调整","content":"从下周起会员每天的主动搭讪额度提高到 30 次。","audience":"vip"}`)) + if send.Code != 200 { + t.Fatalf("HTTP %d: %s", send.Code, send.Body.String()) + } + var envelope struct { + Data struct { + ID int64 `json:"id"` + RecipientCount int `json:"recipientCount"` + } `json:"data"` + } + _ = json.Unmarshal(send.Body.Bytes(), &envelope) + if envelope.Data.RecipientCount != 1 { + t.Fatalf("收件人数 = %d", envelope.Data.RecipientCount) + } + var forVIP, forOthers int + if err := db.QueryRow(`SELECT COUNT(*) FROM notifications WHERE biz_type='broadcast' AND user_id=2`).Scan(&forVIP); err != nil { + t.Fatal(err) + } + if err := db.QueryRow(`SELECT COUNT(*) FROM notifications WHERE biz_type='broadcast' AND user_id<>2`).Scan(&forOthers); err != nil { + t.Fatal(err) + } + if forVIP != 1 || forOthers != 0 { + t.Fatalf("只有会员该收到: vip=%d others=%d", forVIP, forOthers) + } + }) + + t.Run("empty or oversized announcements are refused", func(t *testing.T) { + for _, body := range []string{ + `{"title":"","content":"内容够长了但没有标题","audience":"all"}`, + `{"title":"太短","content":"两个字","audience":"all"}`, + `{"title":"人群不存在","content":"这条应该发不出去","audience":"martians"}`, + `{"title":"没选人","content":"指定用户但一个都没给","audience":"users"}`, + } { + w := httptest.NewRecorder() + a.adminSendBroadcast(w, imTestRequest(1, "POST", "/admin/v1/notifications/broadcasts", body)) + if w.Code != 400 { + t.Fatalf("应当被拒绝: %s -> HTTP %d", body, w.Code) + } + } + }) + + t.Run("revoking takes back only what nobody has read", func(t *testing.T) { + send := httptest.NewRecorder() + a.adminSendBroadcast(send, imTestRequest(1, "POST", "/admin/v1/notifications/broadcasts", + `{"title":"系统维护","content":"今晚 23:00 到 24:00 之间服务会短暂中断，给你带来不便很抱歉。","audience":"all"}`)) + if send.Code != 200 { + t.Fatalf("HTTP %d: %s", send.Code, send.Body.String()) + } + var envelope struct { + Data struct { + ID int64 `json:"id"` + } `json:"data"` + } + _ = json.Unmarshal(send.Body.Bytes(), &envelope) + // 一个人已经读了，撤回不该把他那条也抹掉。 + if _, err := db.Exec(`UPDATE notifications SET read_at=NOW(3) WHERE biz_type='broadcast' AND biz_id=? AND user_id=1`, envelope.Data.ID); err != nil { + t.Fatal(err) + } + revoke := httptest.NewRecorder() + a.adminRevokeBroadcast(revoke, imTestRequest(1, "POST", fmt.Sprintf("/admin/v1/notifications/broadcasts/%d/revoke", envelope.Data.ID), ``)) + if revoke.Code != 200 { + t.Fatalf("HTTP %d: %s", revoke.Code, revoke.Body.String()) + } + var left int + if err := db.QueryRow(`SELECT COUNT(*) FROM notifications WHERE biz_type='broadcast' AND biz_id=?`, envelope.Data.ID).Scan(&left); err != nil { + t.Fatal(err) + } + if left != 1 { + t.Fatalf("已读的那条应当留着: 还剩 %d 条", left) + } + again := httptest.NewRecorder() + a.adminRevokeBroadcast(again, imTestRequest(1, "POST", fmt.Sprintf("/admin/v1/notifications/broadcasts/%d/revoke", envelope.Data.ID), ``)) + if again.Code != 400 { + t.Fatalf("重复撤回应当被拒绝: HTTP %d", again.Code) + } + }) +} + +func TestAdminKeywordsMySQL(t *testing.T) { + db := isolatedIMDatabase(t) + a := &App{db: db, hub: NewHub(), config: Config{Environment: "development", JWTSecret: "isolated-im-regression-secret-only"}} + invalidateKeywordCache() + + create := httptest.NewRecorder() + a.adminCreateKeyword(create, imTestRequest(1, "POST", "/admin/v1/moderation/keywords", + `{"word":"加我微信","scope":"message","action":"block","category":"引流"}`)) + if create.Code != 200 { + t.Fatalf("HTTP %d: %s", create.Code, create.Body.String()) + } + duplicate := httptest.NewRecorder() + a.adminCreateKeyword(duplicate, imTestRequest(1, "POST", "/admin/v1/moderation/keywords", + `{"word":"加我微信","scope":"message","action":"review"}`)) + if duplicate.Code != 400 { + t.Fatalf("同范围下的重复词应当被拒绝: HTTP %d", duplicate.Code) + } + review := httptest.NewRecorder() + a.adminCreateKeyword(review, imTestRequest(1, "POST", "/admin/v1/moderation/keywords", + `{"word":"兼职","scope":"all","action":"review","category":"疑似广告"}`)) + if review.Code != 200 { + t.Fatalf("HTTP %d: %s", review.Code, review.Body.String()) + } + invalidateKeywordCache() + + t.Run("blocking beats review, and scope is respected", func(t *testing.T) { + word, action := a.matchKeyword(context.Background(), "message", "你好，加我微信聊，有兼职") + if action != "block" { + t.Fatalf("同时命中拦截与转人工时应当以拦截为准: %s %s", word, action) + } + word, action = a.matchKeyword(context.Background(), "post", "招人做兼职") + if word != "兼职" || action != "review" { + t.Fatalf("全局词应当在动态里生效: %q %q", word, action) + } + // message 范围的词不该影响动态。 + if word, _ = a.matchKeyword(context.Background(), "post", "加我微信"); word != "" { + t.Fatalf("只对消息生效的词泄漏到了动态: %q", word) + } + }) + + t.Run("a message that trips a blocking word never reaches the database", func(t *testing.T) { + // 用真实的会话入口建会话，免得测试自己拼一套和线上不一样的数据。 + var created struct { + ID int64 `json:"id"` + } + if err := json.Unmarshal(imTestCall(t, a.directConversation, 1, "POST", "/api/v1/im/conversations", + `{"userId":2}`, 200), &created); err != nil { + t.Fatal(err) + } + conversation := created.ID + _, _, err := a.persistMessageContext(context.Background(), conversation, 1, "client-block-1", 1, + map[string]any{"text": "加我微信详聊"}) + if err == nil { + t.Fatal("命中拦截词的消息不该写进去") + } + var stored int + if queryErr := db.QueryRow(`SELECT COUNT(*) FROM im_messages WHERE conversation_id=?`, conversation).Scan(&stored); queryErr != nil { + t.Fatal(queryErr) + } + if stored != 0 { + t.Fatalf("库里不该有这条消息: %d", stored) + } + // 转人工的词照常发出去，但要在风险中心留一条。 + if _, _, err = a.persistMessageContext(context.Background(), conversation, 1, "client-review-1", 1, + map[string]any{"text": "有兼职可以做吗"}); err != nil { + t.Fatalf("转人工的词不该拦下消息: %v", err) + } + var events int + if queryErr := db.QueryRow(`SELECT COUNT(*) FROM risk_events WHERE user_id=1 AND event_type='keyword_message'`).Scan(&events); queryErr != nil { + t.Fatal(queryErr) + } + if events != 1 { + t.Fatalf("风险事件 = %d，转人工的命中应当留痕", events) + } + }) + + t.Run("the test box answers before anyone complains", func(t *testing.T) { + w := httptest.NewRecorder() + a.adminTestKeyword(w, imTestRequest(1, "POST", "/admin/v1/moderation/keywords/test", + `{"scope":"message","text":"加我微信"}`)) + if w.Code != 200 || !strings.Contains(w.Body.String(), `"action":"block"`) { + t.Fatalf("HTTP %d: %s", w.Code, w.Body.String()) + } + }) +} + +func TestAdminRiskAndBatchMySQL(t *testing.T) { + db := isolatedIMDatabase(t) + a := &App{db: db, hub: NewHub(), config: Config{Environment: "development", JWTSecret: "isolated-im-regression-secret-only"}} + + if _, err := db.Exec(`INSERT INTO user_risk_profiles(user_id,risk_score,risk_level) VALUES(2,80,2)`); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`INSERT INTO risk_events(user_id,event_type,score_delta) VALUES(2,'spam_message',10)`); err != nil { + t.Fatal(err) + } + + t.Run("whitelisting needs a reason and is written down", func(t *testing.T) { + bare := httptest.NewRecorder() + a.adminWhitelistRiskUser(bare, imTestRequest(1, "POST", "/admin/v1/risk/users/2/whitelist", `{"on":true}`)) + if bare.Code != 400 { + t.Fatalf("加白必须写理由: HTTP %d", bare.Code) + } + w := httptest.NewRecorder() + a.adminWhitelistRiskUser(w, imTestRequest(1, "POST", "/admin/v1/risk/users/2/whitelist", + `{"on":true,"note":"官方运营账号，批量发消息属正常"}`)) + if w.Code != 200 { + t.Fatalf("HTTP %d: %s", w.Code, w.Body.String()) + } + var whitelisted bool + var note string + if err := db.QueryRow(`SELECT whitelisted,whitelist_note FROM user_risk_profiles WHERE user_id=2`).Scan(&whitelisted, ¬e); err != nil { + t.Fatal(err) + } + if !whitelisted || note == "" { + t.Fatalf("whitelisted=%v note=%q", whitelisted, note) + } + var logged int + if err := db.QueryRow(`SELECT COUNT(*) FROM admin_audit_logs WHERE action='risk_whitelist' AND target_id=2`).Scan(&logged); err != nil { + t.Fatal(err) + } + if logged != 1 { + t.Fatalf("加白要留审计: %d", logged) + } + }) + + t.Run("an event can be closed once", func(t *testing.T) { + var eventID int64 + if err := db.QueryRow(`SELECT id FROM risk_events WHERE user_id=2`).Scan(&eventID); err != nil { + t.Fatal(err) + } + w := httptest.NewRecorder() + a.adminHandleRiskEvent(w, imTestRequest(1, "POST", fmt.Sprintf("/admin/v1/risk/events/%d/handle", eventID), `{"note":"核实是正常运营行为"}`)) + if w.Code != 200 { + t.Fatalf("HTTP %d: %s", w.Code, w.Body.String()) + } + again := httptest.NewRecorder() + a.adminHandleRiskEvent(again, imTestRequest(1, "POST", fmt.Sprintf("/admin/v1/risk/events/%d/handle", eventID), `{"note":"再点一次"}`)) + if again.Code != 400 { + t.Fatalf("重复处理应当被拒绝: HTTP %d", again.Code) + } + }) + + t.Run("batch actions are capped and leave a trace", func(t *testing.T) { + ids := []int64{} + for index := 0; index < 3; index++ { + result, err := db.Exec(`INSERT INTO posts(user_id,content,visibility) VALUES(1,?,1)`, fmt.Sprintf("批量测试 %d", index)) + if err != nil { + t.Fatal(err) + } + id, _ := result.LastInsertId() + ids = append(ids, id) + } + payload, _ := json.Marshal(map[string]any{"ids": ids, "action": "remove", "reason": "批量清理垃圾内容"}) + w := httptest.NewRecorder() + a.adminBatchPosts(w, imTestRequest(1, "POST", "/admin/v1/posts/batch", string(payload))) + if w.Code != 200 || !strings.Contains(w.Body.String(), `"affected":3`) { + t.Fatalf("HTTP %d: %s", w.Code, w.Body.String()) + } + var live int + if err := db.QueryRow(`SELECT COUNT(*) FROM posts WHERE status=1`).Scan(&live); err != nil { + t.Fatal(err) + } + if live != 0 { + t.Fatalf("还剩 %d 条没下架", live) + } + // 恢复也走同一个入口。 + restore, _ := json.Marshal(map[string]any{"ids": ids, "action": "restore"}) + back := httptest.NewRecorder() + a.adminBatchPosts(back, imTestRequest(1, "POST", "/admin/v1/posts/batch", string(restore))) + if back.Code != 200 { + t.Fatalf("HTTP %d: %s", back.Code, back.Body.String()) + } + // 空选择与超量都要挡住。 + empty := httptest.NewRecorder() + a.adminBatchPosts(empty, imTestRequest(1, "POST", "/admin/v1/posts/batch", `{"ids":[],"action":"remove"}`)) + if empty.Code != 400 { + t.Fatalf("空勾选应当被拒绝: HTTP %d", empty.Code) + } + many := make([]int64, 101) + for index := range many { + many[index] = int64(index + 1) + } + oversized, _ := json.Marshal(map[string]any{"ids": many, "action": "remove"}) + over := httptest.NewRecorder() + a.adminBatchPosts(over, imTestRequest(1, "POST", "/admin/v1/posts/batch", string(oversized))) + if over.Code != 400 { + t.Fatalf("超过 100 条应当被拒绝: HTTP %d", over.Code) + } + var logged int + if err := db.QueryRow(`SELECT COUNT(*) FROM admin_audit_logs WHERE action LIKE 'batch_%'`).Scan(&logged); err != nil { + t.Fatal(err) + } + if logged < 2 { + t.Fatalf("批量操作要留审计: %d", logged) + } + }) +} diff --git a/im/backend/internal/app/admin_risk_batch.go b/im/backend/internal/app/admin_risk_batch.go new file mode 100644 index 0000000..a224517 --- /dev/null +++ b/im/backend/internal/app/admin_risk_batch.go @@ -0,0 +1,247 @@ +package app + +import ( + "fmt" + "net/http" + "strings" +) + +// Two things the console could not do: act on a risk score, and act on more +// than one row at a time. A moderator looking at forty spam posts had to click +// forty times, which is how backlogs turn into "we stopped looking". + +// --- 风险中心 ------------------------------------------------------------- + +func (a *App) adminWhitelistRiskUser(w http.ResponseWriter, r *http.Request) { + id, err := pathID(r) + if err != nil { + fail(w, http.StatusBadRequest, 20001, "编号无效") + return + } + var req struct { + On bool `json:"on"` + Note string `json:"note"` + } + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "参数错误") + return + } + req.Note = strings.TrimSpace(req.Note) + // 加白等于对后面所有告警说"这个人我看过了"，没有理由就没法复核。 + if req.On && req.Note == "" { + fail(w, http.StatusBadRequest, 20001, "加白必须写明理由") + return + } + if len([]rune(req.Note)) > 200 { + fail(w, http.StatusBadRequest, 20001, "理由过长") + return + } + result, err := a.db.ExecContext(r.Context(), `UPDATE user_risk_profiles + SET whitelisted=?,whitelist_note=?,reviewed_by=?,reviewed_at=NOW(3) WHERE user_id=?`, + req.On, req.Note, current(r).ID, id) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "操作失败") + return + } + if affected, _ := result.RowsAffected(); affected == 0 { + fail(w, http.StatusNotFound, 30001, "这个用户没有风险档案") + return + } + a.audit(r, "risk_whitelist", "user", id, map[string]any{"on": req.On, "note": req.Note}) + reply(w, map[string]bool{"success": true}) +} + +func (a *App) adminHandleRiskEvent(w http.ResponseWriter, r *http.Request) { + id, err := pathID(r) + if err != nil { + fail(w, http.StatusBadRequest, 20001, "编号无效") + return + } + var req struct { + Note string `json:"note"` + } + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "参数错误") + return + } + req.Note = strings.TrimSpace(req.Note) + if len([]rune(req.Note)) > 200 { + fail(w, http.StatusBadRequest, 20001, "备注过长") + return + } + result, err := a.db.ExecContext(r.Context(), `UPDATE risk_events SET handled_at=NOW(3),handled_by=?,handled_note=? + WHERE id=? AND handled_at IS NULL`, current(r).ID, req.Note, id) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "操作失败") + return + } + if affected, _ := result.RowsAffected(); affected == 0 { + fail(w, http.StatusBadRequest, 20001, "这条事件已经处理过了") + return + } + a.audit(r, "risk_event_handled", "risk_event", id, map[string]any{"note": req.Note}) + reply(w, map[string]bool{"success": true}) +} + +// --- 批量处置 ------------------------------------------------------------- + +type batchRequest struct { + IDs []int64 `json:"ids"` + Action string `json:"action"` + Reason string `json:"reason"` +} + +func validBatch(w http.ResponseWriter, req *batchRequest, allowed map[string]bool) bool { + req.Reason = strings.TrimSpace(req.Reason) + if len(req.IDs) == 0 { + fail(w, http.StatusBadRequest, 20001, "请先勾选要处理的内容") + return false + } + // 一次几百条的批量下架，点错一次就要一条条撤回；限量能把误操作的代价压住。 + if len(req.IDs) > 100 { + fail(w, http.StatusBadRequest, 20001, "一次最多处理 100 条") + return false + } + if !allowed[req.Action] { + fail(w, http.StatusBadRequest, 20001, "操作无效") + return false + } + if len([]rune(req.Reason)) > 200 { + fail(w, http.StatusBadRequest, 20001, "原因过长") + return false + } + return true +} + +func placeholdersFor(ids []int64) (string, []any) { + marks := make([]string, 0, len(ids)) + args := make([]any, 0, len(ids)) + for _, id := range ids { + marks = append(marks, "?") + args = append(args, id) + } + return strings.Join(marks, ","), args +} + +func (a *App) adminBatchPosts(w http.ResponseWriter, r *http.Request) { + var req batchRequest + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "参数错误") + return + } + if !validBatch(w, &req, map[string]bool{"remove": true, "restore": true}) { + return + } + marks, args := placeholdersFor(req.IDs) + query := `UPDATE posts SET status=0,deleted_at=NOW(3) WHERE status=1 AND id IN (` + marks + `)` + if req.Action == "restore" { + query = `UPDATE posts SET status=1,deleted_at=NULL WHERE status=0 AND id IN (` + marks + `)` + } + result, err := a.db.ExecContext(r.Context(), query, args...) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "批量操作失败") + return + } + affected, _ := result.RowsAffected() + a.audit(r, "batch_"+req.Action, "post", 0, map[string]any{"ids": req.IDs, "affected": affected, "reason": req.Reason}) + reply(w, map[string]any{"success": true, "affected": affected}) +} + +func (a *App) adminBatchMessages(w http.ResponseWriter, r *http.Request) { + var req batchRequest + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "参数错误") + return + } + if !validBatch(w, &req, map[string]bool{"remove": true, "restore": true}) { + return + } + if req.Action == "remove" && req.Reason == "" { + fail(w, http.StatusBadRequest, 20001, "批量下架消息必须写明原因") + return + } + marks, args := placeholdersFor(req.IDs) + query := `UPDATE im_messages SET admin_removed_at=NOW(3),admin_remove_reason=? WHERE admin_removed_at IS NULL AND id IN (` + marks + `)` + params := append([]any{req.Reason}, args...) + if req.Action == "restore" { + query = `UPDATE im_messages SET admin_removed_at=NULL,admin_remove_reason='' WHERE admin_removed_at IS NOT NULL AND id IN (` + marks + `)` + params = args + } + result, err := a.db.ExecContext(r.Context(), query, params...) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "批量操作失败") + return + } + affected, _ := result.RowsAffected() + a.audit(r, "batch_"+req.Action, "message", 0, map[string]any{"ids": req.IDs, "affected": affected, "reason": req.Reason}) + reply(w, map[string]any{"success": true, "affected": affected}) +} + +// adminBatchReports only closes reports without punishing anyone. Sanctions +// stay one at a time on purpose: batching a punishment is how the wrong person +// gets banned. +func (a *App) adminBatchReports(w http.ResponseWriter, r *http.Request) { + var req batchRequest + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "参数错误") + return + } + if !validBatch(w, &req, map[string]bool{"dismiss": true}) { + return + } + if req.Reason == "" { + fail(w, http.StatusBadRequest, 20001, "批量驳回必须写明理由") + return + } + marks, args := placeholdersFor(req.IDs) + params := append([]any{req.Reason, current(r).ID}, args...) + result, err := a.db.ExecContext(r.Context(), `UPDATE reports SET status='REJECTED',action_type='NONE',handle_remark=?, + handled_by=?,handled_at=NOW(3) WHERE status='PENDING' AND id IN (`+marks+`)`, params...) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "批量操作失败") + return + } + affected, _ := result.RowsAffected() + a.audit(r, "batch_dismiss", "report", 0, map[string]any{"ids": req.IDs, "affected": affected, "reason": req.Reason}) + reply(w, map[string]any{"success": true, "affected": affected}) +} + +// adminBatchPetListings mirrors the same shape for the pet module queue. +func (a *App) adminBatchPetListings(w http.ResponseWriter, r *http.Request) { + var req struct { + batchRequest + Kind string `json:"kind"` + } + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "参数错误") + return + } + table, ok := petListingTables[strings.TrimSpace(req.Kind)] + if !ok { + fail(w, http.StatusBadRequest, 20001, "类型无效") + return + } + if !validBatch(w, &req.batchRequest, map[string]bool{"approve": true, "reject": true}) { + return + } + if req.Action == "reject" && req.Reason == "" { + fail(w, http.StatusBadRequest, 20001, "批量驳回必须写明原因") + return + } + marks, args := placeholdersFor(req.IDs) + status := 1 + if req.Action == "reject" { + status = 2 + } + params := append([]any{status, req.Reason}, args...) + result, err := a.db.ExecContext(r.Context(), fmt.Sprintf( + `UPDATE %s SET moderation_status=?,moderation_note=? WHERE moderation_status=0 AND deleted_at IS NULL AND id IN (%s)`, + table, marks), params...) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "批量操作失败") + return + } + affected, _ := result.RowsAffected() + a.audit(r, "batch_"+req.Action, "pet_"+req.Kind, 0, map[string]any{"ids": req.IDs, "affected": affected, "reason": req.Reason}) + reply(w, map[string]any{"success": true, "affected": affected}) +} diff --git a/im/backend/internal/app/app.go b/im/backend/internal/app/app.go index 1240749..d0ac035 100644 --- a/im/backend/internal/app/app.go +++ b/im/backend/internal/app/app.go @@ -353,6 +353,17 @@ func (a *App) adminRoutes() []rest.Route { {Method: http.MethodPost, Path: "/admin/v1/reports/:id/handle", Handler: permit("reports:handle", a.adminHandleReport)}, {Method: http.MethodGet, Path: "/admin/v1/risk/users", Handler: permit("risk:view", a.adminRiskUsers)}, {Method: http.MethodGet, Path: "/admin/v1/risk/events", Handler: permit("risk:view", a.adminRiskEvents)}, + {Method: http.MethodPost, Path: "/admin/v1/risk/users/:id/whitelist", Handler: permit("risk:view", a.adminWhitelistRiskUser)}, + {Method: http.MethodPost, Path: "/admin/v1/risk/events/:id/handle", Handler: permit("risk:view", a.adminHandleRiskEvent)}, + {Method: http.MethodPost, Path: "/admin/v1/posts/batch", Handler: permit("content:manage", a.adminBatchPosts)}, + {Method: http.MethodPost, Path: "/admin/v1/messages/batch", Handler: permit("messages:manage", a.adminBatchMessages)}, + {Method: http.MethodPost, Path: "/admin/v1/reports/batch", Handler: permit("reports:handle", a.adminBatchReports)}, + {Method: http.MethodPost, Path: "/admin/v1/pet/listings/batch", Handler: permit("users:manage", a.adminBatchPetListings)}, + {Method: http.MethodGet, Path: "/admin/v1/moderation/keywords", Handler: permit("system:manage", a.adminKeywords)}, + {Method: http.MethodPost, Path: "/admin/v1/moderation/keywords", Handler: permit("system:manage", a.adminCreateKeyword)}, + {Method: http.MethodPut, Path: "/admin/v1/moderation/keywords/:id", Handler: permit("system:manage", a.adminUpdateKeyword)}, + {Method: http.MethodDelete, Path: "/admin/v1/moderation/keywords/:id", Handler: permit("system:manage", a.adminDeleteKeyword)}, + {Method: http.MethodPost, Path: "/admin/v1/moderation/keywords/test", Handler: permit("system:manage", a.adminTestKeyword)}, {Method: http.MethodGet, Path: "/admin/v1/membership/plans", Handler: permit("membership:manage", a.adminPlans)}, {Method: http.MethodPost, Path: "/admin/v1/membership/plans", Handler: permit("membership:manage", a.adminCreatePlan)}, {Method: http.MethodPatch, Path: "/admin/v1/membership/plans/:id", Handler: permit("membership:manage", a.adminUpdatePlan)}, @@ -363,7 +374,13 @@ func (a *App) adminRoutes() []rest.Route { {Method: http.MethodDelete, Path: "/admin/v1/orders/:id", Handler: permit("orders:manage", a.adminDeleteOrder)}, {Method: http.MethodPost, Path: "/admin/v1/orders/:id/pay", Handler: permit("orders:manage", a.adminMarkOrderPaid)}, {Method: http.MethodPost, Path: "/admin/v1/orders/:id/close", Handler: permit("orders:manage", a.adminCloseOrder)}, - {Method: http.MethodPost, Path: "/admin/v1/orders/:id/refund", Handler: permit("orders:manage", a.adminRefund)}, + {Method: http.MethodPost, Path: "/admin/v1/orders/:id/refund", Handler: permit("orders:manage", a.adminRefundOrder)}, + {Method: http.MethodGet, Path: "/admin/v1/orders/:id/refunds", Handler: permit("orders:view", a.adminOrderRefunds)}, + {Method: http.MethodGet, Path: "/admin/v1/finance/overview", Handler: permit("orders:view", a.adminFinanceOverview)}, + {Method: http.MethodGet, Path: "/admin/v1/notifications/broadcasts", Handler: permit("users:view", a.adminBroadcasts)}, + {Method: http.MethodGet, Path: "/admin/v1/notifications/broadcasts/preview", Handler: permit("users:view", a.adminBroadcastPreview)}, + {Method: http.MethodPost, Path: "/admin/v1/notifications/broadcasts", Handler: permit("users:manage", a.adminSendBroadcast)}, + {Method: http.MethodPost, Path: "/admin/v1/notifications/broadcasts/:id/revoke", Handler: permit("users:manage", a.adminRevokeBroadcast)}, {Method: http.MethodGet, Path: "/admin/v1/messages", Handler: permit("messages:view", a.adminMessages)}, {Method: http.MethodPost, Path: "/admin/v1/messages/:id/moderate", Handler: permit("messages:manage", a.adminModerateMessage)}, {Method: http.MethodGet, Path: "/admin/v1/pets", Handler: permit("users:view", a.adminPets)}, diff --git a/im/backend/internal/app/content_keywords.go b/im/backend/internal/app/content_keywords.go new file mode 100644 index 0000000..91eb97c --- /dev/null +++ b/im/backend/internal/app/content_keywords.go @@ -0,0 +1,298 @@ +package app + +import ( + "context" + "net/http" + "strings" + "sync" + "time" +) + +// One keyword list for the whole product. Before this, messages had no check at +// all, posts had none, and adoption had its own comma-separated config string — +// three different answers to the same question. The list is cached because it +// is read on every message and written a few times a week. + +type contentKeyword struct { + ID int64 + Word string + Scope string + Action string +} + +type keywordCache struct { + mutex sync.RWMutex + words []contentKeyword + loadedAt time.Time +} + +var keywords = &keywordCache{} + +const keywordCacheTTL = time.Minute + +func (a *App) loadKeywords(ctx context.Context) []contentKeyword { + keywords.mutex.RLock() + if time.Since(keywords.loadedAt) < keywordCacheTTL { + defer keywords.mutex.RUnlock() + return keywords.words + } + keywords.mutex.RUnlock() + + rows, err := a.db.QueryContext(ctx, `SELECT id,word,scope,action FROM content_keywords WHERE status=1`) + if err != nil { + return nil + } + defer rows.Close() + loaded := []contentKeyword{} + for rows.Next() { + var item contentKeyword + if rows.Scan(&item.ID, &item.Word, &item.Scope, &item.Action) == nil && strings.TrimSpace(item.Word) != "" { + item.Word = strings.ToLower(strings.TrimSpace(item.Word)) + loaded = append(loaded, item) + } + } + keywords.mutex.Lock() + keywords.words = loaded + keywords.loadedAt = time.Now() + keywords.mutex.Unlock() + return loaded +} + +func invalidateKeywordCache() { + keywords.mutex.Lock() + keywords.loadedAt = time.Time{} + keywords.mutex.Unlock() +} + +// matchKeyword returns the first word this text trips and what to do about it. +// A "block" anywhere wins over a "review": the stricter rule is the one the +// operator meant. +func (a *App) matchKeyword(ctx context.Context, scope, text string) (string, string) { + if strings.TrimSpace(text) == "" || !a.configBool(ctx, "moderation.keywords_enabled", true) { + return "", "" + } + lowered := strings.ToLower(text) + hit, action := "", "" + for _, item := range a.loadKeywords(ctx) { + if item.Scope != "all" && item.Scope != scope { + continue + } + if !strings.Contains(lowered, item.Word) { + continue + } + if item.Action == "block" { + a.countKeywordHit(ctx, item.ID) + return item.Word, "block" + } + if hit == "" { + hit, action = item.Word, "review" + } + } + if hit != "" { + for _, item := range a.loadKeywords(ctx) { + if item.Word == hit { + a.countKeywordHit(ctx, item.ID) + break + } + } + } + return hit, action +} + +// 命中次数是运营调词库的依据：一个从来不命中的词和一个天天命中的词，价值不一样。 +func (a *App) countKeywordHit(ctx context.Context, id int64) { + go func() { + background, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _, _ = a.db.ExecContext(background, `UPDATE content_keywords SET hit_count=hit_count+1,last_hit_at=NOW(3) WHERE id=?`, id) + }() +} + +// --- 管理端 --------------------------------------------------------------- + +func (a *App) adminKeywords(w http.ResponseWriter, r *http.Request) { + page, size, offset := pagination(r) + where := " WHERE 1=1" + args := []any{} + if scope := strings.TrimSpace(r.URL.Query().Get("scope")); scope != "" { + where += " AND scope=?" + args = append(args, scope) + } + if action := strings.TrimSpace(r.URL.Query().Get("action")); action != "" { + where += " AND action=?" + args = append(args, action) + } + if keyword := strings.TrimSpace(r.URL.Query().Get("keyword")); keyword != "" { + where += " AND (word LIKE ? OR category LIKE ?)" + like := "%" + keyword + "%" + args = append(args, like, like) + } + var total int64 + _ = a.db.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM content_keywords`+where, args...).Scan(&total) + rows, err := a.db.QueryContext(r.Context(), `SELECT id,word,scope,action,category,note,status,hit_count,last_hit_at,created_at + FROM content_keywords`+where+` ORDER BY hit_count DESC,id DESC LIMIT ? OFFSET ?`, append(args, size, offset)...) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "查询词库失败") + return + } + defer rows.Close() + items := []map[string]any{} + for rows.Next() { + var id int64 + var word, scope, action, category, note string + var status, hits int + var lastHit any + var created time.Time + if rows.Scan(&id, &word, &scope, &action, &category, ¬e, &status, &hits, &lastHit, &created) != nil { + continue + } + items = append(items, map[string]any{ + "id": id, "word": word, "scope": scope, "action": action, "category": category, + "note": note, "status": status, "hitCount": hits, "lastHitAt": lastHit, + "createdAt": created.Format(time.RFC3339), + }) + } + reply(w, map[string]any{"items": items, "total": total, "page": page, "size": size}) +} + +type keywordRequest struct { + Word string `json:"word"` + Scope string `json:"scope"` + Action string `json:"action"` + Category string `json:"category"` + Note string `json:"note"` + Status int `json:"status"` +} + +var keywordScopes = map[string]bool{"all": true, "message": true, "post": true, "adoption": true, "nickname": true} + +func validKeyword(req *keywordRequest) string { + req.Word = strings.TrimSpace(req.Word) + req.Category = strings.TrimSpace(req.Category) + req.Note = strings.TrimSpace(req.Note) + if req.Scope == "" { + req.Scope = "all" + } + if req.Action == "" { + req.Action = "review" + } + if len([]rune(req.Word)) < 1 || len([]rune(req.Word)) > 30 { + return "关键词为 1 到 30 个字" + } + if !keywordScopes[req.Scope] { + return "适用范围无效" + } + if req.Action != "block" && req.Action != "review" { + return "命中动作只能是拦截或转人工" + } + if len([]rune(req.Category)) > 20 || len([]rune(req.Note)) > 100 { + return "分类或备注过长" + } + return "" +} + +func (a *App) adminCreateKeyword(w http.ResponseWriter, r *http.Request) { + var req keywordRequest + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "格式错误") + return + } + if message := validKeyword(&req); message != "" { + fail(w, http.StatusBadRequest, 20001, message) + return + } + result, err := a.db.ExecContext(r.Context(), `INSERT INTO content_keywords(word,scope,action,category,note,status,created_by) + VALUES(?,?,?,?,?,1,?)`, req.Word, req.Scope, req.Action, req.Category, req.Note, current(r).ID) + if err != nil { + fail(w, http.StatusBadRequest, 20001, "这个词在该范围下已经存在") + return + } + id, _ := result.LastInsertId() + invalidateKeywordCache() + a.audit(r, "create", "content_keyword", id, map[string]any{"word": req.Word, "scope": req.Scope, "action": req.Action}) + reply(w, map[string]any{"id": id}) +} + +func (a *App) adminUpdateKeyword(w http.ResponseWriter, r *http.Request) { + id, err := pathID(r) + if err != nil { + fail(w, http.StatusBadRequest, 20001, "编号无效") + return + } + var req keywordRequest + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "格式错误") + return + } + if message := validKeyword(&req); message != "" { + fail(w, http.StatusBadRequest, 20001, message) + return + } + status := 1 + if req.Status == 0 { + status = 0 + } + result, err := a.db.ExecContext(r.Context(), `UPDATE content_keywords SET word=?,scope=?,action=?,category=?,note=?,status=? WHERE id=?`, + req.Word, req.Scope, req.Action, req.Category, req.Note, status, id) + if err != nil { + fail(w, http.StatusBadRequest, 20001, "保存失败，可能与已有词重复") + return + } + if affected, _ := result.RowsAffected(); affected == 0 { + fail(w, http.StatusNotFound, 30001, "词条不存在") + return + } + invalidateKeywordCache() + a.audit(r, "update", "content_keyword", id, map[string]any{"word": req.Word, "status": status}) + reply(w, map[string]bool{"success": true}) +} + +func (a *App) adminDeleteKeyword(w http.ResponseWriter, r *http.Request) { + id, err := pathID(r) + if err != nil { + fail(w, http.StatusBadRequest, 20001, "编号无效") + return + } + result, err := a.db.ExecContext(r.Context(), `DELETE FROM content_keywords WHERE id=?`, id) + if err != nil { + fail(w, http.StatusInternalServerError, 50001, "删除失败") + return + } + if affected, _ := result.RowsAffected(); affected == 0 { + fail(w, http.StatusNotFound, 30001, "词条不存在") + return + } + invalidateKeywordCache() + a.audit(r, "delete", "content_keyword", id, nil) + reply(w, map[string]bool{"success": true}) +} + +// adminTestKeyword lets an operator paste a real message and see what the list +// would do with it, instead of finding out from a complaint. +func (a *App) adminTestKeyword(w http.ResponseWriter, r *http.Request) { + var req struct { + Scope string `json:"scope"` + Text string `json:"text"` + } + if decode(r, &req) != nil { + fail(w, http.StatusBadRequest, 20001, "格式错误") + return + } + if req.Scope == "" { + req.Scope = "message" + } + word, action := a.matchKeyword(r.Context(), req.Scope, req.Text) + reply(w, map[string]any{"word": word, "action": action, "matched": word != ""}) +} + +// recordKeywordRisk turns a soft keyword hit into something visible: the risk +// centre is where a moderator would look, so that is where it goes. +func (a *App) recordKeywordRisk(ctx context.Context, userID int64, scope, word string) { + if userID <= 0 { + return + } + _, _ = a.db.ExecContext(ctx, `INSERT INTO risk_events(user_id,event_type,score_delta,metadata) + VALUES(?,?,?,JSON_OBJECT('scope',?,'word',?))`, userID, "keyword_"+scope, 2, scope, word) + _, _ = a.db.ExecContext(ctx, `INSERT INTO user_risk_profiles(user_id,risk_score,message_score) + VALUES(?,2,2) ON DUPLICATE KEY UPDATE risk_score=risk_score+2,message_score=message_score+2`, userID) +} diff --git a/im/backend/internal/app/im.go b/im/backend/internal/app/im.go index b3947ce..42f6efb 100644 --- a/im/backend/internal/app/im.go +++ b/im/backend/internal/app/im.go @@ -604,6 +604,16 @@ func (a *App) persistMessageContext(ctx context.Context, conversationID, senderI if err = a.ensureMessageMembership(ctx, senderID, messageType); err != nil { return item, nil, err } + // 词库对文字消息生效：拦截词直接发不出去，转人工的词放行但记一条风险事件， + // 免得把用户挡在门外却没人知道发生过什么。 + if messageType == 1 { + if word, action := a.matchKeyword(ctx, "message", string(body)); word != "" { + if action == "block" { + return item, nil, fmt.Errorf("消息包含平台不允许的内容") + } + a.recordKeywordRisk(ctx, senderID, "message", word) + } + } var mediaAssetID int64 if messageType == 2 || messageType == 3 { contentMap, _ := content.(map[string]any) diff --git a/im/backend/internal/app/pet_adoption.go b/im/backend/internal/app/pet_adoption.go index 972e5c2..e268ee3 100644 --- a/im/backend/internal/app/pet_adoption.go +++ b/im/backend/internal/app/pet_adoption.go @@ -56,6 +56,10 @@ var adoptionQuestions = []struct { } func (a *App) adoptionReviewReason(ctx context.Context, ownerID int64, text string) string { + // 统一词库优先；配置里那份旧词表留着兜底，免得词库被清空后送养就没人看了。 + if word, _ := a.matchKeyword(ctx, "adoption", text); word != "" { + return "文案命中敏感词：" + word + } words := strings.Split(a.configPlain(ctx, "pet.adoption_review_words", ""), ",") lowered := strings.ToLower(text) for _, word := range words { diff --git a/im/backend/internal/app/pet_feed_pricing.go b/im/backend/internal/app/pet_feed_pricing.go index b46346a..4ece074 100644 --- a/im/backend/internal/app/pet_feed_pricing.go +++ b/im/backend/internal/app/pet_feed_pricing.go @@ -14,11 +14,11 @@ import ( // starting point, not a rule. type priceSuggestConfig struct { - Base map[string]int `json:"base"` // 物种基准单价（分） - Cities map[string]int `json:"cities"` // 城市系数（百分比），键是城市编码前两位或全码 - Minutes map[string]int `json:"minutes"` // 每次时长加价（分） - DogWalk int `json:"dogWalk"` // 含遛狗时的额外加价（分） - Rounding int `json:"rounding"` // 取整到多少分 + Base map[string]int `json:"base"` // 物种基准单价（分） + Cities map[string]int `json:"cities"` // 城市系数（百分比），键是城市编码前两位或全码 + Minutes map[string]int `json:"minutes"` // 每次时长加价（分） + DogWalk int `json:"dogWalk"` // 含遛狗时的额外加价（分） + Rounding int `json:"rounding"` // 取整到多少分 } const defaultPriceSuggest = `{ diff --git a/im/backend/internal/app/social.go b/im/backend/internal/app/social.go index ec90be7..be101ea 100644 --- a/im/backend/internal/app/social.go +++ b/im/backend/internal/app/social.go @@ -564,6 +564,16 @@ func (a *App) createPost(w http.ResponseWriter, r *http.Request) { if req.Visibility != 2 { req.Visibility = 1 } + // 动态和消息共用一套词库：拦截词直接拒绝，转人工的词先发出去但排进审核队列。 + moderation := 1 + if word, action := a.matchKeyword(r.Context(), "post", req.Content); word != "" { + if action == "block" { + fail(w, 400, 20001, "动态包含平台不允许的内容") + return + } + moderation = 0 + a.recordKeywordRisk(r.Context(), current(r).ID, "post", word) + } tx, err := a.db.BeginTx(r.Context(), nil) if err != nil { fail(w, 500, 50001, "发布失败") @@ -578,7 +588,7 @@ func (a *App) createPost(w http.ResponseWriter, r *http.Request) { return } } - result, err := tx.ExecContext(r.Context(), `INSERT INTO posts (user_id,content,visibility,city_code,location_text) VALUES (?,?,?,'310100',?)`, current(r).ID, req.Content, req.Visibility, req.Location) + result, err := tx.ExecContext(r.Context(), `INSERT INTO posts (user_id,content,visibility,city_code,location_text,moderation_status) VALUES (?,?,?,'310100',?,?)`, current(r).ID, req.Content, req.Visibility, req.Location, moderation) if err != nil { _ = tx.Rollback() fail(w, 500, 50001, "发布失败") diff --git a/im/backend/migrations/042_admin_broadcast_refund.sql b/im/backend/migrations/042_admin_broadcast_refund.sql new file mode 100644 index 0000000..b73f727 --- /dev/null +++ b/im/backend/migrations/042_admin_broadcast_refund.sql @@ -0,0 +1,43 @@ +SET NAMES utf8mb4; + +-- 公告与系统通知群发。发出去的每一条都记一行，因为"给谁发过什么"是事后 +-- 唯一能对账的东西；撤回也只撤这一批里还没读的那些。 +CREATE TABLE IF NOT EXISTS notification_broadcasts ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, + title VARCHAR(100) NOT NULL, + content VARCHAR(1000) NOT NULL, + audience VARCHAR(20) NOT NULL COMMENT 'all/users/vip/city/sitters/owners', + audience_ref VARCHAR(500) NOT NULL DEFAULT '' COMMENT '城市编码或用户编号列表', + with_push TINYINT(1) NOT NULL DEFAULT 0, + recipient_count INT UNSIGNED NOT NULL DEFAULT 0, + push_count INT UNSIGNED NOT NULL DEFAULT 0, + status VARCHAR(20) NOT NULL DEFAULT 'SENT' COMMENT 'SENT/REVOKED', + revoked_count INT UNSIGNED NOT NULL DEFAULT 0, + created_by BIGINT UNSIGNED NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + revoked_at DATETIME(3) NULL, + PRIMARY KEY (id), + KEY idx_broadcast_created (created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +-- 订单退款流水。代喂那张 pet_feed_refunds 记的是任务口径，这张记的是订单 +-- 口径：会员、保证金、代喂都在这里，一笔订单可以退很多次。 +CREATE TABLE IF NOT EXISTS order_refunds ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, + order_id BIGINT UNSIGNED NOT NULL, + refund_no VARCHAR(60) NOT NULL, + user_id BIGINT UNSIGNED NOT NULL, + amount_cent INT UNSIGNED NOT NULL, + channel VARCHAR(20) NOT NULL DEFAULT 'gateway' COMMENT 'gateway 原路退回 / offline 线下已退', + reason VARCHAR(500) NOT NULL DEFAULT '', + handled_by BIGINT UNSIGNED NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + PRIMARY KEY (id), + UNIQUE KEY uk_order_refund_no (refund_no), + KEY idx_order_refunds_order (order_id, id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES + ('notification.broadcast_max_recipients','20000','integer','单次公告的最大接收人数，超过要分批发'), + ('notification.broadcast_push_enabled','true','boolean','公告是否允许同时走离线推送') +ON DUPLICATE KEY UPDATE description=VALUES(description); diff --git a/im/backend/migrations/043_admin_risk_keywords.sql b/im/backend/migrations/043_admin_risk_keywords.sql new file mode 100644 index 0000000..6bb2fdf --- /dev/null +++ b/im/backend/migrations/043_admin_risk_keywords.sql @@ -0,0 +1,47 @@ +SET NAMES utf8mb4; + +-- 风险中心原来只能看。加白是"这个人我们看过了，别再报"，事件处理是"这条我 +-- 跟进完了"，两件事都得留下是谁做的。 +ALTER TABLE user_risk_profiles + ADD COLUMN whitelisted TINYINT(1) NOT NULL DEFAULT 0 COMMENT '人工加白后不再进风险列表', + ADD COLUMN whitelist_note VARCHAR(255) NOT NULL DEFAULT '', + ADD COLUMN reviewed_by BIGINT UNSIGNED NULL, + ADD COLUMN reviewed_at DATETIME(3) NULL; + +ALTER TABLE risk_events + ADD COLUMN handled_at DATETIME(3) NULL, + ADD COLUMN handled_by BIGINT UNSIGNED NULL, + ADD COLUMN handled_note VARCHAR(255) NOT NULL DEFAULT '', + ADD KEY idx_risk_events_handled (handled_at, id); + +-- 统一词库：消息、动态、送养原来各有各的判断（送养那组还写在配置里）。 +-- 命中之后是拦下来还是转人工，是运营的判断，所以写在数据里而不是代码里。 +CREATE TABLE IF NOT EXISTS content_keywords ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, + word VARCHAR(60) NOT NULL, + scope VARCHAR(20) NOT NULL DEFAULT 'all' COMMENT 'all/message/post/adoption/nickname', + action VARCHAR(10) NOT NULL DEFAULT 'review' COMMENT 'block 直接拦下 / review 转人工', + category VARCHAR(30) NOT NULL DEFAULT '' COMMENT '分类，便于成组开关', + note VARCHAR(200) NOT NULL DEFAULT '', + status TINYINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '0 停用 1 启用', + hit_count INT UNSIGNED NOT NULL DEFAULT 0, + last_hit_at DATETIME(3) NULL, + created_by BIGINT UNSIGNED NULL, + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), + PRIMARY KEY (id), + UNIQUE KEY uk_keyword_scope (word, scope), + KEY idx_keywords_status (status, scope) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +-- 把送养那组已经在用的词搬进词库，行为保持原样（转人工，不直接拦）。 +INSERT IGNORE INTO content_keywords(word,scope,action,category,note) +SELECT TRIM(word),'adoption','review','送养收费','从 pet.adoption_review_words 迁移' FROM ( + SELECT '元' AS word UNION ALL SELECT '钱' UNION ALL SELECT '价' UNION ALL SELECT '押金' + UNION ALL SELECT '费用' UNION ALL SELECT '红包' UNION ALL SELECT '转账' UNION ALL SELECT '收费' + UNION ALL SELECT '出售' UNION ALL SELECT '买' UNION ALL SELECT '卖' +) AS legacy; + +INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES + ('moderation.keywords_enabled','true','boolean','是否启用统一词库；关闭后消息与动态不做关键词判断') +ON DUPLICATE KEY UPDATE description=VALUES(description);