EXIF scanner: multi-photo gallery with GPS detection
This commit is contained in:
@@ -0,0 +1,144 @@
|
||||
"""EXIF + OCR test backend — validate that GPS survives upload pipeline."""
|
||||
import io, json, re, uuid
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from PIL import Image as PILImage
|
||||
|
||||
try:
|
||||
import pytesseract
|
||||
HAS_TESSERACT = True
|
||||
except ImportError:
|
||||
HAS_TESSERACT = False
|
||||
|
||||
UPLOADS = Path(__file__).parent / "uploads"
|
||||
UPLOADS.mkdir(exist_ok=True)
|
||||
|
||||
app = FastAPI(title="EXIF Test")
|
||||
|
||||
|
||||
def _dms_to_decimal(dms, ref):
|
||||
"""Convert EXIF DMS tuple to decimal degrees."""
|
||||
try:
|
||||
deg, minutes, sec = float(dms[0]), float(dms[1]), float(dms[2])
|
||||
decimal = deg + minutes / 60.0 + sec / 3600.0
|
||||
if ref in ("S", "W"):
|
||||
decimal = -decimal
|
||||
return round(decimal, 7)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def extract_exif(image_bytes: bytes) -> dict:
|
||||
"""Pull all useful EXIF fields + GPS from raw image bytes."""
|
||||
result = {"has_exif": False, "tags": {}, "gps": None}
|
||||
try:
|
||||
img = PILImage.open(io.BytesIO(image_bytes))
|
||||
exif = img.getexif()
|
||||
if not exif:
|
||||
return result
|
||||
|
||||
# Standard EXIF tags
|
||||
for tag_id, value in exif.items():
|
||||
tag_name = PILImage.ExifTags.TAGS.get(tag_id, f"0x{tag_id:04x}")
|
||||
# Skip binary/thumbnail blobs
|
||||
if tag_name in ("MakerNote", "UserComment", "PrintImageMatching"):
|
||||
continue
|
||||
result["tags"][tag_name] = str(value)[:300]
|
||||
|
||||
if result["tags"]:
|
||||
result["has_exif"] = True
|
||||
|
||||
# GPS IFD
|
||||
gps_ifd = exif.get_ifd(0x8825)
|
||||
if gps_ifd:
|
||||
lat_ref = gps_ifd.get(1, "N")
|
||||
lat_dms = gps_ifd.get(2)
|
||||
lng_ref = gps_ifd.get(3, "E")
|
||||
lng_dms = gps_ifd.get(4)
|
||||
|
||||
if lat_dms and lng_dms:
|
||||
lat = _dms_to_decimal(lat_dms, lat_ref)
|
||||
lng = _dms_to_decimal(lng_dms, lng_ref)
|
||||
if lat is not None and lng is not None:
|
||||
result["gps"] = {
|
||||
"lat": lat,
|
||||
"lng": lng,
|
||||
"lat_ref": str(lat_ref),
|
||||
"lng_ref": str(lng_ref),
|
||||
"raw_lat_dms": [float(v) for v in lat_dms],
|
||||
"raw_lng_dms": [float(v) for v in lng_dms],
|
||||
}
|
||||
except Exception as e:
|
||||
result["error"] = str(e)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def run_ocr(image_bytes: bytes) -> dict:
|
||||
"""Run Tesseract OCR on the image."""
|
||||
if not HAS_TESSERACT:
|
||||
return {"available": False, "text": "", "error": "pytesseract not installed"}
|
||||
|
||||
tmp_path = UPLOADS / f"ocr_{uuid.uuid4().hex}.jpg"
|
||||
tmp_path.write_bytes(image_bytes)
|
||||
try:
|
||||
img = PILImage.open(tmp_path)
|
||||
img_gray = img.convert("L")
|
||||
text = pytesseract.image_to_string(img_gray, config="--psm 6")
|
||||
# Extract machine ID patterns
|
||||
match_5dash = re.search(r"(\d{5})[-\s]*(\d{6,})", text)
|
||||
match_5plus = re.search(r"(\d{5,})", text)
|
||||
return {
|
||||
"available": True,
|
||||
"raw_text": text.strip()[:500],
|
||||
"match_5dash6": match_5dash.group(0) if match_5dash else None,
|
||||
"match_5plus": match_5plus.group(0) if match_5plus else None,
|
||||
}
|
||||
finally:
|
||||
tmp_path.unlink(missing_ok=True)
|
||||
|
||||
|
||||
@app.post("/api/analyze")
|
||||
async def analyze_photo(file: UploadFile = File(...)):
|
||||
"""Upload a photo, get back EXIF + OCR results."""
|
||||
contents = await file.read()
|
||||
file_size = len(contents)
|
||||
|
||||
# Save
|
||||
ext = Path(file.filename or "photo.jpg").suffix.lower()
|
||||
if ext not in {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tiff", ".tif", ".dng"}:
|
||||
ext = ".jpg"
|
||||
fname = f"{uuid.uuid4().hex}{ext}"
|
||||
(UPLOADS / fname).write_bytes(contents)
|
||||
|
||||
exif_result = extract_exif(contents)
|
||||
ocr_result = run_ocr(contents) if HAS_TESSERACT else {"available": False, "text": ""}
|
||||
|
||||
return {
|
||||
"filename": file.filename,
|
||||
"saved_as": fname,
|
||||
"file_size": file_size,
|
||||
"file_size_kb": round(file_size / 1024, 1),
|
||||
"exif": exif_result,
|
||||
"ocr": ocr_result,
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/uploads")
|
||||
async def list_uploads():
|
||||
"""List previously uploaded files."""
|
||||
files = sorted(UPLOADS.glob("*"), key=lambda p: p.stat().st_mtime, reverse=True)
|
||||
return [
|
||||
{"name": f.name, "size_kb": round(f.stat().st_size / 1024, 1)}
|
||||
for f in files[:20]
|
||||
]
|
||||
|
||||
|
||||
# Mount static LAST
|
||||
app.mount("/", StaticFiles(directory="static", html=True), name="static")
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
uvicorn.run(app, host="0.0.0.0", port=8903)
|
||||
Reference in New Issue
Block a user