e79690518e
Some checks failed
Build and Deploy (service.xpcool.com) / build-and-deploy (push) Failing after 54s
- 修正 status 列默认值为 0，避免历史失败记录被误标为成功
- 历史数据回填改为幂等更新，可重复执行
- 过滤未解析的 ${...} 配置占位符，优化 Bark 推送错误提示
- LogDetail 记录不存在时统一返回参数错误，避免暴露 SQL 细节
238 lines
7.9 KiB
Go
238 lines
7.9 KiB
Go
package recruitment
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/os/gtime"
|
||
|
||
"service.xpcool.com/internal/dao"
|
||
"service.xpcool.com/internal/model/do"
|
||
"service.xpcool.com/internal/model/dto"
|
||
"service.xpcool.com/internal/model/entity"
|
||
)
|
||
|
||
// barkPayload 对应自建 Bark 服务的 /push 接口入参。
|
||
type barkPayload struct {
|
||
DeviceKey string `json:"device_key"`
|
||
Title string `json:"title"`
|
||
Body string `json:"body"`
|
||
Group string `json:"group"` // 通知分组
|
||
Level string `json:"level"` // 优先级
|
||
URL string `json:"url"` // 点击跳转
|
||
}
|
||
|
||
// defaultDeviceKey 读取环境变量注入的兜底设备密钥（BARK_DEVICE_KEY -> bark.deviceKey）。
|
||
// 当数据库没有任何订阅记录时，推送与测试推送回退到该密钥，做到「填了 env 即可推送」。
|
||
func defaultDeviceKey(ctx context.Context) string {
|
||
return resolveCfgValue(g.Cfg().MustGet(ctx, "bark.deviceKey", "").String())
|
||
}
|
||
|
||
// resolveCfgValue 过滤未解析的 ${...} 占位符。
|
||
//
|
||
// 背景：gf v2.10.2 不再自动替换配置中的 ${ENV}，需由 cmd.injectEnv 显式注入。
|
||
// 若某环境变量未设置（如本地未配 BARK_BASE_URL），配置项会保留字面量
|
||
// "${BARK_BASE_URL}"，直接使用会产生 "invalid character { in host name" 之类的
|
||
// 误导性错误。此处统一把未解析占位符视为空值，使错误信息更准确。
|
||
func resolveCfgValue(v string) string {
|
||
v = strings.TrimSpace(v)
|
||
if strings.HasPrefix(v, "${") && strings.HasSuffix(v, "}") {
|
||
return ""
|
||
}
|
||
return v
|
||
}
|
||
|
||
// defaultSubscription 构造一个使用兜底密钥的默认订阅（全地区/全分类/仅新公告）。
|
||
func defaultSubscription(ctx context.Context) *entity.PushSubscription {
|
||
key := defaultDeviceKey(ctx)
|
||
if key == "" {
|
||
return nil
|
||
}
|
||
return &entity.PushSubscription{
|
||
Id: 0, Name: "默认(Bark环境变量)", DeviceKey: key,
|
||
Regions: "[]", Categories: "[]", OnlyNew: 1, PushTime: "08:00", Enabled: 1,
|
||
}
|
||
}
|
||
|
||
// barkPush 向指定设备密钥发送 Bark 推送（自建服务：POST {baseUrl}/push）。
|
||
func barkPush(ctx context.Context, deviceKey, title, body string) (bool, string, error) {
|
||
baseURL := resolveCfgValue(g.Cfg().MustGet(ctx, "bark.baseUrl", "").String())
|
||
if baseURL == "" {
|
||
return false, "未配置 bark.baseUrl", fmt.Errorf("bark baseUrl empty")
|
||
}
|
||
if deviceKey == "" {
|
||
// 未显式传入则回退到环境变量兜底密钥。
|
||
deviceKey = defaultDeviceKey(ctx)
|
||
}
|
||
if deviceKey == "" {
|
||
return false, "设备密钥为空", fmt.Errorf("device key empty")
|
||
}
|
||
payload := barkPayload{
|
||
DeviceKey: deviceKey,
|
||
Title: title,
|
||
Body: body,
|
||
Group: "recruit_daily",
|
||
Level: "active",
|
||
}
|
||
client := g.Client()
|
||
client.SetTimeout(10 * time.Second)
|
||
resp, err := client.Post(ctx, strings.TrimRight(baseURL, "/")+"/push", payload)
|
||
if err != nil {
|
||
return false, err.Error(), err
|
||
}
|
||
defer resp.Close()
|
||
var out struct {
|
||
Code int `json:"code"`
|
||
Message string `json:"message"`
|
||
}
|
||
_ = json.Unmarshal(resp.ReadAll(), &out)
|
||
ok := resp.StatusCode == 200 && out.Code == 200
|
||
msg := out.Message
|
||
if msg == "" {
|
||
msg = fmt.Sprintf("HTTP %d", resp.StatusCode)
|
||
}
|
||
return ok, msg, nil
|
||
}
|
||
|
||
// sendDailyDigest 对所有启用订阅发送「招聘早报」汇总（自建 Bark）。
|
||
// 若库内无任何订阅，则回退到环境变量兜底密钥（BARK_DEVICE_KEY）发一份全量早报。
|
||
func sendDailyDigest(ctx context.Context) {
|
||
var subs []entity.PushSubscription
|
||
if err := dao.PushSubscription.Ctx(ctx).Where("enabled", 1).Scan(&subs); err != nil {
|
||
g.Log().Errorf(ctx, "加载推送订阅失败: %v", err)
|
||
return
|
||
}
|
||
if len(subs) == 0 {
|
||
def := defaultSubscription(ctx)
|
||
if def == nil {
|
||
g.Log().Infof(ctx, "无启用订阅且未配置兜底设备密钥，跳过早报推送")
|
||
return
|
||
}
|
||
subs = []entity.PushSubscription{*def}
|
||
}
|
||
for _, sub := range subs {
|
||
title, body := buildDigest(ctx, sub)
|
||
if title == "" && body == "" {
|
||
// 仅推送新公告且今日无新：跳过，避免打扰。
|
||
continue
|
||
}
|
||
ok, msg, _ := barkPush(ctx, sub.DeviceKey, title, body)
|
||
recordPushLog(ctx, sub.Id, title, body, ok, msg)
|
||
}
|
||
}
|
||
|
||
// buildDigest 按订阅过滤条件生成早报内容；仅推送新公告且无新时返回空串。
|
||
func buildDigest(ctx context.Context, sub entity.PushSubscription) (string, string) {
|
||
regions := parseJSONStrings(sub.Regions)
|
||
cats := parseJSONInts(sub.Categories)
|
||
since := gtime.Now().StartOfDay().String()
|
||
m := dao.RecruitmentInfo.Ctx(ctx).Where("status", g.Slice{0, 1}).Where("created_at >= ?", since)
|
||
if len(regions) > 0 {
|
||
m = m.Where("region IN (?)", regions)
|
||
}
|
||
if len(cats) > 0 {
|
||
m = m.Where("category IN (?)", cats)
|
||
}
|
||
var list []dto.RecruitmentInfoVO
|
||
if err := m.Fields("title, category, region, deadline").OrderDesc("publish_date").Scan(&list); err != nil {
|
||
g.Log().Errorf(ctx, "早报内容查询失败: %v", err)
|
||
return "", ""
|
||
}
|
||
if sub.OnlyNew == 1 && len(list) == 0 {
|
||
return "", "" // 无新公告则跳过
|
||
}
|
||
counts := map[int]int{}
|
||
for _, v := range list {
|
||
counts[v.Category]++
|
||
}
|
||
var sb strings.Builder
|
||
sb.WriteString("今日新增：")
|
||
parts := make([]string, 0, len(counts))
|
||
for c, n := range counts {
|
||
parts = append(parts, dto.CategoryName(c)+" "+itoa(n))
|
||
}
|
||
sb.WriteString(strings.Join(parts, " · "))
|
||
sb.WriteString("\n")
|
||
limit := 8
|
||
for i, v := range list {
|
||
if i >= limit {
|
||
break
|
||
}
|
||
line := "· " + v.Title
|
||
if v.Deadline != "" {
|
||
line += "（截止 " + v.Deadline + "）"
|
||
}
|
||
sb.WriteString(line + "\n")
|
||
}
|
||
title := "贵州招聘早报 · " + gtime.Now().Format("Y-m-d")
|
||
if len(regions) == 1 {
|
||
title += " · " + regions[0]
|
||
}
|
||
return title, strings.TrimRight(sb.String(), "\n")
|
||
}
|
||
|
||
// recordPushLog 写推送记录。
|
||
func recordPushLog(ctx context.Context, subId uint64, title, body string, ok bool, msg string) {
|
||
res := 0
|
||
if ok {
|
||
res = 1
|
||
}
|
||
if _, err := dao.PushLog.Ctx(ctx).Data(do.PushLog{
|
||
SubscriptionId: subId, PushAt: gtime.Now(), Title: title, Body: body,
|
||
Result: res, Error: msg, CreatedAt: gtime.Now(),
|
||
}).Insert(); err != nil {
|
||
g.Log().Errorf(ctx, "写入推送日志失败: %v", err)
|
||
}
|
||
}
|
||
|
||
// ---------- 数据源告警 ----------
|
||
|
||
// notifySourceFailing 推送「数据源连续失败」预警（尚未禁用，提示人工关注）。
|
||
func notifySourceFailing(ctx context.Context, src entity.CrawlSource, failCount int, errMsg string) {
|
||
title := "抓取源预警 · " + src.Name
|
||
body := fmt.Sprintf("数据源「%s」已连续失败 %d 次，暂未禁用。\n最近错误：%s",
|
||
src.Name, failCount, truncateStr(errMsg, 200))
|
||
sendAlert(ctx, title, body)
|
||
}
|
||
|
||
// notifySourceDisabled 推送「数据源已自动禁用」告警。
|
||
func notifySourceDisabled(ctx context.Context, src entity.CrawlSource, failCount int, errMsg string) {
|
||
title := "抓取源已禁用 · " + src.Name
|
||
body := fmt.Sprintf("数据源「%s」连续失败 %d 次，已自动禁用。\n最近错误：%s\n请在后台确认后手动重新启用。",
|
||
src.Name, failCount, truncateStr(errMsg, 200))
|
||
sendAlert(ctx, title, body)
|
||
}
|
||
|
||
// sendAlert 向所有启用订阅（或兜底密钥）发送一条系统告警。
|
||
// 与早报不同，告警不受订阅的地区/分类过滤影响——运维信息应送达全部订阅者。
|
||
func sendAlert(ctx context.Context, title, body string) {
|
||
keys := make([]string, 0, 4)
|
||
var subs []entity.PushSubscription
|
||
if err := dao.PushSubscription.Ctx(ctx).Where("enabled", 1).Scan(&subs); err == nil {
|
||
for _, s := range subs {
|
||
if s.DeviceKey != "" {
|
||
keys = append(keys, s.DeviceKey)
|
||
}
|
||
}
|
||
}
|
||
// 库内无订阅时回退到环境变量兜底密钥。
|
||
if len(keys) == 0 {
|
||
if k := defaultDeviceKey(ctx); k != "" {
|
||
keys = append(keys, k)
|
||
}
|
||
}
|
||
if len(keys) == 0 {
|
||
g.Log().Warningf(ctx, "无可用设备密钥，告警未送达：%s", title)
|
||
return
|
||
}
|
||
for _, k := range keys {
|
||
if ok, msg, _ := barkPush(ctx, k, title, body); !ok {
|
||
g.Log().Warningf(ctx, "告警推送失败 device=%s: %s", k, msg)
|
||
}
|
||
}
|
||
}
|