feat: 断点续传与色情识别提速/修复,支持 Linux 交叉编译

- feat: 新增续传表 scan_log,重扫跳过已归类文件(尺寸/头尾指纹校验)
- fix: 色情识别记录批量入库超 SQLite 999 变量上限,改每 100 条一批
- perf: NudeNet 检测改为多线程推理池(每线程独立模型实例)
- perf: 明暗分析 JPEG(YCbCr) 快路径;SQLite 读并发放开;加阶段耗时统计
- fix: 前端色情识别状态由后端如实上报(配置开关+检测服务是否起来)
- feat: ffprobe 内嵌按平台拆分,Linux 构建回退系统 ffprobe,支持交叉编译
- docs: README 补充断点续传与 CentOS 交叉编译步骤
- chore: organize.workers 默认 8
This commit is contained in:
2025-06-19 21:07:35 +08:00
parent ea134afa27
commit 957dfe1134
18 changed files with 908 additions and 111 deletions
+137 -41
View File
@@ -15,6 +15,7 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net"
@@ -48,17 +49,20 @@ const requestTimeout = 30 * time.Second
// Detector 封装一个本地 python 服务的客户端。
// 多个探测协程会并发调 Detect,底层 HTTP 客户端线程安全;
// 服务端单线程处理,请求自己排队,不用额外加锁。
// 服务端用固定大小线程池并行推理(每个 worker 一份独立模型实例),请求不必排队。
type Detector struct {
url string // 本地服务地址,如 http://127.0.0.1:54321
threshold float64 // 判定阈值 0~1
client *http.Client
cmd *exec.Cmd // python 进程句柄(Stop 时杀掉)
scriptDir string // 临时脚本所在目录(Stop 时清理)
mu sync.Mutex // 保护下面的关闭状态
python string // python 可执行文件路径(服务重启时要用)
modelDir string // 模型目录 lib(服务重启时要用)
mu sync.Mutex // 保护 stopped / 进程句柄
stopped bool // 已经 Stop 过了
cancel context.CancelFunc
done chan struct{} // 进程退出时关闭(Wait 协程负责)
restartMu sync.Mutex // 串行化重启:同一时刻只一个 goroutine 重启,其余排队等
}
// Start 拉起本地 python 检测服务并等它就绪。
@@ -87,13 +91,6 @@ func Start(pythonPath string, threshold float64) (*Detector, error) {
return nil, fmt.Errorf("写入检测脚本失败: %v", err)
}
// 找一个空闲端口
port, err := freePort()
if err != nil {
_ = os.RemoveAll(tmpDir)
return nil, fmt.Errorf("找空闲端口失败: %v", err)
}
// 模型目录:程序运行目录下的 lib(和 application.properties / media-info.db 同一根目录)。
// 服务脚本优先加载 lib/320n.onnx;本地没有会自己下载到那里,下载失败再回退用 pip 自带模型。
modelDir := ""
@@ -101,27 +98,51 @@ func Start(pythonPath string, threshold float64) (*Detector, error) {
modelDir = filepath.Join(wd, "lib")
}
// 拉起服务。标准输出/错误接到程序自身的输出,方便看 python 报错。
d := &Detector{
threshold: threshold,
client: &http.Client{Timeout: requestTimeout},
python: py,
modelDir: modelDir,
scriptDir: tmpDir,
}
// 拉起服务(同样一套逻辑也用于崩溃后的重启)
if err := d.launch(); err != nil {
_ = os.RemoveAll(tmpDir)
return nil, err
}
return d, nil
}
// launch 启动(或重启)python 检测服务,并等它就绪(模型加载完成)。
// 调用方需要保证持有 mu,且 stopped 为 false。
func (d *Detector) launch() error {
// 先把上一次的进程收拾掉(首次启动时没有,直接跳过)
d.teardown()
// 找一个空闲端口
port, err := freePort()
if err != nil {
return fmt.Errorf("找空闲端口失败: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
cmd := exec.CommandContext(ctx, py, scriptPath, fmt.Sprintf("%d", port), modelDir)
cmd := exec.CommandContext(ctx, d.python, filepath.Join(d.scriptDir, "server.py"), fmt.Sprintf("%d", port), d.modelDir)
// 标准输出/错误都接到程序自身的输出:
// - 扫描期间 os.Stdout 会被接管进"运行日志",这样 python 的报错(崩溃原因)也能在前端看到
// - 方便排查服务为什么挂掉
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
cmd.Stderr = os.Stdout
if err := cmd.Start(); err != nil {
cancel()
_ = os.RemoveAll(tmpDir)
return nil, fmt.Errorf("启动检测服务失败: %v", err)
return fmt.Errorf("启动检测服务失败: %v", err)
}
done := make(chan struct{})
d := &Detector{
url: fmt.Sprintf("http://127.0.0.1:%d", port),
threshold: threshold,
client: &http.Client{Timeout: requestTimeout},
cmd: cmd,
scriptDir: tmpDir,
cancel: cancel,
done: done,
}
d.url = fmt.Sprintf("http://127.0.0.1:%d", port)
d.cmd = cmd
d.cancel = cancel
d.done = done
// 后台等进程退出:ProcessState 要 Wait 之后才有值,这里用 done 通知,
// 这样 waitReady 能及时发现"python 没装 nudenet 直接退掉"的情况。
go func() {
@@ -131,10 +152,32 @@ func Start(pythonPath string, threshold float64) (*Detector, error) {
// 等服务就绪(模型加载完成)。就绪前先探测 /health。
if err := d.waitReady(); err != nil {
d.Stop()
return nil, err
d.teardown()
return err
}
return d, nil
return nil
}
// teardown 停掉当前 python 进程并清理句柄(不删临时目录)。
// 调用方需要保证持有 mu。进程可能已经死了(崩溃),kill 失败也无所谓。
func (d *Detector) teardown() {
if d.cancel != nil {
d.cancel()
d.cancel = nil
}
if d.cmd != nil && d.cmd.Process != nil {
_ = d.cmd.Process.Kill()
}
if d.done != nil {
// 等 Wait 协程回收进程,避免僵尸进程;进程已死则立刻返回。
// 等超过 3 秒就放弃,防止极端情况下卡死重启流程。
select {
case <-d.done:
case <-time.After(3 * time.Second):
}
d.done = nil
}
d.cmd = nil
}
// waitReady 轮询本地服务,直到能正常应答。
@@ -168,14 +211,73 @@ func (d *Detector) waitReady() error {
// - score:命中的敏感类里最高分数(没有命中为 0)
// - nsfw:是否判定为色情(有敏感类且分数 >= 阈值)
// - err:检测失败(服务挂了/图读不出等),由调用方兜底按未命中处理
//
// 服务失联(连接被拒,多半是 python 进程在处理某张图时崩了)时,会串行化地
// 重启服务并重试一次:多个并发 worker 里只有一个负责重启,其余排队,等重启完
// 也重试自己的图——保证服务短暂挂掉不会漏掉一整批图片。
func (d *Detector) Detect(path string) (score float64, nsfw bool, err error) {
score, nsfw, err = d.detectOnce(path)
if err == nil {
return score, nsfw, nil
}
// 只有"发请求都失败"(连接被拒/重置,说明服务失联)才走重启;
// 服务正常返回的明确错误(如某张图读不了)不重启。
var ce *connError
if !errors.As(err, &ce) {
return score, nsfw, err
}
d.restartMu.Lock()
defer d.restartMu.Unlock()
if d.isStopped() {
return 0, false, err
}
// 可能排在前面的 goroutine 已经把服务重启好了:先探活,活着就直接重试,别反复重启
if d.healthy() {
return d.detectOnce(path)
}
fmt.Printf("=== 色情检测服务失联(%v), 正在重启…\n", ce.err)
if rerr := d.launch(); rerr != nil {
return 0, false, fmt.Errorf("检测服务重启失败: %v", rerr)
}
fmt.Println("=== 色情检测服务已重启,继续识别")
return d.detectOnce(path)
}
// healthy 快速探活:本地服务是否还能正常应答(带短超时,避免拖慢扫描)。
func (d *Detector) healthy() bool {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, d.url+"/health", nil)
if err != nil {
return false
}
resp, err := d.client.Do(req)
if err != nil {
return false
}
defer resp.Body.Close()
return resp.StatusCode == http.StatusOK
}
// isStopped 判断检测器是否已被 Stop(线程安全)。
func (d *Detector) isStopped() bool {
d.mu.Lock()
defer d.mu.Unlock()
return d.stopped
}
// detectOnce 发一次检测请求(不带重启逻辑)。
func (d *Detector) detectOnce(path string) (score float64, nsfw bool, err error) {
body, _ := json.Marshal(map[string]interface{}{
"path": path,
"threshold": d.threshold,
})
resp, err := d.client.Post(d.url+"/detect", "application/json", bytes.NewReader(body))
if err != nil {
return 0, false, err
// 打上"连接失败"标记,供 Detect 判断是否要重启
return 0, false, &connError{err}
}
defer resp.Body.Close()
data, err := io.ReadAll(resp.Body)
@@ -197,27 +299,21 @@ func (d *Detector) Detect(path string) (score float64, nsfw bool, err error) {
return out.Score, out.Nsfw, nil
}
// connError 标记"请求根本没发出去/连不上本地服务"的错误(服务失联)。
type connError struct{ err error }
func (e *connError) Error() string { return e.err.Error() }
func (e *connError) Unwrap() error { return e.err }
// Stop 停掉检测服务并清理临时文件。可以安全地重复调用。
func (d *Detector) Stop() {
d.mu.Lock()
defer d.mu.Unlock()
if d.stopped {
d.mu.Unlock()
return
}
d.stopped = true
d.mu.Unlock()
if d.cancel != nil {
d.cancel()
}
if d.cmd != nil && d.cmd.Process != nil {
// 进程活着才 kill;kill 失败也无所谓,服务很快就没了
_ = d.cmd.Process.Kill()
}
if d.done != nil {
// 等 Wait 协程回收进程,避免僵尸进程;进程已死则立刻返回
<-d.done
}
d.teardown()
if d.scriptDir != "" {
_ = os.RemoveAll(d.scriptDir)
}
+47
View File
@@ -0,0 +1,47 @@
package nsfw
import (
"encoding/base64"
"os"
"path/filepath"
"testing"
"time"
)
// 1x1 透明 PNG,当测试图片用(NudeNet 会自己缩放)。
const testPNG = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
// TestDetectRestart 验证:python 检测服务中途崩溃后,Detect 会自动重启并恢复正常。
func TestDetectRestart(t *testing.T) {
py := resolvePython("")
if py == "" {
t.Skip("无 python")
}
img := filepath.Join(t.TempDir(), "test.png")
if data, err := base64.StdEncoding.DecodeString(testPNG); err != nil {
t.Fatalf("解码测试图片失败: %v", err)
} else if err := os.WriteFile(img, data, 0644); err != nil {
t.Fatalf("写测试图片失败: %v", err)
}
d, err := Start(py, 0.5)
if err != nil {
t.Fatalf("Start: %v", err)
}
defer d.Stop()
if _, _, err := d.Detect(img); err != nil {
t.Fatalf("崩溃前检测失败: %v", err)
}
// 直接杀掉 python 进程,模拟处理某张图时崩溃
if d.cmd != nil && d.cmd.Process != nil {
_ = d.cmd.Process.Kill()
}
time.Sleep(2 * time.Second) // 等进程退出、端口释放
if _, _, err := d.Detect(img); err != nil {
t.Fatalf("崩溃后检测失败(应自动重启): %v", err)
}
t.Log("重启后检测成功")
}
+81 -21
View File
@@ -1,8 +1,10 @@
# -*- coding: utf-8 -*-
"""NudeNet 本地检测服务:由 Go 侧启动,常驻提供 /health 和 /detect。
只依赖 Python 标准库 + nudenet 包,模型在进程里只加载一次。
收到 /detect 就把图片喂给 NudeDetector,返回结果 JSON。
只依赖 Python 标准库 + nudenet 包。
onnxruntime 的 InferenceSession 不是线程安全的,同一个实例并发推理会崩,
所以这里用一个固定大小的推理线程池:每个 worker 线程各持一份独立模型实例,
N 个请求可以真·并行推理。收到 /detect 就把图片喂给 NudeDetector,返回结果 JSON。
模型加载优先级(模型放在 <程序运行目录>/lib/320n.onnx):
1. 本地已有该文件 → 直接加载(离线可用,适合随程序分发);
@@ -15,12 +17,15 @@ Windows 超长路径说明:OpenCV 的 imread 走老式 C API,路径超过 26
再把字节直接交给 NudeNet(v3 的 detect 支持 bytes)——从内存解码,彻底
绕开路径限制,同时兼容老版 v2(只收路径时落到临时短路径文件再试)。
"""
import concurrent.futures
import json
import os
import shutil
import sys
import tempfile
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from nudenet import NudeDetector
@@ -49,6 +54,17 @@ MODEL_DIR = sys.argv[2] if len(sys.argv) > 2 else ""
_detector = None
# 推理并发数:onnxruntime 的 InferenceSession.run() 不是线程安全的,
# 同一个实例并发推理会崩,所以每个 worker 线程各持有一个独立实例(NudeDetector),
# N 个 worker 即可真·并行推理。默认 4,可用环境变量 KM_NSFW_WORKERS 调整。
INFER_WORKERS = int(os.environ.get("KM_NSFW_WORKERS", "4") or "4")
if INFER_WORKERS < 1:
INFER_WORKERS = 1
_executor = None # 推理线程池(固定 N 个 worker,模型各持一份)
_thread_local = threading.local() # 每个 worker 线程自己的检测器
_MODEL_PATH = None # 启动时解析好的模型路径(None 表示用包自带模型)
def _looks_like_model(path):
"""粗略校验文件是不是真 ONNX 模型:开头不是 HTML、大小不小于 1MB(320n 模型约 11.6MB)。
@@ -65,6 +81,13 @@ def _looks_like_model(path):
return False
def _download_model(local):
"""带超时地把模型下载到 local。urlretrieve 没有超时,网络挂起会卡死整个服务。"""
req = urllib.request.Request(MODEL_URL)
with urllib.request.urlopen(req, timeout=30) as resp, open(local, "wb") as f:
shutil.copyfileobj(resp, f)
def _ensure_model(model_dir):
"""返回要加载的模型路径;返回 None 表示用 nudenet 包自带的模型。"""
if not model_dir:
@@ -73,11 +96,11 @@ def _ensure_model(model_dir):
if os.path.isfile(local) and _looks_like_model(local):
print("=== 使用本地模型: %s" % local, flush=True)
return local
# 本地没有(或文件无效)→ 下载到 lib 目录(仅第一次需要网络)
# 本地没有(或文件无效)→ 下载到 lib 目录(仅第一次需要网络,带 30 秒超时)
try:
os.makedirs(model_dir, exist_ok=True)
print("=== 本地未找到有效模型, 正在下载: %s" % local, flush=True)
urllib.request.urlretrieve(MODEL_URL, local)
_download_model(local)
if _looks_like_model(local):
print("=== 模型下载完成: %s" % local, flush=True)
return local
@@ -93,22 +116,33 @@ def _ensure_model(model_dir):
def get_detector():
global _detector
"""确保模型已就绪(下载/加载),返回一个检测器实例。
启动时调用一次:把模型下载/解析好,/health 就绪即代表模型可用。
真正推理用的独立实例由 _new_detector() 按线程各建一份,共享这份模型路径。
"""
global _detector, _MODEL_PATH
if _detector is None:
mp = _ensure_model(MODEL_DIR)
try:
_detector = NudeDetector(model_path=mp)
except Exception as exc: # noqa: BLE001 - 指定路径加载失败就回退包自带模型,别崩
print("=== 本地模型加载失败(%s), 回退用包自带模型" % exc, flush=True)
if mp:
try:
os.remove(mp)
except OSError:
pass
_detector = NudeDetector()
_MODEL_PATH = _ensure_model(MODEL_DIR)
_detector = _new_detector()
return _detector
def _new_detector():
"""创建一份独立检测器实例(每个推理 worker 各持一份,可并行推理)。"""
mp = _MODEL_PATH
try:
return NudeDetector(model_path=mp) if mp else NudeDetector()
except Exception as exc: # noqa: BLE001 - 指定路径加载失败就回退包自带模型,别崩
print("=== 本地模型加载失败(%s), 回退用包自带模型" % exc, flush=True)
if mp:
try:
os.remove(mp)
except OSError:
pass
return NudeDetector()
def _longpath(path):
"""Windows 下给绝对路径加 \\\\?\\ 前缀,让 Python 原生 open 也能处理超长路径。"""
if os.name != "nt":
@@ -136,7 +170,10 @@ def _read_bytes(path):
def _detect_bytes(det, data, path):
"""把图片字节喂给 NudeNet。v3 直接吃 bytes;老版 v2 只收路径时落到临时短路径重试。"""
"""把图片字节喂给 NudeNet。v3 直接吃 bytes;老版 v2 只收路径时落到临时短路径重试。
用线程自己的检测器实例推理,各实例互不干扰,可以真·并行。
"""
try:
return det.detect(data)
except (TypeError, AttributeError):
@@ -154,6 +191,19 @@ def _detect_bytes(det, data, path):
pass
def _infer(path):
"""整个推理任务(读文件 + 检测),提交到固定线程池执行。
读文件在池外并行进行;推理用线程各自的独立检测器,N 个 worker 真·并行。
"""
data = _read_bytes(path)
det = getattr(_thread_local, "detector", None)
if det is None:
det = _new_detector()
_thread_local.detector = det
return _detect_bytes(det, data, path)
class Handler(BaseHTTPRequestHandler):
def _send(self, obj, status=200):
data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
@@ -179,8 +229,8 @@ class Handler(BaseHTTPRequestHandler):
threshold = float(body.get("threshold", 0.5))
if not path:
return self._send({"ok": False, "error": "缺少图片路径"})
data = _read_bytes(path)
detections = _detect_bytes(get_detector(), data, path)
# 丢给推理线程池:读文件并行、推理由 N 个 worker 并行处理
detections = _executor.submit(_infer, path).result()
except Exception as exc: # noqa: BLE001 - 服务端任何异常都要回给 Go,别让进程崩
return self._send({"ok": False, "error": str(exc)})
@@ -200,9 +250,19 @@ class Handler(BaseHTTPRequestHandler):
pass
# 多线程接收连接 + 大监听队列:Go 侧会并发发请求(worker 数可到 8),
# 单线程 HTTPServer 默认 backlog 只有 5,并发一高连接就被内核拒绝(报 10061 连接拒绝)。
# ThreadingHTTPServer 让连接即时被 accept;推理再交给固定线程池并行执行
# (每个 worker 一份独立模型实例,onnxruntime 同一实例并发推理会崩)。
class Server(ThreadingHTTPServer):
request_queue_size = 128
if __name__ == "__main__":
port = int(sys.argv[1]) if len(sys.argv) > 1 else 0
# 启动前先把模型准备好(本地加载 / 下载 / 回退),/health 就绪即代表模型可用,
# 这样 Go 侧 waitReady 等到的就是"模型已就绪"的服务。
get_detector()
HTTPServer(("127.0.0.1", port), Handler).serve_forever()
# 固定大小推理线程池:每个 worker 线程持一份独立模型实例,可并行推理
_executor = concurrent.futures.ThreadPoolExecutor(max_workers=INFER_WORKERS)
Server(("127.0.0.1", port), Handler).serve_forever()