#!/usr/bin/env python3 """ photophetch-ios-helper.py Async wrapper around pymobiledevice3 AFC for PhotoPhetch iOS connectivity. Called by IosConnector as a subprocess. Usage: python3 photophetch-ios-helper.py list-devices python3 photophetch-ios-helper.py list-photos python3 photophetch-ios-helper.py pull /DCIM/148APPLE/IMG_001.HEIC /tmp/dest.heic python3 photophetch-ios-helper.py delete /DCIM/148APPLE/IMG_001.HEIC Output is JSON on stdout. Errors go to stderr with exit code 1. """ import sys import json import os import asyncio from datetime import datetime MEDIA_EXTENSIONS = { 'jpg', 'jpeg', 'heic', 'heif', 'avif', 'webp', 'png', 'dng', 'raw', 'arw', 'cr2', 'cr3', 'nef', 'orf', 'rw2', 'tiff', 'tif', 'bmp', 'mp4', 'm4v', 'mov', 'mpeg', 'mpg', '3gp', '3g2', 'avi', 'mkv', 'webm', } VIDEO_EXTENSIONS = { 'mp4', 'm4v', 'mov', 'mpeg', 'mpg', '3gp', '3g2', 'avi', 'mkv', 'webm', } async def get_lockdown(): from pymobiledevice3.lockdown import create_using_usbmux return await create_using_usbmux() async def get_afc(lockdown): from pymobiledevice3.services.afc import AfcService return AfcService(lockdown) async def cmd_list_devices(): try: lockdown = await get_lockdown() result = [{ 'udid': lockdown.udid, 'name': lockdown.display_name or lockdown.udid, }] print(json.dumps(result)) except Exception as e: print(f'Error: {e}', file=sys.stderr) # Return empty list rather than failing — device not connected print(json.dumps([])) async def cmd_list_photos(): try: lockdown = await get_lockdown() afc = await get_afc(lockdown) try: dcim_entries = await afc.listdir('/DCIM') except Exception as e: print(f'Error listing DCIM: {e}', file=sys.stderr) print(json.dumps([])) return photos = [] debug_logged = False # log raw stat for first file to help debug date issues for folder in sorted(dcim_entries): folder_path = f'/DCIM/{folder}' try: entries = await afc.listdir(folder_path) except Exception: continue for filename in entries: ext = filename.rsplit('.', 1)[-1].lower() if '.' in filename else '' if ext not in MEDIA_EXTENSIONS: continue device_path = f'{folder_path}/{filename}' # Get file metadata via stat size_bytes = 0 date_iso = None try: info = await afc.stat(device_path) if not debug_logged: print(f'DEBUG stat: {dict(info)}', file=sys.stderr) debug_logged = True size_bytes = int(info.get('st_size', 0)) # st_mtime is a datetime.datetime object in this version dt = info.get('st_mtime') or info.get('datetime') or info.get('st_birthtime') if dt is not None: if hasattr(dt, 'isoformat'): date_iso = dt.isoformat() elif isinstance(dt, (list, tuple)) and len(dt) >= 6: date_iso = datetime(int(dt[0]), int(dt[1]), int(dt[2]), int(dt[3]), int(dt[4]), int(dt[5])).isoformat() except Exception: pass photos.append({ 'devicePath': device_path, 'filename': filename, 'sizeBytes': size_bytes, 'dateIso': date_iso, 'sourceFolder': folder, 'mediaType': 'VIDEO' if ext in VIDEO_EXTENSIONS else 'PHOTO', 'isHeic': ext in ('heic', 'heif'), }) print(json.dumps(photos)) except Exception as e: print(f'Error: {e}', file=sys.stderr) sys.exit(1) async def cmd_pull(device_path: str, local_path: str): try: lockdown = await get_lockdown() afc = await get_afc(lockdown) os.makedirs(os.path.dirname(os.path.abspath(local_path)), exist_ok=True) # afc.pull(src, dst) — copies remote file to local path await afc.pull(device_path, local_path) print(json.dumps({'ok': True, 'path': local_path})) except Exception as e: print(f'Error: {e}', file=sys.stderr) sys.exit(1) async def cmd_pull_thumb(device_path: str, local_path: str): """Pull first 512KB — enough for EXIF thumbnail extraction via sips.""" try: lockdown = await get_lockdown() afc = await get_afc(lockdown) os.makedirs(os.path.dirname(os.path.abspath(local_path)), exist_ok=True) try: # fread(path, size, offset) reads partial file — much faster than full pull data = await afc.fread(device_path, 512 * 1024, 0) except Exception: # fread may not exist or may fail — fall back to full pull await afc.pull(device_path, local_path) print(json.dumps({'ok': True, 'path': local_path})) return with open(local_path, 'wb') as f: f.write(data) print(json.dumps({'ok': True, 'path': local_path})) except Exception as e: print(f'Error: {e}', file=sys.stderr) sys.exit(1) async def cmd_batch_thumbs(pairs_json: str): """ Pull thumbnails for multiple files in a single AFC session. Input: JSON array of [device_path, local_path] pairs. Output: JSON array of {devicePath, ok} results. Strategy by format: - HEIC/HEIF: read first 1MB (thumbnail track is near file start), extract embedded thumbnail via sips. Falls back to full pull if needed. - JPEG/PNG: read first 512KB (sufficient for sips resample). """ try: pairs = json.loads(pairs_json) lockdown = await get_lockdown() afc = await get_afc(lockdown) results = [] for device_path, local_path in pairs: ext = device_path.rsplit('.', 1)[-1].lower() if '.' in device_path else '' is_heic = ext in ('heic', 'heif') try: os.makedirs(os.path.dirname(os.path.abspath(local_path)), exist_ok=True) if is_heic: # HEIC: pull full file — sips needs the full image to convert to JPEG # Cached permanently so only transferred once await afc.pull(device_path, local_path) else: # JPEG/PNG: 512KB partial read is sufficient for sips resample try: data = await afc.fread(device_path, 512 * 1024, 0) with open(local_path, 'wb') as f: f.write(data) except Exception: await afc.pull(device_path, local_path) results.append({'devicePath': device_path, 'ok': True}) except Exception as e: results.append({'devicePath': device_path, 'ok': False, 'error': str(e)}) print(json.dumps(results)) except Exception as e: print(f'Error: {e}', file=sys.stderr) sys.exit(1) async def cmd_delete(device_path: str): try: lockdown = await get_lockdown() afc = await get_afc(lockdown) await afc.rm(device_path) print(json.dumps({'ok': True})) except Exception as e: print(f'Error: {e}', file=sys.stderr) sys.exit(1) if __name__ == '__main__': if len(sys.argv) < 2: print('Usage: photophetch-ios-helper.py [args]', file=sys.stderr) sys.exit(1) command = sys.argv[1] if command == 'list-devices': asyncio.run(cmd_list_devices()) elif command == 'list-photos': asyncio.run(cmd_list_photos()) elif command == 'pull' and len(sys.argv) == 4: asyncio.run(cmd_pull(sys.argv[2], sys.argv[3])) elif command == 'pull-thumb' and len(sys.argv) == 4: asyncio.run(cmd_pull_thumb(sys.argv[2], sys.argv[3])) elif command == 'batch-thumbs' and len(sys.argv) == 3: asyncio.run(cmd_batch_thumbs(sys.argv[2])) elif command == 'delete' and len(sys.argv) == 3: asyncio.run(cmd_delete(sys.argv[2])) else: print(f'Unknown command: {command}', file=sys.stderr) sys.exit(1)