海康电表维护

doum

10 小时以前 ce44d803b73a65b2cc31db5bcc662139029463d3

海康电表维护

 admin/src/api/business/collectionMedia.js

@@ -102,3 +102,16 @@
    return new Blob([blob], { type: 'video/mp4' })
  })
}

export function analyzeMediaSnapshot (id) {
  return request.post(`/visitsAdmin/cloudService/business/collectionStation/media/snapshot/analyze/${id}`, {})
}

export function fetchMediaSnapshots (id, cacheBust) {
  const config = cacheBust ? { params: { _t: cacheBust } } : {}
  return request.get(`/visitsAdmin/cloudService/business/collectionStation/media/snapshot/${id}`, config)
}

export function saveManualMediaSnapshot (data) {
  return request.post('/visitsAdmin/cloudService/business/collectionStation/media/snapshot/manual', data)
}

 admin/src/views/business/collectionMedia.vue

@@ -51,19 +51,38 @@
        <el-table-column prop="endTime" label="结束时间" min-width="160" />
        <el-table-column prop="downloadStatus" label="下载状态" width="90">
          <template slot-scope="{row}">
            <span v-if="row.downloadStatus === 1">已下载</span>
            <span v-else-if="row.downloadStatus === 2">失败</span>
            <span v-else-if="row.downloadStatus === 3">下载中</span>
            <span v-else>待下载</span>
            <span :class="downloadStatusClass(row)">{{ downloadStatusLabel(row) }}</span>
          </template>
        </el-table-column>
        <el-table-column prop="filePathLocal" label="本地路径" min-width="160" show-overflow-tooltip />
        <el-table-column label="操作" width="220" fixed="right">
        <el-table-column prop="snapshotStatus" label="快照" width="90">
          <template slot-scope="{row}">
            <el-button type="text" v-if="canPreview(row)" @click="handlePreview(row)">预览</el-button>
            <span :class="snapshotStatusClass(row)">{{ snapshotStatusLabel(row) }}</span>
          </template>
        </el-table-column>
        <el-table-column prop="filePathLocal" label="本地路径" min-width="120">
          <template slot-scope="{row}">
            <div v-if="isDownloadedVideo(row)" class="list-video-cell" @click="handlePreview(row)">
              <video
                :key="'thumb-' + row.id"
                :src="buildListVideoSrc(row)"
                class="list-video-thumb"
                muted
                preload="metadata"
                playsinline
              />
            </div>
            <span v-else class="path-text" :title="row.filePathLocal">{{ row.filePathLocal || '-' }}</span>
          </template>
        </el-table-column>
        <el-table-column label="操作" width="240" fixed="right">
          <template slot-scope="{row}">
            <el-button type="text" v-if="canDownloadToStation(row)" v-permissions="['business:collectionMedia:download', 'business:collectionStation:sync']"
              @click="handleDownload(row)">下载</el-button>
            <el-button type="text" v-if="row.downloadStatus === 1" @click="handleSaveLocal(row)">下载到本地</el-button>
            <el-button type="text" v-if="row.downloadStatus === 1 && row.snapshotStatus !== 1"
              @click="handleAnalyzeSnapshot(row)">生成快照</el-button>
            <el-button type="text" v-if="row.downloadStatus === 1" @click="openManualCorrect(row)">人工纠正</el-button>
            <el-button type="text" v-if="row.snapshotStatus === 2" @click="handleViewSnapshot(row)">查看快照</el-button>
          </template>
        </el-table-column>
      </el-table>
@@ -82,6 +101,44 @@
        <div v-else-if="!previewLoading" class="preview-empty">暂不支持该类型预览</div>
      </div>
    </el-dialog>

    <el-dialog :title="snapshotTitle" :visible.sync="snapshotVisible" width="720px" append-to-body>
      <div v-loading="snapshotLoading" class="snapshot-wrap">
        <div v-if="snapshotList.length" class="snapshot-grid">
          <div v-for="item in snapshotList" :key="item.id + '-' + snapshotCacheVersion" class="snapshot-item">
            <div class="snapshot-label">{{ snapshotTypeLabel(item.snapshotType) }}</div>
            <el-image v-if="item.fileUrlFull" :src="buildSnapshotImageUrl(item.fileUrlFull)" fit="contain"
              class="snapshot-image" :preview-src-list="[buildSnapshotImageUrl(item.fileUrlFull)]" />
            <div class="snapshot-meta">
              <span>时刻: {{ item.timestampSec }}s</span>
              <span v-if="item.confidence">置信度: {{ item.confidence }}</span>
              <span>来源: {{ item.source || '-' }}</span>
            </div>
          </div>
        </div>
        <div v-else-if="!snapshotLoading" class="preview-empty">暂无快照，请先点击「生成快照」</div>
        <div v-if="snapshotMediaRow" class="snapshot-actions">
          <el-button type="primary" plain @click="openManualCorrect(snapshotMediaRow)">人工纠正时刻</el-button>
        </div>
      </div>
    </el-dialog>

    <el-dialog title="人工纠正快照时刻" :visible.sync="manualCorrectVisible" width="860px" append-to-body
      @close="closeManualCorrect">
      <div v-loading="manualCorrectLoading" class="manual-correct-wrap">
        <video v-if="manualPreviewSrc" ref="manualVideo" :src="manualPreviewSrc" controls preload="metadata"
          playsinline class="preview-video" @loadedmetadata="onManualVideoLoaded" @timeupdate="onManualTimeUpdate" />
        <div class="manual-slider">
          <span>当前时刻: {{ manualTimestampSec.toFixed(1) }}s</span>
          <el-slider v-model="manualTimestampSec" :min="0" :max="manualDurationSec" :step="0.5"
            @input="seekManualVideo" />
        </div>
        <div class="manual-buttons">
          <el-button type="primary" :loading="manualSaving" @click="saveManualSnapshot(1)">设为门头图</el-button>
          <el-button type="success" :loading="manualSaving" @click="saveManualSnapshot(2)">设为交付图</el-button>
        </div>
      </div>
    </el-dialog>
  </TableLayout>
</template>

@@ -90,7 +147,16 @@
import TableLayout from '@/layouts/TableLayout'
import Pagination from '@/components/common/Pagination'
import { syncMedia, downloadMedia, batchDownloadMedia, list as fetchStationList } from '@/api/business/collectionStation'
import { fetchPreviewText, fetchPreviewBlob, fetchMediaFile, ensureMp4Blob, buildPreviewUrl } from '@/api/business/collectionMedia'
import {
  fetchPreviewText,
  fetchPreviewBlob,
  fetchMediaFile,
  ensureMp4Blob,
  buildPreviewUrl,
  analyzeMediaSnapshot,
  fetchMediaSnapshots,
  saveManualMediaSnapshot
} from '@/api/business/collectionMedia'

export default {
  name: 'CollectionMedia',
@@ -113,7 +179,23 @@
      previewRow: null,
      previewUseDirectUrl: false,
      previewBlobUrl: '',
      downloadPollTimer: null
      downloadPollTimer: null,
      snapshotVisible: false,
      snapshotLoading: false,
      snapshotTitle: '配送快照',
      snapshotList: [],
      snapshotPollTimer: null,
      snapshotMediaRow: null,
      manualCorrectVisible: false,
      manualCorrectLoading: false,
      manualCorrectRow: null,
      manualPreviewSrc: '',
      manualBlobUrl: '',
      manualTimestampSec: 0,
      manualDurationSec: 600,
      manualSaving: false,
      manualSeekFromSlider: false,
      snapshotCacheVersion: 0
    }
  },
  created () {
@@ -131,7 +213,9 @@
  },
  beforeDestroy () {
    this.revokePreviewUrl()
    this.revokeManualPreviewUrl()
    this.stopDownloadPoll()
    this.stopSnapshotPoll()
  },
  methods: {
    loadStations () {
@@ -151,8 +235,11 @@
      if (row.mediaType === 2) return '音频'
      return '视频'
    },
    canPreview (row) {
      return row.downloadStatus === 1 && (row.fileUrlFull || row.filePathLocal)
    isDownloadedVideo (row) {
      return row.downloadStatus === 1 && row.filePathLocal && this.resolvePreviewMode(row) === 'video'
    },
    buildListVideoSrc (row) {
      return row.fileUrlFull || buildPreviewUrl(row.id)
    },
    canDownloadToStation (row) {
      return row.downloadStatus !== 1 && row.downloadStatus !== 3
@@ -199,6 +286,94 @@
      }).catch(err => {
        this.$message.error(err.message || '下载到本地失败')
      })
    },
    snapshotTypeLabel (type) {
      return type === 2 ? '货品交付图' : '到店门头图'
    },
    downloadStatusLabel (row) {
      const map = { 0: '待下载', 1: '已下载', 2: '失败', 3: '下载中' }
      return map[row.downloadStatus] != null ? map[row.downloadStatus] : '待下载'
    },
    downloadStatusClass (row) {
      const map = { 0: 'status-info', 1: 'status-success', 2: 'status-danger', 3: 'status-primary' }
      return map[row.downloadStatus] || 'status-info'
    },
    snapshotStatusLabel (row) {
      if (row.snapshotStatus === 2) return '已完成'
      if (row.snapshotStatus === 1) return '分析中'
      if (row.snapshotStatus === 3) return '失败'
      if (row.downloadStatus === 1) return '未分析'
      return '-'
    },
    snapshotStatusClass (row) {
      if (row.snapshotStatus === 2) return 'status-success'
      if (row.snapshotStatus === 1) return 'status-warning'
      if (row.snapshotStatus === 3) return 'status-danger'
      if (row.downloadStatus === 1) return 'status-info'
      return 'status-muted'
    },
    buildSnapshotImageUrl (url) {
      if (!url) return ''
      const sep = url.indexOf('?') >= 0 ? '&' : '?'
      return `${url}${sep}_t=${this.snapshotCacheVersion}`
    },
    handleAnalyzeSnapshot (row) {
      analyzeMediaSnapshot(row.id).then(res => {
        this.$message.success(res || '已提交快照分析')
        this.search()
        this.startSnapshotPoll(row.id)
      })
    },
    handleViewSnapshot (row) {
      this.snapshotMediaRow = row
      this.snapshotTitle = (row.fileName || '媒体') + ' - é…é€å¿«ç…§'
      this.snapshotVisible = true
      this.loadSnapshots(row.id)
    },
    loadSnapshots (mediaId) {
      this.snapshotLoading = true
      this.snapshotList = []
      this.snapshotCacheVersion = Date.now()
      fetchMediaSnapshots(mediaId, this.snapshotCacheVersion).then(list => {
        this.snapshotList = list || []
        this.snapshotLoading = false
      }).catch(err => {
        this.snapshotLoading = false
        this.$message.error(err.message || '加载快照失败')
      })
    },
    startSnapshotPoll (mediaId) {
      this.stopSnapshotPoll()
      let count = 0
      this.snapshotPollTimer = setInterval(() => {
        count++
        if (count > 40) {
          this.stopSnapshotPoll()
          return
        }
        this.api.fetchList({
          page: this.tableData.pagination.pageIndex,
          capacity: this.tableData.pagination.pageSize,
          model: this.searchForm,
          sorts: this.tableData.sorts
        }).then(data => {
          this.tableData.list = data.records
          this.tableData.pagination.total = data.total
          const row = (data.records || []).find(item => item.id === mediaId)
          if (row && row.snapshotStatus !== 1) {
            this.stopSnapshotPoll()
            if (row.snapshotStatus === 2 && this.snapshotVisible) {
              this.loadSnapshots(mediaId)
            }
          }
        }).catch(() => {})
      }, 3000)
    },
    stopSnapshotPoll () {
      if (this.snapshotPollTimer) {
        clearInterval(this.snapshotPollTimer)
        this.snapshotPollTimer = null
      }
    },
    startDownloadPoll () {
      this.stopDownloadPoll()
@@ -333,6 +508,92 @@
        URL.revokeObjectURL(this.previewBlobUrl)
        this.previewBlobUrl = ''
      }
    },
    openManualCorrect (row) {
      if (row.downloadStatus !== 1) {
        this.$message.warning('请先下载媒体文件')
        return
      }
      this.manualCorrectRow = row
      this.manualCorrectVisible = true
      this.manualCorrectLoading = true
      this.manualTimestampSec = 0
      this.manualDurationSec = 600
      this.revokeManualPreviewUrl()
      fetchPreviewBlob(row.id)
        .then(blob => ensureMp4Blob(blob))
        .then(blob => {
          this.manualBlobUrl = URL.createObjectURL(blob)
          this.manualPreviewSrc = this.manualBlobUrl
          this.manualCorrectLoading = false
        })
        .catch(err => {
          this.manualCorrectLoading = false
          this.$message.error(err.message || '加载视频失败')
        })
    },
    closeManualCorrect () {
      this.revokeManualPreviewUrl()
      this.manualPreviewSrc = ''
      this.manualCorrectRow = null
    },
    revokeManualPreviewUrl () {
      if (this.manualBlobUrl) {
        URL.revokeObjectURL(this.manualBlobUrl)
        this.manualBlobUrl = ''
      }
    },
    onManualVideoLoaded () {
      const video = this.$refs.manualVideo
      if (video && video.duration && isFinite(video.duration)) {
        this.manualDurationSec = Math.max(1, Math.floor(video.duration))
      }
    },
    onManualTimeUpdate () {
      if (this.manualSeekFromSlider) {
        return
      }
      const video = this.$refs.manualVideo
      if (video) {
        this.manualTimestampSec = Math.round(video.currentTime * 2) / 2
      }
    },
    seekManualVideo (val) {
      const video = this.$refs.manualVideo
      if (!video) {
        return
      }
      this.manualSeekFromSlider = true
      video.currentTime = val
      setTimeout(() => {
        this.manualSeekFromSlider = false
      }, 200)
    },
    saveManualSnapshot (snapshotType) {
      if (!this.manualCorrectRow) {
        return
      }
      this.manualSaving = true
      saveManualMediaSnapshot({
        mediaId: this.manualCorrectRow.id,
        snapshotType,
        timestampSec: this.manualTimestampSec
      }).then(res => {
        this.manualSaving = false
        this.$message.success(res || '保存成功')
        const mediaId = this.manualCorrectRow.id
        this.snapshotCacheVersion = Date.now()
        this.search()
        if (this.snapshotMediaRow && this.snapshotMediaRow.id === mediaId) {
          this.snapshotMediaRow = { ...this.snapshotMediaRow, snapshotStatus: 2 }
        }
        if (this.snapshotVisible && this.snapshotMediaRow && this.snapshotMediaRow.id === mediaId) {
          this.loadSnapshots(mediaId)
        }
      }).catch(err => {
        this.manualSaving = false
        this.$message.error(err.message || '保存失败')
      })
    }
  }
}
@@ -369,4 +630,91 @@
  color: #909399;
  padding: 40px 0;
}
.snapshot-wrap {
  min-height: 120px;
}
.snapshot-grid {
  display: flex;
  gap: 16px;
  flex-wrap: wrap;
}
.snapshot-item {
  flex: 1;
  min-width: 280px;
}
.snapshot-label {
  font-weight: 600;
  margin-bottom: 8px;
}
.snapshot-image {
  width: 100%;
  max-height: 280px;
  background: #f5f7fa;
}
.snapshot-meta {
  margin-top: 8px;
  font-size: 12px;
  color: #606266;
  display: flex;
  flex-direction: column;
  gap: 4px;
}
.snapshot-actions {
  margin-top: 16px;
  text-align: center;
}
.manual-correct-wrap {
  min-height: 200px;
}
.manual-slider {
  margin-top: 16px;
}
.manual-buttons {
  margin-top: 16px;
  display: flex;
  gap: 12px;
  justify-content: center;
}
.list-video-cell {
  display: inline-block;
  cursor: pointer;
  line-height: 0;
}
.list-video-thumb {
  width: 70px;
  height: 70px;
  object-fit: cover;
  background: #000;
  border-radius: 4px;
  vertical-align: middle;
}
.path-text {
  display: inline-block;
  max-width: 160px;
  overflow: hidden;
  text-overflow: ellipsis;
  white-space: nowrap;
}
.status-success {
  color: #67c23a;
  font-weight: 500;
}
.status-danger {
  color: #f56c6c;
  font-weight: 500;
}
.status-warning {
  color: #e6a23c;
  font-weight: 500;
}
.status-primary {
  color: #409eff;
  font-weight: 500;
}
.status-info {
  color: #909399;
}
.status-muted {
  color: #c0c4cc;
}
</style>

 server/db/business.collection_media.snapshot.alter.sql

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,5 @@
-- collection_media å¢žåŠ å¿«ç…§åˆ†æžçŠ¶æ€
ALTER TABLE `collection_media`
  ADD COLUMN `snapshot_status` int(11) DEFAULT '0' COMMENT '0未分析 1分析中 2完成 3失败' AFTER `download_time`,
  ADD COLUMN `snapshot_time` datetime DEFAULT NULL COMMENT '快照分析完成时间' AFTER `snapshot_status`,
  ADD COLUMN `snapshot_message` varchar(512) DEFAULT NULL COMMENT '快照分析失败原因' AFTER `snapshot_time`;

 server/db/business.collection_media.sql

