dtoro/mule-image
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: editable taken_at + folder-based date repair and filter

Browse Source 
Lets operators fix corrupted capture dates at scale. Adds an editable
Date Taken field with a folder/filename-derived suggestion hint, a bulk
Date Taken section in the multi-select sidebar that either applies one
date to the whole selection or infers a per-photo date from each path,
a warning badge on thumbnails whose stored date disagrees with the
path, and a "Date issues" filter pill so suspicious photos can be
surfaced and fixed as a group. Edits are written back to EXIF on disk
so rescans don't clobber the fix.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
authentik Default Admin
2026-04-11 11:48:55 +02:00
parent 339e1be510
commit 30d03d8d4d
19 changed files with 1281 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
backend/app/models/photos.py
View File

@@ -46,6 +46,13 @@ class Photo(Base):
# column is maintained by two places: the scanner sets it on new
# rows, and POST /folders/{id}/hide recomputes it on toggle.
is_hidden = Column(Boolean, nullable=False, default=False, server_default='false', index=True)
# "Capture date is probably wrong" — denormalized from the folder/filename
# date-guesser. Set at scan time and recomputed on every taken_at edit so
# the filter bar can query it directly. See services/date_guess.py for
# the heuristic; kept as a stored column because recomputing on every
# list query would mean running the regex stack across thousands of rows.
has_date_warning = Column(Boolean, nullable=False, default=False, server_default='false', index=True)
# Thumbnail paths
thumb_small = Column(String) # path to 240px thumb

85
backend/app/routers/photos.py
View File

