-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
111 lines (94 loc) · 3.2 KB
/
main.py
File metadata and controls
111 lines (94 loc) · 3.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
from app.api.endpoints import pdf
from app.utils.ghostscript_inkcov import gs_available
import os
import subprocess
import time
enable_docs = settings.enable_docs or settings.environment in ("dev", "staging")
app = FastAPI(
title=settings.api_title,
version=settings.api_version,
description=settings.api_description,
docs_url="/docs" if enable_docs else None,
redoc_url="/redoc" if enable_docs else None,
openapi_url="/openapi.json" if enable_docs else None,
)
START_TIME = time.time()
def _git_commit_short() -> str:
# Prefer env var if provided
commit = os.getenv("GIT_COMMIT")
if commit:
return commit[:7]
# Try git command
try:
out = subprocess.run([
"git", "rev-parse", "--short", "HEAD"
], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
return out.stdout.decode().strip()
except Exception:
return "unknown"
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=settings.cors_allow_credentials,
allow_methods=settings.cors_allow_methods,
allow_headers=settings.cors_allow_headers,
)
# Include routers
app.include_router(pdf.router)
@app.get("/")
async def root():
return {
"message": "PDF Dieline Processor API",
"version": settings.api_version,
"endpoints": {
"analyze": "/api/pdf/analyze",
"process": "/api/pdf/process",
"process_with_json": "/api/pdf/process-with-json-file"
}
}
@app.get("/hello/{name}")
async def say_hello(name: str):
return {"message": f"Hello {name}"}
def _pkg_version(modname: str) -> tuple[bool, str | None]:
try:
mod = __import__(modname)
ver = getattr(mod, "__version__", None)
if ver is None and modname == "fitz":
# PyMuPDF historically exposes the version via VersionBind / docstring
ver = getattr(mod, "VersionBind", None) or getattr(mod, "version", (None,))[0]
return True, ver
except Exception:
return False, None
# Prime Ghostscript probe at import time (cached); never blocks if missing
_GS_PROBE = gs_available()
@app.get("/healthz")
async def healthz():
pikepdf_ok, pikepdf_ver = _pkg_version("pikepdf")
fitz_ok, fitz_ver = _pkg_version("fitz")
pypdf_ok, pypdf_ver = _pkg_version("pypdf")
gs_ok, gs_ver = _GS_PROBE
return {
"status": "ok",
"uptime_seconds": round(time.time() - START_TIME, 2),
"dependencies": {
"pikepdf": {"available": pikepdf_ok, "version": pikepdf_ver},
"pymupdf": {"available": fitz_ok, "version": fitz_ver},
"pypdf": {"available": pypdf_ok, "version": pypdf_ver},
"ghostscript": {"available": gs_ok, "version": gs_ver},
},
"modes": {
"color_analysis_heuristic": True,
"color_analysis_measured": gs_ok,
},
}
@app.get("/version")
async def version():
return {
"name": settings.api_title,
"version": settings.api_version,
"git_commit": _git_commit_short(),
}