@@ -15,6 +15,9 @@
  `file_path_local` varchar(512) DEFAULT NULL COMMENT 'FTP本地相对路径',
  `download_status` int(11) DEFAULT '0' COMMENT '0待下载 1已下载 2失败 3下载中',
  `download_time` datetime DEFAULT NULL COMMENT '下载完成时间',
  `snapshot_status` int(11) DEFAULT '0' COMMENT '0未分析 1分析中 2完成 3失败',
  `snapshot_time` datetime DEFAULT NULL COMMENT '快照分析完成时间',
  `snapshot_message` varchar(512) DEFAULT NULL COMMENT '快照分析失败原因',
  `create_date` datetime DEFAULT NULL COMMENT '创建时间',
  `isdeleted` int(11) DEFAULT '0' COMMENT '是否删除0否 1是',
  PRIMARY KEY (`id`),

 server/db/business.delivery_media_snapshot.sql

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,16 @@
CREATE TABLE IF NOT EXISTS `delivery_media_snapshot` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `media_id` int(11) NOT NULL COMMENT 'collection_media.id',
  `transport_code` varchar(64) DEFAULT NULL COMMENT '运单号(可选)',
  `snapshot_type` int(11) NOT NULL COMMENT '1门头 2交付',
  `timestamp_sec` decimal(10,2) DEFAULT NULL COMMENT '视频内秒数',
  `file_path` varchar(512) DEFAULT NULL COMMENT 'FTP相对路径',
  `confidence` decimal(5,4) DEFAULT NULL COMMENT '置信度',
  `source` varchar(16) DEFAULT NULL COMMENT 'ai/asr/hybrid/voice/manual/mock',
  `model_version` varchar(32) DEFAULT NULL COMMENT '模型版本',
  `create_date` datetime DEFAULT NULL COMMENT '创建时间',
  `isdeleted` int(11) DEFAULT '0' COMMENT '0否 1是',
  PRIMARY KEY (`id`),
  KEY `idx_media_id` (`media_id`),
  KEY `idx_transport_code` (`transport_code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='配送媒体快照(门头/交付)';

 server/db/business.delivery_media_snapshot_feedback.sql

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,13 @@
CREATE TABLE IF NOT EXISTS `delivery_media_snapshot_feedback` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `media_id` int(11) NOT NULL COMMENT 'collection_media.id',
  `snapshot_type` int(11) NOT NULL COMMENT '1门头 2交付',
  `ai_time_sec` decimal(10,2) DEFAULT NULL COMMENT 'AI原时刻',
  `manual_time_sec` decimal(10,2) NOT NULL COMMENT '人工纠正时刻',
  `model_version` varchar(32) DEFAULT NULL COMMENT '纠正时模型版本',
  `create_date` datetime DEFAULT NULL COMMENT '创建时间',
  `isdeleted` int(11) DEFAULT '0' COMMENT '0否 1是',
  PRIMARY KEY (`id`),
  KEY `idx_media_id` (`media_id`),
  KEY `idx_create_date` (`create_date`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='快照人工纠正反馈(用于增量训练)';

 server/snapshot_infer/.gitignore

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,10 @@
models/*.onnx
models/*.pt
models/*.pth
data/frames/
data/raw/
__pycache__/
*.pyc
.venv/
venv/
*.whl

 server/snapshot_infer/README.md

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,127 @@
# snapshot-infer

配送任务视频 **门头 / äº¤ä»˜** æ—¶åˆ»æ£€æµ‹æŽ¨ç†æœåŠ¡ï¼ˆONNX è§†è§‰ + faster-whisper ASR èžåˆï¼‰ã€‚

## æž¶æž„

1. Java `dmvisit_service` ä¸‹è½½ MP4 åˆ° FTP åŽï¼Œå¼‚步调用 `POST /analyze`
2. æœ¬æœåŠ¡ä¸‹è½½è§†é¢‘ â†’ å¹¶è¡Œ **抽帧 ONNX æ‰“分** + **ASR å…³é”®è¯** â†’ èžåˆæ—¶åˆ»
3. Java ä¾§ FFmpeg æˆªå¸§å¹¶ä¸Šä¼  FTP（`delivery_media_snapshot` è¡¨ï¼‰

## çŽ¯å¢ƒè¦æ±‚

- Python 3.9+（Windows ç”¨ `py` å¯åŠ¨å™¨ï¼‰
- **ffmpeg / ffprobe** åœ¨ PATH æˆ–设置 `FFMPEG_DIR`
- `models/` ç›®å½•ä¸‹æ”¾ç½®è®­ç»ƒå¯¼å‡ºçš„ ONNX：
  - `storefront_int8.onnx` / `handover_int8.onnx`（或 float ç‰ˆï¼‰
  - `version.json`

## å®‰è£…与启动（Windows CPU）

```powershell
cd server/snapshot_infer
py -m pip install -r requirements.txt
# è®­ç»ƒä¾èµ–（仅训练机）
# py -m pip install -r training/requirements-train.txt

$env:SNAPSHOT_MODEL_DIR = "./models"
$env:WHISPER_MODEL = "tiny"
$env:SNAPSHOT_SAMPLE_FPS = "0.5"
py -m uvicorn app.main:app --host 0.0.0.0 --port 8095
```

Linux:

```bash
cd server/snapshot_infer
python3 -m pip install -r requirements.txt
export SNAPSHOT_MODEL_DIR=./models WHISPER_MODEL=tiny SNAPSHOT_SAMPLE_FPS=0.5
python3 -m uvicorn app.main:app --host 0.0.0.0 --port 8095
```

## API

| æ–¹æ³• | è·¯å¾„ | è¯´æ˜Ž |
|------|------|------|
| GET | `/health` | è¿”回 `status`, `modelVersion`, ONNX/ASR åŠ è½½çŠ¶æ€ |
| POST | `/analyze` | ä¸»æŽ¨ç†ï¼ˆè¯·æ±‚/响应 camelCase，与 Java DTO å¯¹é½ï¼‰ |

### `/health` ç¤ºä¾‹

```json
{
  "status": "ok",
  "modelVersion": "1.0.0",
  "onnxStorefrontLoaded": true,
  "onnxHandoverLoaded": true,
  "asrAvailable": true
}
```

`status=degraded` è¡¨ç¤º ONNX æœªåŠ è½½å®Œæ•´ï¼Œéœ€éƒ¨ç½²æ¨¡åž‹æ–‡ä»¶ã€‚

## dmvisit é…ç½®

`application-dev.yml` / `application-pro.yml`:

```yaml
snapshot:
  infer:
    base-url: http://127.0.0.1:8095
    sample-fps: 0.5
    fail-open-mock: false   # ç”Ÿäº§å¿…须为 false
```

- **dev** å¯è®¾ `fail-open-mock: true` åœ¨æ—  ONNX æ—¶å›žé€€ 25%/75% mock
- **pro** å·²é»˜è®¤ `fail-open-mock: false`，推理失败写入 `snapshot_status=3`

## è®­ç»ƒæµç¨‹

详见 [`docs/annotation_spec.md`](docs/annotation_spec.md)。

```powershell
cd server/snapshot_infer/training
py -m pip install -r requirements-train.txt

# 1. å‡†å¤‡æ ‡æ³¨ JSONL -> data/annotations.jsonl
# 2. æŠ½å¸§æ‰“æ ‡
py prepare_dataset.py -c config.yaml

# 3. è®­ç»ƒ
py train.py -c config.yaml

# 4. å¯¼å‡º ONNX（Windows è‹¥ INT8 é‡åŒ–崩溃，会自动使用 float .onnx）
py export_onnx.py -c config.yaml --version 1.0.0
# æˆ–跳过量化：py export_onnx.py -c config.yaml --version 1.0.0 --no-quantize

# 5. è¯„ä¼°
py evaluate.py -c config.yaml --annotations ../data/annotations.jsonl
```

### Windows æ— æ³•ç”Ÿæˆ `*_int8.onnx`

部分 Windows çŽ¯å¢ƒä¸Š `onnxruntime.quantization` ä¼šè¿›ç¨‹å´©æºƒï¼ˆexit `3221225477`）。**不影响使用**：

- å¯¼å‡ºè„šæœ¬ä¼šè‡ªåŠ¨å›žé€€åˆ° `storefront.onnx` / `handover.onnx`（float32）
- `version.json` ä¼šè®°å½•å®žé™…文件名
- æŽ¨ç†æœåŠ¡åŒæ ·å¯åŠ è½½ float æ¨¡åž‹ï¼Œä»…速度略慢于 INT8

若必须 INT8，可在 Linux è®­ç»ƒæœºæ‰§è¡Œ `export_onnx.py`，或尝试 `py -m pip install onnxruntime==1.16.3` åŽé‡è¯•ã€‚

## å·¥å…·

| è„šæœ¬ | ç”¨é€” |
|------|------|
| `tools/export_media_list.py --mysql` | ä»Ž DB å¯¼å‡ºå¾…标注清单 |
| `tools/export_feedback.py` | å¯¼å‡ºäººå·¥çº æ­£åé¦ˆä¸ºè®­ç»ƒ JSONL |
| `tools/benchmark_cpu.py <video_url>` | CPU å•æ¡åŽ‹æµ‹ |

## ç›‘控

- å¯åŠ¨åŽè®¿é—® `GET http://127.0.0.1:8095/health`
- `onnxStorefrontLoaded=false` â†’ æ£€æŸ¥ `models/` è·¯å¾„与环境变量 `SNAPSHOT_MODEL_DIR`
- åˆ†æžå¤±è´¥åŽŸå› è§ `collection_media.snapshot_message`

## æ•°æ®åº“（需执行 SQL）

- `server/db/business.delivery_media_snapshot_feedback.sql` â€” äººå·¥çº æ­£åé¦ˆè¡¨

 server/snapshot_infer/app/__init__.py

 server/snapshot_infer/app/asr.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,81 @@
# -*- coding: utf-8 -*-
import logging
import os
import subprocess
import tempfile
from typing import List, Optional, Tuple

from app.schemas import AsrHit, KeywordConfig
from app.video_io import get_ffmpeg_cmd

logger = logging.getLogger(__name__)

_whisper_model = None


def _get_whisper():
    global _whisper_model
    if _whisper_model is None:
        from faster_whisper import WhisperModel
        model_size = os.environ.get("WHISPER_MODEL", "tiny")
        _whisper_model = WhisperModel(model_size, device="cpu", compute_type="int8")
        logger.info("加载 Whisper æ¨¡åž‹: %s", model_size)
    return _whisper_model


def extract_audio_wav(video_path: str) -> str:
    out = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
    out.close()
    cmd = [
        get_ffmpeg_cmd("ffmpeg"), "-y", "-i", video_path,
        "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1",
        out.name,
    ]
    subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    return out.name


def transcribe(video_path: str) -> List[Tuple[str, float, float]]:
    wav = extract_audio_wav(video_path)
    try:
        model = _get_whisper()
        segments, _ = model.transcribe(wav, language="zh", vad_filter=True)
        return [(seg.text.strip(), seg.start, seg.end) for seg in segments if seg.text.strip()]
    except Exception as e:
        logger.warning("ASR å¤±è´¥: %s", e)
        return []
    finally:
        if os.path.isfile(wav):
            os.remove(wav)


def match_keywords(
    segments: List[Tuple[str, float, float]],
    keywords: KeywordConfig,
) -> List[AsrHit]:
    hits: List[AsrHit] = []
    for text, start, _end in segments:
        for kw in keywords.storefront:
            if kw in text:
                hits.append(AsrHit(keyword=kw, time_sec=round(start, 2)))
                break
        for kw in keywords.handover:
            if kw in text:
                hits.append(AsrHit(keyword=kw, time_sec=round(start, 2)))
                break
    return hits


def best_asr_time(hits: List[AsrHit], keywords: List[str]) -> Optional[float]:
    for hit in hits:
        if hit.keyword in keywords:
            return hit.time_sec
    return None


def asr_available() -> bool:
    try:
        import faster_whisper  # noqa: F401
        return True
    except ImportError:
        return False

 server/snapshot_infer/app/frame_sampler.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,45 @@
# -*- coding: utf-8 -*-
import logging
import os
import subprocess
import tempfile
from typing import List, Tuple

from app.video_io import get_ffmpeg_cmd

logger = logging.getLogger(__name__)


def sample_frames(video_path: str, sample_fps: float, duration: float) -> List[Tuple[float, str]]:
    """按 sample_fps æŠ½å¸§ï¼Œè¿”回 [(time_sec, jpg_path), ...]"""
    if sample_fps <= 0:
        sample_fps = 0.5
    step = 1.0 / sample_fps
    out_dir = tempfile.mkdtemp(prefix="snap_frames_")
    frames: List[Tuple[float, str]] = []
    t = 0.0
    idx = 0
    ffmpeg = get_ffmpeg_cmd("ffmpeg")
    while t <= duration:
        out_path = os.path.join(out_dir, f"{idx:06d}.jpg")
        cmd = [
            ffmpeg, "-y", "-ss", f"{t:.3f}",
            "-i", video_path,
            "-frames:v", "1", "-q:v", "2",
            out_path,
        ]
        print(f"cmd{cmd}")
        subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        if os.path.isfile(out_path) and os.path.getsize(out_path) > 0:
            frames.append((round(t, 3), out_path))
        t += step
        idx += 1
    return frames


def cleanup_frames(frames: List[Tuple[float, str]]) -> None:
    if not frames:
        return
    out_dir = os.path.dirname(frames[0][1])
    import shutil
    shutil.rmtree(out_dir, ignore_errors=True)

 server/snapshot_infer/app/fusion.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,52 @@
# -*- coding: utf-8 -*-
from typing import List, Optional, Tuple

from app.schemas import AsrHit, KeywordConfig, SnapshotHit


def fuse_hit(
    vision: Optional[Tuple[float, float]],
    asr_time: Optional[float],
    asr_weight: float = 0.7,
    window: float = 2.0,
) -> Optional[SnapshotHit]:
    if vision and asr_time is not None:
        vt, vc = vision
        if abs(vt - asr_time) <= window * 3:
            t = round(asr_weight * asr_time + (1 - asr_weight) * vt, 2)
            conf = min(0.99, vc + 0.1)
            return SnapshotHit(time_sec=t, confidence=round(conf, 4), source="hybrid")
        t = round(asr_weight * asr_time + (1 - asr_weight) * vt, 2)
        return SnapshotHit(time_sec=t, confidence=round(vc, 4), source="hybrid")
    if asr_time is not None:
        return SnapshotHit(time_sec=round(asr_time, 2), confidence=0.75, source="asr")
    if vision:
        return SnapshotHit(time_sec=round(vision[0], 2), confidence=round(vision[1], 4), source="ai")
    return None


def fuse_results(
    sf_vision: Optional[Tuple[float, float]],
    ho_vision: Optional[Tuple[float, float]],
    asr_hits: List[AsrHit],
    keywords: KeywordConfig,
    duration: float,
) -> Tuple[Optional[SnapshotHit], Optional[SnapshotHit]]:
    from app.asr import best_asr_time

    sf_asr = best_asr_time(asr_hits, keywords.storefront)
    ho_asr = best_asr_time(asr_hits, keywords.handover)

    storefront = fuse_hit(sf_vision, sf_asr)
    min_ho = (storefront.time_sec + 30.0) if storefront else 0.0
    if ho_vision and ho_vision[0] < min_ho:
        ho_vision = None
    if ho_asr is not None and ho_asr < min_ho:
        ho_asr = None

    handover = fuse_hit(ho_vision, ho_asr)

    if storefront and handover and handover.time_sec <= storefront.time_sec:
        handover.time_sec = round(min(duration - 1, storefront.time_sec + 60), 2)

    return storefront, handover

 server/snapshot_infer/app/main.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,38 @@
# -*- coding: utf-8 -*-
"""配送视频门头/交付时刻检测推理服务（ONNX + ASR）。"""
import logging
import os

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel

from app.asr import asr_available
from app.pipeline import get_registry, run_analyze
from app.schemas import AnalyzeRequest, HealthResponse

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")

app = FastAPI(title="snapshot-infer", version="2.0.0")


def _camel_json(model: BaseModel) -> JSONResponse:
    return JSONResponse(content=model.model_dump(by_alias=True, exclude_none=True))


@app.get("/health")
def health():
    reg = get_registry()
    body = HealthResponse(
        status="ok" if reg.ready else "degraded",
        model_version=reg.version,
        onnx_storefront_loaded=reg.storefront.loaded if reg.storefront else False,
        onnx_handover_loaded=reg.handover.loaded if reg.handover else False,
        asr_available=asr_available(),
    )
    return _camel_json(body)


@app.post("/analyze")
def analyze(req: AnalyzeRequest):
    return _camel_json(run_analyze(req))

 server/snapshot_infer/app/onnx_infer.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,100 @@
# -*- coding: utf-8 -*-
import json
import logging
import os
from typing import List, Optional, Tuple

import numpy as np
import onnxruntime as ort
from PIL import Image

logger = logging.getLogger(__name__)

IMAGENET_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32)
IMAGENET_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32)


class OnnxClassifier:
    def __init__(self, model_path: str, image_size: int = 224):
        self.image_size = image_size
        self.session: Optional[ort.InferenceSession] = None
        if model_path and os.path.isfile(model_path):
            self.session = ort.InferenceSession(
                model_path,
                providers=["CPUExecutionProvider"],
            )
            logger.info("加载 ONNX: %s", model_path)

    @property
    def loaded(self) -> bool:
        return self.session is not None

    def preprocess(self, image_path: str) -> np.ndarray:
        img = Image.open(image_path).convert("RGB").resize((self.image_size, self.image_size))
        arr = np.array(img).astype(np.float32) / 255.0
        arr = (arr - IMAGENET_MEAN) / IMAGENET_STD
        return arr.transpose(2, 0, 1)[None]

    def predict_batch(self, image_paths: List[str], batch_size: int = 16) -> List[float]:
        if not self.session:
            return [0.0] * len(image_paths)
        scores = []
        for i in range(0, len(image_paths), batch_size):
            batch_paths = image_paths[i : i + batch_size]
            batch = np.concatenate([self.preprocess(p) for p in batch_paths], axis=0)
            logits = self.session.run(None, {"input": batch})[0]
            for logit in logits:
                scores.append(float(1.0 / (1.0 + np.exp(-logit[0]))))
        return scores


class ModelRegistry:
    def __init__(self, model_dir: str):
        self.model_dir = model_dir
        self.version = "unknown"
        self.image_size = 224
        self.storefront: Optional[OnnxClassifier] = None
        self.handover: Optional[OnnxClassifier] = None
        self._load()

    def _load(self):
        version_path = os.path.join(self.model_dir, "version.json")
        sf_name = ho_name = None
        if os.path.isfile(version_path):
            with open(version_path, encoding="utf-8") as f:
                meta = json.load(f)
            self.version = meta.get("model_version", "1.0.0")
            self.image_size = int(meta.get("image_size", 224))
            sf_name = meta.get("storefront_model")
            ho_name = meta.get("handover_model")

        def resolve(name: str, fallback: str) -> str:
            if name:
                path = os.path.join(self.model_dir, name)
                if os.path.isfile(path):
                    return path
            for suffix in (f"{fallback}_int8.onnx", f"{fallback}.onnx"):
                path = os.path.join(self.model_dir, suffix)
                if os.path.isfile(path):
                    return path
            return ""

        sf_path = resolve(sf_name, "storefront")
        ho_path = resolve(ho_name, "handover")
        self.storefront = OnnxClassifier(sf_path, self.image_size) if sf_path else OnnxClassifier("", self.image_size)
        self.handover = OnnxClassifier(ho_path, self.image_size) if ho_path else OnnxClassifier("", self.image_size)

    @property
    def ready(self) -> bool:
        return self.storefront.loaded and self.handover.loaded


def score_frames(
    registry: ModelRegistry,
    frames: List[Tuple[float, str]],
) -> Tuple[List[Tuple[float, float]], List[Tuple[float, float]]]:
    times = [f[0] for f in frames]
    paths = [f[1] for f in frames]
    sf_scores = registry.storefront.predict_batch(paths)
    ho_scores = registry.handover.predict_batch(paths)
    return list(zip(times, sf_scores)), list(zip(times, ho_scores))

 server/snapshot_infer/app/pipeline.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,112 @@
# -*- coding: utf-8 -*-
import logging
import os
from concurrent.futures import ThreadPoolExecutor, as_completed

from app.asr import asr_available, match_keywords, transcribe
from app.frame_sampler import cleanup_frames, sample_frames
from app.fusion import fuse_results
from app.onnx_infer import ModelRegistry, score_frames
from app.quality import refine_time_in_window
from app.schemas import AnalyzeRequest, AnalyzeResponse, KeywordConfig
from app.temporal import find_peaks_ordered
from app.video_io import temp_video

logger = logging.getLogger(__name__)

_registry: ModelRegistry = None


def get_registry() -> ModelRegistry:
    global _registry
    if _registry is None:
        model_dir = os.environ.get("SNAPSHOT_MODEL_DIR", os.path.join(os.path.dirname(__file__), "..", "models"))
        _registry = ModelRegistry(os.path.abspath(model_dir))
    return _registry


