I Unleashed Apple's Free Local AI on My Photos Library

export PHOTOS_LIBRARY_PATH="$HOME/Pictures/Photos Library.photoslibrary"

python3 Photos-Auto-Geotag.py

python3 Photos-AI-Organizer.py

Enter Photos album name: 2026 - Trick or Treat

ALBUM: 2026 - Trick or Treat
TOTAL PHOTOS: 9
MISSING GPS: 0

GPS UPDATE MODE:
1. Tag only photographs missing GPS
2. Re-tag ALL photographs in this album

Select 1 or 2: 1
All photographs already have GPS.

Enter Photos album name: 2023 - Annapolis

ALBUM: 2023 - Annapolis
TOTAL PHOTOS: 63
MISSING GPS: 49

GPS UPDATE MODE:
1. Tag only photographs missing GPS
2. Re-tag ALL photographs in this album

Select 1 or 2: 2
RE-TAG MODE ENABLED
PHOTOGRAPHS SELECTED: 63
Existing GPS coordinates will be replaced after approval.

LOCATION MODE:
1. Use one known location for the entire album
2. Analyze each photograph individually with FM

You may also enter a location directly.

Selection or location: 2

4/63 ANALYZING: IMG_5528.HEIC
   Annapolis | Maryland State House | HIGH

0003. IMG_5528.HEIC | Annapolis | Maryland State House | 38.934000, -76.606000
Type YES to apply ALL listed locations, or anything else to cancel: YES
3/61 UPDATED: IMG_5528.HEIC

Analyze this photograph for an Apple Photos organizer.

USER REQUEST:
Find photos of the tidal basin in March 2026

