78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from fastapi import FastAPI, File, HTTPException, Request, UploadFile
|
||
|
|
from fastapi.responses import FileResponse, HTMLResponse
|
||
|
|
from fastapi.staticfiles import StaticFiles
|
||
|
|
from fastapi.templating import Jinja2Templates
|
||
|
|
from pydantic import BaseModel, Field
|
||
|
|
|
||
|
|
from .config import ROOT_DIR
|
||
|
|
from .job_service import analyze_attachment_urls, job_store, process_uploads
|
||
|
|
|
||
|
|
|
||
|
|
app = FastAPI(title="PCF 焊接数据导入生成工具")
|
||
|
|
templates = Jinja2Templates(directory=str(ROOT_DIR / "templates"))
|
||
|
|
app.mount("/static", StaticFiles(directory=str(ROOT_DIR / "static")), name="static")
|
||
|
|
|
||
|
|
|
||
|
|
class ParseUrlRequest(BaseModel):
|
||
|
|
attachment_url: str | None = Field(default=None, description="单个附件 URL，支持 .pcf 或 .zip")
|
||
|
|
attachment_urls: list[str] | None = Field(default=None, description="多个附件 URL，支持 .pcf 或 .zip")
|
||
|
|
|
||
|
|
def urls(self) -> list[str]:
|
||
|
|
urls: list[str] = []
|
||
|
|
if self.attachment_url:
|
||
|
|
urls.append(self.attachment_url)
|
||
|
|
if self.attachment_urls:
|
||
|
|
urls.extend(self.attachment_urls)
|
||
|
|
return [url for url in urls if url]
|
||
|
|
|
||
|
|
|
||
|
|
@app.get("/", response_class=HTMLResponse)
|
||
|
|
async def index(request: Request):
|
||
|
|
return templates.TemplateResponse("index.html", {"request": request})
|
||
|
|
|
||
|
|
|
||
|
|
@app.post("/api/jobs")
|
||
|
|
async def create_job(files: list[UploadFile] = File(...)):
|
||
|
|
if not files:
|
||
|
|
raise HTTPException(status_code=400, detail="请至少上传一个 PCF 文件")
|
||
|
|
result = await process_uploads(files)
|
||
|
|
return result.to_public_dict()
|
||
|
|
|
||
|
|
|
||
|
|
@app.post("/api/AnalysisPcf")
|
||
|
|
async def analysis_pcf(request: ParseUrlRequest):
|
||
|
|
urls = request.urls()
|
||
|
|
if not urls:
|
||
|
|
raise HTTPException(status_code=400, detail="请提供 attachment_url 或 attachment_urls")
|
||
|
|
return await analyze_attachment_urls(urls)
|
||
|
|
|
||
|
|
|
||
|
|
@app.get("/api/jobs/{job_id}")
|
||
|
|
async def get_job(job_id: str):
|
||
|
|
result = job_store.get(job_id)
|
||
|
|
if not result:
|
||
|
|
raise HTTPException(status_code=404, detail="任务不存在或服务已重启")
|
||
|
|
return result.to_public_dict()
|
||
|
|
|
||
|
|
|
||
|
|
@app.get("/api/jobs/{job_id}/download/{kind}")
|
||
|
|
async def download(job_id: str, kind: str):
|
||
|
|
result = job_store.get(job_id)
|
||
|
|
if not result:
|
||
|
|
raise HTTPException(status_code=404, detail="任务不存在或服务已重启")
|
||
|
|
files = {
|
||
|
|
"xls": (result.output_xls, "application/vnd.ms-excel"),
|
||
|
|
"issues": (result.issues_csv, "text/csv; charset=utf-8"),
|
||
|
|
"report": (result.report_json, "application/json"),
|
||
|
|
}
|
||
|
|
if kind not in files:
|
||
|
|
raise HTTPException(status_code=404, detail="下载类型不存在")
|
||
|
|
path, media_type = files[kind]
|
||
|
|
if not Path(path).exists():
|
||
|
|
raise HTTPException(status_code=404, detail="文件不存在")
|
||
|
|
return FileResponse(path, media_type=media_type, filename=Path(path).name)
|