def run_analyze(req: AnalyzeRequest) -> AnalyzeResponse:
    registry = get_registry()
    if not registry.ready:
        return AnalyzeResponse(
            success=False,
            model_version=registry.version,
            message="ONNX æ¨¡åž‹æœªåŠ è½½ï¼Œè¯·å°† storefront/handover ONNX æ”¾å…¥ models/ ç›®å½•",
        )
    if not req.video_url:
        return AnalyzeResponse(success=False, model_version=registry.version, message="video_url ä¸èƒ½ä¸ºç©º")

    keywords = req.keywords or KeywordConfig()
    sample_fps = req.sample_fps if req.sample_fps and req.sample_fps > 0 else float(os.environ.get("SNAPSHOT_SAMPLE_FPS", "0.5"))

    try:
        with temp_video(req.video_url, req.duration_sec or 0.0) as (video_path, duration):
            if req.duration_sec and req.duration_sec > 0:
                duration = req.duration_sec

            asr_hits = []
            sf_vision = ho_vision = None

            def vision_task():
                frames = sample_frames(video_path, sample_fps, duration)
                try:
                    sf_scores, ho_scores = score_frames(registry, frames)
                    return find_peaks_ordered(sf_scores, ho_scores, duration), frames
                finally:
                    cleanup_frames(frames)

            def asr_task():
                if not req.enable_asr or not asr_available():
                    return []
                segments = transcribe(video_path)
                return match_keywords(segments, keywords)

            with ThreadPoolExecutor(max_workers=2) as pool:
                futures = {pool.submit(vision_task): "vision"}
                if req.enable_asr:
                    futures[pool.submit(asr_task)] = "asr"
                vision_result = None
                for fut in as_completed(futures):
                    if futures[fut] == "vision":
                        vision_result, _ = fut.result()
                    else:
                        asr_hits = fut.result()

            if vision_result:
                sf_vision, ho_vision = vision_result

            storefront, handover = fuse_results(sf_vision, ho_vision, asr_hits, keywords, duration)

            if storefront:
                t, _ = refine_time_in_window(video_path, storefront.time_sec)
                storefront.time_sec = t
            if handover:
                t, _ = refine_time_in_window(video_path, handover.time_sec)
                handover.time_sec = t

            if not storefront or not handover:
                return AnalyzeResponse(
                    success=False,
                    model_version=registry.version,
                    duration_sec=duration,
                    storefront=storefront,
                    handover=handover,
                    asr_hits=asr_hits,
                    message="未能检测到门头或交付时刻",
                )

            return AnalyzeResponse(
                success=True,
                model_version=registry.version,
                duration_sec=round(duration, 2),
                storefront=storefront,
                handover=handover,
                asr_hits=asr_hits,
            )
    except Exception as e:
        logger.exception("分析失败 media_id=%s", req.media_id)
        return AnalyzeResponse(
            success=False,
            model_version=registry.version,
            message=str(e),
        )

 server/snapshot_infer/app/quality.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
import logging
import subprocess
from typing import Optional, Tuple

import cv2
import numpy as np

from app.video_io import get_ffmpeg_cmd

logger = logging.getLogger(__name__)


def laplacian_score(image_path: str) -> float:
    img = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)
    if img is None:
        return 0.0
    return float(cv2.Laplacian(img, cv2.CV_64F).var())


def refine_time_in_window(
    video_path: str,
    center_sec: float,
    window_sec: float = 2.0,
    step: float = 0.5,
) -> Tuple[float, float]:
    """在 [center-window, center+window] å†…选清晰度最高帧，返回 (best_sec, score)。"""
    import os
    import tempfile

    best_t = center_sec
    best_score = 0.0
    ffmpeg = get_ffmpeg_cmd("ffmpeg")
    t = max(0.0, center_sec - window_sec)
    end = center_sec + window_sec
    tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
    tmp.close()
    try:
        while t <= end:
            cmd = [
                ffmpeg, "-y", "-ss", f"{t:.3f}", "-i", video_path,
                "-frames:v", "1", "-q:v", "2", tmp.name,
            ]
            subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            if os.path.isfile(tmp.name) and os.path.getsize(tmp.name) > 0:
                score = laplacian_score(tmp.name)
                if score > best_score:
                    best_score = score
                    best_t = t
            t += step
    finally:
        if os.path.isfile(tmp.name):
            os.remove(tmp.name)
    return round(best_t, 2), best_score

 server/snapshot_infer/app/schemas.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,53 @@
# -*- coding: utf-8 -*-
from typing import List, Optional

from pydantic import BaseModel, ConfigDict, Field


