167 lines
6.5 KiB
Python
167 lines
6.5 KiB
Python
# coding=utf-8
|
||
"""识别订单上报失败落盘队列：网络恢复后自动重发，避免订单丢失。"""
|
||
import json
|
||
import os
|
||
import threading
|
||
import time
|
||
from datetime import datetime
|
||
|
||
|
||
class ResultOutbox:
|
||
"""JSON 文件出站队列（按 request_id+kind 去重）。"""
|
||
|
||
def __init__(self, root_dir):
|
||
self.root_dir = os.path.abspath(root_dir or "data/result_outbox")
|
||
self.index_path = os.path.join(self.root_dir, "queue.json")
|
||
self.lock = threading.RLock()
|
||
os.makedirs(self.root_dir, exist_ok=True)
|
||
self._items = []
|
||
self._load()
|
||
|
||
def _load(self):
|
||
if not os.path.isfile(self.index_path):
|
||
self._items = []
|
||
return
|
||
try:
|
||
with open(self.index_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
self._items = list(data.get("items") or [])
|
||
except Exception as exc:
|
||
print(f"[outbox] load failed: {exc}")
|
||
self._items = []
|
||
|
||
def _save(self):
|
||
tmp = self.index_path + ".tmp"
|
||
try:
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(
|
||
{"updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "items": self._items},
|
||
f,
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
os.replace(tmp, self.index_path)
|
||
except Exception as exc:
|
||
print(f"[outbox] save failed: {exc}")
|
||
try:
|
||
if os.path.isfile(tmp):
|
||
os.remove(tmp)
|
||
except Exception:
|
||
pass
|
||
|
||
@staticmethod
|
||
def _key(kind, request_id):
|
||
return f"{kind}:{request_id or ''}"
|
||
|
||
def enqueue(self, kind, payload, error=None, next_delay=15.0):
|
||
"""写入/更新一条待重发记录。返回 True 表示新建。"""
|
||
request_id = (payload or {}).get("request_id") or ""
|
||
if not request_id:
|
||
request_id = f"REQ_LOCAL_{int(time.time() * 1000)}"
|
||
payload = dict(payload or {})
|
||
payload["request_id"] = request_id
|
||
kind = (kind or "recognition").strip() or "recognition"
|
||
now = time.time()
|
||
with self.lock:
|
||
key = self._key(kind, request_id)
|
||
for item in self._items:
|
||
if self._key(item.get("kind"), item.get("request_id")) == key:
|
||
item["payload"] = payload
|
||
item["last_error"] = str(error or "")[:500]
|
||
item["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
item["next_retry_at"] = now + float(next_delay or 15)
|
||
self._save()
|
||
return False
|
||
self._items.append({
|
||
"kind": kind,
|
||
"request_id": request_id,
|
||
"payload": payload,
|
||
"retry_count": 0,
|
||
"last_error": str(error or "")[:500],
|
||
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"next_retry_at": now + float(next_delay or 15),
|
||
})
|
||
self._save()
|
||
print(f"[outbox] queued {kind} request_id={request_id} err={error}")
|
||
return True
|
||
|
||
def ack(self, kind, request_id):
|
||
kind = (kind or "recognition").strip() or "recognition"
|
||
with self.lock:
|
||
before = len(self._items)
|
||
self._items = [
|
||
it for it in self._items
|
||
if not (
|
||
it.get("kind") == kind
|
||
and it.get("request_id") == request_id
|
||
)
|
||
]
|
||
if len(self._items) != before:
|
||
self._save()
|
||
print(f"[outbox] acked {kind} request_id={request_id}")
|
||
return True
|
||
return False
|
||
|
||
def bump(self, kind, request_id, error=None, next_delay=30.0, max_attempts=0):
|
||
"""失败后增加重试计数；超过上限则标记 dead（仍保留便于人工处理）。"""
|
||
kind = (kind or "recognition").strip() or "recognition"
|
||
now = time.time()
|
||
with self.lock:
|
||
for item in self._items:
|
||
if item.get("kind") != kind or item.get("request_id") != request_id:
|
||
continue
|
||
item["retry_count"] = int(item.get("retry_count") or 0) + 1
|
||
item["last_error"] = str(error or "")[:500]
|
||
item["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
max_n = int(max_attempts or 0)
|
||
if max_n > 0 and item["retry_count"] >= max_n:
|
||
item["status"] = "dead"
|
||
item["next_retry_at"] = now + 86400 * 365
|
||
print(
|
||
f"[outbox] dead {kind} request_id={request_id} "
|
||
f"retries={item['retry_count']} err={error}"
|
||
)
|
||
else:
|
||
item["status"] = "pending"
|
||
# 指数退避，最长 10 分钟
|
||
delay = min(float(next_delay or 30) * (2 ** min(item["retry_count"] - 1, 5)), 600.0)
|
||
item["next_retry_at"] = now + delay
|
||
print(
|
||
f"[outbox] retry#{item['retry_count']} {kind} "
|
||
f"request_id={request_id} next={delay:.0f}s err={error}"
|
||
)
|
||
self._save()
|
||
return item["retry_count"]
|
||
return 0
|
||
|
||
def due_items(self, limit=20):
|
||
now = time.time()
|
||
with self.lock:
|
||
due = []
|
||
for item in self._items:
|
||
if (item.get("status") or "pending") == "dead":
|
||
continue
|
||
if float(item.get("next_retry_at") or 0) > now:
|
||
continue
|
||
due.append(item)
|
||
due.sort(key=lambda x: float(x.get("next_retry_at") or 0))
|
||
return [dict(it) for it in due[: max(1, int(limit or 20))]]
|
||
|
||
def pending_count(self):
|
||
with self.lock:
|
||
return sum(
|
||
1 for it in self._items
|
||
if (it.get("status") or "pending") != "dead"
|
||
)
|
||
|
||
def stats(self):
|
||
with self.lock:
|
||
pending = dead = 0
|
||
for it in self._items:
|
||
if (it.get("status") or "pending") == "dead":
|
||
dead += 1
|
||
else:
|
||
pending += 1
|
||
return {"pending": pending, "dead": dead, "total": len(self._items)}
|