feat: initial commit
This commit is contained in:
root
@@ -0,0 +1,226 @@
|
||||
<?php
|
||||
/**
|
||||
* 访客日志队列消费（INSERT / enrich UPDATE）
|
||||
*/
|
||||
require_once dirname(__DIR__) . '/DbHelper.php';
|
||||
require_once __DIR__ . '/VisitorLogSchema.php';
|
||||
require_once __DIR__ . '/VisitorLogQueue.php';
|
||||
|
||||
class VisitorLogWorker
|
||||
{
|
||||
/**
|
||||
* @return array{processed:int, failed:int}
|
||||
*/
|
||||
public static function processBatch(int $limit = 100)
|
||||
{
|
||||
$jobs = VisitorLogQueue::popBatch($limit);
|
||||
$processed = 0;
|
||||
$failed = 0;
|
||||
|
||||
foreach ($jobs as $job) {
|
||||
$ok = false;
|
||||
for ($attempt = 1; $attempt <= 3; $attempt++) {
|
||||
if (self::processJob($job)) {
|
||||
$ok = true;
|
||||
break;
|
||||
}
|
||||
usleep(100000 * $attempt);
|
||||
}
|
||||
if ($ok) {
|
||||
$processed++;
|
||||
} else {
|
||||
$failed++;
|
||||
VisitorLogQueue::moveToDead($job, 'process failed after retries');
|
||||
}
|
||||
}
|
||||
|
||||
return ['processed' => $processed, 'failed' => $failed];
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array $job
|
||||
*/
|
||||
public static function processJob(array $job)
|
||||
{
|
||||
$type = $job['type'] ?? '';
|
||||
if ($type === 'insert_full') {
|
||||
return self::insertFull($job['logs'] ?? []) !== null;
|
||||
}
|
||||
if ($type === 'enrich') {
|
||||
return self::enrich((int) ($job['log_id'] ?? 0), $job['logs'] ?? []);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array $logs
|
||||
* @return int|null insert_id
|
||||
*/
|
||||
public static function insertFull(array $logs)
|
||||
{
|
||||
$conn = self::connect();
|
||||
if (!$conn) {
|
||||
return null;
|
||||
}
|
||||
VisitorLogSchema::ensureColumns($conn);
|
||||
VisitorLogSchema::ensureIndexes($conn);
|
||||
|
||||
$data = self::escapeLogs($conn, $logs);
|
||||
$d = static function ($key) use ($data, $conn) {
|
||||
return array_key_exists($key, $data) ? $data[$key] : cloak_db_escape($conn, '');
|
||||
};
|
||||
|
||||
$sql = "INSERT INTO `visitor_logs`(`campagin_id`, `visit_md5_code`, `domain`, `visit_date`, `IP`, `country`, `result`, `reason`, `referer`, `client`, `browser`, `page`, `language`, `user_agent`, `http_referer`, `accept_language_raw`, `judge_timing`, `fp_url`, `device`) VALUES ('"
|
||||
. $d('campagin_id') . "','"
|
||||
. $d('visit_md5_code') . "','"
|
||||
. $d('site_name') . "','"
|
||||
. ($data['visit_date'] ?? date('Y-m-d H:i:s')) . "','"
|
||||
. $d('customers_ip') . "','"
|
||||
. $d('country') . "','"
|
||||
. $d('result') . "','"
|
||||
. $d('reason') . "','"
|
||||
. $d('v_referer') . "','"
|
||||
. $d('Client') . "','"
|
||||
. $d('v_Browser') . "','"
|
||||
. $d('v_PageURL') . "','"
|
||||
. $d('accept_language') . "','"
|
||||
. $d('user_agent') . "','"
|
||||
. $d('http_referer') . "','"
|
||||
. $d('accept_language_raw') . "','"
|
||||
. $d('judge_timing') . "','"
|
||||
. $d('fp_url') . "','"
|
||||
. $d('device') . "')";
|
||||
|
||||
if ($conn->query($sql) !== true) {
|
||||
VisitorLogQueue::logError('insert_full: ' . $conn->error . ' | ' . $sql);
|
||||
$conn->close();
|
||||
return null;
|
||||
}
|
||||
$id = (int) $conn->insert_id;
|
||||
$conn->close();
|
||||
return $id > 0 ? $id : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步最小 INSERT（二次风控 wait，需立即返回 log_id）
|
||||
*
|
||||
* @param array $logs
|
||||
* @return int|null
|
||||
*/
|
||||
public static function insertMinimal(array $logs)
|
||||
{
|
||||
$conn = self::connect();
|
||||
if (!$conn) {
|
||||
return null;
|
||||
}
|
||||
VisitorLogSchema::ensureColumns($conn);
|
||||
VisitorLogSchema::ensureIndexes($conn);
|
||||
|
||||
$data = self::escapeLogs($conn, $logs);
|
||||
$d = static function ($key) use ($data, $conn) {
|
||||
return array_key_exists($key, $data) ? $data[$key] : cloak_db_escape($conn, '');
|
||||
};
|
||||
|
||||
$sql = "INSERT INTO `visitor_logs`(`campagin_id`, `visit_md5_code`, `domain`, `visit_date`, `IP`, `country`, `result`, `reason`, `client`, `browser`, `language`, `fp_url`, `device`) VALUES ('"
|
||||
. $d('campagin_id') . "','"
|
||||
. $d('visit_md5_code') . "','"
|
||||
. $d('site_name') . "','"
|
||||
. date('Y-m-d H:i:s') . "','"
|
||||
. $d('customers_ip') . "','"
|
||||
. $d('country') . "','"
|
||||
. $d('result') . "','"
|
||||
. $d('reason') . "','"
|
||||
. $d('Client') . "','"
|
||||
. $d('v_Browser') . "','"
|
||||
. $d('accept_language') . "','"
|
||||
. $d('fp_url') . "','"
|
||||
. $d('device') . "')";
|
||||
|
||||
if ($conn->query($sql) !== true) {
|
||||
VisitorLogQueue::logError('insertMinimal: ' . $conn->error);
|
||||
$conn->close();
|
||||
return null;
|
||||
}
|
||||
$id = (int) $conn->insert_id;
|
||||
$conn->close();
|
||||
return $id > 0 ? $id : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param int $logId
|
||||
* @param array $logs
|
||||
*/
|
||||
public static function enrich($logId, array $logs)
|
||||
{
|
||||
if ($logId <= 0) {
|
||||
return false;
|
||||
}
|
||||
$conn = self::connect();
|
||||
if (!$conn) {
|
||||
return false;
|
||||
}
|
||||
|
||||
$data = self::escapeLogs($conn, $logs);
|
||||
$sets = [];
|
||||
$map = [
|
||||
'referer' => 'v_referer',
|
||||
'page' => 'v_PageURL',
|
||||
'user_agent' => 'user_agent',
|
||||
'http_referer' => 'http_referer',
|
||||
'accept_language_raw' => 'accept_language_raw',
|
||||
'judge_timing' => 'judge_timing',
|
||||
];
|
||||
foreach ($map as $col => $logKey) {
|
||||
if (array_key_exists($logKey, $data)) {
|
||||
$sets[] = "`{$col}`='" . $data[$logKey] . "'";
|
||||
} elseif (array_key_exists($col, $data)) {
|
||||
$sets[] = "`{$col}`='" . $data[$col] . "'";
|
||||
}
|
||||
}
|
||||
if ($sets === []) {
|
||||
$conn->close();
|
||||
return true;
|
||||
}
|
||||
|
||||
$sql = 'UPDATE `visitor_logs` SET ' . implode(', ', $sets) . ' WHERE `id`=' . (int) $logId;
|
||||
$ok = $conn->query($sql) === true;
|
||||
if (!$ok) {
|
||||
VisitorLogQueue::logError('enrich: ' . $conn->error);
|
||||
}
|
||||
$conn->close();
|
||||
return $ok;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param mysqli $conn
|
||||
* @param array $logs
|
||||
* @return array<string, string>
|
||||
*/
|
||||
private static function escapeLogs($conn, array $logs)
|
||||
{
|
||||
$data = [];
|
||||
foreach ($logs as $key => $value) {
|
||||
$data[$key] = cloak_db_escape($conn, $value);
|
||||
}
|
||||
if (!isset($data['visit_date'])) {
|
||||
$data['visit_date'] = date('Y-m-d H:i:s');
|
||||
}
|
||||
return $data;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return mysqli|null
|
||||
*/
|
||||
private static function connect()
|
||||
{
|
||||
if (!defined('DB_USERNAME') || !defined('DB_PASSWORD') || !defined('DB_NAME')) {
|
||||
return null;
|
||||
}
|
||||
$conn = @new mysqli('localhost', DB_USERNAME, DB_PASSWORD, DB_NAME);
|
||||
if ($conn->connect_error) {
|
||||
VisitorLogQueue::logError('DB connect: ' . $conn->connect_error);
|
||||
return null;
|
||||
}
|
||||
return $conn;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user