class KeywordConfig(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    storefront: List[str] = Field(default_factory=lambda: ["到店", "到达", "门头"])
    handover: List[str] = Field(default_factory=lambda: ["交付", "交货", "签收"])


class AnalyzeRequest(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    media_id: int = Field(alias="mediaId")
    video_url: Optional[str] = Field(default=None, alias="videoUrl")
    sample_fps: float = Field(default=0.5, alias="sampleFps")
    enable_asr: bool = Field(default=True, alias="enableAsr")
    keywords: Optional[KeywordConfig] = None
    duration_sec: Optional[float] = Field(default=None, alias="durationSec")


class SnapshotHit(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    time_sec: float = Field(alias="timeSec")
    confidence: float
    source: str


class AsrHit(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    keyword: str
    time_sec: float = Field(alias="timeSec")


class AnalyzeResponse(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    success: bool = True
    model_version: str = Field(default="1.0.0", alias="modelVersion")
    duration_sec: float = Field(default=0.0, alias="durationSec")
    storefront: Optional[SnapshotHit] = None
    handover: Optional[SnapshotHit] = None
    asr_hits: List[AsrHit] = Field(default_factory=list, alias="asrHits")
    message: Optional[str] = None


class HealthResponse(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    status: str
    model_version: str = Field(alias="modelVersion")
    onnx_storefront_loaded: bool = Field(default=False, alias="onnxStorefrontLoaded")
    onnx_handover_loaded: bool = Field(default=False, alias="onnxHandoverLoaded")
    asr_available: bool = Field(default=False, alias="asrAvailable")

 server/snapshot_infer/app/temporal.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,47 @@
# -*- coding: utf-8 -*-
from typing import List, Optional, Tuple


def smooth_scores(time_scores: List[Tuple[float, float]], window: int = 3) -> List[Tuple[float, float]]:
    if len(time_scores) <= 1 or window <= 1:
        return time_scores
    half = window // 2
    smoothed = []
    for i, (t, _) in enumerate(time_scores):
        lo = max(0, i - half)
        hi = min(len(time_scores), i + half + 1)
        avg = sum(s for _, s in time_scores[lo:hi]) / (hi - lo)
        smoothed.append((t, avg))
    return smoothed


def find_peak(
    time_scores: List[Tuple[float, float]],
    min_time: float = 0.0,
    min_confidence: float = 0.3,
) -> Optional[Tuple[float, float]]:
    if not time_scores:
        return None
    candidates = [(t, s) for t, s in time_scores if t >= min_time and s >= min_confidence]
    if not candidates:
        candidates = time_scores
    best = max(candidates, key=lambda x: x[1])
    if best[1] < 0.1:
        return None
    return best


def find_peaks_ordered(
    sf_scores: List[Tuple[float, float]],
    ho_scores: List[Tuple[float, float]],
    duration: float,
    min_gap: float = 30.0,
) -> Tuple[Optional[Tuple[float, float]], Optional[Tuple[float, float]]]:
    sf = find_peak(smooth_scores(sf_scores))
    min_ho = (sf[0] + min_gap) if sf else 0.0
    ho = find_peak(smooth_scores(ho_scores), min_time=min_ho)
    if sf and ho and ho[0] <= sf[0]:
        ho = find_peak(smooth_scores(ho_scores), min_time=sf[0] + 1.0)
    if sf and not ho and duration > sf[0] + min_gap:
        ho = find_peak(smooth_scores(ho_scores), min_time=sf[0] + min_gap)
    return sf, ho

 server/snapshot_infer/app/video_io.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,67 @@
# -*- coding: utf-8 -*-
import json
import logging
import os
import shutil
import subprocess
import tempfile
from contextlib import contextmanager
from typing import Optional

import httpx

logger = logging.getLogger(__name__)


def get_ffmpeg_cmd(name: str = "ffmpeg") -> str:
    ffmpeg_dir = "D:/code/ffmpeg/"
#     ffmpeg_dir = os.environ.get("FFMPEG_DIR", "")
    print(f"ffmpeg_dir:{ffmpeg_dir}")
    if ffmpeg_dir:
        exe = "ffmpeg.exe" if os.name == "nt" else "ffmpeg"
        print(f"exe:{exe}")
        if name == "ffprobe":
            exe = "ffprobe.exe" if os.name == "nt" else "ffprobe"
        return os.path.join(ffmpeg_dir, exe)
    return name


def probe_duration(video_path: str) -> float:
    cmd = [
        get_ffmpeg_cmd("ffprobe"),
        "-v", "error",
        "-show_entries", "format=duration",
        "-of", "default=noprint_wrappers=1:nokey=1",
        video_path,
    ]
    try:
        out = subprocess.check_output(cmd, stderr=subprocess.DEVNULL, text=True).strip()
        return float(out) if out else 0.0
    except Exception as e:
        logger.warning("ffprobe å¤±è´¥: %s", e)
        return 0.0


def download_video(url: str, dest_path: str, timeout: float = 600.0) -> None:
    with httpx.stream("GET", url, timeout=timeout, follow_redirects=True) as resp:
        resp.raise_for_status()
        with open(dest_path, "wb") as f:
            for chunk in resp.iter_bytes(chunk_size=65536):
                f.write(chunk)


@contextmanager
def temp_video(video_url: Optional[str], duration_hint: float = 0.0):
    tmp_dir = tempfile.mkdtemp(prefix="snap_infer_")
    video_path = os.path.join(tmp_dir, "video.mp4")
    try:
        if video_url:
            download_video(video_url, video_path)
        else:
            raise ValueError("video_url ä¸èƒ½ä¸ºç©º")
        duration = probe_duration(video_path)
        if duration <= 0:
            duration = duration_hint if duration_hint > 0 else 1200.0
        yield video_path, duration
    finally:
        shutil.rmtree(tmp_dir, ignore_errors=True)

 server/snapshot_infer/data/annotations.jsonl

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,4 @@
{"media_id": 1, "video_path": "http://192.168.0.3/file/collection_media/20260611/0986a054-ccbe-408b-86fc-b834e9c2f4dd.mp4", "storefront_time_sec": 25.5, "handover_time_sec": 49.5, "store_type": "公司", "has_voice_marker": false, "driver_date": "demo_20250601", "split": "train", "notes": "豆米科技派送样例1"}
{"media_id": 2, "video_path": "http://192.168.0.3/file/collection_media/20260611/6808c257-8df7-43f7-b8dc-03789e184929.mp4", "storefront_time_sec": 18.5, "handover_time_sec": 37.5, "store_type": "公司", "has_voice_marker": false, "driver_date": "demo_20250603", "split": "val", "notes": "豆米科技派送样例3"}
{"media_id": 3, "video_path": "http://192.168.0.3/file/collection_media/20260611/0986a054-ccbe-408b-86fc-b834e9c2f4dd.mp4", "storefront_time_sec": 25.5, "handover_time_sec": 49.5, "store_type": "公司", "has_voice_marker": false, "driver_date": "demo_20250602", "split": "val", "notes": "豆米科技派送样例2"}
{"media_id": 4, "video_path": "http://192.168.0.3/file/collection_media/20260611/6808c257-8df7-43f7-b8dc-03789e184929.mp4", "storefront_time_sec": 18.5, "handover_time_sec": 37.5, "store_type": "公司", "has_voice_marker": false, "driver_date": "demo_20250603", "split": "train", "notes": "豆米科技派送样例4"}

 server/snapshot_infer/data/annotations_poc_sample.jsonl

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,2 @@
{"media_id": 1, "video_path": "http://192.168.0.3/file/collection_media/20260611/6808c257-8df7-43f7-b8dc-03789e184929.mp4", "storefront_time_sec": 25.5, "handover_time_sec": 49.5, "store_type": "公司", "has_voice_marker": true, "driver_date": "demo_20250601", "split": "train", "notes": "豆米科技派送样例1"}
{"media_id": 2, "video_path": "http://192.168.0.3/file/collection_media/20260611/17911eb5-ee71-421f-af72-bd804d563201.mp4", "storefront_time_sec": 18.5, "handover_time_sec": 37.5, "store_type": "公司", "has_voice_marker": false, "driver_date": "demo_20250602", "split": "val", "notes": "豆米科技派送样例2"}

 server/snapshot_infer/data/labels.csv

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,291 @@
frame_path,label,task,split
1/21000.jpg,1,storefront,train
1/22000.jpg,1,storefront,train
1/23000.jpg,1,storefront,train
1/24000.jpg,1,storefront,train
1/25000.jpg,1,storefront,train
1/26000.jpg,1,storefront,train
1/27000.jpg,1,storefront,train
1/28000.jpg,1,storefront,train
1/29000.jpg,1,storefront,train
1/30000.jpg,1,storefront,train
1/45000.jpg,1,handover,train
1/46000.jpg,1,handover,train
1/47000.jpg,1,handover,train
1/48000.jpg,1,handover,train
1/49000.jpg,1,handover,train
1/50000.jpg,1,handover,train
1/51000.jpg,1,handover,train
1/52000.jpg,1,handover,train
1/53000.jpg,1,handover,train
1/54000.jpg,1,handover,train
1/0.jpg,0,other,train
1/1000.jpg,0,other,train
1/2000.jpg,0,other,train
1/3000.jpg,0,other,train
1/4000.jpg,0,other,train
1/5000.jpg,0,other,train
1/6000.jpg,0,other,train
1/7000.jpg,0,other,train
1/8000.jpg,0,other,train
1/9000.jpg,0,other,train
1/10000.jpg,0,other,train
1/11000.jpg,0,other,train
1/12000.jpg,0,other,train
1/13000.jpg,0,other,train
1/14000.jpg,0,other,train
1/15000.jpg,0,other,train
1/16000.jpg,0,other,train
1/17000.jpg,0,other,train
1/18000.jpg,0,other,train
1/19000.jpg,0,other,train
1/20000.jpg,0,other,train
1/31000.jpg,0,other,train
1/32000.jpg,0,other,train
1/33000.jpg,0,other,train
1/34000.jpg,0,other,train
1/35000.jpg,0,other,train
1/36000.jpg,0,other,train
1/37000.jpg,0,other,train
1/38000.jpg,0,other,train
1/39000.jpg,0,other,train
1/40000.jpg,0,other,train
1/41000.jpg,0,other,train
1/42000.jpg,0,other,train
1/43000.jpg,0,other,train
1/44000.jpg,0,other,train
1/55000.jpg,0,other,train
1/56000.jpg,0,other,train
1/57000.jpg,0,other,train
1/58000.jpg,0,other,train
1/59000.jpg,0,other,train
1/60000.jpg,0,other,train
1/61000.jpg,0,other,train
1/62000.jpg,0,other,train
1/63000.jpg,0,other,train
1/64000.jpg,0,other,train
1/65000.jpg,0,other,train
1/66000.jpg,0,other,train
1/67000.jpg,0,other,train
1/68000.jpg,0,other,train
1/69000.jpg,0,other,train
1/70000.jpg,0,other,train
1/71000.jpg,0,other,train
1/72000.jpg,0,other,train
1/73000.jpg,0,other,train
1/74000.jpg,0,other,train
1/75000.jpg,0,other,train
1/76000.jpg,0,other,train
1/77000.jpg,0,other,train
1/78000.jpg,0,other,train
1/79000.jpg,0,other,train
1/80000.jpg,0,other,train
1/81000.jpg,0,other,train
1/82000.jpg,0,other,train
1/83000.jpg,0,other,train
1/84000.jpg,0,other,train
1/85000.jpg,0,other,train
1/86000.jpg,0,other,train
1/87000.jpg,0,other,train
1/88000.jpg,0,other,train
1/89000.jpg,0,other,train
1/90000.jpg,0,other,train
1/91000.jpg,0,other,train
1/92000.jpg,0,other,train
1/93000.jpg,0,other,train
1/94000.jpg,0,other,train
1/95000.jpg,0,other,train
1/96000.jpg,0,other,train
1/97000.jpg,0,other,train
2/14000.jpg,1,storefront,val
2/15000.jpg,1,storefront,val
2/16000.jpg,1,storefront,val
2/17000.jpg,1,storefront,val
2/18000.jpg,1,storefront,val
2/19000.jpg,1,storefront,val
2/20000.jpg,1,storefront,val
2/21000.jpg,1,storefront,val
2/22000.jpg,1,storefront,val
2/23000.jpg,1,storefront,val
2/33000.jpg,1,handover,val
2/34000.jpg,1,handover,val
2/35000.jpg,1,handover,val
2/36000.jpg,1,handover,val
2/37000.jpg,1,handover,val
2/38000.jpg,1,handover,val
2/39000.jpg,1,handover,val
2/40000.jpg,1,handover,val
2/41000.jpg,1,handover,val
2/42000.jpg,1,handover,val
2/0.jpg,0,other,val
2/1000.jpg,0,other,val
2/2000.jpg,0,other,val
2/3000.jpg,0,other,val
2/4000.jpg,0,other,val
2/5000.jpg,0,other,val
2/6000.jpg,0,other,val
2/7000.jpg,0,other,val
2/8000.jpg,0,other,val
2/9000.jpg,0,other,val
2/10000.jpg,0,other,val
2/11000.jpg,0,other,val
2/12000.jpg,0,other,val
2/13000.jpg,0,other,val
2/24000.jpg,0,other,val
2/25000.jpg,0,other,val
2/26000.jpg,0,other,val
2/27000.jpg,0,other,val
2/28000.jpg,0,other,val
2/29000.jpg,0,other,val
2/30000.jpg,0,other,val
2/31000.jpg,0,other,val
2/32000.jpg,0,other,val
2/43000.jpg,0,other,val
2/44000.jpg,0,other,val
2/45000.jpg,0,other,val
2/46000.jpg,0,other,val
3/21000.jpg,1,storefront,val
3/22000.jpg,1,storefront,val
3/23000.jpg,1,storefront,val
3/24000.jpg,1,storefront,val
3/25000.jpg,1,storefront,val
3/26000.jpg,1,storefront,val
3/27000.jpg,1,storefront,val
3/28000.jpg,1,storefront,val
3/29000.jpg,1,storefront,val
3/30000.jpg,1,storefront,val
3/45000.jpg,1,handover,val
3/46000.jpg,1,handover,val
3/47000.jpg,1,handover,val
3/48000.jpg,1,handover,val
3/49000.jpg,1,handover,val
3/50000.jpg,1,handover,val
3/51000.jpg,1,handover,val
3/52000.jpg,1,handover,val
3/53000.jpg,1,handover,val
3/54000.jpg,1,handover,val
3/0.jpg,0,other,val
3/1000.jpg,0,other,val
3/2000.jpg,0,other,val
3/3000.jpg,0,other,val
3/4000.jpg,0,other,val
3/5000.jpg,0,other,val
3/6000.jpg,0,other,val
3/7000.jpg,0,other,val
3/8000.jpg,0,other,val
3/9000.jpg,0,other,val
3/10000.jpg,0,other,val
3/11000.jpg,0,other,val
3/12000.jpg,0,other,val
3/13000.jpg,0,other,val
3/14000.jpg,0,other,val
3/15000.jpg,0,other,val
3/16000.jpg,0,other,val
3/17000.jpg,0,other,val
3/18000.jpg,0,other,val
3/19000.jpg,0,other,val
3/20000.jpg,0,other,val
3/31000.jpg,0,other,val
3/32000.jpg,0,other,val
3/33000.jpg,0,other,val
3/34000.jpg,0,other,val
3/35000.jpg,0,other,val
3/36000.jpg,0,other,val
3/37000.jpg,0,other,val
3/38000.jpg,0,other,val
3/39000.jpg,0,other,val
3/40000.jpg,0,other,val
3/41000.jpg,0,other,val
3/42000.jpg,0,other,val
3/43000.jpg,0,other,val
3/44000.jpg,0,other,val
3/55000.jpg,0,other,val
3/56000.jpg,0,other,val
3/57000.jpg,0,other,val
3/58000.jpg,0,other,val
3/59000.jpg,0,other,val
3/60000.jpg,0,other,val
3/61000.jpg,0,other,val
3/62000.jpg,0,other,val
3/63000.jpg,0,other,val
3/64000.jpg,0,other,val
3/65000.jpg,0,other,val
3/66000.jpg,0,other,val
3/67000.jpg,0,other,val
3/68000.jpg,0,other,val
3/69000.jpg,0,other,val
3/70000.jpg,0,other,val
3/71000.jpg,0,other,val
3/72000.jpg,0,other,val
3/73000.jpg,0,other,val
3/74000.jpg,0,other,val
3/75000.jpg,0,other,val
3/76000.jpg,0,other,val
3/77000.jpg,0,other,val
3/78000.jpg,0,other,val
3/79000.jpg,0,other,val
3/80000.jpg,0,other,val
3/81000.jpg,0,other,val
3/82000.jpg,0,other,val
3/83000.jpg,0,other,val
3/84000.jpg,0,other,val
3/85000.jpg,0,other,val
3/86000.jpg,0,other,val
3/87000.jpg,0,other,val
3/88000.jpg,0,other,val
3/89000.jpg,0,other,val
3/90000.jpg,0,other,val
3/91000.jpg,0,other,val
3/92000.jpg,0,other,val
3/93000.jpg,0,other,val
3/94000.jpg,0,other,val
3/95000.jpg,0,other,val
3/96000.jpg,0,other,val
3/97000.jpg,0,other,val
4/14000.jpg,1,storefront,train
4/15000.jpg,1,storefront,train
4/16000.jpg,1,storefront,train
4/17000.jpg,1,storefront,train
4/18000.jpg,1,storefront,train
4/19000.jpg,1,storefront,train
4/20000.jpg,1,storefront,train
4/21000.jpg,1,storefront,train
4/22000.jpg,1,storefront,train
4/23000.jpg,1,storefront,train
4/33000.jpg,1,handover,train
4/34000.jpg,1,handover,train
4/35000.jpg,1,handover,train
4/36000.jpg,1,handover,train
4/37000.jpg,1,handover,train
4/38000.jpg,1,handover,train
4/39000.jpg,1,handover,train
4/40000.jpg,1,handover,train
4/41000.jpg,1,handover,train
4/42000.jpg,1,handover,train
4/0.jpg,0,other,train
4/1000.jpg,0,other,train
4/2000.jpg,0,other,train
4/3000.jpg,0,other,train
4/4000.jpg,0,other,train
4/5000.jpg,0,other,train
4/6000.jpg,0,other,train
4/7000.jpg,0,other,train
4/8000.jpg,0,other,train
4/9000.jpg,0,other,train
4/10000.jpg,0,other,train
4/11000.jpg,0,other,train
4/12000.jpg,0,other,train
4/13000.jpg,0,other,train
4/24000.jpg,0,other,train
4/25000.jpg,0,other,train
4/26000.jpg,0,other,train
4/27000.jpg,0,other,train
4/28000.jpg,0,other,train
4/29000.jpg,0,other,train
4/30000.jpg,0,other,train
4/31000.jpg,0,other,train
4/32000.jpg,0,other,train
4/43000.jpg,0,other,train
4/44000.jpg,0,other,train
4/45000.jpg,0,other,train
4/46000.jpg,0,other,train

 server/snapshot_infer/docs/annotation_spec.md

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,58 @@
# é…é€è§†é¢‘门头/交付时刻标注规范

## ç›®æ ‡

为 ONNX å¸§åˆ†ç±»æ¨¡åž‹æä¾›è®­ç»ƒæ ‡ç­¾ï¼šæ¯æ¡è§†é¢‘标注 **门头最佳时刻** ä¸Ž **交付最佳时刻**（各 1 ä¸ªç§’数）。

## å­—段说明

| å­—段 | ç±»åž‹ | å¿…å¡« | è¯´æ˜Ž |
|------|------|------|------|
| `media_id` | int | æ˜¯ | å¯¹åº” `collection_media.id` |
| `video_path` | string | æ˜¯ | æœ¬åœ°è·¯å¾„或 FTP HTTP URL |
| `storefront_time_sec` | float | æ˜¯ | é—¨å¤´æœ€ä½³å¸§æ—¶åˆ»ï¼ˆç§’） |
| `handover_time_sec` | float | æ˜¯ | äº¤ä»˜æœ€ä½³å¸§æ—¶åˆ»ï¼ˆç§’） |
| `store_type` | string | å¦ | ä¾¿åˆ©åº—/超市/餐饮/其他 |
| `has_voice_marker` | bool | å¦ | æ˜¯å¦å«ã€Œåˆ°åº—/交付」语音 |
| `recorder_sn` | string | å¦ | è®¾å¤‡ SN |
| `driver_date` | string | å¦ | å¸æœº+日期，用于 train/val/test åˆ†ç»„ |
| `split` | string | æ˜¯ | `train` / `val` / `test` |
| `notes` | string | å¦ | badcase è¯´æ˜Ž |

## ç±»åˆ«è¾¹ç•Œ

- **storefront（门头）**：店招、门牌、店铺入口为主体；人可入画但货品非主体
- **handover（交付）**：货品/包装在画面中心，可见递交、放置、签收动作
- **other（负样本）**：行车、仓库、店内走动、空镜等（训练时自动从非 Â±5s çª—口采样）

## æ ‡æ³¨æ“ä½œ

1. æ’­æ”¾æ•´æ®µ MP4，暂停在 **最清晰、构图最好** çš„门头画面，记录当前秒数
2. ç»§ç»­æ’­æ”¾ï¼Œåœ¨ **货品交接最清晰** çš„一帧记录秒数
3. çº¦æŸï¼š`handover_time_sec > storefront_time_sec`（通常相差数十秒以上）
4. è‹¥æŸæ¡è§†é¢‘无交付场景（仅到店），在 `notes` æ ‡æ³¨ã€Œæ— äº¤ä»˜ã€ï¼Œè¯¥æ¡æš‚不纳入训练

## å¯¼å‡ºæ ¼å¼ï¼ˆJSONL）

每行一条 JSON：

```json
{"media_id": 123, "video_path": "http://host/collection_media/20250609/123.mp4", "storefront_time_sec": 742.5, "handover_time_sec": 1085.2, "store_type": "便利店", "has_voice_marker": true, "driver_date": "driver001_20250609", "split": "train"}
```

## æ•°æ®åˆ’分

- train / val / test = **70% / 15% / 15%**
- æŒ‰ `driver_date` æˆ– `recorder_sn + æ—¥æœŸ` **分组划分**，避免同司机同天视频泄漏到测试集

## è§„模建议

| é˜¶æ®µ | è§†é¢‘æ•° |
|------|--------|
| POC | 80~100 |
| å†…测 | 300+ |
| ä¸Šçº¿ | 1000+ |

## Label Studio

见 [`label_studio_config.xml`](../tools/label_studio_config.xml)。导入 `tools/export_media_list.py` ç”Ÿæˆçš„ CSV åŽï¼Œæ ‡æ³¨ä¸¤ä¸ªæ—¶é—´ç‚¹å¹¶å¯¼å‡º JSON，再用 `tools/convert_labelstudio.py` è½¬ä¸º JSONL。

 server/snapshot_infer/docs/deploy_checklist.md

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,34 @@
# éƒ¨ç½²ä¸Žè¯„估检查清单

## ä¸Šçº¿å‰

- [ ] æ‰§è¡Œ `business.delivery_media_snapshot_feedback.sql`
- [ ] è®­ç»ƒå¹¶éƒ¨ç½² `models/storefront_int8.onnx`、`models/handover_int8.onnx`、`version.json`
- [ ] æŽ¨ç†æœºå®‰è£… ffmpeg，且能访问 Java æž„造的 `video_url`（FTP HTTP å‰ç¼€ï¼‰
- [ ] `application-pro.yml` ä¸­ `snapshot.infer.fail-open-mock: false`
- [ ] `GET /health` è¿”回 `status=ok` ä¸”两个 ONNX loaded=true

## æ€§èƒ½ç›®æ ‡ï¼ˆ10 åˆ†é’Ÿè§†é¢‘，CPU）

| æŒ‡æ ‡ | POC ç›®æ ‡ |
|------|----------|
| å•æ¡ analyze è€—æ—¶ | < 10 åˆ†é’Ÿ |
| é—¨å¤´ MAE | < 8s |
| äº¤ä»˜ MAE | < 8s |
| é¡ºåºæ­£ç¡®çŽ‡ | > 95% |

压测命令：

```powershell
cd server/snapshot_infer
py tools/benchmark_cpu.py "http://your-ftp-host/collection_media/xxx.mp4" --sample-fps 0.5
```

## æ•…障排查

| çŽ°è±¡ | å¤„理 |
|------|------|
| health degraded | æ£€æŸ¥ ONNX æ–‡ä»¶æ˜¯å¦åœ¨ SNAPSHOT_MODEL_DIR |
| analyze è¶…æ—¶ | å¢žå¤§ Java `read-timeout-ms`；降低 `sample-fps` |
| ASR æ…¢ | ç¡®è®¤ `WHISPER_MODEL=tiny`；或 `enable-asr: false` çº¯è§†è§‰ |
| Java snapshot_status=3 | æŸ¥çœ‹ `snapshot_message` ä¸ŽæŽ¨ç†æœåŠ¡æ—¥å¿— |

 server/snapshot_infer/docs/training_troubleshooting.md

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,96 @@
# è®­ç»ƒæ•ˆæžœä¸ä½³ â€” æŽ’查与改进指南

## å·²ä¿®å¤çš„代码问题（必做）

此前 `train.py` **只读取 `task=storefront/handover` ä¸” `label=1` çš„è¡Œ**，`task=other` çš„负样本全部被忽略，模型相当于「全图都是正类」，验证必然很差。

`prepare_dataset.py` å·²æ”¹ä¸ºï¼š**每一帧同时写入 storefront / handover ä¸¤è¡Œï¼Œå« label 0/1**。

请 **重新生成数据并训练**：

```powershell
cd server/snapshot_infer/training
py prepare_dataset.py -c config.yaml
py train.py -c config.yaml
py export_onnx.py -c config.yaml --version 1.0.1
py evaluate.py -c config.yaml --split val
```

训练开始时会打印 `train pos=... neg=...`，**neg å¿…é¡» > 0**。

---

## ä½ å½“前数据的问题

查看 `data/annotations.jsonl`：

| é—®é¢˜ | è¯´æ˜Ž |
|------|------|
| ä»… **4 æ¡** æ ‡æ³¨ | è¿œä½ŽäºŽ POC å»ºè®® 80~100 æ¡ |
| å®žé™…只有 **2 ä¸ªä¸åŒè§†é¢‘** | media 1/3 åŒä¸€æ–‡ä»¶ï¼Œ2/4 åŒä¸€æ–‡ä»¶ |
| train/val **重复同一视频** | éªŒè¯æŒ‡æ ‡è™šé«˜æˆ–失真，无法反映泛化 |
| åœºæ™¯å•ä¸€ | å…¨æ˜¯ã€Œå…¬å¸ã€å®¤å†…，模型学不到真实门头/交付 |

**结论**：即使代码正确，4 æ¡æ ·æœ¬ä¹Ÿå‡ ä¹Žä¸å¯èƒ½è®­å‡ºå¯ç”¨æ¨¡åž‹ã€‚需要继续标注 **80+ æ¡çœŸå®žé…é€è§†é¢‘**，且 train/val æŒ‰ **司机+日期** åˆ†ç»„，不能同一视频进 train åˆè¿› val。

---

## æ ‡æ³¨è´¨é‡æ£€æŸ¥

1. **门头时刻**：店招/入口最清晰的一帧（不是车内、不是背影）
2. **交付时刻**：货品/交接动作最清晰的一帧
3. çº¦æŸï¼š`handover_time_sec > storefront_time_sec + 20s`
4. æœ‰è¯­éŸ³æ ‡è®°çš„视频，可开 ASR èžåˆï¼ˆ`enable-asr: true`），减轻纯视觉压力

---

## è®­ç»ƒå‚数建议（`config.yaml`）

```yaml
sampling:
  sample_fps: 1.0          # ä¸ŽæŽ¨ç† sample-fps ä¸€è‡´
  positive_window_sec: 3.0   # æ­£æ ·æœ¬çª—口 Â±3s（原 5s è¿‡å®½ä¼šæ¨¡ç³Šå³°å€¼ï¼‰
  other_downsample_ratio: 2  # è´Ÿæ ·æœ¬ä¸è¦å¤ªå°‘

train:
  batch_size: 32
  epochs: 40
  lr: 0.0005               # å…¨é‡å¾®è°ƒæ—¶å¯ç•¥é™
  early_stop_patience: 8
  freeze_backbone: true      # æ ·æœ¬ <200 æ—¶è‡ªåŠ¨å†»ç»“ backbone（已实现）
```

样本量 >300 åŽï¼Œå¯è®¾ `freeze_backbone: false` åšåˆ†å±‚学习率微调。

---

## è¯„估指标怎么看

`evaluate.py` è¾“出：

| æŒ‡æ ‡ | POC ç›®æ ‡ |
|------|----------|
| é—¨å¤´ MAE | < 8s |
| äº¤ä»˜ MAE | < 8s |
| é¡ºåºæ­£ç¡®çŽ‡ | > 95% |
| åŒ 5 ç§’命中率 | > 70% |

若帧级 loss ä½Žä½† MAE ä»å¤§ï¼šæ£€æŸ¥ **sample_fps è®­ç»ƒä¸ŽæŽ¨ç†æ˜¯å¦ä¸€è‡´**、标注时刻是否准。

---

## çŸ­æœŸå¯ç”¨æ–¹æ¡ˆï¼ˆæ•°æ®ä¸å¤Ÿæ—¶ï¼‰

1. **ASR ä¸ºä¸»**：司机到店/交付时清晰说「到店」「交付」，开 `snapshot.infer.enable-asr: true`，视觉仅辅助
2. **人工纠正**：Admin「人工纠正」写入 feedback，积累后再训
3. **dev mock å…œåº•**：`fail-open-mock: true` ä»…用于联调，不是生产方案

---

## æŽ¨èè¿­ä»£é¡ºåº

1. é‡æ–° `prepare_dataset` + `train` + `export`（修复负样本）
2. æ ‡æ³¨æ‰©åˆ° 50~100 æ¡ï¼ŒåŽ»æŽ‰ val ä¸Ž train é‡å¤è§†é¢‘
3. è·‘ `evaluate.py`，记录 MAE
4. ä¸Šçº¿ snapshot-infer，用 10 æ¡çœŸå®žè§†é¢‘人工看「查看快照」效果
5. badcase äººå·¥çº æ­£ â†’ `export_feedback.py` â†’ åˆå¹¶è¿› `annotations.jsonl` å¢žé‡è®­ç»ƒ

 server/snapshot_infer/models/version.json

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,8 @@
{
  "model_version": "1.0.0",
  "storefront_model": "storefront.onnx",
  "handover_model": "handover.onnx",
  "image_size": 224,
  "quantized": false,
  "exported_at": "2026-06-12T09:24:48.314213Z"
}

 server/snapshot_infer/requirements.txt

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,9 @@
fastapi>=0.100.0
uvicorn>=0.22.0
pydantic>=2.0.0
onnxruntime>=1.16.0
opencv-python-headless>=4.8.0
numpy>=1.24.0
pillow>=10.0.0
faster-whisper>=1.0.0
httpx>=0.25.0

 server/snapshot_infer/tools/benchmark_cpu.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,52 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""CPU åŽ‹æµ‹ï¼šå¯¹å•æ¡è§†é¢‘重复调用 pipeline，统计耗时。"""
import argparse
import os
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from app.schemas import AnalyzeRequest
from app.pipeline import run_analyze, get_registry


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("video_url", help="可访问的 MP4 URL")
    parser.add_argument("--media-id", type=int, default=1)
    parser.add_argument("--duration", type=float, default=0)
    parser.add_argument("--sample-fps", type=float, default=0.5)
    parser.add_argument("--no-asr", action="store_true")
    parser.add_argument("--repeat", type=int, default=1)
    args = parser.parse_args()

    reg = get_registry()
    print(f"model_version={reg.version} ready={reg.ready}")
    if not reg.ready:
        print("警告: ONNX æ¨¡åž‹æœªå°±ç»ªï¼Œç»“果可能失败")

    req = AnalyzeRequest(
        media_id=args.media_id,
        video_url=args.video_url,
        sample_fps=args.sample_fps,
        enable_asr=not args.no_asr,
        duration_sec=args.duration if args.duration > 0 else None,
    )
    times = []
    for i in range(args.repeat):
        t0 = time.perf_counter()
        resp = run_analyze(req)
        elapsed = time.perf_counter() - t0
        times.append(elapsed)
        print(f"run {i+1}: success={resp.success} elapsed={elapsed:.1f}s "
              f"storefront={resp.storefront.time_sec if resp.storefront else None} "
              f"handover={resp.handover.time_sec if resp.handover else None} "
              f"msg={resp.message or ''}")
    if times:
        print(f"avg={sum(times)/len(times):.1f}s min={min(times):.1f}s max={max(times):.1f}s")


if __name__ == "__main__":
    main()

 server/snapshot_infer/tools/convert_labelstudio.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,64 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""将 Label Studio å¯¼å‡º JSON è½¬ä¸ºè®­ç»ƒç”¨ JSONL。"""
import argparse
import json
import sys


def convert(in_path, out_path, default_split="train"):
    with open(in_path, encoding="utf-8") as f:
        tasks = json.load(f)
    if isinstance(tasks, dict):
        tasks = tasks.get("tasks") or tasks.get("data") or [tasks]

    count = 0
    with open(out_path, "w", encoding="utf-8") as out:
        for task in tasks:
            data = task.get("data") or task
            media_id = data.get("media_id") or data.get("id")
            video_path = data.get("video_path") or data.get("video")
            annotations = task.get("annotations") or []
            storefront = handover = None
            for ann in annotations:
                for r in ann.get("result") or []:
                    if r.get("type") != "timelinelabels":
                        continue
                    labels = (r.get("value") or {}).get("timelinelabels") or []
                    ranges = (r.get("value") or {}).get("ranges") or []
                    if not ranges:
                        continue
                    t = float(ranges[0].get("start", 0))
                    if "storefront" in labels:
                        storefront = t
                    if "handover" in labels:
                        handover = t
            if storefront is None or handover is None:
                continue
            item = {
                "media_id": int(media_id) if media_id else count,
                "video_path": video_path,
                "storefront_time_sec": round(storefront, 2),
                "handover_time_sec": round(handover, 2),
                "store_type": data.get("store_type", ""),
                "has_voice_marker": bool(data.get("has_voice_marker")),
                "driver_date": data.get("driver_date", ""),
                "split": data.get("split") or default_split,
                "notes": data.get("notes", ""),
            }
            out.write(json.dumps(item, ensure_ascii=False) + "\n")
            count += 1
    print(f"转换 {count} æ¡ -> {out_path}")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("input", help="Label Studio å¯¼å‡º JSON")
    parser.add_argument("-o", "--output", default="data/annotations.jsonl")
    parser.add_argument("--split", default="train")
    args = parser.parse_args()
    convert(args.input, args.output, args.split)


if __name__ == "__main__":
    main()

 server/snapshot_infer/tools/export_feedback.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,80 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""从 MySQL delivery_media_snapshot_feedback å¯¼å‡ºå¢žé‡è®­ç»ƒ JSONL。"""
import argparse
import json
import os
import sys

try:
    import pymysql
except ImportError:
    pymysql = None


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default=os.environ.get("MYSQL_HOST", "127.0.0.1"))
    parser.add_argument("--port", type=int, default=int(os.environ.get("MYSQL_PORT", "3306")))
    parser.add_argument("--user", default=os.environ.get("MYSQL_USER", "root"))
    parser.add_argument("--password", default=os.environ.get("MYSQL_PASSWORD", ""))
    parser.add_argument("--database", default=os.environ.get("MYSQL_DATABASE", "wuhuyancao"))
    parser.add_argument("--ftp-prefix", default=os.environ.get("FTP_RESOURCE_PREFIX", "http://127.0.0.1/files"))
    parser.add_argument("-o", "--output", default="data/feedback_export.jsonl")
    args = parser.parse_args()

    if pymysql is None:
        print("请安装 pymysql", file=sys.stderr)
        sys.exit(1)

    media_folder = os.environ.get("COLLECTION_MEDIA_FOLDER", "/collection_media/")
    conn = pymysql.connect(
        host=args.host, port=args.port, user=args.user,
        password=args.password, database=args.database, charset="utf8mb4",
    )
    sql = """
        SELECT f.media_id, f.snapshot_type, f.ai_time_sec, f.manual_time_sec,
               m.file_path_local, m.start_time, m.recorder_sn
        FROM delivery_media_snapshot_feedback f
        JOIN collection_media m ON m.id = f.media_id AND m.isdeleted = 0
        WHERE f.isdeleted = 0
        ORDER BY f.id
    """
    groups = {}
    with conn.cursor() as cur:
        cur.execute(sql)
        for row in cur.fetchall():
            media_id, snap_type, ai_t, manual_t, path_local, start_time, recorder_sn = row
            if media_id not in groups:
                video_url = args.ftp_prefix.rstrip("/") + "/" + media_folder.strip("/") + "/" + (path_local or "").lstrip("/")
                driver_date = ""
                if start_time:
                    driver_date = f"{recorder_sn or 'unknown'}_{start_time.strftime('%Y%m%d')}"
                groups[media_id] = {
                    "media_id": media_id,
                    "video_path": video_url,
                    "storefront_time_sec": None,
                    "handover_time_sec": None,
                    "driver_date": driver_date,
                    "split": "train",
                    "notes": "from_feedback",
                }
            if snap_type == 1:
                groups[media_id]["storefront_time_sec"] = float(manual_t)
            elif snap_type == 2:
                groups[media_id]["handover_time_sec"] = float(manual_t)
    conn.close()

    os.makedirs(os.path.dirname(args.output) or ".", exist_ok=True)
    count = 0
    with open(args.output, "w", encoding="utf-8") as out:
        for item in groups.values():
            if item["storefront_time_sec"] is None or item["handover_time_sec"] is None:
                continue
            out.write(json.dumps(item, ensure_ascii=False) + "\n")
            count += 1
    print(f"导出 {count} æ¡ -> {args.output}")


if __name__ == "__main__":
    main()

 server/snapshot_infer/tools/export_media_list.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,136 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""从 MySQL collection_media å¯¼å‡ºå¾…标注视频清单（CSV + JSONL æ¨¡æ¿ï¼‰ã€‚"""
import argparse
import csv
import json
import os
import sys

try:
    import pymysql
except ImportError:
    pymysql = None


def export_from_mysql(host, port, user, password, database, ftp_prefix, limit, out_csv, out_jsonl):
    if pymysql is None:
        print("请安装 pymysql: py -m pip install pymysql", file=sys.stderr)
        sys.exit(1)
    conn = pymysql.connect(
        host=host, port=port, user=user, password=password,
        database=database, charset="utf8mb4",
    )
    sql = """
        SELECT id, file_name, file_path_local, start_time, end_time, recorder_sn
        FROM collection_media
        WHERE isdeleted = 0 AND download_status = 1 AND media_type = 0
          AND file_path_local IS NOT NULL AND file_path_local != ''
        ORDER BY id DESC
        LIMIT %s
    """
    with conn.cursor() as cur:
        cur.execute(sql, (limit,))
        rows = cur.fetchall()
    conn.close()

    media_folder = os.environ.get("COLLECTION_MEDIA_FOLDER", "/collection_media/")
    if not media_folder.endswith("/"):
        media_folder += "/"

    records = []
    for row in rows:
        media_id, file_name, file_path_local, start_time, end_time, recorder_sn = row
        video_url = ftp_prefix.rstrip("/") + "/" + media_folder.lstrip("/") + file_path_local.lstrip("/")
        driver_date = ""
        if start_time:
            driver_date = f"{recorder_sn or 'unknown'}_{start_time.strftime('%Y%m%d')}"
        records.append({
            "media_id": media_id,
            "file_name": file_name or "",
            "video_path": video_url,
            "recorder_sn": recorder_sn or "",
            "driver_date": driver_date,
            "storefront_time_sec": "",
            "handover_time_sec": "",
            "store_type": "",
            "has_voice_marker": "",
            "split": "",
            "notes": "",
        })

    os.makedirs(os.path.dirname(out_csv) or ".", exist_ok=True)
    with open(out_csv, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=list(records[0].keys()) if records else [])
        writer.writeheader()
        writer.writerows(records)

    with open(out_jsonl, "w", encoding="utf-8") as f:
        for r in records:
            template = {
                "media_id": r["media_id"],
                "video_path": r["video_path"],
                "storefront_time_sec": 0.0,
                "handover_time_sec": 0.0,
                "store_type": "",
                "has_voice_marker": False,
                "driver_date": r["driver_date"],
                "split": "train",
                "notes": "TODO: å¡«å†™æ ‡æ³¨",
            }
            f.write(json.dumps(template, ensure_ascii=False) + "\n")

    print(f"导出 {len(records)} æ¡ -> {out_csv}, {out_jsonl}")


def export_from_csv(in_csv, out_jsonl):
    records = []
    with open(in_csv, newline="", encoding="utf-8-sig") as f:
        for row in csv.DictReader(f):
            records.append(row)
    with open(out_jsonl, "w", encoding="utf-8") as f:
        for r in records:
            item = {
                "media_id": int(r["media_id"]),
                "video_path": r.get("video_path") or r.get("video_url", ""),
                "storefront_time_sec": float(r["storefront_time_sec"]) if r.get("storefront_time_sec") else 0.0,
                "handover_time_sec": float(r["handover_time_sec"]) if r.get("handover_time_sec") else 0.0,
                "store_type": r.get("store_type", ""),
                "has_voice_marker": str(r.get("has_voice_marker", "")).lower() in ("1", "true", "yes"),
                "driver_date": r.get("driver_date", ""),
                "split": r.get("split") or "train",
                "notes": r.get("notes", ""),
            }
            f.write(json.dumps(item, ensure_ascii=False) + "\n")
    print(f"转换 {len(records)} æ¡ -> {out_jsonl}")


def main():
    parser = argparse.ArgumentParser(description="导出 collection_media å¾…标注清单")
    parser.add_argument("--mysql", action="store_true", help="从 MySQL è¯»å–")
    parser.add_argument("--host", default=os.environ.get("MYSQL_HOST", "127.0.0.1"))
    parser.add_argument("--port", type=int, default=int(os.environ.get("MYSQL_PORT", "3306")))
    parser.add_argument("--user", default=os.environ.get("MYSQL_USER", "root"))
    parser.add_argument("--password", default=os.environ.get("MYSQL_PASSWORD", ""))
    parser.add_argument("--database", default=os.environ.get("MYSQL_DATABASE", "wuhuyancao"))
    parser.add_argument("--ftp-prefix", default=os.environ.get("FTP_RESOURCE_PREFIX", "http://127.0.0.1/files"))
    parser.add_argument("--limit", type=int, default=200)
    parser.add_argument("--out-csv", default="data/annotation_tasks.csv")
    parser.add_argument("--out-jsonl", default="data/annotations_template.jsonl")
    parser.add_argument("--from-csv", help="从已填 CSV è½¬ JSONL")
    args = parser.parse_args()

    if args.from_csv:
        export_from_csv(args.from_csv, args.out_jsonl)
    elif args.mysql:
        export_from_mysql(
            args.host, args.port, args.user, args.password, args.database,
            args.ftp_prefix, args.limit, args.out_csv, args.out_jsonl,
        )
    else:
        print("请指定 --mysql æˆ– --from-csv", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

 server/snapshot_infer/tools/label_studio_config.xml

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,9 @@
<View>
  <Header value="配送视频：门头 / äº¤ä»˜æ—¶åˆ»æ ‡æ³¨"/>
  <Video name="video" value="$video_path" sync="audio"/>
  <Labels name="event" toName="video">
    <Label value="storefront" background="#FFA500"/>
    <Label value="handover" background="#008000"/>
  </Labels>
  <TextArea name="notes" toName="video" placeholder="备注：门店类型、是否有语音标记等"/>
</View>

 server/snapshot_infer/training/config.yaml

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,28 @@
# è®­ç»ƒé…ç½®
data:
  annotations_jsonl: ../data/annotations.jsonl
  frames_dir: ../data/frames
  labels_csv: ../data/labels.csv

sampling:
  sample_fps: 1.0
  positive_window_sec: 3.0
  other_downsample_ratio: 2

model:
  backbone: mobilenet_v3_small
  image_size: 224
  pos_weight: null

train:
  batch_size: 32
  epochs: 40
  lr: 0.0005
  early_stop_patience: 8
  num_workers: 0
  freeze_backbone: true

export:
  opset: 17
  quantize: true
  output_dir: ../models

 server/snapshot_infer/training/evaluate.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,171 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""视频级评估：与推理 pipeline ä¸€è‡´çš„时序平滑 + é¡ºåºçº¦æŸã€‚"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path

import numpy as np
import onnxruntime as ort
import yaml
from PIL import Image

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from app.temporal import find_peaks_ordered, smooth_scores


def load_config(path):
    with open(path, encoding="utf-8") as f:
        return yaml.safe_load(f)


def resolve_model(models_dir, name):
    version_path = models_dir / "version.json"
    if version_path.is_file():
        meta = json.loads(version_path.read_text(encoding="utf-8"))
        key = f"{name}_model"
        if meta.get(key):
            p = models_dir / meta[key]
            if p.is_file():
                return p
    for suffix in (f"{name}_int8.onnx", f"{name}.onnx"):
        p = models_dir / suffix
        if p.is_file():
            return p
    return None


def ffprobe_duration(video_path):
    cmd = [
        "ffprobe", "-v", "error", "-show_entries", "format=duration",
        "-of", "default=noprint_wrappers=1:nokey=1", video_path,
    ]
    try:
        out = subprocess.check_output(cmd, stderr=subprocess.DEVNULL, text=True).strip()
        return float(out) if out else 0.0
    except Exception:
        return 0.0


def download_if_needed(video_path, cache_dir):
    if os.path.isfile(video_path):
        return video_path, None
    if not video_path.startswith("http"):
        return None, None
    import httpx
    os.makedirs(cache_dir, exist_ok=True)
    local = os.path.join(cache_dir, "eval_tmp.mp4")
    with httpx.stream("GET", video_path, timeout=600.0, follow_redirects=True) as r:
        r.raise_for_status()
        with open(local, "wb") as f:
            for chunk in r.iter_bytes():
                f.write(chunk)
    return local, local


def sample_scores(video_path, session, sample_fps, image_size):
    duration = ffprobe_duration(video_path)
    if duration <= 0:
        return [], duration
    times, scores = [], []
    t, step = 0.0, 1.0 / sample_fps
    tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
    tmp.close()
    mean = np.array([0.485, 0.456, 0.406], dtype=np.float32)
    std = np.array([0.229, 0.224, 0.225], dtype=np.float32)
    try:
        while t <= duration:
            subprocess.run(
                ["ffmpeg", "-y", "-ss", str(t), "-i", video_path, "-frames:v", "1", "-q:v", "2", tmp.name],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            )
            if os.path.isfile(tmp.name) and os.path.getsize(tmp.name) > 0:
                img = Image.open(tmp.name).convert("RGB").resize((image_size, image_size))
                arr = (np.array(img).astype(np.float32) / 255.0 - mean) / std
                arr = arr.transpose(2, 0, 1)[None].astype(np.float32)
                logit = session.run(None, {"input": arr})[0][0][0]
                prob = float(1.0 / (1.0 + np.exp(-logit)))
                times.append(round(t, 2))
                scores.append(prob)
            t += step
    finally:
        if os.path.isfile(tmp.name):
            os.remove(tmp.name)
    return list(zip(times, scores)), duration


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-c", "--config", default=str(Path(__file__).parent / "config.yaml"))
    parser.add_argument("--annotations", default="../data/annotations.jsonl")
    parser.add_argument("--models-dir", default="../models")
    parser.add_argument("--sample-fps", type=float, default=0.5)
    parser.add_argument("--split", default="val")
    args = parser.parse_args()
    cfg = load_config(args.config)
    size = cfg["model"]["image_size"]
    models_dir = Path(args.config).resolve().parent / args.models_dir

    sf_path = resolve_model(models_dir, "storefront")
    ho_path = resolve_model(models_dir, "handover")
    if not sf_path or not ho_path:
        print("未找到 ONNX æ¨¡åž‹ï¼Œè¯·å…ˆ export_onnx.py")
        return

    sf_sess = ort.InferenceSession(str(sf_path), providers=["CPUExecutionProvider"])
    ho_sess = ort.InferenceSession(str(ho_path), providers=["CPUExecutionProvider"])

    ann_path = Path(args.config).resolve().parent / args.annotations
    items = []
    with open(ann_path, encoding="utf-8") as f:
        for line in f:
            if line.strip():
                items.append(json.loads(line))
    val_items = [i for i in items if i.get("split") == args.split] or items

    cache_dir = str(Path(args.config).resolve().parent / "../data/eval_cache")
    sf_mae = ho_mae = order_ok = hit5 = n = 0
    for item in val_items:
        vp = item["video_path"]
        local, tmp = download_if_needed(vp, cache_dir)
        if not local:
            print(f"跳过 media_id={item['media_id']}: è§†é¢‘不可访问")
            continue
        sf_scores, duration = sample_scores(local, sf_sess, args.sample_fps, size)
        ho_scores, _ = sample_scores(local, ho_sess, args.sample_fps, size)
        sf_peak, ho_peak = find_peaks_ordered(sf_scores, ho_scores, duration)
        pred_sf = sf_peak[0] if sf_peak else 0.0
        pred_ho = ho_peak[0] if ho_peak else 0.0
        gt_sf = float(item["storefront_time_sec"])
        gt_ho = float(item["handover_time_sec"])
        sf_err = abs(pred_sf - gt_sf)
        ho_err = abs(pred_ho - gt_ho)
        sf_mae += sf_err
        ho_mae += ho_err
        if pred_ho > pred_sf:
            order_ok += 1
        if sf_err <= 5 and ho_err <= 5:
            hit5 += 1
        n += 1
        print(
            f"media_id={item['media_id']} gt_sf={gt_sf}s pred_sf={pred_sf:.1f}s err={sf_err:.1f}s | "
            f"gt_ho={gt_ho}s pred_ho={pred_ho:.1f}s err={ho_err:.1f}s"
        )
        if tmp and os.path.isfile(tmp):
            os.remove(tmp)

    if n == 0:
        print("无可用验证样本")
        return
    print("---")
    print(f"样本数={n}")
    print(f"门头 MAE={sf_mae/n:.2f}s  äº¤ä»˜ MAE={ho_mae/n:.2f}s")
    print(f"顺序正确率={order_ok/n*100:.1f}%  åŒ5秒命中率={hit5/n*100:.1f}%")


if __name__ == "__main__":
    main()

 server/snapshot_infer/training/export_onnx.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,128 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""导出 PyTorch æƒé‡ä¸º ONNX，并可选 INT8 åŠ¨æ€é‡åŒ–。"""
import argparse
import json
import os
import shutil
import subprocess
import sys
from datetime import datetime
from pathlib import Path

import torch
import yaml
from train import build_model


def load_config(path):
    with open(path, encoding="utf-8") as f:
        cfg = yaml.safe_load(f)
    base = Path(path).resolve().parent
    out = cfg["export"]["output_dir"]
    if not os.path.isabs(out):
        cfg["export"]["output_dir"] = str((base / out).resolve())
    return cfg


def try_quantize_subprocess(onnx_path: str, int8_path: str) -> bool:
    """在子进程执行 ORT é‡åŒ–，避免主进程因 Windows ä¸Š ORT bug å´©æºƒã€‚"""
    code = (
        "import sys\n"
        "from onnxruntime.quantization import QuantType, quantize_dynamic\n"
        "quantize_dynamic(sys.argv[1], sys.argv[2], weight_type=QuantType.QUInt8)\n"
        "print('OK')\n"
    )
    try:
        result = subprocess.run(
            [sys.executable, "-c", code, onnx_path, int8_path],
            capture_output=True,
            text=True,
            timeout=600,
        )
    except subprocess.TimeoutExpired:
        print(f"量化超时: {int8_path}")
        return False
    if result.returncode != 0:
        err = (result.stderr or result.stdout or "").strip()
        if err:
            print(f"量化失败 exit={result.returncode}: {err[:500]}")
        else:
            print(f"量化失败 exit={result.returncode}（Windows ä¸Š ORT é‡åŒ–器可能崩溃）")
        return False
    return os.path.isfile(int8_path) and os.path.getsize(int8_path) > 0


def export_one(task, cfg, do_quantize: bool):
    out_dir = cfg["export"]["output_dir"]
    os.makedirs(out_dir, exist_ok=True)
    size = cfg["model"]["image_size"]
    ckpt = os.path.join(out_dir, f"{task}.pt")
    if not os.path.isfile(ckpt):
        raise FileNotFoundError(f"未找到权重 {ckpt}，请先运行 train.py")

    model = build_model()
    state = torch.load(ckpt, map_location="cpu")
    model.load_state_dict(state)
    model.eval()

    dummy = torch.randn(1, 3, size, size)
    onnx_path = os.path.join(out_dir, f"{task}.onnx")
    torch.onnx.export(
        model,
        dummy,
        onnx_path,
        input_names=["input"],
        output_names=["logits"],
        dynamic_axes={"input": {0: "batch"}, "logits": {0: "batch"}},
        opset_version=cfg["export"]["opset"],
    )
    print(f"导出 float ONNX: {onnx_path}")

    int8_path = os.path.join(out_dir, f"{task}_int8.onnx")
    if do_quantize and cfg["export"].get("quantize"):
        if try_quantize_subprocess(onnx_path, int8_path):
            print(f"量化 INT8: {int8_path}")
            return os.path.basename(int8_path)
        print(
            f"WARN: {task}_int8.onnx é‡åŒ–未成功，推理服务将使用 float æ¨¡åž‹ {task}.onnx\n"
            "      å¸¸è§åŽŸå› : Windows ä¸Š onnxruntime.quantization å´©æºƒã€‚"
            " å¯å‡çº§/降级 onnxruntime，或使用 --no-quantize è·³è¿‡é‡åŒ–。"
        )
    elif do_quantize:
        print(f"跳过量化（config quantize=false）")

    return os.path.basename(onnx_path)


def main():
    parser = argparse.ArgumentParser(description="导出 storefront/handover ONNX æ¨¡åž‹")
    parser.add_argument("-c", "--config", default=str(Path(__file__).parent / "config.yaml"))
    parser.add_argument("--version", default="1.0.0")
    parser.add_argument("--no-quantize", action="store_true", help="仅导出 float .onnx，不尝试 INT8")
    args = parser.parse_args()
    cfg = load_config(args.config)
    do_quantize = not args.no_quantize

    paths = {}
    for task in ("storefront", "handover"):
        paths[task] = export_one(task, cfg, do_quantize)

    version_info = {
        "model_version": args.version,
        "storefront_model": paths["storefront"],
        "handover_model": paths["handover"],
        "image_size": cfg["model"]["image_size"],
        "quantized": do_quantize and paths["storefront"].endswith("_int8.onnx"),
        "exported_at": datetime.utcnow().isoformat() + "Z",
    }
    version_path = os.path.join(cfg["export"]["output_dir"], "version.json")
    with open(version_path, "w", encoding="utf-8") as f:
        json.dump(version_info, f, indent=2, ensure_ascii=False)
    print(f"写入 {version_path}")
    print(f"  storefront -> {paths['storefront']}")
    print(f"  handover   -> {paths['handover']}")


if __name__ == "__main__":
    main()

 server/snapshot_infer/training/extract_frames.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""prepare_dataset åˆ«åå…¥å£ã€‚"""
from prepare_dataset import main

if __name__ == "__main__":
    main()

 server/snapshot_infer/training/prepare_dataset.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,150 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""从 annotations.jsonl ç”Ÿæˆ labels.csv（帧路径, label, task）。"""
import argparse
import csv
import json
import os
import random
import subprocess
import sys
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parent.parent


def load_config(path):
    with open(path, encoding="utf-8") as f:
        cfg = yaml.safe_load(f)
    base = Path(path).resolve().parent
    for key in ("annotations_jsonl", "frames_dir", "labels_csv"):
        p = cfg["data"][key]
        if not os.path.isabs(p):
            cfg["data"][key] = str((base / p).resolve())
    cfg["data"]["output_dir"] = cfg["export"]["output_dir"]
    if not os.path.isabs(cfg["export"]["output_dir"]):
        cfg["export"]["output_dir"] = str((base / cfg["export"]["output_dir"]).resolve())
    return cfg


def ffprobe_duration(video_path, ffmpeg_dir=""):
    ffprobe = "ffprobe"
    if ffmpeg_dir:
        ffprobe = os.path.join(ffmpeg_dir, "ffprobe.exe" if os.name == "nt" else "ffprobe")
    cmd = [
        ffprobe, "-v", "error", "-show_entries", "format=duration",
        "-of", "default=noprint_wrappers=1:nokey=1", video_path,
    ]
    try:
        out = subprocess.check_output(cmd, stderr=subprocess.DEVNULL, text=True).strip()
        return float(out) if out else 0.0
    except Exception:
        return 0.0


def extract_frame_at(video_path, out_path, sec, ffmpeg_dir=""):
    ffmpeg = "ffmpeg"
    if ffmpeg_dir:
        ffmpeg = os.path.join(ffmpeg_dir, "ffmpeg.exe" if os.name == "nt" else "ffmpeg")
    os.makedirs(os.path.dirname(out_path), exist_ok=True)
    cmd = [
        ffmpeg, "-y", "-ss", str(sec), "-i", video_path,
        "-frames:v", "1", "-q:v", "2", out_path,
    ]
    subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)


def process_video(item, cfg, ffmpeg_dir):
    media_id = item["media_id"]
    video_path = item["video_path"]
    if not os.path.isfile(video_path) and not video_path.startswith("http"):
        print(f"跳过 media_id={media_id}: è§†é¢‘不存在 {video_path}")
        return []

    local_video = video_path
    tmp_video = None
    if video_path.startswith("http"):
        import httpx
        tmp_video = os.path.join(cfg["data"]["frames_dir"], f"_tmp_{media_id}.mp4")
        os.makedirs(cfg["data"]["frames_dir"], exist_ok=True)
        with httpx.stream("GET", video_path, timeout=600.0) as r:
            r.raise_for_status()
            with open(tmp_video, "wb") as f:
                for chunk in r.iter_bytes():
                    f.write(chunk)
        local_video = tmp_video

    duration = ffprobe_duration(local_video, ffmpeg_dir)
    if duration <= 0:
        duration = max(item.get("handover_time_sec", 600), 600)

    sample_fps = cfg["sampling"]["sample_fps"]
    win = cfg["sampling"]["positive_window_sec"]
    sf_t = float(item["storefront_time_sec"])
    ho_t = float(item["handover_time_sec"])

    rows = []
    t = 0.0
    step = 1.0 / sample_fps
    other_count = 0
    other_budget = max(1, int(duration * sample_fps / cfg["sampling"]["other_downsample_ratio"]))
    while t <= duration:
        rel = f"{media_id}/{int(t * 1000)}.jpg"
        out_path = os.path.join(cfg["data"]["frames_dir"], rel)
        is_sf = abs(t - sf_t) <= win
        is_ho = abs(t - ho_t) <= win
        need_other = not is_sf and not is_ho and other_count < other_budget
        if is_sf or is_ho or need_other:
            extract_frame_at(local_video, out_path, t, ffmpeg_dir)
            if os.path.isfile(out_path):
                split = item.get("split", "train")
                sf_label = 1 if is_sf else 0
                ho_label = 1 if is_ho else 0
                rows.append((rel, sf_label, "storefront", split))
                rows.append((rel, ho_label, "handover", split))
                if need_other:
                    other_count += 1
        t += step
    if tmp_video and os.path.isfile(tmp_video):
        os.remove(tmp_video)
    return rows


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-c", "--config", default=str(Path(__file__).parent / "config.yaml"))
    parser.add_argument("--ffmpeg-dir", default=os.environ.get("FFMPEG_DIR", ""))
    parser.add_argument("--limit", type=int, default=0)
    args = parser.parse_args()
    cfg = load_config(args.config)
    os.makedirs(cfg["data"]["frames_dir"], exist_ok=True)

    items = []
    with open(cfg["data"]["annotations_jsonl"], encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                items.append(json.loads(line))
    if args.limit:
        items = items[: args.limit]

    all_rows = []
    for i, item in enumerate(items):
        if item.get("storefront_time_sec", 0) <= 0 or item.get("handover_time_sec", 0) <= 0:
            print(f"跳过未标注 media_id={item.get('media_id')}")
            continue
        print(f"[{i+1}/{len(items)}] media_id={item['media_id']}")
        all_rows.extend(process_video(item, cfg, args.ffmpeg_dir))

    with open(cfg["data"]["labels_csv"], "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["frame_path", "label", "task", "split"])
        for row in all_rows:
            w.writerow(row)
    print(f"写入 {len(all_rows)} è¡Œ -> {cfg['data']['labels_csv']}")


if __name__ == "__main__":
    main()

 server/snapshot_infer/training/requirements-train.txt

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,10 @@
torch>=2.0.0
torchvision>=0.15.0
onnx>=1.14.0
onnxruntime>=1.16.0
pyyaml>=6.0
pillow>=9.0.0
opencv-python-headless>=4.8.0
numpy>=1.24.0
tqdm>=4.65.0
httpx>=0.25.0

 server/snapshot_infer/training/train.py

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,206 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""训练 storefront / handover äºŒåˆ†ç±»æ¨¡åž‹ï¼ˆMobileNetV3-Small）。"""
import argparse
import csv
import os
from pathlib import Path

import torch
import torch.nn as nn
import yaml
from PIL import Image
from torch.utils.data import DataLoader, Dataset
from torchvision import models, transforms
from tqdm import tqdm


class FrameDataset(Dataset):
    def __init__(self, rows, frames_dir, transform):
        self.rows = rows
        self.frames_dir = frames_dir
        self.transform = transform

    def __len__(self):
        return len(self.rows)

    def __getitem__(self, idx):
        path, label = self.rows[idx]
        img = Image.open(os.path.join(self.frames_dir, path)).convert("RGB")
        return self.transform(img), torch.tensor(label, dtype=torch.float32)


def load_config(path):
    with open(path, encoding="utf-8") as f:
        cfg = yaml.safe_load(f)
    base = Path(path).resolve().parent
    for key in ("frames_dir", "labels_csv"):
        p = cfg["data"][key]
        if not os.path.isabs(p):
            cfg["data"][key] = str((base / p).resolve())
    out = cfg["export"]["output_dir"]
    if not os.path.isabs(out):
        cfg["export"]["output_dir"] = str((base / out).resolve())
    return cfg


def read_labels(csv_path, task, split=None):
    """读取某任务的训练样本：该 task è¡Œå« label 0/1；兼容旧版 task=other è´Ÿæ ·æœ¬ã€‚"""
    rows = []
    with open(csv_path, newline="", encoding="utf-8") as f:
        for r in csv.DictReader(f):
            if split and r.get("split") and r["split"] != split:
                continue
            row_task = r["task"]
            label = int(r["label"])
            if row_task == task:
                rows.append((r["frame_path"], label))
            elif row_task == "other" and label == 0:
                rows.append((r["frame_path"], 0))
    return rows


def count_labels(rows):
    pos = sum(1 for _, y in rows if y == 1)
    neg = len(rows) - pos
    return pos, neg


def build_model(freeze_backbone=False):
    m = models.mobilenet_v3_small(weights=models.MobileNet_V3_Small_Weights.DEFAULT)
    if freeze_backbone:
        for p in m.features.parameters():
            p.requires_grad = False
    in_f = m.classifier[0].in_features
    m.classifier = nn.Sequential(
        nn.Linear(in_f, 128),
        nn.Hardswish(),
        nn.Dropout(0.2),
        nn.Linear(128, 1),
    )
    return m


def build_transforms(size, augment=False):
    if augment:
        return transforms.Compose([
            transforms.Resize((size, size)),
            transforms.RandomApply([
                transforms.ColorJitter(brightness=0.3, contrast=0.3, saturation=0.2),
            ], p=0.7),
            transforms.RandomApply([
                transforms.GaussianBlur(kernel_size=3),
            ], p=0.2),
            transforms.ToTensor(),
            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
        ])
    return transforms.Compose([
        transforms.Resize((size, size)),
        transforms.ToTensor(),
        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
    ])


def train_task(task, cfg):
    frames_dir = cfg["data"]["frames_dir"]
    train_rows = read_labels(cfg["data"]["labels_csv"], task, "train")
    val_rows = read_labels(cfg["data"]["labels_csv"], task, "val")
    if not train_rows:
        train_rows = read_labels(cfg["data"]["labels_csv"], task)
    if not val_rows:
        val_rows = train_rows[: max(1, len(train_rows) // 10)]

    train_pos, train_neg = count_labels(train_rows)
    val_pos, val_neg = count_labels(val_rows)
    print(f"[{task}] train pos={train_pos} neg={train_neg} | val pos={val_pos} neg={val_neg}")
    if train_neg == 0:
        print(f"WARN: [{task}] è®­ç»ƒé›†æ— è´Ÿæ ·æœ¬ï¼è¯·é‡æ–°è¿è¡Œ prepare_dataset.py ç”Ÿæˆ labels.csv")
    if train_pos < 10:
        print(f"WARN: [{task}] æ­£æ ·æœ¬è¿‡å°‘（<10），建议标注至少 80+ æ¡è§†é¢‘")

    size = cfg["model"]["image_size"]
    train_loader = DataLoader(
        FrameDataset(train_rows, frames_dir, build_transforms(size, augment=True)),
        batch_size=cfg["train"]["batch_size"],
        shuffle=True,
        num_workers=cfg["train"]["num_workers"],
    )
    val_loader = DataLoader(
        FrameDataset(val_rows, frames_dir, build_transforms(size, augment=False)),
        batch_size=cfg["train"]["batch_size"],
        shuffle=False,
        num_workers=cfg["train"]["num_workers"],
    )

    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    freeze = cfg["train"].get("freeze_backbone", True) and train_pos < 200
    model = build_model(freeze_backbone=freeze).to(device)
    if freeze:
        print(f"[{task}] å°æ ·æœ¬æ¨¡å¼ï¼šå†»ç»“ backbone，仅训练分类头")

    pos_count = max(train_pos, 1)
    neg_count = max(train_neg, 1)
    auto_pos_weight = min(neg_count / pos_count, 10.0)
    pos_weight_val = cfg["model"].get("pos_weight") or auto_pos_weight
    pos_weight = torch.tensor([pos_weight_val], device=device)
    print(f"[{task}] pos_weight={pos_weight_val:.2f}")

    criterion = nn.BCEWithLogitsLoss(pos_weight=pos_weight)
    lr = cfg["train"]["lr"]
    if freeze:
        optimizer = torch.optim.Adam(filter(lambda p: p.requires_grad, model.parameters()), lr=lr)
    else:
        optimizer = torch.optim.Adam([
            {"params": model.features.parameters(), "lr": lr * 0.1},
            {"params": model.classifier.parameters(), "lr": lr},
        ])

    best_loss = float("inf")
    patience = 0
    os.makedirs(cfg["export"]["output_dir"], exist_ok=True)
    ckpt_path = os.path.join(cfg["export"]["output_dir"], f"{task}.pt")

    for epoch in range(cfg["train"]["epochs"]):
        model.train()
        for x, y in tqdm(train_loader, desc=f"{task} epoch {epoch+1}"):
            x, y = x.to(device), y.to(device).unsqueeze(1)
            optimizer.zero_grad()
            loss = criterion(model(x), y)
            loss.backward()
            optimizer.step()

        model.eval()
        val_loss = 0.0
        n = 0
        with torch.no_grad():
            for x, y in val_loader:
                x, y = x.to(device), y.to(device).unsqueeze(1)
                val_loss += criterion(model(x), y).item() * x.size(0)
                n += x.size(0)
        val_loss /= max(n, 1)
        print(f"{task} epoch {epoch+1} val_loss={val_loss:.4f}")
        if val_loss < best_loss:
            best_loss = val_loss
            patience = 0
            torch.save(model.state_dict(), ckpt_path)
        else:
            patience += 1
            if patience >= cfg["train"]["early_stop_patience"]:
                break
    print(f"保存 {ckpt_path}")
    return ckpt_path


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-c", "--config", default=str(Path(__file__).parent / "config.yaml"))
    parser.add_argument("--task", choices=["storefront", "handover", "both"], default="both")
    args = parser.parse_args()
    cfg = load_config(args.config)
    tasks = ["storefront", "handover"] if args.task == "both" else [args.task]
    for t in tasks:
        train_task(t, cfg)


if __name__ == "__main__":
    main()

 server/system_service/src/main/java/com/doumee/core/utils/Constants.java

@@ -242,8 +242,6 @@
    public static final String CS_DOWNLOAD_BATCH_SIZE = "CS_DOWNLOAD_BATCH_SIZE";
    public static final String CS_FFMPEG_PATH = "CS_FFMPEG_PATH";
    public static final String COLLECTION_MEDIA_FOLDER = "COLLECTION_MEDIA";
    /** åª’体下载状态：下载中 */
    public static final int COLLECTION_MEDIA_DOWNLOADING = 3;
    public static final String SMS ="SMS" ;
    public static final String SMS_COMNAME = "SMS_COMNAME";
    public static final String SMS_IP ="SMS_IP" ;

 server/system_service/src/main/java/com/doumee/core/utils/DateUtil.java

@@ -462,7 +462,7 @@
     * @return String
     * @throws Exception
     */
    public static String getNowShortDate() throws Exception {
    public static String getNowShortDate()  {
        String nowDate = "";
        try {
            java.sql.Date date = null;

 server/visits/dmvisit_admin/src/main/java/com/doumee/cloud/admin/CollectionStationCloudController.java

@@ -5,11 +5,14 @@
import com.doumee.core.annotation.pr.PreventRepeat;
import com.doumee.core.utils.Constants;
import com.doumee.dao.admin.request.CollectionMediaSyncRequest;
import com.doumee.dao.admin.request.DeliverySnapshotManualRequest;
import com.doumee.dao.business.model.CollectionMedia;
import com.doumee.dao.business.model.CollectionDockDevice;
import com.doumee.dao.business.model.CollectionStation;
import com.doumee.dao.business.model.DeliveryMediaSnapshot;
import com.doumee.service.business.CollectionMediaSyncService;
import com.doumee.service.business.CollectionStationService;
import com.doumee.service.business.DeliverySnapshotService;
import com.doumee.service.business.third.model.ApiResponse;
import com.doumee.service.business.third.model.PageData;
import com.doumee.service.business.third.model.PageWrap;
@@ -30,6 +33,8 @@
    private CollectionStationService collectionStationService;
    @Autowired
    private CollectionMediaSyncService collectionMediaSyncService;
    @Autowired
    private DeliverySnapshotService deliverySnapshotService;

    @PreventRepeat
    @ApiOperation("新建采集站")
@@ -175,4 +180,27 @@
                                  javax.servlet.http.HttpServletResponse response) {
        collectionMediaSyncService.downloadMediaFile(id, request, response);
    }

    @PreventRepeat
    @ApiOperation("提交媒体快照分析(门头/交付)")
    @PostMapping("/media/snapshot/analyze/{id}")
    @CloudRequiredPermission({"business:collectionMedia:query", "business:collectionStation:query"})
    public ApiResponse<String> analyzeMediaSnapshot(@PathVariable Integer id) {
        return ApiResponse.success(deliverySnapshotService.submitAnalyze(id));
    }

    @ApiOperation("查询媒体快照(门头/交付)")
    @GetMapping("/media/snapshot/{id}")
    @CloudRequiredPermission({"business:collectionMedia:query", "business:collectionStation:query"})
    public ApiResponse<List<DeliveryMediaSnapshot>> listMediaSnapshot(@PathVariable Integer id) {
        return ApiResponse.success(deliverySnapshotService.listByMediaId(id));
    }

    @PreventRepeat
    @ApiOperation("手动指定媒体快照时间点")
    @PostMapping("/media/snapshot/manual")
    @CloudRequiredPermission({"business:collectionMedia:query", "business:collectionStation:query"})
    public ApiResponse<String> saveManualMediaSnapshot(@RequestBody DeliverySnapshotManualRequest request) {
        return ApiResponse.success(deliverySnapshotService.saveManual(request));
    }
}

 server/visits/dmvisit_service/src/main/java/com/doumee/core/haikang/isapi/model/MediaItemDTO.java

@@ -10,7 +10,7 @@
    private String fileName;
    private String playbackUri;
    private String contentType;
    private int mediaType;
    private Integer mediaType;
    private Long fileSize;
    private Date startTime;
    private Date endTime;

 server/visits/dmvisit_service/src/main/java/com/doumee/dao/admin/request/DeliverySnapshotManualRequest.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,21 @@
package com.doumee.dao.admin.request;

import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;

import java.math.BigDecimal;

@Data
@ApiModel("手动指定媒体快照")
public class DeliverySnapshotManualRequest {

    @ApiModelProperty(value = "媒体ID", required = true)
    private Integer mediaId;

    @ApiModelProperty(value = "1门头 2交付", required = true)
    private Integer snapshotType;

    @ApiModelProperty(value = "视频内秒数", required = true)
    private BigDecimal timestampSec;
}

 server/visits/dmvisit_service/src/main/java/com/doumee/dao/business/DeliveryMediaSnapshotFeedbackMapper.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,7 @@
package com.doumee.dao.business;

import com.doumee.dao.business.model.DeliveryMediaSnapshotFeedback;
import com.github.yulichang.base.MPJBaseMapper;

public interface DeliveryMediaSnapshotFeedbackMapper extends MPJBaseMapper<DeliveryMediaSnapshotFeedback> {
}

 server/visits/dmvisit_service/src/main/java/com/doumee/dao/business/DeliveryMediaSnapshotMapper.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,7 @@
package com.doumee.dao.business;

import com.doumee.dao.business.model.DeliveryMediaSnapshot;
import com.github.yulichang.base.MPJBaseMapper;

public interface DeliveryMediaSnapshotMapper extends MPJBaseMapper<DeliveryMediaSnapshot> {
}

 server/visits/dmvisit_service/src/main/java/com/doumee/dao/business/model/CollectionMedia.java

@@ -52,6 +52,14 @@
    private Integer downloadStatus;

    private Date downloadTime;

    @ApiModelProperty(value = "0未分析 1分析中 2完成 3失败")
    private Integer snapshotStatus;

    private Date snapshotTime;

    private String snapshotMessage;

    private Date createDate;
    private Integer isdeleted;
}

 server/visits/dmvisit_service/src/main/java/com/doumee/dao/business/model/DeliveryMediaSnapshot.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,46 @@
package com.doumee.dao.business.model;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;

import java.math.BigDecimal;
import java.util.Date;

@Data
@ApiModel("配送媒体快照")
@TableName("`delivery_media_snapshot`")
public class DeliveryMediaSnapshot {

    @TableId(type = IdType.AUTO)
    private Integer id;

    private Integer mediaId;

    private String transportCode;

    @ApiModelProperty(value = "1门头 2交付")
    private Integer snapshotType;

    private BigDecimal timestampSec;

    private String filePath;

    private BigDecimal confidence;

    private String source;

    private String modelVersion;

    private Date createDate;

    private Integer isdeleted;

    @TableField(exist = false)
    @ApiModelProperty(value = "FTP完整访问地址")
    private String fileUrlFull;
}

 server/visits/dmvisit_service/src/main/java/com/doumee/dao/business/model/DeliveryMediaSnapshotFeedback.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,31 @@
package com.doumee.dao.business.model;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

import java.math.BigDecimal;
import java.util.Date;

@Data
@TableName("`delivery_media_snapshot_feedback`")
public class DeliveryMediaSnapshotFeedback {

    @TableId(type = IdType.AUTO)
    private Integer id;

    private Integer mediaId;

    private Integer snapshotType;

    private BigDecimal aiTimeSec;

    private BigDecimal manualTimeSec;

    private String modelVersion;

    private Date createDate;

    private Integer isdeleted;
}

 server/visits/dmvisit_service/src/main/java/com/doumee/service/business/DeliverySnapshotService.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,17 @@
package com.doumee.service.business;

import com.doumee.dao.admin.request.DeliverySnapshotManualRequest;
import com.doumee.dao.business.model.DeliveryMediaSnapshot;

import java.util.List;

public interface DeliverySnapshotService {

    String submitAnalyze(Integer mediaId);

    List<DeliveryMediaSnapshot> listByMediaId(Integer mediaId);

    String saveManual(DeliverySnapshotManualRequest request);

    void submitAnalyzeAsync(Integer mediaId);
}

 server/visits/dmvisit_service/src/main/java/com/doumee/service/business/collection/CollectionMediaConstants.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,29 @@
package com.doumee.service.business.collection;

/**
 * é‡‡é›†ç«™åª’体/快照业务常量（定义在 dmvisit_service，避免依赖未安装的 system_service æ–°ç‰ˆæœ¬ï¼‰ã€‚
 */
public final class CollectionMediaConstants {

    /** åª’体下载状态：下载中 */
    public static final int DOWNLOAD_STATUS_DOWNLOADING = 3;

    /** åª’体快照状态：未分析 */
    public static final int SNAPSHOT_STATUS_NONE = 0;
    /** åª’体快照状态：分析中 */
    public static final int SNAPSHOT_STATUS_PROCESSING = 1;
    /** åª’体快照状态：完成 */
    public static final int SNAPSHOT_STATUS_DONE = 2;
    /** åª’体快照状态：失败 */
    public static final int SNAPSHOT_STATUS_FAILED = 3;

    /** å¿«ç…§ç±»åž‹ï¼šé—¨å¤´ */
    public static final int SNAPSHOT_TYPE_STOREFRONT = 1;
    /** å¿«ç…§ç±»åž‹ï¼šäº¤ä»˜ */
    public static final int SNAPSHOT_TYPE_HANDOVER = 2;

    public static final String COLLECTION_SNAPSHOT_FOLDER = "COLLECTION_SNAPSHOT";

    private CollectionMediaConstants() {
    }
}

 server/visits/dmvisit_service/src/main/java/com/doumee/service/business/collection/MediaFrameUtil.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,142 @@
package com.doumee.service.business.collection;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.io.BufferedReader;
import java.io.File;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.TimeUnit;

/**
 * ä»Žè§†é¢‘指定时刻截取 JPEG å¸§ã€‚
 */
@Slf4j
public final class MediaFrameUtil {

    private static final long SNAPSHOT_TIMEOUT_MINUTES = 10;

    private MediaFrameUtil() {
    }

    public static double probeDurationSec(String ffmpegPath, File source) {
        if (source == null || !source.exists() || source.length() <= 0) {
            return 0;
        }
        String ffprobe = resolveFfprobe(ffmpegPath);
        List<String> command = Arrays.asList(
                ffprobe,
                "-v", "error",
                "-show_entries", "format=duration",
                "-of", "default=noprint_wrappers=1:nokey=1",
                source.getAbsolutePath()
        );
        try {
            ProcessBuilder builder = new ProcessBuilder(command);
            builder.redirectErrorStream(true);
            Process process = builder.start();
            String output = readStream(process.getInputStream());
            boolean finished = process.waitFor(2, TimeUnit.MINUTES);
            if (!finished) {
                process.destroyForcibly();
                return 0;
            }
            if (process.exitValue() != 0 || StringUtils.isBlank(output)) {
                return 0;
            }
            return Double.parseDouble(output.trim());
        } catch (Exception e) {
            log.warn("ffprobe è¯»å–时长失败: {}", e.getMessage());
            return 0;
        }
    }

    public static boolean extractFrame(String ffmpegPath, File source, File target, double second) {
        if (source == null || !source.exists() || source.length() <= 0) {
            return false;
        }
        if (second < 0) {
            second = 0;
        }
        String ffmpeg = resolveExecutable(ffmpegPath, "ffmpeg");
        String sec = formatSeconds(second);
        List<String> command = Arrays.asList(
                ffmpeg,
                "-y",
                "-ss", sec,
                "-i", source.getAbsolutePath(),
                "-frames:v", "1",
                "-q:v", "2",
                target.getAbsolutePath()
        );
        try {
            ProcessBuilder builder = new ProcessBuilder(command);
            builder.redirectErrorStream(true);
            Process process = builder.start();
            readStream(process.getInputStream());
            boolean finished = process.waitFor(SNAPSHOT_TIMEOUT_MINUTES, TimeUnit.MINUTES);
            if (!finished) {
                process.destroyForcibly();
                log.error("FFmpeg æˆªå¸§è¶…æ—¶ source={} sec={}", source.getAbsolutePath(), sec);
                return false;
            }
            if (process.exitValue() != 0) {
                log.error("FFmpeg æˆªå¸§å¤±è´¥ exitCode={} source={} sec={}", process.exitValue(), source.getAbsolutePath(), sec);
                return false;
            }
            return target.exists() && target.length() > 0;
        } catch (Exception e) {
            log.error("FFmpeg æˆªå¸§å¼‚常 source={} sec={}: {}", source.getAbsolutePath(), sec, e.getMessage());
            return false;
        }
    }

    private static String formatSeconds(double second) {
        if (second <= 0) {
            return "0";
        }
        return String.format(Locale.US, "%.3f", second);
    }

    private static String readStream(java.io.InputStream in) throws Exception {
        StringBuilder sb = new StringBuilder();
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
            String line;
            while ((line = reader.readLine()) != null) {
                sb.append(line).append('\n');
            }
        }
        return sb.toString();
    }

    private static String resolveExecutable(String configuredPath, String defaultName) {
        if (StringUtils.isNotBlank(configuredPath)) {
            return configuredPath.trim();
        }
        return defaultName;
    }

    private static String resolveFfprobe(String ffmpegPath) {
        if (StringUtils.isBlank(ffmpegPath)) {
            return "ffprobe";
        }
        String path = ffmpegPath.trim();
        if (path.contains("/") || path.contains("\\")) {
            int slash = Math.max(path.lastIndexOf('/'), path.lastIndexOf('\\'));
            if (slash >= 0) {
                String dir = path.substring(0, slash + 1);
                String name = path.substring(slash + 1);
                if (name.toLowerCase().endsWith(".exe")) {
                    return dir + name.replaceAll("(?i)ffmpeg\\.exe$", "ffprobe.exe");
                }
                return dir + name.replaceAll("(?i)ffmpeg$", "ffprobe");
            }
            return path.replaceAll("(?i)ffmpeg", "ffprobe");
        }
        return "ffprobe";
    }
}

 server/visits/dmvisit_service/src/main/java/com/doumee/service/business/impl/collection/CollectionMediaSyncServiceImpl.java

@@ -19,12 +19,13 @@
import com.doumee.dao.business.model.CollectionMedia;
import com.doumee.dao.business.model.CollectionStation;
import com.doumee.service.business.CollectionMediaSyncService;
import com.doumee.service.business.DeliverySnapshotService;
import com.doumee.service.business.collection.CollectionMediaConstants;
import com.doumee.service.business.third.model.PageData;
import com.doumee.service.business.third.model.PageWrap;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
@@ -54,6 +55,8 @@
    private CollectionMediaMapper collectionMediaMapper;
    @Autowired
    private CollectionStationMapper collectionStationMapper;
    @Autowired
    private DeliverySnapshotService deliverySnapshotService;
    @Autowired
    private SystemDictDataBiz systemDictDataBiz;
    @Resource(name = "asyncExecutor")
@@ -105,13 +108,19 @@
        if (station == null || Constants.equalsInteger(station.getIsdeleted(), Constants.ONE)) {
            throw new BusinessException(ResponseStatus.DATA_EMPTY);
        }
        List<MediaItemDTO> items = isapiClient.searchMediaAll(station, startTime, endTime, trackId, IsapiConstants.MAX_PAGE_RESULTS);
        List<MediaItemDTO> items = isapiClient.searchMediaAll(station, startTime, endTime,
                resolveSyncTrackId(trackId), IsapiConstants.MAX_PAGE_RESULTS);
        log.info("采集站媒体检索 stationId={} ip={} range={}~{} track={} found={}",
                stationId, station.getIp(), startTime, endTime, trackId, items.size());
                stationId, station.getIp(), startTime, endTime, resolveSyncTrackId(trackId), items.size());
        int count = 0;
        int skipped = 0;
        Date now = new Date();
        for (MediaItemDTO item : items) {
            if (StringUtils.isBlank(item.getFileIndex())) {
                continue;
            }
            if (!isSyncableMp4Video(item)) {
                skipped++;
                continue;
            }
            Long exists = collectionMediaMapper.selectCount(new QueryWrapper<CollectionMedia>().lambda()
@@ -140,7 +149,54 @@
            collectionMediaMapper.insert(media);
            count++;
        }
        if (skipped > 0) {
            log.info("采集站媒体索引跳过非MP4视频 stationId={} skipped={}", stationId, skipped);
        }
        return count;
    }

    /** åŒæ­¥ç´¢å¼•ä»…检索主码流录像 track（默认 101），不包含抓图 track 103 */
    private String resolveSyncTrackId(String trackId) {
        if (StringUtils.isNotBlank(trackId)) {
            String val = trackId.trim();
            if (!"auto".equalsIgnoreCase(val) && !"*".equals(val) && !"0".equals(val)) {
                return val;
            }
        }
        return IsapiConstants.DEFAULT_TRACK_ID;
    }

    /** åŒæ­¥å…¥åº“：仅 MP4 è§†é¢‘（扩展名 .mp4，且非图片/音频类型） */
    private static boolean isSyncableMp4Video(MediaItemDTO item) {
        if (item.getMediaType() != null && item.getMediaType() != 0) {
            return false;
        }
        String trackId = StringUtils.defaultString(item.getTrackId());
        if (trackId.endsWith("3")) {
            return false;
        }
        return isMp4FileName(resolveItemFileName(item));
    }

    private static String resolveItemFileName(MediaItemDTO item) {
        if (StringUtils.isNotBlank(item.getFileName())) {
            return item.getFileName();
        }
        String uri = item.getPlaybackUri();
        if (StringUtils.isBlank(uri)) {
            return null;
        }
        java.util.regex.Matcher m = java.util.regex.Pattern
                .compile("[?&](?:name|filename|fileName)=([^&\\s]+)", java.util.regex.Pattern.CASE_INSENSITIVE)
                .matcher(uri.replace("&amp;", "&"));
        if (m.find()) {
            try {
                return java.net.URLDecoder.decode(m.group(1), StandardCharsets.UTF_8.name());
            } catch (Exception e) {
                return m.group(1);
            }
        }
        return null;
    }

    @Override
@@ -152,7 +208,7 @@
        if (Constants.equalsInteger(media.getDownloadStatus(), Constants.ONE)) {
            throw new BusinessException(ResponseStatus.BAD_REQUEST.getCode(), "文件已下载，无需重复下载");
        }
        if (Constants.equalsInteger(media.getDownloadStatus(), Constants.COLLECTION_MEDIA_DOWNLOADING)) {
        if (Constants.equalsInteger(media.getDownloadStatus(), CollectionMediaConstants.DOWNLOAD_STATUS_DOWNLOADING)) {
            throw new BusinessException(ResponseStatus.BAD_REQUEST.getCode(), "文件正在下载中，请稍后刷新");
        }
        CollectionStation station = collectionStationMapper.selectById(media.getStationId());
@@ -160,7 +216,7 @@
            throw new BusinessException(ResponseStatus.DATA_EMPTY);
        }
        CollectionMedia downloading = new CollectionMedia();
        downloading.setDownloadStatus(Constants.COLLECTION_MEDIA_DOWNLOADING);
        downloading.setDownloadStatus(CollectionMediaConstants.DOWNLOAD_STATUS_DOWNLOADING);
        int updated = collectionMediaMapper.update(downloading, new QueryWrapper<CollectionMedia>().lambda()
                .eq(CollectionMedia::getId, mediaId)
                .in(CollectionMedia::getDownloadStatus, Constants.ZERO, 2));
@@ -195,6 +251,7 @@
            update.setDownloadTime(new Date());
            collectionMediaMapper.updateById(update);
            log.info("异步下载成功 mediaId={} path={}", mediaId, path);
            deliverySnapshotService.submitAnalyzeAsync(mediaId);
        } catch (Exception e) {
            markDownloadFailed(mediaId);
            log.error("异步下载异常 mediaId={}: {}", mediaId, e.getMessage(), e);

 server/visits/dmvisit_service/src/main/java/com/doumee/service/business/impl/collection/DeliverySnapshotServiceImpl.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,396 @@
package com.doumee.service.business.impl.collection;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.doumee.biz.system.SystemDictDataBiz;
import com.doumee.core.constants.ResponseStatus;
import com.doumee.core.exception.BusinessException;
import com.doumee.core.utils.Constants;
import com.doumee.core.utils.DateUtil;
import com.doumee.core.utils.FtpUtil;
import com.doumee.service.business.collection.CollectionMediaConstants;
import com.doumee.service.business.collection.MediaFrameUtil;
import com.doumee.dao.admin.request.DeliverySnapshotManualRequest;
import com.doumee.dao.business.CollectionMediaMapper;
import com.doumee.dao.business.DeliveryMediaSnapshotFeedbackMapper;
import com.doumee.dao.business.DeliveryMediaSnapshotMapper;
import com.doumee.dao.business.model.CollectionMedia;
import com.doumee.dao.business.model.DeliveryMediaSnapshot;
import com.doumee.dao.business.model.DeliveryMediaSnapshotFeedback;
import com.doumee.service.business.DeliverySnapshotService;
import com.doumee.service.business.snapshot.SnapshotAnalyzeRequest;
import com.doumee.service.business.snapshot.SnapshotAnalyzeResponse;
import com.doumee.service.business.snapshot.SnapshotInferClient;
import com.doumee.service.business.snapshot.SnapshotInferProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executor;

@Slf4j
@Service
public class DeliverySnapshotServiceImpl implements DeliverySnapshotService {

    @Autowired
    private CollectionMediaMapper collectionMediaMapper;
    @Autowired
    private DeliveryMediaSnapshotMapper deliveryMediaSnapshotMapper;
    @Autowired
    private DeliveryMediaSnapshotFeedbackMapper deliveryMediaSnapshotFeedbackMapper;
    @Autowired
    private SystemDictDataBiz systemDictDataBiz;
    @Autowired
    private SnapshotInferClient snapshotInferClient;
    @Autowired
    private SnapshotInferProperties snapshotInferProperties;
    @Resource(name = "asyncExecutor")
    private Executor asyncExecutor;

    @Override
    public String submitAnalyze(Integer mediaId) {
        CollectionMedia media = requireDownloadedMedia(mediaId);
        if (Constants.equalsInteger(media.getSnapshotStatus(), CollectionMediaConstants.SNAPSHOT_STATUS_PROCESSING)) {
            throw new BusinessException(ResponseStatus.BAD_REQUEST.getCode(), "快照分析进行中，请稍后刷新");
        }
        markSnapshotProcessing(mediaId);
        asyncExecutor.execute(() -> executeAnalyze(mediaId));
        return "已提交快照分析任务，请稍后刷新查看";
    }

    @Override
    public void submitAnalyzeAsync(Integer mediaId) {
        if (!snapshotInferProperties.isAutoOnDownload()) {
            return;
        }
        try {
            CollectionMedia media = collectionMediaMapper.selectById(mediaId);
            if (media == null || Constants.equalsInteger(media.getIsdeleted(), Constants.ONE)) {
                return;
            }
            if (!Constants.equalsInteger(media.getDownloadStatus(), Constants.ONE)
                    || StringUtils.isBlank(media.getFilePathLocal())) {
                return;
            }
            if (Constants.equalsInteger(media.getSnapshotStatus(), CollectionMediaConstants.SNAPSHOT_STATUS_PROCESSING)
                    || Constants.equalsInteger(media.getSnapshotStatus(), CollectionMediaConstants.SNAPSHOT_STATUS_DONE)) {
                return;
            }
            if (media.getMediaType() != null && media.getMediaType() != 0) {
                return;
            }
            markSnapshotProcessing(mediaId);
            asyncExecutor.execute(() -> executeAnalyze(mediaId));
        } catch (Exception e) {
            log.warn("自动提交快照分析失败 mediaId={}: {}", mediaId, e.getMessage());
        }
    }

    @Override
    public List<DeliveryMediaSnapshot> listByMediaId(Integer mediaId) {
        List<DeliveryMediaSnapshot> list = deliveryMediaSnapshotMapper.selectList(new QueryWrapper<DeliveryMediaSnapshot>().lambda()
                .eq(DeliveryMediaSnapshot::getMediaId, mediaId)
                .eq(DeliveryMediaSnapshot::getIsdeleted, Constants.ZERO)
                .orderByAsc(DeliveryMediaSnapshot::getSnapshotType));
        list.forEach(this::fillSnapshotUrl);
        return list;
    }

    @Override
    public String saveManual(DeliverySnapshotManualRequest request) {
        if (request == null || request.getMediaId() == null || request.getSnapshotType() == null
                || request.getTimestampSec() == null) {
            throw new BusinessException(ResponseStatus.BAD_REQUEST.getCode(), "参数不完整");
        }
        if (!Constants.equalsInteger(request.getSnapshotType(), CollectionMediaConstants.SNAPSHOT_TYPE_STOREFRONT)
                && !Constants.equalsInteger(request.getSnapshotType(), CollectionMediaConstants.SNAPSHOT_TYPE_HANDOVER)) {
            throw new BusinessException(ResponseStatus.BAD_REQUEST.getCode(), "快照类型无效");
        }
        CollectionMedia media = requireDownloadedMedia(request.getMediaId());
        File videoFile = null;
        File frameFile = null;
        try {
            videoFile = downloadMediaToTemp(media);
            frameFile = File.createTempFile("hk_snapshot_", ".jpg");
            if (!MediaFrameUtil.extractFrame(getFfmpegPath(), videoFile, frameFile, request.getTimestampSec().doubleValue())) {
                throw new BusinessException(ResponseStatus.SERVER_ERROR.getCode(), "截帧失败");
            }
            String relativePath = uploadSnapshot(frameFile, media.getId(), request.getSnapshotType());
            saveFeedback(media.getId(), request.getSnapshotType(), request.getTimestampSec());
            upsertSnapshot(media.getId(), request.getSnapshotType(), request.getTimestampSec(),
                    relativePath, null, "manual", null);
            CollectionMedia update = new CollectionMedia();
            update.setId(media.getId());
            update.setSnapshotStatus(CollectionMediaConstants.SNAPSHOT_STATUS_DONE);
            update.setSnapshotTime(new Date());
            update.setSnapshotMessage(null);
            collectionMediaMapper.updateById(update);
            return "保存成功";
        } catch (BusinessException e) {
            throw e;
        } catch (Exception e) {
            log.error("手动保存快照失败 mediaId={}: {}", request.getMediaId(), e.getMessage(), e);
            throw new BusinessException(ResponseStatus.SERVER_ERROR.getCode(), "保存快照失败");
        } finally {
            deleteQuietly(videoFile);
            deleteQuietly(frameFile);
        }
    }

    private void executeAnalyze(Integer mediaId) {
        CollectionMedia media = collectionMediaMapper.selectById(mediaId);
        if (media == null || Constants.equalsInteger(media.getIsdeleted(), Constants.ONE)) {
            return;
        }
        File videoFile = null;
        File storefrontFrame = null;
        File handoverFrame = null;
        try {
            videoFile = downloadMediaToTemp(media);
            double duration = MediaFrameUtil.probeDurationSec(getFfmpegPath(), videoFile);
            if (duration <= 0) {
                duration = estimateDuration(media);
            }
            SnapshotAnalyzeRequest analyzeRequest = new SnapshotAnalyzeRequest();
            analyzeRequest.setMediaId(mediaId);
            analyzeRequest.setVideoUrl(buildVideoUrl(media));
            analyzeRequest.setSampleFps(snapshotInferProperties.getSampleFps());
            analyzeRequest.setEnableAsr(snapshotInferProperties.isEnableAsr());
            analyzeRequest.setDurationSec(duration);
            SnapshotAnalyzeResponse analyzeResponse = snapshotInferClient.analyze(analyzeRequest);
            if (Boolean.FALSE.equals(analyzeResponse.getSuccess())) {
                throw new BusinessException(ResponseStatus.SERVER_ERROR.getCode(),
                        StringUtils.defaultIfBlank(analyzeResponse.getMessage(), "快照推理失败"));
            }

            double storefrontSec = analyzeResponse.getStorefront().getTimeSec();
            double handoverSec = analyzeResponse.getHandover().getTimeSec();
            if (handoverSec <= storefrontSec) {
                handoverSec = Math.min(duration > 0 ? duration - 1 : storefrontSec + 60, storefrontSec + 60);
            }

            storefrontFrame = File.createTempFile("hk_storefront_", ".jpg");
            handoverFrame = File.createTempFile("hk_handover_", ".jpg");
            if (!MediaFrameUtil.extractFrame(getFfmpegPath(), videoFile, storefrontFrame, storefrontSec)) {
                throw new BusinessException(ResponseStatus.SERVER_ERROR.getCode(), "门头图截帧失败");
            }
            if (!MediaFrameUtil.extractFrame(getFfmpegPath(), videoFile, handoverFrame, handoverSec)) {
                throw new BusinessException(ResponseStatus.SERVER_ERROR.getCode(), "交付图截帧失败");
            }

            String storefrontPath = uploadSnapshot(storefrontFrame, mediaId, CollectionMediaConstants.SNAPSHOT_TYPE_STOREFRONT);
            String handoverPath = uploadSnapshot(handoverFrame, mediaId, CollectionMediaConstants.SNAPSHOT_TYPE_HANDOVER);

            upsertSnapshot(mediaId, CollectionMediaConstants.SNAPSHOT_TYPE_STOREFRONT,
                    BigDecimal.valueOf(storefrontSec).setScale(2, RoundingMode.HALF_UP),
                    storefrontPath,
                    analyzeResponse.getStorefront().getConfidence(),
                    StringUtils.defaultIfBlank(analyzeResponse.getStorefront().getSource(), "ai"),
                    analyzeResponse.getModelVersion());
            upsertSnapshot(mediaId, CollectionMediaConstants.SNAPSHOT_TYPE_HANDOVER,
                    BigDecimal.valueOf(handoverSec).setScale(2, RoundingMode.HALF_UP),
                    handoverPath,
                    analyzeResponse.getHandover().getConfidence(),
                    StringUtils.defaultIfBlank(analyzeResponse.getHandover().getSource(), "ai"),
                    analyzeResponse.getModelVersion());

            CollectionMedia done = new CollectionMedia();
            done.setId(mediaId);
            done.setSnapshotStatus(CollectionMediaConstants.SNAPSHOT_STATUS_DONE);
            done.setSnapshotTime(new Date());
            done.setSnapshotMessage(null);
            collectionMediaMapper.updateById(done);
            log.info("快照分析完成 mediaId={} storefrontSec={} handoverSec={}", mediaId, storefrontSec, handoverSec);
        } catch (Exception e) {
            log.error("快照分析失败 mediaId={}: {}", mediaId, e.getMessage(), e);
            CollectionMedia fail = new CollectionMedia();
            fail.setId(mediaId);
            fail.setSnapshotStatus(CollectionMediaConstants.SNAPSHOT_STATUS_FAILED);
            fail.setSnapshotMessage(StringUtils.left(e.getMessage(), 500));
            collectionMediaMapper.updateById(fail);
        } finally {
            deleteQuietly(videoFile);
            deleteQuietly(storefrontFrame);
            deleteQuietly(handoverFrame);
        }
    }

    private void markSnapshotProcessing(Integer mediaId) {
        CollectionMedia processing = new CollectionMedia();
        processing.setId(mediaId);
        processing.setSnapshotStatus(CollectionMediaConstants.SNAPSHOT_STATUS_PROCESSING);
        processing.setSnapshotMessage(null);
        collectionMediaMapper.updateById(processing);
    }

    private CollectionMedia requireDownloadedMedia(Integer mediaId) {
        CollectionMedia media = collectionMediaMapper.selectById(mediaId);
        if (media == null || Constants.equalsInteger(media.getIsdeleted(), Constants.ONE)) {
            throw new BusinessException(ResponseStatus.DATA_EMPTY);
        }
        if (!Constants.equalsInteger(media.getDownloadStatus(), Constants.ONE)
                || StringUtils.isBlank(media.getFilePathLocal())) {
            throw new BusinessException(ResponseStatus.BAD_REQUEST.getCode(), "请先下载媒体文件");
        }
        return media;
    }

    private File downloadMediaToTemp(CollectionMedia media) throws IOException {
        FtpUtil ftp = createFtpClient();
        File temp = File.createTempFile("hk_media_snap_", resolveSuffix(media));
        String remote = getMediaFolder() + media.getFilePathLocal();
        String local = ftp.download(remote, temp.getAbsolutePath());
        if (StringUtils.isBlank(local)) {
            ftp.disconnect();
            throw new BusinessException(ResponseStatus.SERVER_ERROR.getCode(), "下载媒体文件失败");
        }
        ftp.disconnect();
        return temp;
    }

    private String uploadSnapshot(File frameFile, Integer mediaId, int snapshotType) throws IOException {
        FtpUtil ftp = createFtpClient();
        try {
            String folder = getSnapshotFolder();
            String suffix = snapshotType == CollectionMediaConstants.SNAPSHOT_TYPE_STOREFRONT ? "_storefront.jpg" : "_handover.jpg";
            String relative = DateUtil.getNowShortDate() + "/" + mediaId + suffix;
            String remote = folder + relative;
            try (FileInputStream in = new FileInputStream(frameFile)) {
                if (!ftp.uploadInputstream(in, remote)) {
                    throw new BusinessException(ResponseStatus.SERVER_ERROR.getCode(), "上传快照失败");
                }
            }
            return relative;
        } finally {
            ftp.disconnect();
        }
    }

    private void upsertSnapshot(Integer mediaId, int snapshotType, BigDecimal timestampSec, String filePath,
                                Double confidence, String source, String modelVersion) {
        deliveryMediaSnapshotMapper.update(null, new UpdateWrapper<DeliveryMediaSnapshot>().lambda()
                .eq(DeliveryMediaSnapshot::getMediaId, mediaId)
                .eq(DeliveryMediaSnapshot::getSnapshotType, snapshotType)
                .set(DeliveryMediaSnapshot::getIsdeleted, Constants.ONE));
        DeliveryMediaSnapshot row = new DeliveryMediaSnapshot();
        row.setMediaId(mediaId);
        row.setSnapshotType(snapshotType);
        row.setTimestampSec(timestampSec);
        row.setFilePath(filePath);
        if (confidence != null) {
            row.setConfidence(BigDecimal.valueOf(confidence).setScale(4, RoundingMode.HALF_UP));
        }
        row.setSource(source);
        row.setModelVersion(modelVersion);
        row.setCreateDate(new Date());
        row.setIsdeleted(Constants.ZERO);
        deliveryMediaSnapshotMapper.insert(row);
    }

    private void fillSnapshotUrl(DeliveryMediaSnapshot snapshot) {
        if (StringUtils.isBlank(snapshot.getFilePath())) {
            return;
        }
        try {
            String prefix = systemDictDataBiz.queryByCode(Constants.FTP, Constants.FTP_RESOURCE_PATH).getCode();
            snapshot.setFileUrlFull(prefix + getSnapshotFolder() + snapshot.getFilePath());
        } catch (Exception e) {
            log.warn("构建快照URL失败 id={}: {}", snapshot.getId(), e.getMessage());
        }
    }

    private String buildVideoUrl(CollectionMedia media) {
        try {
            String prefix = systemDictDataBiz.queryByCode(Constants.FTP, Constants.FTP_RESOURCE_PATH).getCode();
            return prefix + getMediaFolder() + media.getFilePathLocal();
        } catch (Exception e) {
            return null;
        }
    }

    private double estimateDuration(CollectionMedia media) {
        if (media.getStartTime() != null && media.getEndTime() != null) {
            long ms = media.getEndTime().getTime() - media.getStartTime().getTime();
            if (ms > 0) {
                return ms / 1000.0;
            }
        }
        return 1200.0;
    }

    private FtpUtil createFtpClient() throws IOException {
        return new FtpUtil(
                systemDictDataBiz.queryByCode(Constants.FTP, Constants.FTP_HOST).getCode(),
                Integer.parseInt(systemDictDataBiz.queryByCode(Constants.FTP, Constants.FTP_PORT).getCode()),
                systemDictDataBiz.queryByCode(Constants.FTP, Constants.FTP_USERNAME).getCode(),
                systemDictDataBiz.queryByCode(Constants.FTP, Constants.FTP_PWD).getCode());
    }

    private String getMediaFolder() {
        try {
            return systemDictDataBiz.queryByCode(Constants.FTP, Constants.COLLECTION_MEDIA_FOLDER).getCode();
        } catch (Exception e) {
            return "/collection_media/";
        }
    }

    private String getSnapshotFolder() {
        try {
            return systemDictDataBiz.queryByCode(Constants.FTP, CollectionMediaConstants.COLLECTION_SNAPSHOT_FOLDER).getCode();
        } catch (Exception e) {
            return "/collection_snapshot/";
        }
    }

    private String getFfmpegPath() {
        try {
            return systemDictDataBiz.queryByCode(Constants.CS_PARAM, Constants.CS_FFMPEG_PATH).getCode();
        } catch (Exception e) {
            return "ffmpeg";
        }
    }

    private String resolveSuffix(CollectionMedia media) {
        if (StringUtils.isNotBlank(media.getFileName()) && media.getFileName().contains(".")) {
            return media.getFileName().substring(media.getFileName().lastIndexOf('.')).toLowerCase();
        }
        return ".mp4";
    }

    private void saveFeedback(Integer mediaId, int snapshotType, BigDecimal manualTimeSec) {
        DeliveryMediaSnapshot existing = deliveryMediaSnapshotMapper.selectOne(new QueryWrapper<DeliveryMediaSnapshot>().lambda()
                .eq(DeliveryMediaSnapshot::getMediaId, mediaId)
                .eq(DeliveryMediaSnapshot::getSnapshotType, snapshotType)
                .eq(DeliveryMediaSnapshot::getIsdeleted, Constants.ZERO)
                .orderByDesc(DeliveryMediaSnapshot::getId)
                .last("LIMIT 1"));
        DeliveryMediaSnapshotFeedback feedback = new DeliveryMediaSnapshotFeedback();
        feedback.setMediaId(mediaId);
        feedback.setSnapshotType(snapshotType);
        if (existing != null && existing.getTimestampSec() != null) {
            feedback.setAiTimeSec(existing.getTimestampSec());
            feedback.setModelVersion(existing.getModelVersion());
        }
        feedback.setManualTimeSec(manualTimeSec);
        feedback.setCreateDate(new Date());
        feedback.setIsdeleted(Constants.ZERO);
        deliveryMediaSnapshotFeedbackMapper.insert(feedback);
    }

    private void deleteQuietly(File file) {
        if (file != null && file.exists() && !file.delete()) {
            log.warn("临时文件删除失败: {}", file.getAbsolutePath());
        }
    }

}

 server/visits/dmvisit_service/src/main/java/com/doumee/service/business/snapshot/SnapshotAnalyzeRequest.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,23 @@
package com.doumee.service.business.snapshot;

import lombok.Data;

import java.util.Arrays;
import java.util.List;

@Data
public class SnapshotAnalyzeRequest {

    private Integer mediaId;
    private String videoUrl;
    private Double sampleFps;
    private Boolean enableAsr;
    private Double durationSec;
    private KeywordConfig keywords = new KeywordConfig();

    @Data
    public static class KeywordConfig {
        private List<String> storefront = Arrays.asList("到店", "到达", "门头");
        private List<String> handover = Arrays.asList("交付", "交货", "签收");
    }
}

 server/visits/dmvisit_service/src/main/java/com/doumee/service/business/snapshot/SnapshotAnalyzeResponse.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,62 @@
package com.doumee.service.business.snapshot;



import lombok.Data;



import java.util.ArrayList;

import java.util.List;



@Data

public class SnapshotAnalyzeResponse {



    private Boolean success;

    private String modelVersion;

    private Double durationSec;

    private SnapshotHit storefront;

    private SnapshotHit handover;

    private String message;

    private List<AsrHit> asrHits = new ArrayList<>();



    @Data

    public static class SnapshotHit {

        private Double timeSec;

        private Double confidence;

        private String source;

    }



    @Data

    public static class AsrHit {

        private String keyword;

        private Double timeSec;

    }

}


 server/visits/dmvisit_service/src/main/java/com/doumee/service/business/snapshot/SnapshotInferClient.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,122 @@
package com.doumee.service.business.snapshot;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class SnapshotInferClient {

    @Autowired
    private SnapshotInferProperties properties;

    public boolean healthCheck() {
        if (StringUtils.isBlank(properties.getBaseUrl())) {
            return false;
        }
        String url = normalizeBaseUrl() + "/health";
        try {
            HttpResponse response = HttpRequest.get(url)
                    .timeout(properties.getConnectTimeoutMs())
                    .execute();
            if (response.getStatus() != 200) {
                return false;
            }
            JSONObject json = JSON.parseObject(response.body());
            return json != null && "ok".equalsIgnoreCase(json.getString("status"));
        } catch (Exception e) {
            log.warn("snapshot-infer health å¤±è´¥: {}", e.getMessage());
            return false;
        }
    }

    public SnapshotAnalyzeResponse analyze(SnapshotAnalyzeRequest request) {
        if (StringUtils.isBlank(properties.getBaseUrl())) {
            return resolveFailure(request, "snapshot.infer.base-url æœªé…ç½®");
        }
        String url = normalizeBaseUrl() + "/analyze";
        String payload = JSON.toJSONString(request);
        try {
            HttpResponse response = HttpRequest.post(url)
                    .body(payload)
                    .contentType("application/json")
                    .setConnectionTimeout(properties.getConnectTimeoutMs())
                    .timeout(properties.getReadTimeoutMs())
                    .execute();
            String body = response.body();
            if (response.getStatus() != 200) {
                log.warn("snapshot-infer analyze HTTP {} body={}", response.getStatus(), body);
                return resolveFailure(request, "推理服务 HTTP " + response.getStatus());
            }
            SnapshotAnalyzeResponse parsed = JSON.parseObject(body, SnapshotAnalyzeResponse.class);
            if (parsed == null) {
                return resolveFailure(request, "推理响应为空");
            }
            if (Boolean.FALSE.equals(parsed.getSuccess())) {
                String msg = StringUtils.defaultIfBlank(parsed.getMessage(), "推理未检测到有效时刻");
                log.warn("snapshot-infer analyze å¤±è´¥: {}", msg);
                return resolveFailure(request, msg);
            }
            if (parsed.getStorefront() == null || parsed.getHandover() == null) {
                return resolveFailure(request, "推理响应缺少门头或交付时刻");
            }
            if (parsed.getSuccess() == null) {
                parsed.setSuccess(true);
            }
            if (parsed.getAsrHits() == null) {
                parsed.setAsrHits(new java.util.ArrayList<>());
            }
            return parsed;
        } catch (Exception e) {
            log.warn("snapshot-infer analyze å¼‚常: {}", e.getMessage());
            return resolveFailure(request, e.getMessage());
        }
    }

    private SnapshotAnalyzeResponse resolveFailure(SnapshotAnalyzeRequest request, String message) {
        if (properties.isFailOpenMock()) {
            log.warn("fail-open-mock å¯ç”¨ï¼Œä½¿ç”¨æœ¬åœ° mock: {}", message);
            return buildLocalMock(request);
        }
        SnapshotAnalyzeResponse fail = new SnapshotAnalyzeResponse();
        fail.setSuccess(false);
        fail.setMessage(message);
        fail.setModelVersion("unavailable");
        return fail;
    }

    private SnapshotAnalyzeResponse buildLocalMock(SnapshotAnalyzeRequest request) {
        double duration = request.getDurationSec() != null && request.getDurationSec() > 0
                ? request.getDurationSec() : 1200.0;
        double t1 = Math.round(duration * 0.25 * 100.0) / 100.0;
        double t2 = Math.round(duration * 0.75 * 100.0) / 100.0;
        if (t2 <= t1) {
            t2 = t1 + 60.0;
        }
        SnapshotAnalyzeResponse response = new SnapshotAnalyzeResponse();
        response.setSuccess(true);
        response.setModelVersion("local-mock");
        response.setDurationSec(duration);
        SnapshotAnalyzeResponse.SnapshotHit storefront = new SnapshotAnalyzeResponse.SnapshotHit();
        storefront.setTimeSec(t1);
        storefront.setConfidence(0.5);
        storefront.setSource("local-mock");
        SnapshotAnalyzeResponse.SnapshotHit handover = new SnapshotAnalyzeResponse.SnapshotHit();
        handover.setTimeSec(t2);
        handover.setConfidence(0.5);
        handover.setSource("local-mock");
        response.setStorefront(storefront);
        response.setHandover(handover);
        return response;
    }

    private String normalizeBaseUrl() {
        return properties.getBaseUrl().trim().replaceAll("/+$", "");
    }
}

 server/visits/dmvisit_service/src/main/java/com/doumee/service/business/snapshot/SnapshotInferProperties.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,40 @@
package com.doumee.service.business.snapshot;



import lombok.Data;

import org.springframework.boot.context.properties.ConfigurationProperties;

import org.springframework.stereotype.Component;



@Data

@Component

@ConfigurationProperties(prefix = "snapshot.infer")

public class SnapshotInferProperties {



    private String baseUrl = "http://127.0.0.1:8095";

    private int connectTimeoutMs = 5000;

    private int readTimeoutMs = 600000;

    private boolean enableAsr = true;

    private boolean autoOnDownload = true;

    private double sampleFps = 0.5;

    /** æŽ¨ç†å¤±è´¥æ—¶æ˜¯å¦å›žé€€åˆ°æœ¬åœ° 25%/75% mock（生产环境应为 false） */

    private boolean failOpenMock = false;

}


 server/visits/dmvisit_service/src/main/resources/application-dev.yml

@@ -19,6 +19,16 @@

debug_model: true

snapshot:
  infer:
    base-url: http://127.0.0.1:8095
    connect-timeout-ms: 5000
    read-timeout-ms: 600000
    enable-asr: true
    auto-on-download: true
    sample-fps: 0.5
    fail-open-mock: true

########################同步数据模式  ########################
data-sync:
  org-user-data-origin: 0 #组织数据 0自建 2以海康为主 1华晟ERP系统

 server/visits/dmvisit_service/src/main/resources/application-pro.yml

@@ -17,6 +17,16 @@

debug_model: false

snapshot:
  infer:
    base-url: http://127.0.0.1:8095
    connect-timeout-ms: 5000
    read-timeout-ms: 600000
    enable-asr: true
    auto-on-download: true
    sample-fps: 0.5
    fail-open-mock: false

########################同步数据模式  ########################
data-sync:
  org-user-data-origin: 3 #组织数据 0自建 2以海康为主 1华晟ERP系统 3简道云 4钉钉