diff --git a/api_server.py b/api_server.py index 485abc7..ee1b639 100644 --- a/api_server.py +++ b/api_server.py @@ -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"]) diff --git a/wanzheng.py b/wanzheng.py index 4eb2d19..bd7cfa4 100644 --- a/wanzheng.py +++ b/wanzheng.py @@ -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():