You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
333 lines
10 KiB
333 lines
10 KiB
package service |
|
|
|
import ( |
|
"context" |
|
"encoding/json" |
|
"fmt" |
|
"strings" |
|
"time" |
|
|
|
"go-common/app/interface/main/push-archive/dao" |
|
"go-common/app/interface/main/push-archive/model" |
|
accmdl "go-common/app/service/main/account/model" |
|
"go-common/library/log" |
|
) |
|
|
|
const ( |
|
_insertAct = "insert" |
|
_updateAct = "update" |
|
_deleteAct = "delete" |
|
|
|
_archiveTable = "archive" |
|
|
|
_pushRetryTimes = 3 |
|
_pushPartSize = 500000 // 每次请求最多推50w mid |
|
|
|
_hbaseBatch = 100 |
|
) |
|
|
|
func (s *Service) consumeArchiveproc() { |
|
defer s.wg.Done() |
|
var ( |
|
err error |
|
msgs = s.archiveSub.Messages() |
|
) |
|
for { |
|
msg, ok := <-msgs |
|
if !ok { |
|
log.Warn("s.ArchiveSub has been closed.") |
|
return |
|
} |
|
msg.Commit() |
|
s.arcMo++ |
|
log.Info("consume archive key(%s) offset(%d) message(%s)", msg.Key, msg.Offset, msg.Value) |
|
m := new(model.Message) |
|
if err = json.Unmarshal(msg.Value, &m); err != nil { |
|
log.Error("json.Unmarshal(%v) error(%v)", string(msg.Value), err) |
|
continue |
|
} |
|
if m.Table != _archiveTable { |
|
continue |
|
} |
|
n := new(model.Archive) |
|
if err = json.Unmarshal(m.New, n); err != nil { |
|
log.Error("json.Unmarshal(%s) error(%v)", m.New, err) |
|
continue |
|
} |
|
// 稿件过审 |
|
switch m.Action { |
|
case _insertAct: |
|
if !n.IsNormal() { |
|
log.Info("archive (%d) upper(%d) is not normal when insert, no need to push", n.ID, n.Mid) |
|
continue |
|
} |
|
case _updateAct: |
|
o := new(model.Archive) |
|
if err = json.Unmarshal(m.Old, o); err != nil { |
|
log.Error("json.Unmarshal(%s) error(%v)", m.Old, err) |
|
continue |
|
} |
|
if !n.IsNormal() || o.State == n.State || o.PubTime == n.PubTime { |
|
log.Info("archive (%d) upper(%d) is not normal when update, no need to push", n.ID, n.Mid) |
|
continue |
|
} |
|
default: |
|
continue |
|
} |
|
dao.PromInfo("archive_push") |
|
// 在指定周期内，up主只推送一次 |
|
if s.limit(n.Mid) { |
|
log.Info("upper's push limit upper(%d) aid(%d)", n.Mid, n.ID) |
|
continue |
|
} |
|
log.Info("noticeFans start, mid(%d) aid(%d)", n.Mid, n.ID) |
|
err = s.noticeFans(n) // notice fans |
|
log.Info("noticeFans end, mid(%d) aid(%d) error(%v)", n.Mid, n.ID, err) |
|
} |
|
} |
|
|
|
// isForbidTime 当前时间是否被禁止推送消息 |
|
func (s *Service) isForbidTime() (forbid bool, err error) { |
|
var forbidStart, forbidEnd time.Time |
|
now := time.Now() |
|
for _, f := range s.ForbidTimes { |
|
forbidStart, err = s.getTodayTime(f.PushForbidStartTime) |
|
if err != nil { |
|
log.Error("isForbidTime getTodayTime(%s) error(%v)", f.PushForbidStartTime, err) |
|
return |
|
} |
|
forbidEnd, err = s.getTodayTime(f.PushForbidEndTime) |
|
if err != nil { |
|
log.Error("isForbidTime getTodayTime(%s) error(%v)", f.PushForbidEndTime, err) |
|
return |
|
} |
|
if now.After(forbidStart) && now.Before(forbidEnd) { |
|
forbid = true |
|
break |
|
} |
|
} |
|
return |
|
} |
|
|
|
func (s *Service) noticeFans(arc *model.Archive) (err error) { |
|
isPGC := s.isPGC(arc) |
|
mids, counter, err := s.fans(arc.Mid, arc.ID, isPGC) |
|
if err != nil { |
|
return |
|
} |
|
log.Info("noticeFans get fans mid(%d) aid(%d) fans number(%d)", arc.Mid, arc.ID, counter) |
|
arc.Author = s.name(arc.Mid) |
|
for gKey, list := range mids { |
|
g := s.dao.FanGroups[gKey] |
|
params := model.NewBatchParam(map[string]interface{}{ |
|
"relationType": g.RelationType, |
|
"archive": arc, |
|
"group": s.getGroupParam(g), |
|
"msgTemplate": g.MsgTemplate, |
|
}, model.PushParamHandler) |
|
dao.Batch(&list, _pushPartSize, _pushRetryTimes, params, s.dao.NoticeFans) |
|
} |
|
return |
|
} |
|
|
|
func (s *Service) fans(upper int64, aid int64, isPGC bool) (mids map[string][]int64, counter int, err error) { |
|
mids = make(map[string][]int64) |
|
// 从 HBase 获取upper粉丝数据 |
|
fans, err := s.dao.Fans(context.TODO(), upper, isPGC) |
|
ln := len(fans) |
|
if err != nil { |
|
log.Error("fans from hbase upper(%d) error(%v)", upper, err) |
|
return |
|
} |
|
log.Info("filter fans -- get all fans of upper(%d), fans num(%d)", upper, ln) |
|
if ln == 0 { |
|
return |
|
} |
|
dao.PromInfoAdd("archive_fans", int64(ln)) |
|
// 筛选粉丝，分配到对应的实验组 |
|
hitFans := s.group(upper, fans) |
|
hour := time.Now().Hour() |
|
for gKey, hits := range hitFans { |
|
log.Info("before filter by active time, upper(%d) group(%s) fans(%d)", upper, gKey, len(hits)) |
|
if len(hits) <= 0 { |
|
continue |
|
} |
|
g := s.dao.FanGroups[gKey] |
|
var activeFans, excluded []int64 |
|
if g.RelationType == model.RelationAttention { |
|
activeFans, excluded = s.dao.FansByActiveTime(hour, &hits) |
|
noLimitFans := s.noPushLimitFans(upper, gKey, &activeFans) |
|
for _, mid := range activeFans { |
|
if !s.filterUserSetting(mid, g.RelationType) { |
|
excluded = append(excluded, mid) |
|
log.Info("filter fans: excluded by usersettings, fan(%d), relationtype(%d), upper(%d) table(%s)", mid, g.RelationType, upper, g.HBaseTable) |
|
continue |
|
} |
|
if !s.pushLimit(mid, upper, g, &noLimitFans) { |
|
excluded = append(excluded, mid) |
|
continue |
|
} |
|
mids[gKey] = append(mids[gKey], mid) |
|
dao.PromInfo(fmt.Sprintf("%s__send", g.Name)) |
|
log.Info("filter fans: included by pushlimit: upper(%d)'s fan(%d) group.name(%s)", upper, mid, g.Name) |
|
} |
|
} else { |
|
mids[gKey] = hits |
|
} |
|
counter += len(mids[gKey]) |
|
log.Info("upper(%d) aid(%d) group(%s) fans(%d)", upper, aid, gKey, len(mids[gKey])) |
|
// 统计数据落表 |
|
s.formStatisticsProc(aid, g.Name, mids[gKey], &excluded) |
|
} |
|
return |
|
} |
|
|
|
// group 粉丝分组: 分组优先级(优先级 & 分组的可用性)-->分组规则(默认default,hbase过滤走hbase) |
|
func (s *Service) group(upper int64, fans map[int64]int) (list map[string][]int64) { |
|
list = make(map[string][]int64) |
|
// 将粉丝分为 特殊关注 和 普通关注 两类 |
|
attentions, specials := s.dao.FansByProportion(upper, fans) |
|
log.Info("group count upper(%d) attentions(%d) specials(%d)", upper, len(attentions), len(specials)) |
|
// 禁止推送时间判断, 免打扰用户(针对普通关注) |
|
isForbid, err := s.isForbidTime() |
|
if err != nil { |
|
log.Error("isForbidTime upper(%d) error(%v)", upper, err) |
|
} |
|
// 按顺序筛选出每个分组的数据 |
|
for _, gkey := range s.dao.GroupOrder { |
|
g := s.dao.FanGroups[gkey] |
|
if g.RelationType == model.RelationAttention && isForbid { |
|
log.Info("forbid by time upper(%d) group(%+v)", upper, g) |
|
continue |
|
} |
|
// 优先处理abtest的流量 |
|
if g.Hitby == model.GroupDataTypeAbtest { |
|
switch g.RelationType { |
|
case model.RelationAttention: |
|
var pool, exists, ex []int64 |
|
pool, attentions = s.fansByAbtest(g, attentions) |
|
for _, gk := range s.dao.GroupOrder { |
|
gg := s.dao.FanGroups[gk] |
|
if gg.RelationType != model.RelationAttention || |
|
gg.Hitby != model.GroupDataTypeHBase || |
|
gg.Name == "ai:pushlist_follow_recent" { |
|
continue |
|
} |
|
ex, pool = s.dao.FansByHBase(upper, gk, &pool) |
|
exists = append(exists, ex...) |
|
} |
|
list[gkey] = s.filterByBlacklist(upper, exists) |
|
case model.RelationSpecial: |
|
list[gkey], specials = s.fansByAbtest(g, specials) |
|
} |
|
continue |
|
} |
|
if g.Hitby == model.GroupDataTypeAbComparison { |
|
// 对照组保持原来逻辑 |
|
switch g.RelationType { |
|
case model.RelationAttention: |
|
var pool, exists []int64 |
|
pool, attentions = s.fansByAbtest(g, attentions) |
|
for _, gk := range s.dao.GroupOrder { |
|
gg := s.dao.FanGroups[gk] |
|
if gg.RelationType != model.RelationAttention || gg.Hitby != model.GroupDataTypeHBase { |
|
continue |
|
} |
|
exists, pool = s.dao.FansByHBase(upper, gk, &pool) |
|
list[gkey] = append(list[gkey], exists...) |
|
} |
|
case model.RelationSpecial: |
|
list[gkey], specials = s.fansByAbtest(g, specials) |
|
} |
|
continue |
|
} |
|
switch { |
|
case g.RelationType == model.RelationAttention && g.Hitby == model.GroupDataTypeDefault: |
|
list[gkey] = attentions |
|
attentions = []int64{} |
|
case g.RelationType == model.RelationSpecial && g.Hitby == model.GroupDataTypeDefault: |
|
list[gkey] = specials |
|
specials = []int64{} |
|
case g.RelationType == model.RelationAttention && g.Hitby == model.GroupDataTypeHBase: |
|
list[gkey], attentions = s.dao.FansByHBase(upper, gkey, &attentions) |
|
case g.RelationType == model.RelationSpecial && g.Hitby == model.GroupDataTypeHBase: |
|
list[gkey], specials = s.dao.FansByHBase(upper, gkey, &specials) |
|
default: |
|
log.Error("group failed for grouporder(%s) & group.relationtype(%d) & hitby(%s)", gkey, g.RelationType, g.Hitby) |
|
} |
|
} |
|
return |
|
} |
|
|
|
func (s *Service) filterByBlacklist(upper int64, fans []int64) (result []int64) { |
|
for { |
|
var mids []int64 |
|
l := len(fans) |
|
if l == 0 { |
|
return |
|
} |
|
if l <= _hbaseBatch { |
|
mids = fans[:] |
|
fans = []int64{} |
|
} else { |
|
mids = fans[:_hbaseBatch] |
|
fans = fans[_hbaseBatch:] |
|
} |
|
exists, notExists := s.dao.ExistsInBlacklist(context.Background(), upper, mids) |
|
result = append(result, notExists...) // 不在黑名单中的用户直接可推 |
|
exists, notExists = s.dao.ExistsInWhitelist(context.Background(), upper, exists) |
|
result = append(result, exists...) // 存在黑名单中但是后来被恢复到白名单中的用户可推 |
|
for _, mid := range notExists { |
|
// 只出现在黑名单中的用户不推 |
|
log.Warn("filter by blacklist, mid(%d)", mid) |
|
} |
|
} |
|
} |
|
|
|
// getGroupParam g.name被_分隔后的第1个/2个值（比如：ai:pushlist_offline_up的是offline, special的就是special） |
|
func (s *Service) getGroupParam(g *dao.FanGroup) string { |
|
p := strings.Split(g.Name, "_") |
|
if len(p) == 1 { |
|
return p[0] |
|
} |
|
if g.RelationType == model.RelationSpecial { |
|
return "s_" + p[1] |
|
} |
|
return p[1] |
|
} |
|
|
|
// filterUserSetting 普通关注up主的粉丝中，推送开启配置为全部推送的人很少，因此，只要其开启了特殊关注/全部关注／没有设置的就发送；特殊关注等同 |
|
func (s *Service) filterUserSetting(mid int64, relationType int) bool { |
|
if relationType == model.RelationSpecial && s.userSettings[mid] != nil && s.userSettings[mid].Type == model.PushTypeForbid { |
|
return false |
|
} |
|
if relationType == model.RelationAttention && s.userSettings[mid] != nil && s.userSettings[mid].Type == model.PushTypeForbid { |
|
return false |
|
} |
|
return true |
|
} |
|
|
|
func (s *Service) name(mid int64) (name string) { |
|
arg := &accmdl.ArgMid{Mid: mid} |
|
info, err := s.accRPC.Info3(context.TODO(), arg) |
|
if err != nil { |
|
dao.PromError("archive:获取作者信息") |
|
log.Error("s.accRPC.Info3(%+v) error(%v)", arg, err) |
|
return |
|
} |
|
name = info.Name |
|
return |
|
} |
|
|
|
func (s *Service) isPGC(arc *model.Archive) bool { |
|
return (arc.Attribute >> model.AttrBitIsPGC & 1) == 1 |
|
} |
|
|
|
// Test for test |
|
// TODO delete, for test |
|
func (s *Service) Test(arc *model.Archive) { |
|
if arc.ID == 0 || arc.Mid == 0 { |
|
return |
|
} |
|
go s.noticeFans(arc) |
|
} // TODO delete, for test
|
|
|