chore: test callback time
This commit is contained in:
+8
-1
@@ -25,7 +25,8 @@ from resident_worker import ResidentPool
|
||||
# ==================== 日志配置 ====================
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s | %(levelname)s | %(message)s'
|
||||
format='%(asctime)s.%(msecs)03d | %(levelname)s | %(message)s',
|
||||
datefmt='%Y-%m-%d %H:%M:%S'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -96,7 +97,10 @@ def health_check():
|
||||
|
||||
@app.post("/api/v1/analysis/submit", response_model=TaskResponse)
|
||||
def submit_task(request: TaskRequest):
|
||||
received_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
|
||||
task_id = request.task_id if request.task_id else Path(request.folder_name).name
|
||||
logger.info("API 接收到任务时间：%s | task_id=%s | folder_name=%s",
|
||||
received_at, task_id, request.folder_name)
|
||||
if not task_id or len(task_id) < 3:
|
||||
raise HTTPException(status_code=400, detail="任务 ID 格式错误，至少 3 个字符")
|
||||
# Resolve the date at submission, not after a queued task starts tomorrow.
|
||||
@@ -110,6 +114,9 @@ def submit_task(request: TaskRequest):
|
||||
except QueueFull as exc:
|
||||
raise HTTPException(status_code=429, detail=str(exc),
|
||||
headers={"Retry-After": "5"}) from exc
|
||||
logger.info("API 任务入队时间：%s | task_id=%s | status=%s | created=%s",
|
||||
datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3],
|
||||
task_id, task["status"], created)
|
||||
return TaskResponse(code=200,
|
||||
message="任务已加入队列" if created else "任务已存在，正在等待或运行中",
|
||||
task_id=task_id, status=task["status"])
|
||||
|
||||
+28
-9
@@ -14,6 +14,7 @@ import os
|
||||
import argparse
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from tqdm import tqdm
|
||||
@@ -29,12 +30,16 @@ def setup_logging(verbose: bool = False):
|
||||
logging.basicConfig(
|
||||
level=level,
|
||||
format='%(asctime)s | %(levelname)s | %(message)s',
|
||||
datefmt='%H:%M:%S'
|
||||
datefmt='%Y-%m-%d %H:%M:%S'
|
||||
)
|
||||
return logging.getLogger(__name__)
|
||||
|
||||
logger = setup_logging()
|
||||
|
||||
|
||||
def now_text() -> str:
|
||||
return datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
|
||||
|
||||
# ==================== 支持中文路径的读写函数 ====================
|
||||
def cv2_imread_chinese(file_path: str) -> Optional[np.ndarray]:
|
||||
if not os.path.exists(file_path):
|
||||
@@ -551,20 +556,27 @@ def save_text_report(results: List[Dict], texts_dir: Path):
|
||||
# ==================== 回调后端接口 ====================
|
||||
def callback_api(task_id: str, status: int, api_url: str = API_HOST+API_CALLBACK):
|
||||
"""完成后回调后端接口"""
|
||||
started = time.perf_counter()
|
||||
callback_at = now_text()
|
||||
logger.info("任务 %s 开始回调 Go：%s | url=%s | status=%s",
|
||||
task_id, callback_at, api_url, status)
|
||||
try:
|
||||
payload = {
|
||||
"task_id": task_id,
|
||||
"status": status # 1=成功，0=失败
|
||||
}
|
||||
response = requests.post(api_url, json=payload, timeout=10)
|
||||
elapsed = time.perf_counter() - started
|
||||
if response.status_code == 200:
|
||||
logger.info(f"回调成功：{task_id}")
|
||||
logger.info("任务 %s 回调 Go 成功：%s | 回调耗时：%.3f 秒",
|
||||
task_id, now_text(), elapsed)
|
||||
return True
|
||||
else:
|
||||
logger.warning(f"回调失败：{response.status_code, api_url}")
|
||||
return False
|
||||
logger.warning("任务 %s 回调 Go 失败：%s | status=%s | 回调耗时：%.3f 秒 | url=%s",
|
||||
task_id, now_text(), response.status_code, elapsed, api_url)
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"回调异常：{e}")
|
||||
logger.error("任务 %s 回调 Go 异常：%s | 时间：%s | 回调耗时：%.3f 秒",
|
||||
task_id, e, now_text(), time.perf_counter() - started)
|
||||
return False
|
||||
|
||||
class SafeYOLO:
|
||||
@@ -583,10 +595,11 @@ class SafeYOLO:
|
||||
with self._lock:
|
||||
return self._model(*args, **kwargs)
|
||||
|
||||
|
||||
def analyze_directory(input_dir, task_id, model, workers=4, api_url=None):
|
||||
"""Reuse a warmed model. Return structured status to the resident parent."""
|
||||
received_at = now_text()
|
||||
started = time.perf_counter()
|
||||
logger.info("任务 %s 接收到任务时间：%s | input_dir=%s", task_id, received_at, input_dir)
|
||||
input_dir = Path(input_dir)
|
||||
results = []
|
||||
error = None
|
||||
@@ -616,15 +629,21 @@ def analyze_directory(input_dir, task_id, model, workers=4, api_url=None):
|
||||
logger.exception("分析任务失败：%s", task_id)
|
||||
error = str(exc) or type(exc).__name__
|
||||
analysis_seconds = time.perf_counter() - started
|
||||
logger.info("任务 %s 分析结束：%.3f 秒，成功 %d，未完整成功 %d", task_id, analysis_seconds,
|
||||
finished_at = now_text()
|
||||
logger.info("任务 %s 处理完成时间：%s | 处理用时：%.3f 秒 | 成功 %d，未完整成功 %d",
|
||||
task_id, finished_at, analysis_seconds,
|
||||
sum(r['status'] == 'success' for r in results),
|
||||
sum(r['status'] != 'success' for r in results))
|
||||
callback_ok = None
|
||||
if api_url:
|
||||
callback_ok = callback_api(task_id, 0 if error else 1, api_url)
|
||||
total_seconds = time.perf_counter() - started
|
||||
logger.info("任务 %s 全流程结束：%s | 接收：%s | 处理用时：%.3f 秒 | 含回调总用时：%.3f 秒",
|
||||
task_id, now_text(), received_at, analysis_seconds, total_seconds)
|
||||
return {'task_id': task_id, 'error': error, 'images': results,
|
||||
'analysis_seconds': analysis_seconds,
|
||||
'total_seconds': time.perf_counter() - started, 'callback_ok': callback_ok}
|
||||
'total_seconds': total_seconds, 'callback_ok': callback_ok,
|
||||
'received_at': received_at, 'finished_at': finished_at}
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
Reference in New Issue
Block a user