Salastil/odysseus
1
0
Fork 0
mirror of https://github.com/pewdiepie-archdaemon/odysseus.git synced 2026-06-28 15:45:22 -04:00
Code Issues Packages Projects Releases Wiki Activity

refactor(routes): move gallery domain into routes/gallery subpackage (#4903)

Browse Source 
Move the gallery route domain into routes/gallery/ while preserving backward-compatible legacy import shims.

- app imports the canonical gallery route module
- canonical gallery route code imports canonical gallery helpers
- legacy gallery route/helper paths remain compatibility aliases
- add shim regression coverage for module identity and monkeypatch behavior
- repoint gallery source-introspection tests to the canonical paths

No intended behavior change.
This commit is contained in:
Tal.Yuan
2026-06-28 17:40:34 +08:00
committed by GitHub
parent 5b8bfdabab
commit 8066a8e0cd
17 changed files with 2165 additions and 2077 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+1 -1
View File
@@ -685,7 +685,7 @@ from routes.signature_routes import setup_signature_routes
app.include_router(setup_signature_routes())
# Gallery (image library)
from routes.gallery_routes import setup_gallery_routes
from routes.gallery.gallery_routes import setup_gallery_routes
app.include_router(setup_gallery_routes())
# Persisted image-editor drafts (server-backed projects)
routes/gallery/__init__.py
+6
View File
@@ -0,0 +1,6 @@
"""Gallery route domain package (slice 2a, #4082/#4071).
Contains gallery_routes.py and gallery_helpers.py, migrated from the flat
routes/ directory. Backward-compat shims at routes/gallery_routes.py and
routes/gallery_helpers.py re-export from here.
"""
routes/gallery/gallery_helpers.py
+144
View File
@@ -0,0 +1,144 @@
"""gallery_helpers.py — extracted helpers, models, and small utilities.
Imported by gallery_routes.py."""
"""Gallery routes — browsable library for photos and AI-generated images."""
import logging
from datetime import datetime
from typing import Dict, Any, Optional
from pydantic import BaseModel
from core.database import GalleryImage
from src.auth_helpers import _auth_disabled
logger = logging.getLogger(__name__)
# ---- Request schemas ----
class GalleryPatch(BaseModel):
tags: Optional[str] = None
favorite: Optional[bool] = None
album_id: Optional[str] = None
# ---- EXIF extraction ----
def _extract_exif(content: bytes) -> dict:
"""Extract EXIF metadata from image bytes. Returns dict of fields."""
result = {"width": None, "height": None}
try:
from PIL import Image
from io import BytesIO
img = Image.open(BytesIO(content))
# Read the raw EXIF before any transpose: exif_transpose strips the
# orientation tag and with it the parsed EXIF view.
exif = img._getexif() if hasattr(img, '_getexif') else None
# Record DISPLAY dimensions (EXIF-rotated), matching upload_handler.
# A phone photo with Orientation 6/8 is stored landscape but shown
# portrait, so the raw width/height swap the aspect ratio.
try:
from PIL import ImageOps
img = ImageOps.exif_transpose(img) or img
except Exception:
pass
result["width"] = img.width
result["height"] = img.height
if not exif:
return result
# EXIF tag IDs
# 271=Make, 272=Model, 306=DateTime, 36867=DateTimeOriginal
# 34853=GPSInfo
result["camera_make"] = str(exif.get(271, "")).strip() or None
result["camera_model"] = str(exif.get(272, "")).strip() or None
# Date taken
for tag_id in (36867, 36868, 306): # DateTimeOriginal, DateTimeDigitized, DateTime
raw = exif.get(tag_id)
if raw:
try:
result["taken_at"] = datetime.strptime(str(raw).strip(), "%Y:%m:%d %H:%M:%S")
break
except (ValueError, TypeError):
pass
# GPS
gps_info = exif.get(34853)
if gps_info and isinstance(gps_info, dict):
try:
def _to_deg(vals):
d, m, s = [float(v) for v in vals]
return d + m / 60 + s / 3600
if 2 in gps_info and 4 in gps_info:
lat = _to_deg(gps_info[2])
lng = _to_deg(gps_info[4])
if gps_info.get(1) == 'S': lat = -lat
if gps_info.get(3) == 'W': lng = -lng
result["gps_lat"] = f"{lat:.6f}"
result["gps_lng"] = f"{lng:.6f}"
except Exception:
pass
except Exception as e:
# User-visible failure (photo loses metadata): surface at WARNING
# and record on the result so the upload endpoint can pass it back.
logger.warning(f"EXIF extraction failed: {e}")
result["exif_error"] = str(e)
return result
# ---- Helpers ----
def _image_to_dict(img: GalleryImage, session_name: str = None) -> Dict[str, Any]:
return {
"id": img.id,
"filename": img.filename,
"url": f"/api/generated-image/{img.filename}",
"prompt": img.prompt,
"model": img.model,
"size": img.size,
"quality": img.quality,
"tags": img.tags or "",
"ai_tags": img.ai_tags or "",
"user_tags": img.tags or "",
"session_id": img.session_id,
"session_name": session_name,
"album_id": img.album_id,
"is_active": img.is_active,
"favorite": img.favorite or False,
"taken_at": img.taken_at.isoformat() if img.taken_at else None,
"camera": f"{img.camera_make or ''} {img.camera_model or ''}".strip() or None,
"gps": {"lat": img.gps_lat, "lng": img.gps_lng} if img.gps_lat else None,
"width": img.width,
"height": img.height,
"file_size": img.file_size,
"created_at": img.created_at.isoformat() if img.created_at else None,
"updated_at": img.updated_at.isoformat() if img.updated_at else None,
}
def _owner_filter(q, user, model_cls=GalleryImage):
"""Apply owner filtering to a gallery query.
``get_current_user`` returns None both in auth-disabled single-user mode
and when auth is enabled but no current user was resolved. Preserve the
single-user behavior, but fail closed for auth-enabled null-user states.
"""
if user is not None:
return q.filter(model_cls.owner == user)
if _auth_disabled():
return q
return q.filter(False)
def _human_size(nbytes):
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if abs(nbytes) < 1024:
return f"{nbytes:.1f} {unit}"
nbytes /= 1024
return f"{nbytes:.1f} PB"
routes/gallery/gallery_routes.py
+1926
View File
File diff suppressed because it is too large Load Diff
routes/gallery_helpers.py
+10 -140
View File
@@ -1,144 +1,14 @@
"""gallery_helpers.py — extracted helpers, models, and small utilities.
"""Backward-compat shim — canonical location is routes/gallery/gallery_helpers.py.
Imported by gallery_routes.py."""
This module is replaced in ``sys.modules`` by the canonical module object so
that ``import routes.gallery_helpers``, ``from routes.gallery_helpers import X``,
``importlib.import_module("routes.gallery_helpers")``, and
``monkeypatch.setattr(routes.gallery_helpers, ...)`` all operate on the *same*
object. Keeps existing import paths working after slice 2a (#4082/#4071).
"""
"""Gallery routes — browsable library for photos and AI-generated images."""
import sys as _sys
import logging
from datetime import datetime
from typing import Dict, Any, Optional
from routes.gallery import gallery_helpers as _canonical # noqa: F401
from pydantic import BaseModel
from core.database import GalleryImage
from src.auth_helpers import _auth_disabled
logger = logging.getLogger(__name__)
# ---- Request schemas ----
class GalleryPatch(BaseModel):
tags: Optional[str] = None
favorite: Optional[bool] = None
album_id: Optional[str] = None
# ---- EXIF extraction ----
def _extract_exif(content: bytes) -> dict:
"""Extract EXIF metadata from image bytes. Returns dict of fields."""
result = {"width": None, "height": None}
try:
from PIL import Image
from io import BytesIO
img = Image.open(BytesIO(content))
# Read the raw EXIF before any transpose: exif_transpose strips the
# orientation tag and with it the parsed EXIF view.
exif = img._getexif() if hasattr(img, '_getexif') else None
# Record DISPLAY dimensions (EXIF-rotated), matching upload_handler.
# A phone photo with Orientation 6/8 is stored landscape but shown
# portrait, so the raw width/height swap the aspect ratio.
try:
from PIL import ImageOps
img = ImageOps.exif_transpose(img) or img
except Exception:
pass
result["width"] = img.width
result["height"] = img.height
if not exif:
return result
# EXIF tag IDs
# 271=Make, 272=Model, 306=DateTime, 36867=DateTimeOriginal
# 34853=GPSInfo
result["camera_make"] = str(exif.get(271, "")).strip() or None
result["camera_model"] = str(exif.get(272, "")).strip() or None
# Date taken
for tag_id in (36867, 36868, 306): # DateTimeOriginal, DateTimeDigitized, DateTime
raw = exif.get(tag_id)
if raw:
try:
result["taken_at"] = datetime.strptime(str(raw).strip(), "%Y:%m:%d %H:%M:%S")
break
except (ValueError, TypeError):
pass
# GPS
gps_info = exif.get(34853)
if gps_info and isinstance(gps_info, dict):
try:
def _to_deg(vals):
d, m, s = [float(v) for v in vals]
return d + m / 60 + s / 3600
if 2 in gps_info and 4 in gps_info:
lat = _to_deg(gps_info[2])
lng = _to_deg(gps_info[4])
if gps_info.get(1) == 'S': lat = -lat
if gps_info.get(3) == 'W': lng = -lng
result["gps_lat"] = f"{lat:.6f}"
result["gps_lng"] = f"{lng:.6f}"
except Exception:
pass
except Exception as e:
# User-visible failure (photo loses metadata): surface at WARNING
# and record on the result so the upload endpoint can pass it back.
logger.warning(f"EXIF extraction failed: {e}")
result["exif_error"] = str(e)
return result
# ---- Helpers ----
def _image_to_dict(img: GalleryImage, session_name: str = None) -> Dict[str, Any]:
return {
"id": img.id,
"filename": img.filename,
"url": f"/api/generated-image/{img.filename}",
"prompt": img.prompt,
"model": img.model,
"size": img.size,
"quality": img.quality,
"tags": img.tags or "",
"ai_tags": img.ai_tags or "",
"user_tags": img.tags or "",
"session_id": img.session_id,
"session_name": session_name,
"album_id": img.album_id,
"is_active": img.is_active,
"favorite": img.favorite or False,
"taken_at": img.taken_at.isoformat() if img.taken_at else None,
"camera": f"{img.camera_make or ''} {img.camera_model or ''}".strip() or None,
"gps": {"lat": img.gps_lat, "lng": img.gps_lng} if img.gps_lat else None,
"width": img.width,
"height": img.height,
"file_size": img.file_size,
"created_at": img.created_at.isoformat() if img.created_at else None,
"updated_at": img.updated_at.isoformat() if img.updated_at else None,
}
def _owner_filter(q, user, model_cls=GalleryImage):
"""Apply owner filtering to a gallery query.
``get_current_user`` returns None both in auth-disabled single-user mode
and when auth is enabled but no current user was resolved. Preserve the
single-user behavior, but fail closed for auth-enabled null-user states.
"""
if user is not None:
return q.filter(model_cls.owner == user)
if _auth_disabled():
return q
return q.filter(False)
def _human_size(nbytes):
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if abs(nbytes) < 1024:
return f"{nbytes:.1f} {unit}"
nbytes /= 1024
return f"{nbytes:.1f} PB"
_sys.modules[__name__] = _canonical
routes/gallery_routes.py
+12 -1922
View File
File diff suppressed because it is too large Load Diff
tests/test_direct_upload_limits.py
+1 -1
View File
@@ -40,7 +40,7 @@ def test_direct_upload_routes_use_bounded_reads():
"routes/stt_routes.py": [
"read_upload_limited(file, STT_MAX_AUDIO_BYTES",
],
"routes/gallery_routes.py": [
"routes/gallery/gallery_routes.py": [
"read_upload_limited(file, GALLERY_UPLOAD_MAX_BYTES",
"read_upload_limited(file, GALLERY_TRANSFORM_UPLOAD_MAX_BYTES",
],
tests/test_endpoint_owner_scope_followup.py
+1 -1
View File
@@ -377,7 +377,7 @@ def test_compare_endpoint_key_lookup_is_owner_scoped():
def test_gallery_image_endpoint_lookups_are_owner_scoped():
body = Path("routes/gallery_routes.py").read_text(encoding="utf-8")
body = Path("routes/gallery/gallery_routes.py").read_text(encoding="utf-8")
helper_body = body.split("def _visible_image_endpoint_query", 1)[1].split(
"def _first_visible_image_endpoint", 1
)[0]
tests/test_gallery_album_owner_scope.py
+1 -1
View File
@@ -12,7 +12,7 @@ from pathlib import Path
def _function_sources():
source = Path("routes/gallery_routes.py").read_text(encoding="utf-8")
source = Path("routes/gallery/gallery_routes.py").read_text(encoding="utf-8")
tree = ast.parse(source)
return {
node.name: ast.get_source_segment(source, node) or ""
tests/test_gallery_endpoint_ssrf.py
+1 -1
View File
@@ -15,7 +15,7 @@ metadata range.
import ast
from pathlib import Path
SRC = Path(__file__).resolve().parent.parent / "routes" / "gallery_routes.py"
SRC = Path(__file__).resolve().parent.parent / "routes" / "gallery" / "gallery_routes.py"
def _function_source(src_text: str, func_name: str) -> str:
tests/test_gallery_exif_orientation.py
+2 -2
View File
@@ -32,8 +32,8 @@ def extract_exif(monkeypatch):
return MagicMock()
monkeypatch.setitem(sys.modules, "core.database", _DBStub("core.database"))
monkeypatch.delitem(sys.modules, "routes.gallery_helpers", raising=False)
mod = importlib.import_module("routes.gallery_helpers")
monkeypatch.delitem(sys.modules, "routes.gallery.gallery_helpers", raising=False)
mod = importlib.import_module("routes.gallery.gallery_helpers")
return mod._extract_exif
tests/test_gallery_filename_confinement.py
+1 -1
View File
@@ -128,7 +128,7 @@ def test_gallery_replace_rejects_symlink_escape(tmp_path, monkeypatch):
def test_gallery_file_operations_use_confining_resolver():
source = Path("routes/gallery_routes.py").read_text(encoding="utf-8")
source = Path("routes/gallery/gallery_routes.py").read_text(encoding="utf-8")
assert 'Path("data/generated_images") / img.filename' not in source
assert 'os.path.join("data", "generated_images", img.filename)' not in source
tests/test_gallery_image_privileges.py
+1 -1
View File
@@ -15,7 +15,7 @@ GATED_IMAGE_FUNCTIONS = {
def _gallery_source():
return Path("routes/gallery_routes.py").read_text(encoding="utf-8")
return Path("routes/gallery/gallery_routes.py").read_text(encoding="utf-8")
def _function_sources(source):
tests/test_gallery_routes_shim.py
+52
View File
@@ -0,0 +1,52 @@
"""Regression test for the gallery route shim (slice 2a, #4082/#4071).
The backward-compat shims at ``routes/gallery_routes.py`` and
``routes/gallery_helpers.py`` use ``sys.modules`` replacement so the legacy
import path and the canonical ``routes.gallery.*`` path resolve to the *same*
module object. This test pins that contract: if the shim is ever changed to a
plain ``from ... import *`` (or removed), these assertions catch it before the
monkeypatch-based gallery tests silently start patching the wrong module.
"""
import importlib
import routes.gallery_routes as _shim_routes # noqa: F401
import routes.gallery_helpers as _shim_helpers # noqa: F401
def test_legacy_and_canonical_route_module_are_same_object():
"""``import routes.gallery_routes`` must alias the canonical module."""
legacy = importlib.import_module("routes.gallery_routes")
canonical = importlib.import_module("routes.gallery.gallery_routes")
assert legacy is canonical, (
"routes.gallery_routes shim must resolve to the canonical "
"routes.gallery.gallery_routes module object"
)
def test_legacy_and_canonical_helpers_module_are_same_object():
"""``import routes.gallery_helpers`` must alias the canonical module."""
legacy = importlib.import_module("routes.gallery_helpers")
canonical = importlib.import_module("routes.gallery.gallery_helpers")
assert legacy is canonical, (
"routes.gallery_helpers shim must resolve to the canonical "
"routes.gallery.gallery_helpers module object"
)
def test_monkeypatch_via_legacy_path_affects_canonical(monkeypatch):
"""Patching through the legacy path must reach the canonical module.
Several gallery tests do ``import routes.gallery_routes as gr`` followed by
``monkeypatch.setattr(gr, "get_current_user", ...)``. For that to take
effect at runtime, the legacy module object and the canonical one must be
identical.
"""
legacy = importlib.import_module("routes.gallery_routes")
canonical = importlib.import_module("routes.gallery.gallery_routes")
sentinel = object()
monkeypatch.setattr(legacy, "setup_gallery_routes", sentinel)
assert canonical.setup_gallery_routes is sentinel, (
"monkeypatch via legacy path did not reach the canonical module"
)
tests/test_security_regressions.py
+3 -3
View File
@@ -1099,9 +1099,9 @@ def _import_session_routes_for_filename():
def _import_gallery_routes_for_filename():
# Same rationale as the session route helper: import _sanitize_gallery_filename
# against the real core.database and leave a clean, real module cached.
_drop_route_module_cache("routes.gallery_routes")
_drop_route_module_cache("routes.gallery_helpers")
return importlib.import_module("routes.gallery_routes")
_drop_route_module_cache("routes.gallery.gallery_routes")
_drop_route_module_cache("routes.gallery.gallery_helpers")
return importlib.import_module("routes.gallery.gallery_routes")
def test_export_filename_sanitizer_blocks_header_and_path_chars():
tests/test_upload_limits_centralized.py
+2 -2
View File
@@ -80,7 +80,7 @@ def test_non_positive_env_rejected(monkeypatch, env):
def test_routes_import_from_upload_limits_not_local_defs():
"""Routes must import the constant, not redefine it via raw getenv / literal."""
forbidden = {
"routes/gallery_routes.py": [
"routes/gallery/gallery_routes.py": [
'int(os.getenv("ODYSSEUS_GALLERY_UPLOAD_MAX_BYTES"',
'int(os.getenv("ODYSSEUS_GALLERY_TRANSFORM_UPLOAD_MAX_BYTES"',
],
@@ -97,7 +97,7 @@ def test_routes_import_from_upload_limits_not_local_defs():
# And each imports from upload_limits.
imports = {
"routes/gallery_routes.py": "GALLERY_UPLOAD_MAX_BYTES",
"routes/gallery/gallery_routes.py": "GALLERY_UPLOAD_MAX_BYTES",
"routes/memory_routes.py": "MEMORY_IMPORT_MAX_BYTES",
"routes/personal_routes.py": "PERSONAL_UPLOAD_MAX_BYTES",
"routes/email_routes.py": "EMAIL_COMPOSE_UPLOAD_MAX_BYTES",
tests/test_vision_owner_scope.py
+1 -1
View File
@@ -89,7 +89,7 @@ def test_request_vision_call_sites_pass_owner():
processor_source = (ROOT / "src" / "document_processor.py").read_text()
upload_source = (ROOT / "routes" / "upload_routes.py").read_text()
document_source = (ROOT / "routes" / "document_routes.py").read_text()
gallery_source = (ROOT / "routes" / "gallery_routes.py").read_text()
gallery_source = (ROOT / "routes" / "gallery" / "gallery_routes.py").read_text()
memory_source = (ROOT / "routes" / "memory_routes.py").read_text()
assert 'analyze_image_with_vl_result(file_info["path"], owner=owner)' in chat_source