@@ -22,6 +22,8 @@ from app.models.folders import SourceRoot
from app.models.heaps import heap_photos
from app.models.tags import photo_tags
from app.schemas.photos import PhotoResponse, PhotoUpdate, PhotoListResponse, BulkAction
from app.services.exif_writer import ExifWriteError, write_taken_at
from app.services.date_guess import has_date_warning as compute_date_warning
from app.config import settings
router = APIRouter()
@@ -39,6 +41,7 @@ async def list_photos(
color_label: Optional[str] = None,
is_discarded: Optional[bool] = False,
is_duplicate: Optional[bool] = None,
has_date_warning: Optional[bool] = None,
heap_id: Optional[str] = None,
sort: str = "taken_at",
order: str = "desc",
@@ -149,6 +152,8 @@ async def list_photos(
# view shows everything regardless of duplicate status.
if is_duplicate is not None:
filters.append(Photo.is_duplicate == is_duplicate)
if has_date_warning is not None:
filters.append(Photo.has_date_warning == has_date_warning)
# Heap membership filter — restrict to photos that belong to the heap.
if heap_id:
@@ -642,6 +647,28 @@ async def update_photo(
photo.filename = new_name
photo.filepath = new_path
# taken_at edits write EXIF first, DB second — we'd rather surface a
# failure than leave the DB ahead of the file on disk. On success the
# source flips to 'manual' so the UI can render a badge and the next
# rescan knows not to overwrite it.
if 'taken_at' in update_data:
new_dt = update_data.pop('taken_at')
if new_dt is not None:
try:
await write_taken_at(photo.filepath, new_dt)
except ExifWriteError as exc:
raise HTTPException(
status_code=500,
detail=f"Failed to write EXIF: {exc}",
)
photo.taken_at = new_dt
photo.taken_at_source = 'manual'
# Recompute warning: a manual edit usually clears it (user just
# told us the right date), but if they set it to something that
# still disagrees with the folder path we'd rather keep the
# flag up than pretend the problem's gone.
photo.has_date_warning = compute_date_warning(photo.filepath, new_dt)
# Apply remaining updates
for field, value in update_data.items():
setattr(photo, field, value)
@@ -921,6 +948,64 @@ async def bulk_action(
elif action.action == 'set_color':
for photo in photos:
photo.color_label = action.value
elif action.action in ('set_taken_at', 'set_taken_at_map'):
# Two shapes share one code path:
# set_taken_at → value is one ISO datetime, applied to every id
# set_taken_at_map → value is {photo_id: iso datetime}, per-photo
# The per-photo variant is what the "guess from folder" bulk flow
# uses when every selected photo gets a different date.
if action.action == 'set_taken_at':
if not isinstance(action.value, str) or not action.value:
raise HTTPException(
status_code=400,
detail="set_taken_at requires an ISO datetime string",
)
try:
uniform_dt = datetime.fromisoformat(action.value)
except ValueError:
raise HTTPException(
status_code=400,
detail="Invalid ISO datetime for set_taken_at",
)
date_map = {p.id: uniform_dt for p in photos}
else:
if not isinstance(action.value, dict) or not action.value:
raise HTTPException(
status_code=400,
detail="set_taken_at_map requires a {id: iso} mapping",
)
date_map = {}
for pid, raw in action.value.items():
if not isinstance(raw, str):
continue
try:
date_map[pid] = datetime.fromisoformat(raw)
except ValueError:
continue
updated = 0
errors: list[dict[str, str]] = []
for photo in photos:
new_dt = date_map.get(photo.id)
if new_dt is None:
continue
try:
await write_taken_at(photo.filepath, new_dt)
except ExifWriteError as exc:
errors.append({"id": photo.id, "message": str(exc)})
continue
photo.taken_at = new_dt
photo.taken_at_source = 'manual'
photo.has_date_warning = compute_date_warning(photo.filepath, new_dt)
updated += 1
await db.commit()
return {
"status": "success",
"updated": updated,
"skipped": len(photos) - updated - len(errors),
"errors": errors,
}
elif action.action == 'add_tags':
# value is a list of tag ids. We bulk-insert (photo_id, tag_id)
# rows for every (photo, tag) combination that doesn't already

1
backend/app/schemas/photos.py
View File

@@ -39,6 +39,7 @@ class PhotoResponse(PhotoBase):
latitude: Optional[float] = None
longitude: Optional[float] = None
is_duplicate: bool = False
has_date_warning: bool = False
live_photo_video_id: Optional[str] = None
# tags: List[Dict[str, Any]] = [] # TODO: Enable when using eager loading

214
backend/app/services/date_guess.py Normal file
View File

@@ -0,0 +1,214 @@
"""
Folder/filename-based date guessing and "taken_at looks wrong" detection.
Direct Python port of `frontend/src/lib/guessDateFromPath.ts` — the logic
must stay in sync because the frontend renders the suggestion hint in the
info panel while the backend owns the `has_date_warning` flag that the
filter bar queries. Any heuristic change has to be applied to both files.
The guesser walks a filepath, tries the filename first as the source of
truth, then falls back to folder segments (deepest first) and multi-
segment layouts. Returns ``None`` when no recognisable date can be
extracted. `has_date_warning()` compares the guess to a stored `taken_at`
and reports whether the difference is large enough to flag.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from datetime import datetime
from typing import Literal, Optional
Confidence = Literal["high", "medium", "low"]
Source = Literal["folder", "filename"]
@dataclass(frozen=True)
class DateGuess:
date: datetime
confidence: Confidence
matched: str
source: Source
_MIN_YEAR = 1970
# Bump the ceiling annually via `datetime.now()` rather than a literal so
# we don't ship a time bomb. `+1` allows near-future timestamps (cameras
# with a slightly advanced clock at year end) without opening the door to
# 4-digit serial numbers that happen to start with "30xx".
def _max_year() -> int:
return datetime.now().year + 1
def _valid_year(y: int) -> bool:
return _MIN_YEAR <= y <= _max_year()
def _make_date(y: int, m: int, d: int) -> Optional[datetime]:
if not _valid_year(y):
return None
if not (1 <= m <= 12):
return None
if not (1 <= d <= 31):
return None
try:
# Noon local so downstream day-bucketing is stable across timezone
# rounding. The frontend mirrors this.
return datetime(y, m, d, 12, 0, 0)
except ValueError:
return None
def _segments(filepath: str) -> list[str]:
return [s for s in re.split(r"[\\/]+", filepath) if s]
_COMPACT_RE = re.compile(r"(?<!\d)(\d{4})(\d{2})(\d{2})(?!\d)")
_DASHED_RE = re.compile(r"(?<!\d)(\d{4})[-_.](\d{1,2})[-_.](\d{1,2})(?!\d)")
_MONTH_RE = re.compile(r"(?<!\d)(\d{4})[-_.](\d{1,2})(?!\d)")
_YEAR_RE = re.compile(r"(?<!\d)(\d{4})(?!\d)")
_FOUR_DIGITS = re.compile(r"^\d{4}$")
_ONE_OR_TWO = re.compile(r"^\d{1,2}$")
def _guess_from_string(
input: str,
source: Source,
allow_year_only: bool,
) -> Optional[DateGuess]:
if not input:
return None
m = _COMPACT_RE.search(input)
if m:
d = _make_date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
if d:
return DateGuess(
date=d,
confidence="high",
matched=f"{m.group(1)}-{m.group(2)}-{m.group(3)}",
source=source,
)
m = _DASHED_RE.search(input)
if m:
d = _make_date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
if d:
return DateGuess(
date=d,
confidence="high",
matched=f"{m.group(1)}-{m.group(2)}-{m.group(3)}",
source=source,
)
m = _MONTH_RE.search(input)
if m:
d = _make_date(int(m.group(1)), int(m.group(2)), 15)
if d:
return DateGuess(
date=d,
confidence="medium",
matched=f"{m.group(1)}-{m.group(2)}",
source=source,
)
if allow_year_only:
m = _YEAR_RE.search(input)
if m:
d = _make_date(int(m.group(1)), 7, 1)
if d:
return DateGuess(
date=d,
confidence="low",
matched=m.group(1),
source=source,
)
return None
def _guess_from_folder_layout(folders: list[str]) -> Optional[DateGuess]:
# YYYY / MM / DD
for i in range(len(folders) - 2):
a, b, c = folders[i], folders[i + 1], folders[i + 2]
if _FOUR_DIGITS.match(a) and _ONE_OR_TWO.match(b) and _ONE_OR_TWO.match(c):
d = _make_date(int(a), int(b), int(c))
if d:
return DateGuess(
date=d,
confidence="high",
matched=f"{a}/{b}/{c}",
source="folder",
)
# YYYY / MM
for i in range(len(folders) - 1):
a, b = folders[i], folders[i + 1]
if _FOUR_DIGITS.match(a) and _ONE_OR_TWO.match(b):
d = _make_date(int(a), int(b), 15)
if d:
return DateGuess(
date=d,
confidence="medium",
matched=f"{a}/{b}",
source="folder",
)
return None
_CONFIDENCE_RANK: dict[Confidence, int] = {"high": 3, "medium": 2, "low": 1}
def guess_date_from_path(filepath: str) -> Optional[DateGuess]:
"""Filename wins when it has any viable match; otherwise walk folder
segments deepest-first and pick the strongest hit."""
if not filepath:
return None
segs = _segments(filepath)
if not segs:
return None
filename = segs[-1]
folders = segs[:-1]
from_filename = _guess_from_string(filename, "filename", allow_year_only=False)
if from_filename:
return from_filename
best: Optional[DateGuess] = None
for seg in reversed(folders):
hit = _guess_from_string(seg, "folder", allow_year_only=True)
if not hit:
continue
if not best or _CONFIDENCE_RANK[hit.confidence] > _CONFIDENCE_RANK[best.confidence]:
best = hit
if hit.confidence == "high":
break
from_layout = _guess_from_folder_layout(folders)
if from_layout and (
not best or _CONFIDENCE_RANK[from_layout.confidence] > _CONFIDENCE_RANK[best.confidence]
):
best = from_layout
return best
_ONE_DAY = 24 * 60 * 60
def has_date_warning(filepath: str, taken_at: Optional[datetime]) -> bool:
"""True when the path-based guess disagrees with ``taken_at`` by more
than 24h, or when ``taken_at`` is missing and the path would supply
one. This is the authoritative flag stored on `photos.has_date_warning`
and queried by the timeline filter."""
guess = guess_date_from_path(filepath)
if not guess:
return False
if taken_at is None:
return True
try:
diff = abs((taken_at - guess.date).total_seconds())
except (TypeError, ValueError):
return False
return diff > _ONE_DAY

72
backend/app/services/exif_writer.py Normal file
View File

@@ -0,0 +1,72 @@
"""
EXIF write-back helpers.
The rest of the app reads EXIF at scan time and stashes the result in Postgres
(see `services/metadata.py`). This module handles the reverse direction: when
the user corrects a date in the UI we also rewrite the relevant EXIF tags on
disk so a later rescan won't clobber the fix and external tools see the same
truth the DB does.
"""
import asyncio
import logging
import subprocess
from datetime import datetime
from pathlib import Path
logger = logging.getLogger(__name__)
EXIFTOOL_TIMEOUT_SECONDS = 30
class ExifWriteError(RuntimeError):
"""Raised when exiftool fails to write tags to a file."""
def _format_exif_dt(dt: datetime) -> str:
return dt.strftime("%Y:%m:%d %H:%M:%S")
async def write_taken_at(filepath: str, dt: datetime) -> None:
"""Rewrite DateTimeOriginal / CreateDate / ModifyDate on the file.
- ``-overwrite_original`` so we don't litter the library with
``<name>_original`` sidecars.
- ``-P`` preserves the file's mtime so the scanner's mtime-based
change detection stays quiet.
- We set all three common date tags together because different viewers
read different ones; keeping them in lockstep avoids confusing
downstream tools and our own re-extraction pass.
"""
if not Path(filepath).exists():
raise ExifWriteError(f"File not found: {filepath}")
stamp = _format_exif_dt(dt)
cmd = [
"exiftool",
"-overwrite_original",
"-P",
f"-DateTimeOriginal={stamp}",
f"-CreateDate={stamp}",
f"-ModifyDate={stamp}",
filepath,
]
def _run() -> subprocess.CompletedProcess:
return subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=EXIFTOOL_TIMEOUT_SECONDS,
)
try:
result = await asyncio.to_thread(_run)
except subprocess.TimeoutExpired as exc:
raise ExifWriteError(f"exiftool timed out writing {filepath}") from exc
except FileNotFoundError as exc:
raise ExifWriteError("exiftool binary not available") from exc
if result.returncode != 0:
msg = (result.stderr or result.stdout or "unknown error").strip()
logger.warning("exiftool write failed for %s: %s", filepath, msg)
raise ExifWriteError(msg)

9
backend/app/services/metadata.py
View File

@@ -15,6 +15,7 @@ from sqlalchemy import select
from app.database import AsyncSessionLocal
from app.models import Photo
from app.services.date_guess import has_date_warning
logger = logging.getLogger(__name__)
@@ -238,6 +239,14 @@ async def _extract_metadata_async(photo_id: str):
photo.taken_at = taken_at
photo.taken_at_source = 'exif'
break
# Re-run the path-vs-date heuristic now that we know
# whether EXIF provided a real capture date. A true EXIF
# date that matches the folder clears the warning the
# scanner set during the filesystem-mtime pass.
photo.has_date_warning = has_date_warning(
photo.filepath, photo.taken_at
)
# Extract dimensions if not already set
if not photo.width:

9
backend/app/tasks/scan.py
View File

@@ -21,6 +21,7 @@ from app.models import Photo, Folder, SourceRoot
from app.config import settings
from app.tasks.thumbs import generate_thumbnails
from app.services.metadata import extract_metadata
from app.services.date_guess import has_date_warning
logger = logging.getLogger(__name__)
@@ -231,6 +232,7 @@ async def _scan_folder_async(folder_path: str, source_root_id: Optional[str], ta
effective_hidden = await is_folder_effectively_hidden(folder)
# Create photo entry
mtime_dt = datetime.fromtimestamp(stat.st_mtime)
photo = Photo(
filepath=filepath,
filename=filename,
@@ -239,8 +241,13 @@ async def _scan_folder_async(folder_path: str, source_root_id: Optional[str], ta
media_type=get_media_type(filepath),
original_format=Path(filepath).suffix.upper()[1:],
file_size=stat.st_size,
taken_at=datetime.fromtimestamp(stat.st_mtime),
taken_at=mtime_dt,
taken_at_source='filesystem',
# First-pass flag based on the filesystem mtime;
# metadata.extract_metadata re-runs this once
# EXIF has been parsed so a real DateTimeOriginal
# can clear the warning.
has_date_warning=has_date_warning(filepath, mtime_dt),
is_duplicate=is_dup,
is_hidden=effective_hidden,
processing_status='pending'