feat(nginx-log): NginxLogImport 定时任务 + timer config 注册（Base/App 双层）

app/tools/controller/timer/NginxLogImport.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\tools\controller\timer;
+
+use base\tools\controller\timer\NginxLogImportBase;
+
+/**
+ * nginx 访问日志导入定时任务（App 入口类）.
+ *
+ * 框架路由 /tools/timer.NginxLogImport/do 与 /tools/timer.NginxLogImport/execute
+ * 均通过本类入口实例化（依赖倒置：定时器框架/路由系统直接调本类，不调 Base）。
+ *
+ * 业务侧如需定制导入行为（如改批大小、加自定义过滤、改入库目标表等），
+ * 重写本类对应方法即可拦截，无需改动 extend/base/。
+ */
+class NginxLogImport extends NginxLogImportBase
+{
+}

extend/base/common/command/timer/config.php

@@ -21,4 +21,12 @@ return [
         'frequency' => 86400,
         'concurrency' => 1,
     ],
+    [
+        'name' => 'nginx_log_import',
+        'type' => 'site',
+        'target' => '/tools/timer.NginxLogImport/do',
+        'frequency' => 300,
+        'concurrency' => 1,
+        'run_type' => 'auto',
+    ],
 ];

extend/base/tools/controller/timer/NginxLogImportBase.php

@@ -0,0 +1,281 @@
+<?php
+
+declare(strict_types=1);
+
+namespace base\tools\controller\timer;
+
+use app\admin\model\NginxAccessLog;
+use app\common\controller\TimerController;
+use think\facade\Log;
+
+/**
+ * nginx 访问日志导入定时任务（Base 层）.
+ *
+ * 流程：
+ *   1. env 总开关校验（严格布尔判断，详见 notepad Task 2 #14）
+ *   2. 读取 env 配置的日志文件清单
+ *   3. 依赖倒置：实例化 app 层 Parser / Reader
+ *   4. 每个文件首行格式自检 + 串行增量读取 + 批量入库
+ *   5. savePosition 写 parse_fail_count / parse_fail_samples
+ *
+ * 业务侧如需定制导入逻辑，重写 app/tools/controller/timer/NginxLogImport.php 对应方法即可拦截。
+ */
+class NginxLogImportBase extends TimerController
+{
+    /**
+     * 防刷间隔（秒）.
+     * 仅做控制器侧防刷，不影响定时器侧 frequency 节流。
+     */
+    protected $frequency = 60;
+
+    /**
+     * 并发上限 cap：单分片（id=0）。
+     */
+    protected $concurrency = 1;
+
+    /**
+     * 批量 insert chunk 大小。
+     */
+    protected const INSERT_CHUNK = 500;
+
+    /**
+     * 单次读取最大行数（与 Reader 默认上限一致）。
+     */
+    protected const READ_MAX_LINES = 100000;
+
+    /**
+     * parse_fail_samples 最大记录条数。
+     */
+    protected const FAIL_SAMPLES_MAX = 10;
+
+    public function do()
+    {
+        // 1. 严格布尔判断（env convert 'true'→true / 'false'→false）
+        //    见 notepad Task 2 #14：必须 !== true，不能用 !== 'true' 字符串比较
+        if (env('nginx_log.enable') !== true) {
+            Log::debug('nginx_log disabled, skip');
+            return 'disabled';
+        }
+
+        // 2. 读日志文件清单（分号分隔）
+        $logFilesConfig = (string) env('nginx_log.log_files', '');
+        $files = array_filter(array_map('trim', explode(';', $logFilesConfig)), function ($f) {
+            return $f !== '';
+        });
+        if (empty($files)) {
+            return 'no_files';
+        }
+
+        // 3. 依赖倒置：使用 app 层入口类（业务侧可重写拦截）
+        $parser = new \app\common\service\NginxLogParser();
+        $reader = new \app\common\service\NginxLogReader();
+
+        // 静态资源过滤开关
+        $excludeStatic = (int) sysconfig('nginx_log', 'exclude_static', 1) === 1;
+
+        $totalFiles = 0;
+        $totalLines = 0;
+        $totalFails = 0;
+        $errors = [];
+
+        foreach ($files as $file) {
+            try {
+                if (!is_file($file)) {
+                    throw new \RuntimeException("文件不存在: {$file}");
+                }
+
+                // 4. 首行格式自检（避免对非 nginx 日志文件做无意义解析）
+                //    取首个非空行喂给 parser->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);
+            }
+        }
+    }
+}