four-level-schedule/schedule-manager/parser/parse_service.py

import uvicorn
import time
import threading
import argparse
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field
from typing import Dict, Optional, List
import os
import httpx

# 初始化FastAPI应用
app = FastAPI(title="Python解析服务", version="1.0")

# 存储任务状态：key=task_id, value={status, progress, result, error}
task_status: Dict[str, dict] = {}
# 存储服务状态
service_status = {
    "status": "running",
    "pid": os.getpid(),
    "start_time": time.time()
}
# 锁机制，保证多线程安全
task_lock = threading.Lock()


# ------------------------------
# 请求模型定义
# ------------------------------
class ExecuteRequest(BaseModel):
    file_path: str=Field(alias="filePath")
    task_id:str=Field(alias="taskId")

# ------------------------------
# 接口定义
# ------------------------------
@app.post("/execute", summary="接收解析任务并执行")
async def execute_task(request: ExecuteRequest):
    """
    接收Java端下发的解析任务，后台异步执行
    """
    try:
        # 验证请求参数
        if not request.file_path or not request.task_id:
            return {
                "code": 400,
                "msg": "参数错误：filePath和taskId不能为空",
                "data": None
            }

        # 调用远程解析服务
        remote_url = "http://localhost:1086/execute"

        async with httpx.AsyncClient(timeout=300) as client:
            try:
                response = await client.post(
                    remote_url,
                    json={
                        "filePath": request.file_path,
                        "taskId": request.task_id
                    }
                )
                response.raise_for_status()
                return response.json()
            except Exception as e:
                return {
                    "code": 500,
                    "msg": f"调用远程解析服务失败: {str(e)}",
                    "data": None
                }

    except Exception as e:
        return {
            "code": 500,
            "msg": f"服务内部错误: {str(e)}",
            "data": None
        }

@app.get("/status", summary="状态接口")
async def health_check():
    """
    返回实例健康状态、资源使用情况
    """
    
    
    
    return {
    "code": 200,
    "msg": "success",
    "data": {
        "status": 0,
        "cpu_usage": 0.2,
        "gpu_usage": 0.0,
        "memory_usage": 0.2,
        "gpu_memory": 0.0
    }
}


@app.get("/shutdown", summary="优雅关闭服务（供Java端终结实例调用）")
async def shutdown_service():
    """
    优雅关闭服务，清理资源
    """
    service_status["status"] = "stopping"
    # 可添加清理逻辑：如保存未完成任务、关闭解析器连接等
    return {
        "code": 200,
        "msg": "服务开始优雅关闭",
        "data": None
    }

# ------------------------------
# 启动入口
# ------------------------------
def main():
    parser = argparse.ArgumentParser(description="Python解析服务启动参数")
    parser.add_argument("--host", default="0.0.0.0", help="服务监听地址")
    parser.add_argument("--port", type=int, default=8000, help="服务监听端口")
    args = parser.parse_args()
    
    # 启动FastAPI服务
    uvicorn.run(
        "parse_service:app",
        host=args.host,
        port=args.port,
        reload=False,  # 生产环境关闭热重载
        workers=1  # 单进程运行，避免任务状态共享问题
    )

if __name__ == "__main__":
    main()