70 lines
1.9 KiB
Bash
70 lines
1.9 KiB
Bash
#!/usr/bin/env bash
|
||
set -u
|
||
|
||
INPUT_PATH="${1:-/root/gongxiangp2/output/history-sku-import-full.json}"
|
||
OPERATOR="${2:-admin}"
|
||
PAUSE_MS="${3:-1500}"
|
||
LOOP_PAUSE_SEC="${4:-5}"
|
||
WORKDIR="/root/gongxiangp2"
|
||
LOG_DIR="$WORKDIR/output"
|
||
LOG_PATH="$LOG_DIR/history-import-daemon.log"
|
||
CHECKPOINT_PATH="$LOG_DIR/$(basename "${INPUT_PATH%.*}").queue-progress.json"
|
||
|
||
mkdir -p "$LOG_DIR"
|
||
|
||
log() {
|
||
printf '[%s] %s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$1" | tee -a "$LOG_PATH"
|
||
}
|
||
|
||
pending_count() {
|
||
python3 - "$INPUT_PATH" "$CHECKPOINT_PATH" <<'PY'
|
||
import json, sys
|
||
input_path = sys.argv[1]
|
||
checkpoint_path = sys.argv[2]
|
||
with open(input_path, 'r', encoding='utf-8') as f:
|
||
records = json.load(f)
|
||
candidates = [r for r in records if r.get('status') == 'ready']
|
||
processed = set()
|
||
try:
|
||
with open(checkpoint_path, 'r', encoding='utf-8') as f:
|
||
checkpoint = json.load(f)
|
||
processed = {item.get('fingerprint', '') for item in checkpoint.get('results', []) if item.get('fingerprint')}
|
||
except FileNotFoundError:
|
||
pass
|
||
pending = [r for r in candidates if r.get('fingerprint') not in processed]
|
||
print(len(pending))
|
||
PY
|
||
}
|
||
|
||
log "历史导入守护进程启动"
|
||
log "input=$INPUT_PATH operator=$OPERATOR pause_ms=$PAUSE_MS loop_pause_sec=$LOOP_PAUSE_SEC checkpoint=$CHECKPOINT_PATH"
|
||
|
||
while true; do
|
||
PENDING="$(pending_count)"
|
||
if [ "$PENDING" = "0" ]; then
|
||
log "没有待处理记录，${LOOP_PAUSE_SEC}s 后继续巡检"
|
||
sleep "$LOOP_PAUSE_SEC"
|
||
continue
|
||
fi
|
||
|
||
log "剩余待处理: $PENDING，开始处理下一条"
|
||
(
|
||
cd "$WORKDIR" && \
|
||
npx tsx scripts/run-history-import-queue.ts \
|
||
--input="$INPUT_PATH" \
|
||
--operator="$OPERATOR" \
|
||
--apply \
|
||
--max-items=1 \
|
||
--pause-ms="$PAUSE_MS"
|
||
) >>"$LOG_PATH" 2>&1
|
||
EXIT_CODE=$?
|
||
|
||
if [ $EXIT_CODE -ne 0 ]; then
|
||
log "队列子进程异常退出，exit_code=$EXIT_CODE；${LOOP_PAUSE_SEC}s 后重试"
|
||
else
|
||
log "本轮处理完成，${LOOP_PAUSE_SEC}s 后继续"
|
||
fi
|
||
|
||
sleep "$LOOP_PAUSE_SEC"
|
||
done
|