Files
CKB-Interface/application/cunkebao/service/TrafficPoolService.php
2026-02-04 11:02:33 +08:00

763 lines
24 KiB
PHP
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

<?php
namespace app\cunkebao\service;
use app\common\model\TrafficPoolV2;
use app\common\model\TrafficPoolCompany;
use app\common\model\TrafficPoolSource;
use app\common\model\TrafficPoolBehavior;
use app\common\model\TrafficPoolTag;
use app\common\model\TrafficPoolAllotRecord;
use think\Db;
/**
* 流量池核心服务类
* 处理流量的入池、查询、更新等核心业务逻辑
*/
class TrafficPoolService
{
/**
* 流量入池(核心方法)
*
* @param string $identifier 唯一标识微信ID优先
* @param int $companyId 公司ID
* @param int $sourceType 来源类型
* @param array $poolData 流量总表数据
* @param array $companyData 公司流量数据
* @param array $sourceData 来源数据
* @return array [poolId, poolCompanyId]
*/
public function enterPool(
string $identifier,
int $companyId,
int $sourceType,
array $poolData = [],
array $companyData = [],
array $sourceData = []
) {
Db::startTrans();
try {
// 1. 查找或创建总表记录
$pool = TrafficPoolV2::findOrCreateByIdentifier($identifier, $poolData);
// 2. 查找或创建公司记录
$poolCompany = TrafficPoolCompany::findOrCreateByIdentifierAndCompany(
$identifier,
$companyId,
$pool->id,
$companyData
);
// 3. 创建来源记录
TrafficPoolSource::createSource(
$poolCompany->id,
$identifier,
$companyId,
$sourceType,
$sourceData
);
Db::commit();
return [
'poolId' => $pool->id,
'poolCompanyId' => $poolCompany->id
];
} catch (\Exception $e) {
Db::rollback();
throw $e;
}
}
/**
* 好友通过时同步流量信息
*
* @param string $identifier 微信ID
* @param int $companyId 公司ID
* @param int $wechatFriendId 微信好友表ID
* @param array $friendData 好友数据
* @return TrafficPoolCompany|null
*/
public function syncFriendPass(string $identifier, int $companyId, int $wechatFriendId, array $friendData = [])
{
// 查找流量池记录
$poolCompany = TrafficPoolCompany::where('identifier', $identifier)
->where('companyId', $companyId)
->where('isDel', 0)
->find();
if (!$poolCompany) {
// 如果不存在,先入池
$result = $this->enterPool(
$identifier,
$companyId,
TrafficPoolSource::SOURCE_TYPE_FRIEND_ADD,
$friendData,
array_merge($friendData, [
'wechatFriendId' => $wechatFriendId,
'friendStatus' => TrafficPoolCompany::FRIEND_STATUS_PASSED,
'friendPassTime' => time()
])
);
$poolCompany = TrafficPoolCompany::find($result['poolCompanyId']);
} else {
// 更新现有记录
$poolCompany->save([
'wechatFriendId' => $wechatFriendId,
'friendStatus' => TrafficPoolCompany::FRIEND_STATUS_PASSED,
'friendPassTime' => time(),
'updateTime' => time()
]);
}
// 更新总表基础信息
if (!empty($friendData)) {
$pool = TrafficPoolV2::find($poolCompany->poolId);
if ($pool) {
$pool->updateBasicInfo($friendData);
}
}
return $poolCompany;
}
/**
* 同步微信标签
*
* @param int $poolCompanyId 公司流量ID
* @param array $labels 标签数组
* @return int 同步数量
*/
public function syncWechatTags(int $poolCompanyId, array $labels)
{
$poolCompany = TrafficPoolCompany::find($poolCompanyId);
if (!$poolCompany) {
return 0;
}
return TrafficPoolTag::syncWechatTags(
$poolCompanyId,
$poolCompany->identifier,
$poolCompany->companyId,
$labels
);
}
/**
* 获取流量池列表(分页)
*
* @param int $companyId 公司ID
* @param int $page 页码
* @param int $pageSize 每页数量
* @param array $filters 筛选条件
* @return array
*/
public function getPoolList(int $companyId, int $page = 1, int $pageSize = 10, array $filters = [])
{
$query = TrafficPoolCompany::alias('tpc')
->join('ck_traffic_pool tp', 'tp.id = tpc.poolId', 'LEFT')
->where('tpc.companyId', $companyId)
->where('tpc.isDel', 0);
// 应用筛选条件
if (!empty($filters['keyword'])) {
$keyword = $filters['keyword'];
$query->where(function($q) use ($keyword) {
$q->where('tp.nickname', 'like', "%{$keyword}%")
->whereOr('tp.wechatId', 'like', "%{$keyword}%")
->whereOr('tp.wechatAlias', 'like', "%{$keyword}%")
->whereOr('tp.mobile', 'like', "%{$keyword}%")
->whereOr('tpc.realName', 'like', "%{$keyword}%")
->whereOr('tpc.phone', 'like', "%{$keyword}%");
});
}
if (isset($filters['friendStatus'])) {
$query->where('tpc.friendStatus', $filters['friendStatus']);
}
if (isset($filters['level'])) {
$query->where('tpc.level', $filters['level']);
}
if (isset($filters['lifecycle'])) {
$query->where('tpc.lifecycle', $filters['lifecycle']);
}
if (isset($filters['allocateStatus'])) {
$query->where('tpc.allocateStatus', $filters['allocateStatus']);
}
if (!empty($filters['ownerWechatId'])) {
$query->where('tpc.ownerWechatId', $filters['ownerWechatId']);
}
// RFM 筛选
if (isset($filters['rfmMMin'])) {
$query->where('tpc.rfmM', '>=', $filters['rfmMMin']);
}
if (isset($filters['rfmMMax'])) {
$query->where('tpc.rfmM', '<=', $filters['rfmMMax']);
}
// 统计总数
$total = $query->count();
// 查询列表
$list = $query->field([
'tpc.id',
'tpc.poolId',
'tpc.identifier',
'tpc.companyId',
'tpc.friendStatus',
'tpc.level',
'tpc.intentionLevel',
'tpc.lifecycle',
'tpc.lastInteractTime',
'tpc.rfmF',
'tpc.rfmM',
'tpc.totalMsgCount',
'tpc.totalOrderCount',
'tpc.totalOrderAmount',
'tpc.ownerWechatId',
'tpc.allocateStatus',
'tpc.realName',
'tpc.phone',
'tpc.createTime',
'tp.nickname',
'tp.avatar',
'tp.gender',
'tp.wechatId',
'tp.wechatAlias',
'tp.mobile',
'tp.region'
])
->order('tpc.id DESC')
->page($page, $pageSize)
->select();
// 获取标签
$poolCompanyIds = array_column($list->toArray(), 'id');
$tags = $this->getTagsForPoolCompanies($poolCompanyIds);
// 组装数据
$result = [];
foreach ($list as $item) {
$data = $item->toArray();
// 计算 RFM R 值
$data['rfmR'] = $item->lastInteractTime ? (int)floor((time() - $item->lastInteractTime) / 86400) : 9999;
// 计算 RFM 总分
$data['rfmScore'] = $this->calculateRfmScore($data['rfmR'], $data['rfmF'], $data['rfmM']);
// 添加标签
$data['tags'] = $tags[$item->id] ?? [];
$result[] = $data;
}
return [
'list' => $result,
'total' => $total,
'page' => $page,
'pageSize' => $pageSize
];
}
/**
* 获取流量详情
*
* @param int $poolCompanyId 公司流量ID
* @param int $companyId 公司ID
* @return array|null
*/
public function getPoolDetail(int $poolCompanyId, int $companyId)
{
$poolCompany = TrafficPoolCompany::alias('tpc')
->join('ck_traffic_pool tp', 'tp.id = tpc.poolId', 'LEFT')
->where('tpc.id', $poolCompanyId)
->where('tpc.companyId', $companyId)
->where('tpc.isDel', 0)
->field('tpc.*, tp.nickname, tp.avatar, tp.gender, tp.wechatId, tp.wechatAlias, tp.mobile, tp.region, tp.country, tp.province, tp.city, tp.signature')
->find();
if (!$poolCompany) {
return null;
}
$data = $poolCompany->toArray();
// 获取标签
$data['tags'] = TrafficPoolTag::getTagsByPoolCompany($poolCompanyId)->toArray();
// 获取来源历史带群归属信息限制50条
$data['sources'] = TrafficPoolSource::getSourcesWithOwners($poolCompanyId, 50);
// 获取行为轨迹最近50条
$data['behaviors'] = TrafficPoolBehavior::getUserJourney($poolCompanyId, 50)->toArray();
// 获取分配历史
$data['allotRecords'] = TrafficPoolAllotRecord::getAllotHistory($poolCompanyId)->toArray();
// 如果消息数为0从微信消息表中统计实际消息数
if (empty($data['totalMsgCount']) || $data['totalMsgCount'] == 0) {
$msgCount = 0;
// 优先通过wechatFriendId统计最准确
if (!empty($poolCompany->wechatFriendId)) {
$msgCount = Db::table('s2_wechat_message')
->where('wechatFriendId', $poolCompany->wechatFriendId)
->where('type', 1) // 好友消息type=1
->where('isDeleted', 0)
->count();
}
// 如果wechatFriendId没有统计到尝试通过identifier微信ID统计
if ($msgCount == 0 && !empty($poolCompany->identifier)) {
// 统计发送者或接收者是该微信ID的消息
// 需要关联s2_wechat_friend表通过wechatId匹配
$msgCount = Db::table('s2_wechat_message')
->alias('wm')
->join(['s2_wechat_friend' => 'wf'], 'wm.wechatFriendId = wf.id', 'LEFT')
->where(function($query) use ($poolCompany) {
$query->where('wm.senderWechatId', $poolCompany->identifier)
->whereOr('wf.wechatId', $poolCompany->identifier);
})
->where('wm.type', 1) // 好友消息
->where('wm.isDeleted', 0)
->where('wf.isDeleted', 0)
->count();
}
// 如果从行为表也有记录,取较大值(兼容旧数据)
$behaviorMsgCount = TrafficPoolBehavior::where('poolCompanyId', $poolCompanyId)
->whereIn('behaviorType', [
TrafficPoolBehavior::BEHAVIOR_TYPE_SEND_MSG,
TrafficPoolBehavior::BEHAVIOR_TYPE_RECEIVE_MSG
])
->count();
if ($behaviorMsgCount > $msgCount) {
$msgCount = $behaviorMsgCount;
}
if ($msgCount > 0) {
// 更新数据库中的消息数
$poolCompany->save([
'totalMsgCount' => $msgCount,
'updateTime' => time()
]);
$data['totalMsgCount'] = $msgCount;
}
}
// 计算 RFM
$data['rfmR'] = $poolCompany->lastInteractTime ? (int)floor((time() - $poolCompany->lastInteractTime) / 86400) : 9999;
$data['rfmScore'] = $this->calculateRfmScore($data['rfmR'], $data['rfmF'], $data['rfmM']);
return $data;
}
/**
* 更新流量信息
*
* @param int $poolCompanyId 公司流量ID
* @param int $companyId 公司ID
* @param array $data 更新数据
* @return bool
*/
public function updatePool(int $poolCompanyId, int $companyId, array $data)
{
$poolCompany = TrafficPoolCompany::where('id', $poolCompanyId)
->where('companyId', $companyId)
->where('isDel', 0)
->find();
if (!$poolCompany) {
return false;
}
// 允许更新的字段
$allowFields = [
'realName', 'phone', 'email', 'birthday', 'address',
'company', 'position', 'remark', 'customFields',
'level', 'intentionLevel', 'lifecycle', 'status'
];
$updateData = array_intersect_key($data, array_flip($allowFields));
$updateData['updateTime'] = time();
return $poolCompany->save($updateData);
}
/**
* 批量获取流量的标签
*
* @param array $poolCompanyIds
* @return array
*/
protected function getTagsForPoolCompanies(array $poolCompanyIds)
{
if (empty($poolCompanyIds)) {
return [];
}
$tags = TrafficPoolTag::whereIn('poolCompanyId', $poolCompanyIds)
->where('isDel', 0)
->select();
$result = [];
foreach ($tags as $tag) {
if (!isset($result[$tag->poolCompanyId])) {
$result[$tag->poolCompanyId] = [];
}
$result[$tag->poolCompanyId][] = [
'id' => $tag->id,
'tagDefineId' => $tag->tagDefineId,
'tagName' => $tag->tagName,
'tagType' => $tag->tagType,
'tagValue' => $tag->tagValue
];
}
return $result;
}
/**
* 计算 RFM 评分
*
* @param int $r 最后互动距今天数
* @param int $f 互动频次
* @param float $m 消费金额
* @return array
*/
public function calculateRfmScore($r, $f, $m)
{
// R 评分(天数越少分数越高)
if ($r <= 7) {
$rScore = 5;
} elseif ($r <= 30) {
$rScore = 4;
} elseif ($r <= 90) {
$rScore = 3;
} elseif ($r <= 180) {
$rScore = 2;
} else {
$rScore = 1;
}
// F 评分
if ($f >= 100) {
$fScore = 5;
} elseif ($f >= 50) {
$fScore = 4;
} elseif ($f >= 20) {
$fScore = 3;
} elseif ($f >= 5) {
$fScore = 2;
} else {
$fScore = 1;
}
// M 评分
if ($m >= 10000) {
$mScore = 5;
} elseif ($m >= 5000) {
$mScore = 4;
} elseif ($m >= 1000) {
$mScore = 3;
} elseif ($m >= 100) {
$mScore = 2;
} else {
$mScore = 1;
}
return [
'R' => $rScore,
'F' => $fScore,
'M' => $mScore,
'total' => $rScore + $fScore + $mScore
];
}
/**
* 获取流量统计数据
*
* @param int $companyId 公司ID
* @return array
*/
public function getStatistics(int $companyId)
{
$today = strtotime('today');
$yesterday = strtotime('yesterday');
$thisWeek = strtotime('monday this week');
$thisMonth = strtotime('first day of this month');
// 总流量数
$totalCount = TrafficPoolCompany::where('companyId', $companyId)
->where('isDel', 0)
->count();
// 好友数
$friendCount = TrafficPoolCompany::where('companyId', $companyId)
->where('friendStatus', TrafficPoolCompany::FRIEND_STATUS_PASSED)
->where('isDel', 0)
->count();
// 今日新增
$todayNewCount = TrafficPoolCompany::where('companyId', $companyId)
->where('createTime', '>=', $today)
->where('isDel', 0)
->count();
// 本周新增
$weekNewCount = TrafficPoolCompany::where('companyId', $companyId)
->where('createTime', '>=', $thisWeek)
->where('isDel', 0)
->count();
// 本月新增
$monthNewCount = TrafficPoolCompany::where('companyId', $companyId)
->where('createTime', '>=', $thisMonth)
->where('isDel', 0)
->count();
// 客户等级分布
$levelDistribution = TrafficPoolCompany::where('companyId', $companyId)
->where('isDel', 0)
->group('level')
->field('level, COUNT(*) as count')
->select()
->toArray();
// 生命周期分布
$lifecycleDistribution = TrafficPoolCompany::where('companyId', $companyId)
->where('isDel', 0)
->group('lifecycle')
->field('lifecycle, COUNT(*) as count')
->select()
->toArray();
// 来源分布
$sourceDistribution = TrafficPoolSource::where('companyId', $companyId)
->where('isFirstSource', 1)
->group('sourceType')
->field('sourceType, COUNT(*) as count')
->select()
->toArray();
return [
'totalCount' => $totalCount,
'friendCount' => $friendCount,
'todayNewCount' => $todayNewCount,
'weekNewCount' => $weekNewCount,
'monthNewCount' => $monthNewCount,
'levelDistribution' => $levelDistribution,
'lifecycleDistribution' => $lifecycleDistribution,
'sourceDistribution' => $sourceDistribution
];
}
/**
* 从标签引擎同步用户标签
*
* @param int $poolCompanyId 流量池公司ID
* @param int $companyId 公司ID
* @param int $operatorId 操作人ID
* @return array 同步结果
*/
public function syncTagsFromEngine(int $poolCompanyId, int $companyId, int $operatorId = null)
{
// 获取流量池记录
$poolCompany = TrafficPoolCompany::where('id', $poolCompanyId)
->where('companyId', $companyId)
->where('isDel', 0)
->find();
if (!$poolCompany) {
throw new \Exception('流量不存在');
}
// 获取标识信息用于查询标签引擎
$identifiers = [];
// 微信ID
if (!empty($poolCompany->identifier)) {
$identifiers[] = [
'type' => 'wechat',
'value' => $poolCompany->identifier
];
}
// 手机号
if (!empty($poolCompany->phone)) {
$identifiers[] = [
'type' => 'phone',
'value' => $poolCompany->phone
];
}
if (empty($identifiers)) {
throw new \Exception('无有效标识可用于查询标签');
}
// 调用标签引擎服务
$tagEngineService = new \app\common\service\TagEngineService();
$result = $tagEngineService->queryByIdentifiers($identifiers, [
'mask_identifier' => false
]);
//exit_data($result);
if ($result === false) {
throw new \Exception('标签引擎查询失败');
}
// 检查返回结果
if (isset($result['code']) && $result['code'] !== 0) {
throw new \Exception($result['message'] ?? '标签引擎返回错误');
}
$data = $result['data'] ?? $result;
if (!is_array($data)) {
$data = [];
}
$syncedCount = 0;
$skippedCount = 0;
// 处理返回的标签数据
foreach ($data as $item) {
if (empty($item['found']) || empty($item['tags'])) {
continue;
}
foreach ($item['tags'] as $tagData) {
try {
// 查找或创建标签定义
$tagDefine = $this->findOrCreateTagDefine(
$companyId,
$tagData['tag_code'] ?? '',
$tagData['tag_name'] ?? '',
$tagData['category'] ?? '标签引擎',
$tagData['tag_type'] ?? 'string'
);
if (!$tagDefine) {
$skippedCount++;
continue;
}
// 添加标签到流量池
$tag = TrafficPoolTag::addTag(
$poolCompanyId,
$poolCompany->identifier,
$companyId,
$tagDefine->id,
TrafficPoolTag::SOURCE_AI, // 来源为AI/外部同步
$operatorId,
$tagData['tag_value'] ?? null,
null // score
);
if ($tag) {
$syncedCount++;
} else {
$skippedCount++;
}
} catch (\Exception $e) {
$skippedCount++;
continue;
}
}
}
return [
'syncedCount' => $syncedCount,
'skippedCount' => $skippedCount,
'total' => $syncedCount + $skippedCount
];
}
/**
* 查找或创建标签定义
*
* @param int $companyId 公司ID
* @param string $tagCode 标签代码
* @param string $tagName 标签名称
* @param string $categoryName 分类名称
* @param string $valueType 值类型
* @return \app\common\model\TrafficPoolTagDefine|null
*/
protected function findOrCreateTagDefine(
int $companyId,
string $tagCode,
string $tagName,
string $categoryName,
string $valueType
) {
if (empty($tagName)) {
return null;
}
// 标签类型映射
$typeMap = [
'numeric' => 'number',
'enum' => 'enum',
'string' => 'string',
'boolean' => 'boolean',
'datetime' => 'datetime',
'json' => 'json',
];
$mappedType = $typeMap[$valueType] ?? 'string';
// 先查找是否已存在该标签定义
$tagDefine = \app\common\model\TrafficPoolTagDefine::where('companyId', $companyId)
->where('tagName', $tagName)
->where('tagType', TrafficPoolTag::TAG_TYPE_AI) // AI标签类型
->where('isDel', 0)
->find();
if ($tagDefine) {
return $tagDefine;
}
// 查找或创建分类
$category = \app\common\model\TrafficPoolTagCategory::where('companyId', $companyId)
->where('categoryName', $categoryName)
->where('tagType', TrafficPoolTag::TAG_TYPE_AI)
->where('isDel', 0)
->find();
if (!$category) {
$category = new \app\common\model\TrafficPoolTagCategory();
$category->save([
'companyId' => $companyId,
'tagType' => TrafficPoolTag::TAG_TYPE_AI,
'categoryName' => $categoryName,
'description' => '从标签引擎同步的标签分类',
'sortOrder' => 0,
'isDel' => 0,
'createTime' => time(),
'updateTime' => time()
]);
}
// 创建标签定义
$tagDefine = new \app\common\model\TrafficPoolTagDefine();
$tagDefine->save([
'companyId' => $companyId,
'categoryId' => $category->id,
'tagType' => TrafficPoolTag::TAG_TYPE_AI,
'tagCode' => $tagCode ?: 'engine_' . md5($tagName),
'tagName' => $tagName,
'valueType' => $mappedType,
'description' => '从标签引擎同步',
'isSystem' => 0,
'isDel' => 0,
'createTime' => time(),
'updateTime' => time()
]);
return $tagDefine;
}
}