diff --git a/app/tools/controller/timer/NginxLogImport.php b/app/tools/controller/timer/NginxLogImport.php new file mode 100644 index 0000000..e5e0556 --- /dev/null +++ b/app/tools/controller/timer/NginxLogImport.php @@ -0,0 +1,20 @@ + 86400, 'concurrency' => 1, ], + [ + 'name' => 'nginx_log_import', + 'type' => 'site', + 'target' => '/tools/timer.NginxLogImport/do', + 'frequency' => 300, + 'concurrency' => 1, + 'run_type' => 'auto', + ], ]; diff --git a/extend/base/tools/controller/timer/NginxLogImportBase.php b/extend/base/tools/controller/timer/NginxLogImportBase.php new file mode 100644 index 0000000..60ce301 --- /dev/null +++ b/extend/base/tools/controller/timer/NginxLogImportBase.php @@ -0,0 +1,281 @@ +isFormatMatch + $sampleLine = $this->readFirstNonEmptyLine($file); + if ($sampleLine === null || !$parser->isFormatMatch($sampleLine)) { + Log::error("nginx_log_import: 格式不匹配,跳过文件 {$file}"); + $errors[] = "format_mismatch:{$file}"; + continue; + } + + $totalFiles++; + + // 5. 串行处理:增量读取 + 解析 + 批量入库 + $stats = $this->processFile($parser, $reader, $file, $excludeStatic); + $totalLines += $stats['lines']; + $totalFails += $stats['fails']; + } catch (\Throwable $e) { + Log::error("nginx_log_import: 处理文件异常 {$file} - " . $e->getMessage()); + $errors[] = 'error:' . basename($file) . ':' . $e->getMessage(); + } + } + + return json_encode([ + 'files' => $totalFiles, + 'lines' => $totalLines, + 'fails' => $totalFails, + 'errors' => $errors, + ], JSON_UNESCAPED_UNICODE); + } + + /** + * 处理单个文件:增量读取、解析、过滤、批量入库、savePosition. + * + * @return array{lines:int,fails:int} + */ + protected function processFile( + \app\common\service\NginxLogParser $parser, + \app\common\service\NginxLogReader $reader, + string $file, + bool $excludeStatic + ): array { + $lines = 0; + $fails = 0; + $samples = []; + $batch = []; + $lastLineHash = null; + $inode = 0; + $offset = 0; + + // Reader 是 Generator,让其自然走完(maxLines / EOF 会自动 savePosition) + // 见 notepad Task 6 #2:不要中途 break,否则 maxLines/EOF 的 savePosition 都不执行 + foreach ($reader->read($file, self::READ_MAX_LINES) as $line) { + // 跟踪最后处理的行(用于失败时 savePosition) + $lastLineHash = md5($line); + + $parsed = $parser->parse($line); + if ($parsed === null) { + $fails++; + if (count($samples) < self::FAIL_SAMPLES_MAX) { + $samples[] = mb_substr($line, 0, 500); + } + continue; + } + + // 静态资源排除(在 parse 成功后判断,避免与 parse 失败混淆) + if ($excludeStatic && $parser->isStaticResource($parsed['uri'])) { + continue; + } + + $batch[] = $this->buildRow($parsed, $file); + + if (count($batch) >= self::INSERT_CHUNK) { + $this->insertBatch($batch); + $lines += count($batch); + $batch = []; + } + } + + // flush 剩余 + if (!empty($batch)) { + $this->insertBatch($batch); + $lines += count($batch); + $batch = []; + } + + // 二次 savePosition:把本批解析失败信息写到 position(Reader 已经写过 offset,这里只更新 fail 字段) + // 见 notepad Task 6 #3:Reader 默认写 failCount=0;本批若有失败需要覆盖 + if ($fails > 0) { + $this->updateFailStats($reader, $file, $fails, $samples); + } + + Log::info("nginx_log_import: 文件 {$file} 完成 lines={$lines} fails={$fails}"); + + return ['lines' => $lines, 'fails' => $fails]; + } + + /** + * 构造入库行:只保留 raw 表实际字段(过滤 parser 多余的 referer_domain/ua_type/ua_name)。 + * + * raw 表字段(见 app/admin/scheme/NginxAccessLog.php): + * file_path / remote_addr / remote_user / time_local / method / uri / query_string / + * http_version / status / body_bytes_sent / http_referer / http_user_agent / + * request_time / upstream_response_time / bytes_sent / country / province / city / create_time + */ + protected function buildRow(array $parsed, string $filePath): array + { + $now = time(); + return [ + 'file_path' => $filePath, + 'remote_addr' => $parsed['remote_addr'], + 'remote_user' => $parsed['remote_user'] ?? '', + 'time_local' => $parsed['time_local'], + 'method' => $parsed['method'], + 'uri' => $parsed['uri'], + 'query_string' => $parsed['query_string'] ?? '', + 'http_version' => $parsed['http_version'] ?? '', + 'status' => $parsed['status'], + 'body_bytes_sent' => $parsed['body_bytes_sent'], + 'http_referer' => $parsed['http_referer'] ?? '', + 'http_user_agent' => $parsed['http_user_agent'] ?? '', + 'request_time' => $parsed['request_time'], + 'upstream_response_time' => $parsed['upstream_response_time'], + 'bytes_sent' => $parsed['bytes_sent'], + 'country' => null, + 'province' => null, + 'city' => null, + 'create_time' => $now, + ]; + } + + /** + * 批量入库(使用 model 的 insertAll,自动走默认连接 + 表前缀)。 + */ + protected function insertBatch(array $batch): void + { + if (empty($batch)) { + return; + } + // NginxAccessLog::insertAll 走 model,schema 字段映射可控; + // chunk 由调用方控制(INSERT_CHUNK=500),避免单条 SQL 过大 + NginxAccessLog::insertAll($batch); + } + + /** + * 更新 position 的失败统计(offset/inode 不变,仅覆盖 fail 字段)。 + * + * 注意:Reader 在 maxLines/EOF 已经 savePosition(failCount=0), + * 这里取当前 position 后用相同 offset 重写,仅更新 failCount / failSamples。 + */ + protected function updateFailStats( + \app\common\service\NginxLogReader $reader, + string $file, + int $fails, + array $samples + ): void { + $position = $reader->getLastPosition($file); + if ($position === null) { + return; // Reader 已经保存过,这里取不到说明异常,放弃覆盖 + } + $samplesJson = empty($samples) ? null : json_encode($samples, JSON_UNESCAPED_UNICODE); + $reader->savePosition( + $file, + (int) $position['inode'], + (int) $position['offset'], + $position['last_line_hash'] ?? null, + $fails, + $samplesJson + ); + } + + /** + * 读取文件首个非空行(用于格式自检). + * + * @return string|null 首个非空行(trim 后),空文件返回 null + */ + protected function readFirstNonEmptyLine(string $file): ?string + { + $fp = @fopen($file, 'rb'); + if ($fp === false) { + return null; + } + try { + while (($line = fgets($fp)) !== false) { + $line = trim($line); + if ($line !== '') { + return $line; + } + } + return null; + } finally { + if (is_resource($fp)) { + fclose($fp); + } + } + } +}