fnvtk/ckb-SuperAdmin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' of https://e.coding.net/g-xtcy5189/cunkebao/cunkebao_v3 into develop

Browse Source
This commit is contained in:
wong
2025-05-13 18:20:40 +08:00
parent f1970eef57 12ed5d7983
commit 6774e2d17a
5 changed files with 344 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
Cunkebao/app/traffic-pool/page.tsx
View File

@@ -95,6 +95,22 @@ interface Statistics {
todayAddCount: number
}
interface ConvertedUser {
id: number
nickname: string
avatar: string
wechatId: string
fromd: string
tags: string[]
createTime: string
status: number
}
interface ConvertedResponse {
list: ConvertedUser[]
total: number
}
export default function TrafficPoolPage() {
const router = useRouter()
const searchParams = useSearchParams()
@@ -155,12 +171,10 @@ export default function TrafficPoolPage() {
limit: "30"
})
// 只有在有搜索关键词时才添加 keyword 参数
if (debouncedSearchQuery) {
params.append("keyword", debouncedSearchQuery)
}
// 只有在选择了特定来源时才添加 fromd 参数
if (sourceFilter !== "all") {
const selectedSource = sourceTypes.find(source => source.id.toString() === sourceFilter)
if (selectedSource) {
@@ -168,39 +182,51 @@ export default function TrafficPoolPage() {
}
}
// 只有在选择了特定状态时才添加 status 参数
if (statusFilter !== "all") {
params.append("status", statusFilter)
}
const response = await api.get<ApiResponse<TrafficPoolResponse>>(`/v1/traffic/pool?${params.toString()}`, {
headers: {
Authorization: `Bearer ${localStorage.getItem('token')}`
}
} as any)
const endpoint = activeCategory === "customer"
? '/v1/traffic/pool/converted'
: '/v1/traffic/pool'
const response = await api.get<ApiResponse<any>>(
`${endpoint}?${params.toString()}`,
{
headers: {
Authorization: `Bearer ${localStorage.getItem('token')}`
}
} as any
)
if (response.code === 200) {
const { list, pagination } = response.data
const { list } = response.data
const transformedUsers = list.map(user => ({
const transformedUsers: TrafficUser[] = list.map((user: any) => ({
id: user.id.toString(),
avatar: user.avatar,
nickname: user.name || user.nickname || '未知用户',
nickname: user.nickname || user.name || '未知用户',
wechatId: user.wechatId,
phone: user.phone,
region: user.region,
note: user.note,
status: user.status,
phone: user.phone || '',
region: user.region || '',
note: user.note || '',
status: activeCategory === "customer" ? 3 : user.status,
addTime: formatDateTime(user.createTime),
source: user.fromd || '未知来源',
assignedTo: user.assignedTo,
category: user.category || "potential",
tags: user.tags || []
assignedTo: user.assignedTo || '',
category: activeCategory as "potential" | "customer" | "lost",
tags: Array.isArray(user.tags)
? user.tags.map((tag: string) => ({
id: tag,
name: tag,
color: 'bg-blue-100 text-blue-800'
}))
: []
}))
setUsers(prev => isNewSearch ? transformedUsers : [...prev, ...transformedUsers])
setCurrentPage(page)
setHasMore(list.length > 0 && page < pagination.totalPages)
setHasMore(list.length === 30)
} else {
toast({
title: "获取数据失败",
@@ -222,7 +248,7 @@ export default function TrafficPoolPage() {
setIsFetching(false)
setLoading(false)
}
}, [debouncedSearchQuery, sourceFilter, statusFilter, sourceTypes])
}, [debouncedSearchQuery, sourceFilter, statusFilter, sourceTypes, activeCategory])
const fetchStatusTypes = useCallback(async () => {
try {
@@ -456,25 +482,27 @@ export default function TrafficPoolPage() {
</SelectContent>
</Select>
<Select
value={statusFilter}
onValueChange={(value) => {
setStatusFilter(value)
setCurrentPage(1)
}}
>
<SelectTrigger className="w-[120px]">
<SelectValue placeholder="状态" />
</SelectTrigger>
<SelectContent>
<SelectItem value="all">全部状态</SelectItem>
{statusTypes.map((status) => (
<SelectItem key={status.id} value={status.id.toString()}>
{status.name}
</SelectItem>
))}
</SelectContent>
</Select>
{activeCategory === "potential" && (
<Select
value={statusFilter}
onValueChange={(value) => {
setStatusFilter(value)
setCurrentPage(1)
}}
>
<SelectTrigger className="w-[120px]">
<SelectValue placeholder="状态" />
</SelectTrigger>
<SelectContent>
<SelectItem value="all">全部状态</SelectItem>
{statusTypes.map((status) => (
<SelectItem key={status.id} value={status.id.toString()}>
{status.name}
</SelectItem>
))}
</SelectContent>
</Select>
)}
</div>
{/* 用户列表 */}
@@ -506,14 +534,22 @@ export default function TrafficPoolPage() {
<div className="font-medium truncate">{user.nickname}</div>
<div
className={`text-xs px-2 py-1 rounded-full ${
user.status === 2
activeCategory === "customer"
? "bg-green-100 text-green-800"
: user.status === 1
? "bg-yellow-100 text-yellow-800"
: "bg-red-100 text-red-800"
: user.status === 2
? "bg-green-100 text-green-800"
: user.status === 1
? "bg-yellow-100 text-yellow-800"
: "bg-red-100 text-red-800"
}`}
>
{user.status === 2 ? "已添加" : user.status === 1 ? "待处理" : "已失败"}
{activeCategory === "customer"
? "已通过"
: user.status === 2
? "已添加"
: user.status === 1
? "待处理"
: "已失败"}
</div>
</div>
<div className="text-sm text-gray-500">微信号: {user.wechatId}</div>
@@ -577,14 +613,22 @@ export default function TrafficPoolPage() {
<div className="text-sm text-gray-500">{selectedUser.wechatId}</div>
<Badge
className={`mt-1 ${
selectedUser.status === 2
activeCategory === "customer"
? "bg-green-100 text-green-800"
: selectedUser.status === 1
? "bg-yellow-100 text-yellow-800"
: "bg-red-100 text-red-800"
: selectedUser.status === 2
? "bg-green-100 text-green-800"
: selectedUser.status === 1
? "bg-yellow-100 text-yellow-800"
: "bg-red-100 text-red-800"
}`}
>
{selectedUser.status === 2 ? "已添加" : selectedUser.status === 1 ? "待处理" : "已失败"}
{activeCategory === "customer"
? "已通过"
: selectedUser.status === 2
? "已添加"
: selectedUser.status === 1
? "待处理"
: "已失败"}
</Badge>
</div>
</div>

1
Server/application/cunkebao/config/route.php
View File

@@ -42,6 +42,7 @@ Route::group('v1/', function () {
// 流量池相关
Route::group('traffic/pool', function () {
Route::get('', 'app\cunkebao\controller\traffic\GetPotentialListWithInCompanyV1Controller@index');
Route::get('converted', 'app\cunkebao\controller\traffic\GetConvertedListWithInCompanyV1Controller@index');
Route::get('types', 'app\cunkebao\controller\traffic\GetPotentialTypeSectionV1Controller@index');
Route::get('sources', 'app\cunkebao\controller\traffic\GetTrafficSourceSectionV1Controller@index');
Route::get('statistics', 'app\cunkebao\controller\traffic\GetPoolStatisticsV1Controller@index');

112
Server/application/cunkebao/controller/traffic/GetConvertedListWithInCompanyV1Controller.php Normal file
View File

@@ -0,0 +1,112 @@
<?php
namespace app\cunkebao\controller\traffic;
use app\common\model\TrafficPool as TrafficPoolModel;
use app\common\model\TrafficSource as TrafficSourceModel;
use app\common\model\WechatFriendShip as WechatFriendShipModel;
use app\cunkebao\controller\BaseController;
use library\ResponseHelper;
/**
* 流量池控制器
*/
class GetConvertedListWithInCompanyV1Controller extends BaseController
{
/**
* 构建返回数据
*
* @param \think\Paginator $result
* @return array
*/
protected function makeResultedSet(\think\Paginator $result): array
{
$resultSets = [];
foreach ($result->items() as $item) {
$item->tags = json_decode($item->tags);
array_push($resultSets, $item->toArray());
}
return $resultSets;
}
/**
* 构建查询条件
*
* @param array $params
* @return array
*/
protected function makeWhere(array $params = []): array
{
if (!empty($keyword = $this->request->param('keyword'))) {
$where[] = ['exp', "w.alias LIKE '%{$keyword}%' OR w.nickname LIKE '%{$keyword}%'"];
}
// 来源的筛选
if ($fromd = $this->request->param('fromd')) {
$where['s.fromd'] = $fromd;
}
$where['s.companyId'] = $this->getUserInfo('companyId');
$where['s.status'] = TrafficSourceModel::STATUS_PASSED;
return array_merge($where, $params);
}
/**
* 获取流量池列表
*
* @param array $where
* @return \think\Paginator
*/
protected function getPoolListByCompanyId(array $where): \think\Paginator
{
$query = TrafficSourceModel::alias('s')
->field(
[
'w.id', 'w.nickname', 'w.avatar',
'CASE WHEN w.alias IS NULL OR w.alias = "" THEN w.wechatId ELSE w.alias END AS wechatId',
's.fromd',
'f.tags', 'f.createTime', TrafficSourceModel::STATUS_PASSED . ' status'
]
)
->join('traffic_pool p', 'p.identifier=s.identifier')
->join('wechat_account w', 'p.wechatId=w.wechatId')
->join('wechat_friendship f', 'w.wechatId=f.wechatId and f.deleteTime=0')
->order('s.id desc');
foreach ($where as $key => $value) {
if (is_numeric($key) && is_array($value) && isset($value[0]) && $value[0] === 'exp') {
$query->whereExp('', $value[1]);
continue;
}
$query->where($key, $value);
}
return $query->paginate($this->request->param('limit/d', 10), false, ['page' => $this->request->param('page/d', 1)]);
}
/**
* 获取流量池列表
*
* @return \think\response\Json
*/
public function index()
{
try {
$result = $this->getPoolListByCompanyId( $this->makeWhere() );
return ResponseHelper::success(
[
'list' => $this->makeResultedSet($result),
'total' => $result->total(),
]
);
} catch (\Exception $e) {
return ResponseHelper::error($e->getMessage(), $e->getCode());
}
}
}

8
Server/application/cunkebao/controller/traffic/GetPotentialListWithInCompanyV1Controller.php
View File

@@ -3,6 +3,7 @@
namespace app\cunkebao\controller\traffic;
use app\common\model\TrafficPool as TrafficPoolModel;
use app\common\model\TrafficSource as TrafficSourceModel;
use app\common\model\WechatFriendShip as WechatFriendShipModel;
use app\cunkebao\controller\BaseController;
use library\ResponseHelper;
@@ -27,6 +28,8 @@ class GetPotentialListWithInCompanyV1Controller extends BaseController
// 状态筛选
if ($status = $this->request->param('status')) {
$where['s.status'] = $status;
} else {
$where['s.status'] = array('<>', TrafficSourceModel::STATUS_PASSED);
}
// 来源的筛选
@@ -63,6 +66,11 @@ class GetPotentialListWithInCompanyV1Controller extends BaseController
continue;
}
if (is_array($value)) {
$query->where($key, ...$value);
continue;
}
$query->where($key, $value);
}

227
Server/extend/WeChatDeviceApi/Adapters/ChuKeBao/Adapter.php
View File

@@ -8,6 +8,8 @@ use WeChatDeviceApi\Exceptions\ApiException;
// use WeChatDeviceApi\Adapters\ChuKeBao\Client as ChuKeBaoApiClient;
use think\Db;
use think\facade\Config;
use think\facade\Log;
class Adapter implements WeChatServiceInterface
{
@@ -16,26 +18,29 @@ class Adapter implements WeChatServiceInterface
public function __construct(array $config = [])
{
$this->config = $config;
// $this->config = $config ?: Config::get('wechat_device_api.');
$this->config = $config ?: Config::get('wechat_device_api.adapters.ChuKeBao');
// $this->config = $config;
// $this->apiClient = new ChuKeBaoApiClient($config['api_key'], $config['api_secret'], $config['base_url']);
// 校验配置等...
if (empty($config['api_key']) || empty($config['username']) || empty($config['password'])) {
if (empty($this->config['base_url']) || empty($this->config['username']) || empty($this->config['password'])) {
throw new \InvalidArgumentException("ChuKeBao username and password are required.");
}
}
public function addFriend(string $deviceId, string $targetWxId): bool
{
// 1. 构建请求参数 (VendorA 特定的格式)
// 1. 构建请求参数 (ChuKeBao 特定的格式)
$params = [
'device_identifier' => $deviceId,
'wechat_user_to_add' => $targetWxId,
'username' => $this->config['username'],
'password' => $this->config['password'],
// ... 其他 VendorA 特定参数
// ... 其他 ChuKeBao 特定参数
];
// 2. 调用 VendorA 的 API (例如使用 GuzzleHttp 或 cURL)
// 2. 调用 ChuKeBao 的 API (例如使用 GuzzleHttp 或 cURL)
// $response = $this->apiClient->post('/friend/add', $params);
// 伪代码:
$url = $this->config['base_url'] . '/friend/add';
@@ -44,16 +49,16 @@ class Adapter implements WeChatServiceInterface
// $responseData = json_decode($response->getBody()->getContents(), true);
// 模拟API调用
echo "VendorA: Adding friend {$targetWxId} using device {$deviceId}\n";
echo "ChuKeBao: Adding friend {$targetWxId} using device {$deviceId}\n";
$responseData = ['code' => 0, 'message' => 'Success']; // 假设的响应
// 3. 处理响应，转换为标准结果
if (!isset($responseData['code'])) {
throw new ApiException("VendorA: Invalid API response for addFriend.");
throw new ApiException("ChuKeBao: Invalid API response for addFriend.");
}
if ($responseData['code'] !== 0) {
throw new ApiException("VendorA: Failed to add friend - " . ($responseData['message'] ?? 'Unknown error'));
throw new ApiException("ChuKeBao: Failed to add friend - " . ($responseData['message'] ?? 'Unknown error'));
}
return true;
@@ -61,17 +66,17 @@ class Adapter implements WeChatServiceInterface
public function likeMoment(string $deviceId, string $momentId): bool
{
echo "VendorA: Liking moment {$momentId} using device {$deviceId}\n";
echo "ChuKeBao: Liking moment {$momentId} using device {$deviceId}\n";
// 实现 VendorA 的点赞逻辑
return true;
}
public function getGroupList(string $deviceId): array
{
echo "VendorA: Getting group list for device {$deviceId}\n";
echo "ChuKeBao: Getting group list for device {$deviceId}\n";
// 实现 VendorA 的获取群列表逻辑，并转换数据格式
return [
['id' => 'group1_va', 'name' => 'VendorA Group 1', 'member_count' => 10],
['id' => 'group1_va', 'name' => 'ChuKeBao Group 1', 'member_count' => 10],
];
}
@@ -79,19 +84,19 @@ class Adapter implements WeChatServiceInterface
{
echo "VendorA: Getting friend list for device {$deviceId}\n";
return [
['id' => 'friend1_va', 'nickname' => 'VendorA Friend 1', 'remark' => 'VA-F1'],
['id' => 'friend1_va', 'nickname' => 'ChuKeBao Friend 1', 'remark' => 'VA-F1'],
];
}
public function getDeviceInfo(string $deviceId): array
{
echo "VendorA: Getting device info for device {$deviceId}\n";
echo "ChuKeBao: Getting device info for device {$deviceId}\n";
return ['id' => $deviceId, 'status' => 'online_va', 'battery' => '80%'];
}
public function bindDeviceToCompany(string $deviceId, string $companyId): bool
{
echo "VendorA: Binding device {$deviceId} to company {$companyId}\n";
echo "ChuKeBao: Binding device {$deviceId} to company {$companyId}\n";
return true;
}
@@ -103,7 +108,7 @@ class Adapter implements WeChatServiceInterface
*/
public function getChatroomMemberList(string $deviceId, string $chatroomId): array
{
echo "VendorA: Getting chatroom member list for device {$deviceId}, chatroom {$chatroomId}\n";
echo "ChuKeBao: Getting chatroom member list for device {$deviceId}, chatroom {$chatroomId}\n";
return [
['id' => 'member1_va', 'nickname' => 'VendorA Member 1', 'avatar' => ''],
];
@@ -160,7 +165,6 @@ class Adapter implements WeChatServiceInterface
$offset = 0;
$limit = 2000;
// $usleepTime = 100000;
$usleepTime = 50000;
do {
@@ -179,7 +183,7 @@ class Adapter implements WeChatServiceInterface
SELECT
wechatId,alias,nickname,pyInitial,quanPin,avatar,gender,region,signature,phone,country,privince,city,createTime,updateTime
FROM
s2_wechat_friend GROUP BY wechatId;
s2_wechat_friend GROUP BY wechatId
ON DUPLICATE KEY UPDATE
alias=VALUES(alias),
nickname=VALUES(nickname),
@@ -200,26 +204,125 @@ class Adapter implements WeChatServiceInterface
}
// syncWechatDeviceLoginLog
// public function syncWechatDeviceLoginLog()
// {
// try {
// // 确保使用正确的表名，不要让框架自动添加前缀
// Db::connect()->table('s2_wechat_account')
// ->alias('a')
// ->join(['s2_device' => 'd'], 'd.imei = a.imei')
// ->join(['s2_company_account' => 'c'], 'c.id = d.currentAccountId')
// ->field('d.id as deviceId, a.wechatId, a.wechatAlive as alive, c.departmentId as companyId, a.updateTime as createTime')
// ->chunk(1000, function ($data) {
// try {
// foreach ($data as $item) {
// Log::info("syncWechatDeviceLoginLog: " . json_encode($item));
// try {
// // 检查所有必要字段是否存在，如果不存在则设置默认值
// // if (!isset($item['deviceId']) || !isset($item['wechatId']) ||
// // !isset($item['alive']) || !isset($item['companyId']) ||
// // !isset($item['createTime'])) {
// // \think\facade\Log::warning("Missing required field in syncWechatDeviceLoginLog: " . json_encode($item));
// // // 为缺失字段设置默认值
// // $item['deviceId'] = $item['deviceId'] ?? '';
// // $item['wechatId'] = $item['wechatId'] ?? '';
// // $item['alive'] = $item['alive'] ?? 0;
// // $item['companyId'] = $item['companyId'] ?? 0;
// // $item['createTime'] = $item['createTime'] ?? date('Y-m-d H:i:s');
// // // 如果关键字段仍然为空，则跳过此条记录
// // if (empty($item['deviceId']) || empty($item['wechatId']) || empty($item['createTime'])) {
// // continue;
// // }
// // }
// if (empty($item['deviceId']) || empty($item['wechatId']) || empty($item['createTime'])) {
// continue;
// }
// $exists = Db::connect()->table('ck_device_wechat_login')
// ->where('deviceId', $item['deviceId'])
// ->where('wechatId', $item['wechatId'])
// ->where('createTime', $item['createTime'])
// ->find();
// if (!$exists) {
// Db::connect()->table('ck_device_wechat_login')->insert($item);
// }
// } catch (\Exception $e) {
// \think\facade\Log::error("处理单条数据时出错: " . $e->getMessage() . ", 数据: " . json_encode($item) . ", 堆栈: " . $e->getTraceAsString());
// continue; // 跳过这条出错的记录，继续处理下一条
// }
// }
// } catch (\Exception $e) {
// \think\facade\Log::error("处理批次数据时出错: " . $e->getMessage() . ", 堆栈: " . $e->getTraceAsString());
// // 不抛出异常，让程序继续处理下一批次数据
// }
// });
// } catch (\Exception $e) {
// \think\facade\Log::error("微信好友同步任务异常1: " . $e->getMessage() . ", 堆栈: " . $e->getTraceAsString());
// // 可以选择重新抛出异常或者返回false
// return false;
// }
// return true;
// }
public function syncWechatDeviceLoginLog()
{
Db::table('s2_wechat_account')
->alias('a')
->join('s2_device d', 'd.imei = a.imei')
->join('s2_company_account c', 'c.id = d.currentAccountId')
->field('d.id as deviceId, a.wechatId, a.wechatAlive as alive, c.departmentId as companyId, a.updateTime as createTime')
->chunk(1000, function ($data) {
foreach ($data as $item) {
$exists = Db::table('ck_device_wechat_login')
try {
// 确保使用正确的表名，不要让框架自动添加前缀
$cursor = Db::connect()->table('s2_wechat_account')
->alias('a')
->join(['s2_device' => 'd'], 'd.imei = a.imei')
->join(['s2_company_account' => 'c'], 'c.id = d.currentAccountId')
->field('d.id as deviceId, a.wechatId, a.wechatAlive as alive, c.departmentId as companyId, a.updateTime as createTime')
->cursor();
foreach ($cursor as $item) {
try {
// 检查所有必要字段是否存在，如果不存在则设置默认值
if (
!isset($item['deviceId']) || !isset($item['wechatId']) ||
!isset($item['alive']) || !isset($item['companyId']) ||
!isset($item['createTime'])
) {
Log::warning("Missing required field in syncWechatDeviceLoginLog: " . json_encode($item));
// 为缺失字段设置默认值
$item['deviceId'] = $item['deviceId'] ?? '';
$item['wechatId'] = $item['wechatId'] ?? '';
$item['alive'] = $item['alive'] ?? 0;
$item['companyId'] = $item['companyId'] ?? 0;
$item['createTime'] = $item['createTime'] ?? date('Y-m-d H:i:s');
// 如果关键字段仍然为空，则跳过此条记录
if (empty($item['deviceId']) || empty($item['wechatId']) || empty($item['createTime'])) {
continue;
}
}
$exists = Db::connect()->table('ck_device_wechat_login')
->where('deviceId', $item['deviceId'])
->where('wechatId', $item['wechatId'])
->where('createTime', $item['createTime'])
->find();
if (!$exists) {
Db::table('ck_device_wechat_login')->insert($item);
Db::connect()->table('ck_device_wechat_login')->insert($item);
}
} catch (\Exception $e) {
Log::error("处理单条数据时出错: " . $e->getMessage() . ", 数据: " . json_encode($item) . ", 堆栈: " . $e->getTraceAsString());
continue; // 跳过这条出错的记录，继续处理下一条
}
});
}
return true;
} catch (\Exception $e) {
Log::error("微信好友同步任务异常: " . $e->getMessage() . ", 堆栈: " . $e->getTraceAsString());
return false;
}
}
/**
@@ -230,48 +333,8 @@ class Adapter implements WeChatServiceInterface
* @param int $batchSize 每批处理的数据量
* @return int 影响的行数
*/
// public function syncWechatFriendToTrafficPoolBatch($batchSize = 5000)
// {
// // 1. 获取数据总量
// $total = Db::table('s2_wechat_friend')
// ->group('wechatId')
// ->count();
// // 2. 计算需要处理的批次
// $batchCount = ceil($total / $batchSize);
// $affectedRows = 0;
// // 3. 分批处理
// for ($i = 0; $i < $batchCount; $i++) {
// $offset = $i * $batchSize;
// // 分批查询SQL，使用LIMIT控制每次获取的数据量
// $sql = "INSERT IGNORE INTO ck_traffic_pool(`identifier`, `wechatId`, `mobile`)
// SELECT wechatId identifier, wechatId, phone
// FROM (
// SELECT wechatId, phone
// FROM `s2_wechat_friend`
// GROUP BY wechatId
// LIMIT {$offset}, {$batchSize}
// ) AS temp";
// $affectedRows += Db::execute($sql);
// // 释放内存
// if ($i % 5 == 0) {
// Db::clear();
// gc_collect_cycles();
// }
// }
// return $affectedRows;
// }
public function syncWechatFriendToTrafficPoolBatch($batchSize = 5000)
{
// 1. 先获取去重后的wechatId清单并建立临时索引
Db::execute("CREATE TEMPORARY TABLE IF NOT EXISTS temp_wechat_ids (
wechatId VARCHAR(64) PRIMARY KEY
) ENGINE=MEMORY");
@@ -281,26 +344,15 @@ class Adapter implements WeChatServiceInterface
// 批量插入去重的wechatId
Db::execute("INSERT INTO temp_wechat_ids SELECT DISTINCT wechatId FROM s2_wechat_friend");
// 2. 获取临时表的数据总量
$total = Db::table('temp_wechat_ids')->count();
// 3. 计算需要处理的批次
$batchCount = ceil($total / $batchSize);
$affectedRows = 0;
// 4. 开始事务处理批量数据
try {
for ($i = 0; $i < $batchCount; $i++) {
$offset = $i * $batchSize;
// 使用临时表和JOIN来提高查询效率
// $sql = "INSERT IGNORE INTO ck_traffic_pool(`identifier`, `wechatId`, `mobile`)
// SELECT t.wechatId AS identifier, t.wechatId, f.phone AS mobile
// FROM (
// SELECT wechatId FROM temp_wechat_ids LIMIT {$offset}, {$batchSize}
// ) AS t
// LEFT JOIN s2_wechat_friend f ON t.wechatId = f.wechatId
// GROUP BY t.wechatId";
$sql = "INSERT IGNORE INTO ck_traffic_pool(`identifier`, `wechatId`, `mobile`)
SELECT t.wechatId AS identifier, t.wechatId,
(SELECT phone FROM s2_wechat_friend
@@ -312,38 +364,19 @@ class Adapter implements WeChatServiceInterface
$currentAffected = Db::execute($sql);
$affectedRows += $currentAffected;
// 更频繁地释放内存
// if ($i % 2 == 1) {
if ($i % 5 == 0) {
// Db::clear();
gc_collect_cycles();
}
// 添加短暂休眠，但不要太长以免影响总体执行时间
// 20-50毫秒通常是个不错的平衡点
usleep(30000); // 30毫秒
// 如果批处理大小很大或者当前批影响行数很多，可以适当增加休眠时间
// if ($currentAffected > 1000) {
// usleep(20000); // 额外增加20毫秒
// }
// 输出进度日志
// $progress = round(($i + 1) / $batchCount * 100, 2);
// \think\facade\Log::info("Traffic pool sync progress: {$progress}% completed. Rows affected in this batch: {$currentAffected}");
}
} catch (\Exception $e) {
\think\facade\Log::error("Error in traffic pool sync: " . $e->getMessage());
// 删除临时表
// Db::execute("DROP TEMPORARY TABLE IF EXISTS temp_wechat_ids");
throw $e;
} finally {
Db::execute("DROP TEMPORARY TABLE IF EXISTS temp_wechat_ids");
}
// 5. 清理临时表
// Db::execute("DROP TEMPORARY TABLE IF EXISTS temp_wechat_ids");
return $affectedRows;
}
}