PHOTO DATE:
[this photo's capture date]

Identify where this photograph was taken, the activity or event,
and visible landmarks. Do not invent details.

Return LOCATION, EVENT, EVIDENCE, CONFIDENCE, LATITUDE, LONGITUDE.

What would you like to organize? Find photos of the tidal basin in March 2026

GROUP 6/6 | 18 photos | 2026-03-29
FM AGREEMENT: 10/12 (83.3%)
PROPOSED: 2026 - Tidal Blossom Celebration

Type YES to create ALL proposed albums: YES
PROCESSED: 18/18
VERIFIED IN ALBUM: 0
NOT VERIFIED: 18
APPLESCRIPT LOOKUP ERRORS: 18
STATUS: INCOMPLETE

#!/usr/bin/env python3

import subprocess
import os
import sqlite3
import json
import re
import time
import math
import sys
from pathlib import Path
from datetime import datetime, timezone, timedelta
from collections import Counter

SOURCE_ALBUM = "Not in Album"

LIBRARY = Path(os.environ["PHOTOS_LIBRARY_PATH"]).expanduser()
DB = LIBRARY / "database/Photos.sqlite"

APPLE_EPOCH = 978307200

MIN_GROUP = 6
MAX_GAP_HOURS = 4
MAX_GROUP_SPAN_DAYS = 7
FM_SAMPLE_SIZE = 5

def log(*messages):
    print(*messages, flush=True)

def applescript(script, *args, timeout=600):
    result = subprocess.run(
        ["osascript", "-e", script, *map(str, args)],
        capture_output=True,
        text=True,
        timeout=timeout
    )

    if result.returncode:
        raise RuntimeError(
            result.stderr.strip() or result.stdout.strip()
        )

    return result.stdout.strip()

def get_smart_album_photos(period):
    """
    Query unorganized photographs within the requested date range.
    """

    start = period.get("start")
    end = period.get("end")

    if start:
        start_ts = (
            datetime.fromisoformat(start)
            .replace(tzinfo=timezone.utc)
            .timestamp() - APPLE_EPOCH
        )
    else:
        start_ts = -99999999999

    if end:
        end_ts = (
            datetime.fromisoformat(end)
            .replace(tzinfo=timezone.utc)
            .timestamp() - APPLE_EPOCH + 86400
        )
    else:
        end_ts = 99999999999

    db = sqlite3.connect(DB.as_uri() + "?mode=ro", uri=True)
    db.execute("PRAGMA query_only=ON")

    try:
        rows = db.execute("""
            SELECT a.ZUUID
            FROM ZASSET a
            WHERE a.ZDATECREATED >= ?
              AND a.ZDATECREATED < ?
              AND COALESCE(a.ZTRASHEDSTATE, 0) = 0
              AND NOT EXISTS (
                  SELECT 1
                  FROM Z_34ASSETS m
                  JOIN ZGENERICALBUM g
                    ON g.Z_PK = m.Z_34ALBUMS
                  WHERE m.Z_3ASSETS = a.Z_PK
                    AND COALESCE(g.ZTRASHEDSTATE, 0) = 0
                    AND g.ZKIND = 2
              )
        """, (start_ts, end_ts)).fetchall()

        return {row[0].upper() for row in rows}

    finally:
        db.close()

def read_photos(uuids):
    db = sqlite3.connect(DB.as_uri() + "?mode=ro", uri=True)
    db.execute("PRAGMA query_only=ON")

    try:
        photos = []

        for uuid, captured, lat, lon, filename in db.execute("""
            SELECT
                a.ZUUID,
                a.ZDATECREATED,
                a.ZLATITUDE,
                a.ZLONGITUDE,
                COALESCE(x.ZORIGINALFILENAME, a.ZFILENAME)
            FROM ZASSET a
            LEFT JOIN ZADDITIONALASSETATTRIBUTES x
              ON x.ZASSET = a.Z_PK
            WHERE COALESCE(a.ZTRASHEDSTATE, 0) = 0
        """):
            if uuid.upper() not in uuids or captured is None:
                continue

            date = datetime.fromtimestamp(
                captured + APPLE_EPOCH,
                timezone.utc
            )

            photos.append({
                "uuid": uuid,
                "date": date,
                "filename": filename,
                "latitude": lat,
                "longitude": lon
            })

        return sorted(photos, key=lambda x: x["date"])

    finally:
        db.close()

def original_path(uuid):
    folder = LIBRARY / "originals" / uuid[0].upper()

    if not folder.exists():
        return None

    for path in folder.glob(uuid + ".*"):
        if path.suffix.lower() in (
            ".heic", ".jpg", ".jpeg", ".png"
        ):
            return path

    return None


def parse_request(request, photos=None):
    import calendar

    months = {
        name.lower(): number
        for number, name in enumerate(calendar.month_name)
        if name
    }

    months.update({
        name.lower(): number
        for number, name in enumerate(calendar.month_abbr)
        if name
    })

    pattern = (
        r"\b(" +
        "|".join(map(re.escape, months)) +
        r")\s+(\d{4})\b"
    )

    match = re.search(pattern, request, re.IGNORECASE)

    if not match:
        raise SystemExit(
            "ERROR: Include a month and year, "
            "for example March 2026."
        )

    month = months[match.group(1).lower()]
    year = int(match.group(2))

    last_day = calendar.monthrange(year, month)[1]

    return {
        "start": f"{year:04d}-{month:02d}-01",
        "end": f"{year:04d}-{month:02d}-{last_day:02d}"
    }

def filter_photos(photos, period):
    start = period.get("start")
    end = period.get("end")

    if start:
        start = datetime.fromisoformat(start).date()

    if end:
        end = datetime.fromisoformat(end).date()

    return [
        photo for photo in photos
        if (start is None or photo["date"].date() >= start)
        and (end is None or photo["date"].date() <= end)
    ]


def cluster_photos(photos):
    """
    Discover short photographic events using timestamps.

    FM is not involved in this process.
    """

    from collections import Counter

    if not photos:
        return []

    photos = sorted(photos, key=lambda p: p["date"])

    MIN_PHOTOS = 6
    MAX_GAP_HOURS = 4
    MAX_EVENT_DAYS = 2
    MAX_OVERNIGHT_HOURS = 18

    # Build individual photographic bursts.
    bursts = []
    current = [photos[0]]

    for photo in photos[1:]:

        previous = current[-1]

        gap = (
            photo["date"] - previous["date"]
        ).total_seconds() / 3600

        if gap <= MAX_GAP_HOURS:
            current.append(photo)
        else:
            bursts.append(current)
            current = [photo]

    bursts.append(current)

    log("\nTEMPORAL BURSTS:", len(bursts))

    # Combine nearby bursts from the same day.
    daily = {}

    for burst in bursts:
        for photo in burst:
            day = photo["date"].date()
            daily.setdefault(day, []).append(photo)

    days = sorted(daily)

    events = []
    index = 0

    while index < len(days):

        day = days[index]
        group = sorted(
            daily[day],
            key=lambda p: p["date"]
        )

        # Consider an overnight continuation.
        if index + 1 < len(days):

            next_day = days[index + 1]

            if (next_day - day).days == 1:

                next_group = sorted(
                    daily[next_day],
                    key=lambda p: p["date"]
                )

                overnight_gap = (
                    next_group[0]["date"] -
                    group[-1]["date"]
                ).total_seconds() / 3600

                if (
                    len(group) >= MIN_PHOTOS
                    and len(next_group) >= MIN_PHOTOS
                    and 0 <= overnight_gap <= MAX_OVERNIGHT_HOURS
                ):
                    group.extend(next_group)
                    index += 1

        if len(group) >= MIN_PHOTOS:

            group.sort(key=lambda p: p["date"])

            events.append(group)

            log(
                f"  EVENT {len(events)}:",
                f"{len(group)} photos",
                group[0]["date"].strftime("%Y-%m-%d %H:%M"),
                "to",
                group[-1]["date"].strftime("%Y-%m-%d %H:%M")
            )

        index += 1

    log("\nQUALIFYING EVENT GROUPS:", len(events))

    return events

def fm_image(path, request, date):
    prompt = f"""
Analyze this photograph for an Apple Photos organizer.

USER REQUEST:
{request}

PHOTO DATE:
{date}

Identify:
- Where this photograph was taken, if recognizable.
- The activity, event, or occasion.
- Visible landmarks, signs, or contextual clues.

Do not invent details.

Return exactly:

LOCATION:
EVENT:
EVIDENCE:
CONFIDENCE: HIGH / MEDIUM / LOW / UNKNOWN
LATITUDE:
LONGITUDE:
"""

    result = subprocess.run(
        [
            "fm", "respond",
            "--no-stream",
            "--image", str(path),
            "--text", prompt
        ],
        capture_output=True,
        text=True,
        timeout=180
    )

    if result.returncode:
        raise RuntimeError(result.stderr.strip())

    fields = {}

    for key in (
        "LOCATION", "EVENT", "EVIDENCE",
        "CONFIDENCE", "LATITUDE", "LONGITUDE"
    ):
        match = re.search(
            rf"^{key}:\s*(.*)$",
            result.stdout,
            re.M | re.I
        )
        fields[key.lower()] = (
            match.group(1).strip() if match else "UNKNOWN"
        )

    return fields


def analyze_group(group, request):
    """
    FM validates and names a group.

    It cannot change the group's boundaries.
    """

    import json
    import math

    SAMPLE_SIZE = min(12, len(group))

    indices = sorted({
        round(
            i * (len(group) - 1) /
            max(1, SAMPLE_SIZE - 1)
        )
        for i in range(SAMPLE_SIZE)
    })

    observations = []

    log(
        "  GROUP SIZE:", len(group),
        "| FM SAMPLES:", len(indices)
    )

    for index in indices:

        photo = group[index]
        path = original_path(photo["uuid"])

        if not path:
            log("  ORIGINAL MISSING:", photo["filename"])
            continue

        log("  FM:", photo["filename"])

        try:
            result = fm_image(
                path,
                request,
                photo["date"].isoformat()
            )

            observations.append(result)

            log(
                "    LOCATION:",
                result.get("location", "UNKNOWN"),
                "| EVENT:",
                result.get("event", "UNKNOWN"),
                "| CONFIDENCE:",
                result.get("confidence", "UNKNOWN")
            )

        except Exception as error:
            log("  FM ERROR:", error)

    if len(observations) < 5:
        log("  SKIPPED: Insufficient visual observations")
        return None

    summary = "\n".join(
        json.dumps(o, ensure_ascii=False)
        for o in observations
    )

    year = group[0]["date"].year

    prompt = f"""
You are naming an Apple Photos event album.

The script has ALREADY identified the event group
using photographic timestamps.

You must NOT regroup photographs or change dates.

USER REQUEST:
{request}

YEAR:
{year}

EVENT DATES:
{group[0]['date']} to {group[-1]['date']}

TOTAL PHOTOGRAPHS:
{len(group)}

VISUAL OBSERVATIONS:
{summary}

Determine whether the observations describe
one coherent event, outing, celebration, or trip.

Do not invent locations, landmarks, or activities.

Return exactly:

EVENT:
LOCATION:
SUPPORTING_SAMPLES:
ALBUM_NAME:
DESCRIPTION:

SUPPORTING_SAMPLES must be an integer counting
observations that clearly support the SAME event.

ALBUM_NAME must follow:
{year} - Descriptive Name

If the evidence is insufficient, return UNKNOWN.

Do not return JSON.
"""

    result = subprocess.run(
        [
            "fm",
            "respond",
            "--no-stream",
            "--text",
            prompt
        ],
        capture_output=True,
        text=True,
        timeout=180
    )

    if result.returncode:
        raise RuntimeError(result.stderr.strip())

    def field(name):
        match = re.search(
            rf"^{name}:\s*(.*)$",
            result.stdout,
            re.MULTILINE | re.IGNORECASE
        )

        return match.group(1).strip() if match else "UNKNOWN"

    event = field("EVENT")
    location = field("LOCATION")
    name = field("ALBUM_NAME")
    description = field("DESCRIPTION")

    try:
        supporting = int(field("SUPPORTING_SAMPLES"))
    except ValueError:
        supporting = 0

    if supporting < 0 or supporting > len(observations):
        log(
            "  FM RETURNED INVALID SUPPORT COUNT:",
            supporting,
            "for",
            len(observations),
            "samples"
        )
        return None

    agreement = supporting / len(observations)

    log(
        "  FM AGREEMENT:",
        f"{supporting}/{len(observations)}",
        f"({agreement:.1%})"
    )

    if agreement < 0.80:
        log("  SKIPPED: Below 80% agreement")
        return None

    if (
        event.upper() == "UNKNOWN"
        or name.upper() == "UNKNOWN"
    ):
        log("  SKIPPED: Event not identified")
        return None

    if not re.fullmatch(
        rf"{year} - .+",
        name
    ):
        log("  SKIPPED: Invalid album name")
        return None

    return {
        "accepted": True,
        "agreement": round(agreement, 3),
        "album_name": name,
        "description": description,
        "event": event,
        "location": location,
        "latitude": None,
        "longitude": None,
        "photos": group,
        "observations": observations
    }

CREATE_ALBUM = '''
on run argv
    tell application "Photos"
        with timeout of 300 seconds
            set albumName to item 1 of argv
            set matches to every album whose name is albumName

            if (count of matches) > 0 then
                return id of item 1 of matches
            end if

            set newAlbum to make new album named albumName
            return id of newAlbum
        end timeout
    end tell
end run
'''

ADD_PHOTO = '''
on run argv
    tell application "Photos"
        with timeout of 300 seconds
            set photoID to item 1 of argv
            set albumID to item 2 of argv

            set targetPhoto to media item id photoID
            set targetAlbum to album id albumID

            add {targetPhoto} to targetAlbum

            return "ADDED"
        end timeout
    end tell
end run
'''

SET_GPS = '''
on run argv
    tell application "Photos"
        with timeout of 300 seconds
            set photoID to item 1 of argv
            set lat to (item 2 of argv) as real
            set lon to (item 3 of argv) as real

            set targetPhoto to media item id photoID
            set location of targetPhoto to {lat, lon}

            return "GPS UPDATED"
        end timeout
    end tell
end run
'''


def verify_album_membership(album_id, expected_uuids):
    """
    Read-only verification against the Photos database.

    Returns:
        present: UUIDs currently in the destination album
        missing: UUIDs not currently in the destination album
    """

    album_uuid = album_id.split("/")[0].upper()

    db = sqlite3.connect(
        DB.as_uri() + "?mode=ro",
        uri=True,
        timeout=60
    )

    db.execute("PRAGMA query_only=ON")

    try:
        rows = db.execute("""
            SELECT UPPER(a.ZUUID)
            FROM ZGENERICALBUM g
            JOIN Z_34ASSETS m
              ON m.Z_34ALBUMS = g.Z_PK
            JOIN ZASSET a
              ON a.Z_PK = m.Z_3ASSETS
            WHERE UPPER(g.ZUUID) = ?
              AND COALESCE(g.ZTRASHEDSTATE, 0) = 0
              AND COALESCE(a.ZTRASHEDSTATE, 0) = 0
        """, (album_uuid,)).fetchall()

        actual = {row[0] for row in rows}

        expected = {
            uuid.upper()
            for uuid in expected_uuids
        }

        return expected & actual, expected - actual

    finally:
        db.close()

def main():
    log("========== APPLE PHOTOS AI ORGANIZER ==========")

    request = input("\nWhat would you like to organize? ").strip()

    if not request:
        raise SystemExit("No request entered.")

    log("\nREADING SMART ALBUM: " + SOURCE_ALBUM)

    period = parse_request(request, [])

    log("DATE FILTER: " + str(period))

    log("QUERYING PHOTOS DATABASE...")
    uuids = get_smart_album_photos(period)

    log("SMART ALBUM PHOTOS: " + str(len(uuids)))

    photos = read_photos(uuids)

    log("PHOTOS FOUND IN DATABASE: " + str(len(photos)))

    selected = filter_photos(photos, period)

    log("PHOTOS MATCHING REQUEST: " + str(len(selected)))

    groups = cluster_photos(selected)

    log("CANDIDATE GROUPS: " + str(len(groups)))

    proposals = []

    for index, group in enumerate(groups, 1):
        log("")
        log(
            f"GROUP {index}/{len(groups)} | "
            f"{len(group)} photos | "
            f"{group[0]['date'].date()} "
            f"to {group[-1]['date'].date()}"
        )

        try:
            proposal = analyze_group(group, request)

            if proposal:
                proposals.append(proposal)
                log("  PROPOSED: " + proposal["album_name"])
            else:
                log("  SKIPPED: No coherent high-confidence event")

        except Exception as error:
            log("  ERROR: " + str(error))

    log("\n========== PROPOSED ALBUMS ==========")

    if not proposals:
        log("No qualifying albums found.")
        return

    for index, proposal in enumerate(proposals, 1):
        log("")
        log(f"{index}. {proposal['album_name']}")
        log("   PHOTOS: " + str(len(proposal["photos"])))
        log("   AGREEMENT: " + str(proposal["agreement"]))
        log("   REASON: " + proposal["description"])
        log(
            "   GPS: "
            + str(proposal.get("latitude"))
            + ", "
            + str(proposal.get("longitude"))
        )

    approval = input(
        "\nType YES to create ALL proposed albums: "
    ).strip()

    if approval != "YES":
        log("CANCELLED: No changes made.")
        return

    log("\n========== ORGANIZING PHOTOS ==========")

    for proposal in proposals:
        name = proposal["album_name"]

        log("\nALBUM: " + name)

        album_id = applescript(CREATE_ALBUM, name)

        expected = {
            photo["uuid"].upper()
            for photo in proposal["photos"]
        }

        total = len(expected)

        print("PHOTOS SELECTED:", total)
        print("ADDING PHOTOS...", flush=True)

        successful_requests = 0
        lookup_errors = 0
        other_errors = 0

        for index, photo in enumerate(proposal["photos"], 1):

            photo_id = photo["uuid"] + "/L0/001"

            try:
                result = applescript(
                    ADD_PHOTO,
                    photo_id,
                    album_id
                )

                if result == "ADDED":
                    successful_requests += 1
                else:
                    other_errors += 1

            except Exception as error:
                message = str(error)

                if (
                    "Can't get media item id" in message
                    or "Can’t get media item id" in message
                    or "(-1728)" in message
                ):
                    lookup_errors += 1
                else:
                    other_errors += 1

            if index % 25 == 0 or index == len(proposal["photos"]):
                print(
                    f"  PROCESSED: {index}/{len(proposal['photos'])}",
                    flush=True
                )

        print("\nVERIFYING ALBUM MEMBERSHIP...", flush=True)

        present = set()
        missing = expected

        for attempt in range(1, 6):

            try:
                present, missing = verify_album_membership(
                    album_id,
                    expected
                )

            except Exception as error:
                print(
                    "VERIFICATION ERROR:",
                    str(error)
                )
                break

            if not missing:
                break

            if attempt < 5:
                time.sleep(2)

        print()
        print("========== ALBUM RESULTS ==========")
        print("ALBUM:", name)
        print("EXPECTED PHOTOS:", total)
        print("VERIFIED IN ALBUM:", len(present))
        print("NOT VERIFIED:", len(missing))
        print("APPLESCRIPT LOOKUP ERRORS:", lookup_errors)
        print("OTHER APPLESCRIPT ERRORS:", other_errors)

        if not missing:
            print("STATUS: COMPLETE")
        else:
            print("STATUS: INCOMPLETE")
            print(
                "Some photographs could not be verified "
                "in the destination album."
            )

    log("\n========== COMPLETE ==========")
    log("PHOTOS DELETED: 0")
    log("ALBUMS DELETED: 0")
    log("DATABASE DIRECTLY MODIFIED: NO")
    log("PACKAGES INSTALLED: NONE")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3

import sqlite3
import subprocess
import os
import json
import re
import time
from pathlib import Path
from datetime import datetime, timezone

LIBRARY = Path(os.environ["PHOTOS_LIBRARY_PATH"]).expanduser()
DB = LIBRARY / "database/Photos.sqlite"
APPLE_EPOCH = 978307200

WORK = Path.home() / "Desktop" / "Photos-Geotag-Work"
WORK.mkdir(exist_ok=True)

def run_applescript(script, *args, timeout=300):
    result = subprocess.run(
        ["osascript", "-e", script, *map(str, args)],
        capture_output=True,
        text=True,
        timeout=timeout
    )
    if result.returncode:
        raise RuntimeError(result.stderr.strip() or result.stdout.strip())
    return result.stdout.strip()

def original_path(uuid):
    folder = LIBRARY / "originals" / uuid[0].upper()
    for ext in (".heic", ".HEIC", ".jpg", ".JPG", ".jpeg", ".JPEG", ".png", ".PNG"):
        path = folder / (uuid + ext)
        if path.is_file():
            return path
    return None

def capture_date(value):
    if value is None:
        return "UNKNOWN"
    return datetime.fromtimestamp(
        value + APPLE_EPOCH, timezone.utc
    ).isoformat()

def missing_gps(lat, lon):
    return (
        lat is None or lon is None
        or not (-90 <= lat <= 90 and -180 <= lon <= 180)
        or (lat, lon) == (0, 0)
    )

def get_album():
    name = input("\nEnter Photos album name: ").strip()
    if not name:
        raise SystemExit("No album entered.")

    db = sqlite3.connect(DB.as_uri() + "?mode=ro", uri=True)
    db.execute("PRAGMA query_only=ON")

    try:
        matches = db.execute("""
            SELECT Z_PK, ZUUID, ZTITLE
            FROM ZGENERICALBUM
            WHERE ZTITLE = ?
              AND COALESCE(ZTRASHEDSTATE, 0) = 0
        """, (name,)).fetchall()

        if len(matches) != 1:
            raise SystemExit(
                f"Expected exactly one album named {name!r}; found {len(matches)}."
            )

        pk, album_uuid, album_name = matches[0]

        rows = db.execute("""
            SELECT DISTINCT
                a.ZUUID,
                a.ZDATECREATED,
                a.ZLATITUDE,
                a.ZLONGITUDE,
                x.ZORIGINALFILENAME
            FROM Z_34ASSETS m
            JOIN ZASSET a ON a.Z_PK = m.Z_3ASSETS
            LEFT JOIN ZADDITIONALASSETATTRIBUTES x
              ON x.ZASSET = a.Z_PK
            WHERE m.Z_34ALBUMS = ?
              AND COALESCE(a.ZTRASHEDSTATE, 0) = 0
            ORDER BY a.ZDATECREATED, a.ZUUID
        """, (pk,)).fetchall()

        photos = []
        all_photos = []

        for uuid, date, lat, lon, filename in rows:
            item = {
                "uuid": uuid,
                "filename": filename or uuid,
                "date": capture_date(date)
            }

            all_photos.append(item)

            if missing_gps(lat, lon):
                photos.append(item)

        return album_name, album_uuid, len(rows), photos, all_photos

    finally:
        db.close()

def analyze(photo, album, date):
    prompt = f"""
Analyze this actual photograph for geographic identification.

KNOWN PHOTO ALBUM:
{album}

PHOTO CAPTURE DATE:
{date}

The album provides context about the event or trip.

Examine the actual photograph carefully.

Identify the visible geographic location using:
- Recognizable monuments and landmarks
- Distinctive architecture
- Visible signs and inscriptions
- Surrounding buildings and geography
- Context from the album and capture date

Identify the landmark and city before estimating coordinates.

Use the album context to distinguish visually similar
locations, but do not invent evidence or assume that
every photograph must match the album.

If the photograph is consistent with the album's
known event or location, prefer that interpretation
over an unrelated speculative location.

Do not invent landmarks, cities, or coordinates.

Return exactly these fields, one per line:

COUNTRY:
REGION:
CITY:
LANDMARK:
EVIDENCE:
CONFIDENCE: HIGH / MEDIUM / LOW / UNKNOWN
LATITUDE:
LONGITUDE:

Coordinates must use signed decimal degrees.

North latitude is positive.
South latitude is negative.
East longitude is positive.
West longitude is negative.

For example, western longitude must have a minus sign.

If the location cannot be identified reliably,
return UNKNOWN rather than guessing.
"""

    result = subprocess.run(
        [
            "fm", "respond",
            "--no-stream",
            "--image", str(photo),
            "--text", prompt
        ],
        capture_output=True,
        text=True,
        timeout=180
    )

    if result.returncode:
        raise RuntimeError(result.stderr.strip() or result.stdout.strip())

    raw = result.stdout

    fields = {}

    for key in (
        "COUNTRY", "REGION", "CITY", "LANDMARK",
        "EVIDENCE", "CONFIDENCE", "LATITUDE", "LONGITUDE"
    ):
        match = re.search(
            rf"^{key}:\s*(.*)$",
            raw,
            re.MULTILINE | re.IGNORECASE
        )
        fields[key.lower()] = match.group(1).strip() if match else "UNKNOWN"

    return fields

def coordinate(value, low, high):
    import math
    import re

    if value is None:
        return None

    text = str(value).strip().upper()

    if text in ("", "UNKNOWN", "NONE", "NULL", "N/A"):
        return None

    match = re.fullmatch(
        r"([+-]?\d+(?:\.\d+)?)\s*°?\s*([NSEW])?",
        text
    )

    if not match:
        return None

    number = float(match.group(1))
    direction = match.group(2)

    if direction in ("S", "W"):
        number = -abs(number)
    elif direction in ("N", "E"):
        number = abs(number)

    if not math.isfinite(number):
        return None

    if not low <= number <= high:
        return None

    return number


SET_LOCATION = '''
on run argv
    set photoID to item 1 of argv
    set newLatitude to (item 2 of argv) as real
    set newLongitude to (item 3 of argv) as real

    tell application "Photos"
        with timeout of 300 seconds
            set targetPhoto to media item id photoID
            set location of targetPhoto to {newLatitude, newLongitude}
            return "UPDATED"
        end timeout
    end tell
end run
'''





def trusted_album_location(album, photos, known_location=None):
    import math

    location = known_location or input(
        "\nEnter known location: "
    ).strip()

    if not location:
        print("CANCELLED")
        return

    print("\n========== FM TRUSTED LOCATION ==========")
    print("ALBUM:", album)
    print("EXPECTED LOCATION:", location)
    print("PHOTOS SELECTED:", len(photos))

    sample_count = min(5, len(photos))

    indices = sorted({
        round(i * (len(photos) - 1) / max(1, sample_count - 1))
        for i in range(sample_count)
    })

    def field(text, name):
        match = re.search(
            rf"^\s*{re.escape(name)}:\s*(.*)$",
            text,
            re.MULTILINE | re.IGNORECASE
        )
        return match.group(1).strip() if match else "UNKNOWN"

    def ask_fm(prompt, image=None):
        command = ["fm", "respond", "--no-stream"]

        if image:
            command.extend(["--image", str(image)])

        command.extend(["--text", prompt])

        result = subprocess.run(
            command,
            capture_output=True,
            text=True,
            timeout=180
        )

        if result.returncode:
            raise RuntimeError(
                result.stderr.strip() or result.stdout.strip()
            )

        return result.stdout

    print("\n========== VALIDATING SAMPLE PHOTOS ==========")

    matches = 0
    contradictions = 0
    unknown = 0

    for index, photo_index in enumerate(indices, 1):
        item = photos[photo_index]
        path = original_path(item["uuid"])

        if not path:
            print("ORIGINAL MISSING:", item["filename"])
            unknown += 1
            continue

        print(
            f"{index}/{len(indices)} ANALYZING:",
            item["filename"],
            flush=True
        )

        prompt = f"""
Examine this photograph.

KNOWN ALBUM:
{album}

EXPECTED LOCATION:
{location}

Determine whether the photograph is visually
consistent with the expected location.

Nearby landmarks and buildings may be consistent.

Do not assume a match solely from the album name.
Do not guess GPS coordinates.

Return exactly:

MATCH: YES / NO / UNKNOWN
IDENTIFIED LOCATION:
EVIDENCE:
"""

        try:
            raw = ask_fm(prompt, path)
            match = field(raw, "MATCH").upper()

            print("MATCH:", match)
            print("IDENTIFIED:", field(raw, "IDENTIFIED LOCATION"))
            print("EVIDENCE:", field(raw, "EVIDENCE"))

            if match == "YES":
                matches += 1
            elif match == "NO":
                contradictions += 1
            else:
                unknown += 1

        except Exception as error:
            print("FM ERROR:", error)
            unknown += 1

    print("\n========== SAMPLE RESULTS ==========")
    print("MATCHES:", matches)
    print("CONTRADICTIONS:", contradictions)
    print("UNKNOWN:", unknown)

    required = math.ceil(len(indices) * 0.8)

    if matches < required or contradictions:
        print("STOP: Insufficient location agreement.")
        print("No photographs modified.")
        return

    print("\n========== RESOLVING LOCATION ==========")

    prompt = f"""
Identify approximate GPS coordinates for this
specific named geographic location:

{location}

Use the actual landmark, not the city center.

The location may be anywhere in the world.

Use signed decimal degrees:
North positive, South negative.
East positive, West negative.

Return exactly:

LOCATION:
LATITUDE:
LONGITUDE:
"""

    try:
        raw = ask_fm(prompt)
    except Exception as error:
        print("FM ERROR:", error)
        return

    resolved = field(raw, "LOCATION")

    latitude = coordinate(field(raw, "LATITUDE"), -90, 90)
    longitude = coordinate(field(raw, "LONGITUDE"), -180, 180)

    print("\n========== FM LOCATION PROPOSAL ==========")
    print("LOCATION:", resolved)
    print("LATITUDE:", latitude)
    print("LONGITUDE:", longitude)

    print("\nFM coordinates are estimates.")
    print("Review both numbers, especially longitude signs.")

    print("\nCOORDINATE OPTIONS:")
    print("1. Accept FM coordinates")
    print("2. Enter corrected coordinates")
    print("3. Cancel")

    choice = input("\nSelection: ").strip()

    if choice == "1":
        if latitude is None or longitude is None:
            print("STOP: FM did not provide valid coordinates.")
            return

    elif choice == "2":
        try:
            latitude = float(
                input("Correct latitude: ").strip()
            )
            longitude = float(
                input("Correct longitude: ").strip()
            )
        except ValueError:
            print("STOP: Invalid coordinates.")
            return

    else:
        print("CANCELLED.")
        return

    if not (
        math.isfinite(latitude)
        and math.isfinite(longitude)
        and -90 <= latitude <= 90
        and -180 <= longitude <= 180
    ):
        print("STOP: Coordinates outside valid ranges.")
        return

    print("\n========== FINAL GPS CONFIRMATION ==========")
    print("ALBUM:", album)
    print("LOCATION:", location)
    print("PHOTOS TO UPDATE:", len(photos))
    print("LATITUDE:", latitude)
    print("LONGITUDE:", longitude)
    print(
        "LONGITUDE HEMISPHERE:",
        "WEST" if longitude < 0 else "EAST"
    )

    print("\nAll selected photos will receive these coordinates.")

    approval = input(
        "\nType YES to apply GPS coordinates: "
    ).strip()

    if approval != "YES":
        print("CANCELLED: No changes made.")
        return

    print("\n========== APPLYING GPS ==========")

    updated = 0
    errors = 0

    for index, item in enumerate(photos, 1):
        photo_id = item["uuid"] + "/L0/001"

        try:
            result = run_applescript(
                SET_LOCATION,
                photo_id,
                latitude,
                longitude
            )

            if result == "UPDATED":
                updated += 1
                print(
                    f"{index}/{len(photos)} UPDATED:",
                    item["filename"],
                    flush=True
                )
            else:
                errors += 1
                print(
                    "UNEXPECTED RESULT:",
                    item["filename"],
                    result
                )

        except Exception as error:
            errors += 1
            print("ERROR:", item["filename"], error)

    print("\n========== COMPLETE ==========")
    print("GPS UPDATE REQUESTS SUCCEEDED:", updated)
    print("ERRORS:", errors)
    print("PHOTOS DELETED: 0")
    print("ALBUMS DELETED: 0")
    print("DATABASE DIRECTLY MODIFIED: NO")

def main():
    print("========== APPLE PHOTOS AUTO-GEOTAG ==========")

    album, album_uuid, total, photos, all_photos = get_album()

    print("\nALBUM:", album)
    print("TOTAL PHOTOS:", total)
    print("MISSING GPS:", len(photos))

    print()
    print("GPS UPDATE MODE:")
    print("1. Tag only photographs missing GPS")
    print("2. Re-tag ALL photographs in this album")
    print()

    gps_mode = input("Select 1 or 2: ").strip()

    if gps_mode == "2":
        photos = all_photos
        print()
        print("RE-TAG MODE ENABLED")
        print("PHOTOGRAPHS SELECTED:", len(photos))
        print("Existing GPS coordinates will be replaced after approval.")

    elif gps_mode == "1":
        if not photos:
            print("All photographs already have GPS.")
            return

    else:
        print("CANCELLED.")
        return


    print()
    print("LOCATION MODE:")
    print("1. Use one known location for the entire album")
    print("2. Analyze each photograph individually with FM")
    print()
    print("You may also enter a location directly.")

    mode = input("\nSelection or location: ").strip()

    if mode == "1":
        trusted_album_location(album, photos)
        return

    if mode != "2":
        if not mode:
            print("CANCELLED.")
            return

        trusted_album_location(album, photos, known_location=mode)
        return


    safe_album = re.sub(r"[^A-Za-z0-9_-]+", "_", album)
    cache = WORK / f"{safe_album}-FM-results.jsonl"
    completed = {}

    if cache.exists():
        with cache.open("r", encoding="utf-8") as previous:
            for line in previous:
                try:
                    record = json.loads(line)
                    if "uuid" in record:
                        completed[record["uuid"]] = record
                except (ValueError, KeyError):
                    continue

    print("\n========== ANALYZING PHOTOGRAPHS ==========")

    with cache.open("a", encoding="utf-8") as output:
        for index, item in enumerate(photos, 1):
            uuid = item["uuid"]

            if uuid in completed:
                continue

            path = original_path(uuid)

            if not path:
                print(f"{index}/{len(photos)} ORIGINAL NOT FOUND: {item['filename']}")
                continue

            print(
                f"{index}/{len(photos)} ANALYZING: {item['filename']}",
                flush=True
            )

            try:
                result = analyze(path, album, item["date"])

                record = {**item, **result}

                completed[uuid] = record

                output.write(json.dumps(record, ensure_ascii=False) + "\n")
                output.flush()

                print(
                    "  ",
                    result["city"],
                    "|",
                    result["landmark"],
                    "|",
                    result["confidence"],
                    flush=True
                )

            except Exception as error:
                print("  ERROR:", error, flush=True)

    proposals = []

    for item in photos:
        record = completed.get(item["uuid"])
        if not record:
            continue

        lat = coordinate(record.get("latitude"), -90, 90)
        lon = coordinate(record.get("longitude"), -180, 180)

        if (
            record.get("confidence", "").upper() == "HIGH"
            and lat is not None
            and lon is not None
        ):
            proposals.append((record, lat, lon))

    print("\n========== PROPOSED GPS LOCATIONS ==========")
    print("HIGH-CONFIDENCE PROPOSALS:", len(proposals))
    print("UNRESOLVED:", len(photos) - len(proposals))

    for index, (record, lat, lon) in enumerate(proposals, 1):
        print(
            f"{index:04d}. {record['filename']} | "
            f"{record['city']} | {record['landmark']} | "
            f"{lat:.6f}, {lon:.6f}"
        )

    if not proposals:
        print("\nNo high-confidence locations available.")
        return

    print("\nWARNING: AI locations may be incorrect.")
    print("Review the proposed locations before approving.")

    approval = input(
        "\nType YES to apply ALL listed locations, or anything else to cancel: "
    ).strip()

    if approval != "YES":
        print("CANCELLED. No locations changed.")
        return

    print("\n========== APPLYING LOCATIONS ==========")

    updated = 0
    errors = 0

    for index, (record, lat, lon) in enumerate(proposals, 1):
        photo_id = record["uuid"] + "/L0/001"

        try:
            result = run_applescript(
                SET_LOCATION,
                photo_id,
                lat,
                lon
            )

            if result == "UPDATED":
                updated += 1
                print(
                    f"{index}/{len(proposals)} UPDATED: "
                    f"{record['filename']}",
                    flush=True
                )
            else:
                errors += 1
                print("UNEXPECTED RESULT:", result)

        except Exception as error:
            errors += 1
            print("ERROR:", record["filename"], error)

    print("\n========== COMPLETE ==========")
    print("ALBUM:", album)
    print("LOCATIONS UPDATED:", updated)
    print("ERRORS:", errors)
    print("UNRESOLVED:", len(photos) - len(proposals))
    print("PHOTOS DELETED: 0")
    print("ALBUMS DELETED: 0")
    print("DATABASE DIRECTLY MODIFIED: NO")
    print("PACKAGES INSTALLED: NONE")

if __name__ == "__main__":
    main()