feat: enhance iPhone photo inbox receiver with profile management and clipboard options
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Receive JPEG uploads from iPhone Shortcuts into the workspace inbox."""
|
||||
"""Receive JPEG uploads from iPhone Shortcuts into local Mac inboxes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -7,7 +7,9 @@ import argparse
|
||||
import datetime as dt
|
||||
import json
|
||||
import os
|
||||
import shlex
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from http import HTTPStatus
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
@@ -15,7 +17,47 @@ from tempfile import NamedTemporaryFile
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
|
||||
DEFAULT_OUTPUT_DIR = Path(__file__).resolve().parents[2] / "ai" / "inbox" / "photos"
|
||||
WORKSPACE_DIR = Path(__file__).resolve().parents[2]
|
||||
DEFAULT_OPENCODE_DIR = WORKSPACE_DIR / "ai" / "inbox" / "photos"
|
||||
DEFAULT_MATTERMOST_DIR = Path.home() / "Pictures" / "iPhone Inbox"
|
||||
|
||||
CLIPBOARD_NONE = "none"
|
||||
CLIPBOARD_IMAGE = "image"
|
||||
CLIPBOARD_PATH = "path"
|
||||
CLIPBOARD_TERMINAL_PATH = "terminal-path"
|
||||
CLIPBOARD_FILE = "file"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Profile:
|
||||
name: str
|
||||
output_dir: Path
|
||||
clipboard: str
|
||||
notify: bool = True
|
||||
reveal: bool = False
|
||||
|
||||
|
||||
def env_flag(name: str, default: bool = False) -> bool:
|
||||
value = os.getenv(name)
|
||||
if value is None:
|
||||
return default
|
||||
return value.strip().lower() in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
def env_path(name: str, default: Path) -> Path:
|
||||
value = os.getenv(name)
|
||||
return Path(value).expanduser() if value else default
|
||||
|
||||
|
||||
def profile_defaults() -> dict[str, Profile]:
|
||||
opencode_dir = env_path("IPHONE_PHOTO_OPENCODE_DIR", DEFAULT_OPENCODE_DIR)
|
||||
mattermost_dir = env_path("IPHONE_PHOTO_MATTERMOST_DIR", DEFAULT_MATTERMOST_DIR)
|
||||
general_dir = env_path("IPHONE_PHOTO_GENERAL_DIR", DEFAULT_MATTERMOST_DIR)
|
||||
return {
|
||||
"opencode": Profile("opencode", opencode_dir, CLIPBOARD_TERMINAL_PATH),
|
||||
"mattermost": Profile("mattermost", mattermost_dir, CLIPBOARD_IMAGE),
|
||||
"general": Profile("general", general_dir, CLIPBOARD_NONE),
|
||||
}
|
||||
|
||||
|
||||
def timestamp() -> str:
|
||||
@@ -35,8 +77,8 @@ def looks_like_jpeg(data: bytes) -> bool:
|
||||
return data.startswith(b"\xff\xd8") and data.endswith(b"\xff\xd9")
|
||||
|
||||
|
||||
def env_flag(name: str) -> bool:
|
||||
return os.getenv(name, "").strip().lower() in {"1", "true", "yes", "on"}
|
||||
def normalize_profile(value: str) -> str:
|
||||
return value.strip().lower() or "opencode"
|
||||
|
||||
|
||||
def run_macos_action(command: list[str]) -> bool:
|
||||
@@ -63,6 +105,11 @@ def reveal_in_finder(path: Path) -> bool:
|
||||
return run_macos_action(["open", "-R", str(path)])
|
||||
|
||||
|
||||
def copy_image_to_clipboard(path: Path) -> bool:
|
||||
script = f"set the clipboard to (read (POSIX file {json.dumps(str(path))}) as JPEG picture)"
|
||||
return run_macos_action(["osascript", "-e", script])
|
||||
|
||||
|
||||
def copy_file_to_clipboard(path: Path) -> bool:
|
||||
script = f"""
|
||||
set theFile to POSIX file {json.dumps(str(path))} as alias
|
||||
@@ -73,8 +120,32 @@ end tell
|
||||
return run_macos_action(["osascript", "-e", script])
|
||||
|
||||
|
||||
def copy_text_to_clipboard(text: str) -> bool:
|
||||
script = f"set the clipboard to {json.dumps(text)}"
|
||||
return run_macos_action(["osascript", "-e", script])
|
||||
|
||||
|
||||
def terminal_path_reference(path: Path) -> str:
|
||||
return shlex.quote(str(path))
|
||||
|
||||
|
||||
def apply_clipboard_mode(mode: str, path: Path) -> bool:
|
||||
if mode == CLIPBOARD_NONE:
|
||||
return True
|
||||
if mode == CLIPBOARD_IMAGE:
|
||||
return copy_image_to_clipboard(path)
|
||||
if mode == CLIPBOARD_PATH:
|
||||
return copy_text_to_clipboard(str(path))
|
||||
if mode == CLIPBOARD_TERMINAL_PATH:
|
||||
return copy_text_to_clipboard(terminal_path_reference(path))
|
||||
if mode == CLIPBOARD_FILE:
|
||||
return copy_file_to_clipboard(path)
|
||||
print(f"unsupported clipboard mode: {mode}", flush=True)
|
||||
return False
|
||||
|
||||
|
||||
class UploadHandler(BaseHTTPRequestHandler):
|
||||
server_version = "iPhonePhotoInbox/1.0"
|
||||
server_version = "iPhonePhotoInbox/2.0"
|
||||
|
||||
def do_GET(self) -> None:
|
||||
if self.path == "/health":
|
||||
@@ -88,12 +159,20 @@ class UploadHandler(BaseHTTPRequestHandler):
|
||||
self.send_text(HTTPStatus.NOT_FOUND, "not found\n")
|
||||
return
|
||||
|
||||
query = parse_qs(parsed.query)
|
||||
expected_token = self.server.upload_token
|
||||
supplied_token = parse_qs(parsed.query).get("token", [""])[0]
|
||||
supplied_token = query.get("token", [""])[0]
|
||||
if expected_token and supplied_token != expected_token:
|
||||
self.send_text(HTTPStatus.UNAUTHORIZED, "bad token\n")
|
||||
return
|
||||
|
||||
profile_name = normalize_profile(query.get("profile", [self.server.default_profile])[0])
|
||||
profile = self.server.profiles.get(profile_name)
|
||||
if profile is None:
|
||||
known = ", ".join(sorted(self.server.profiles))
|
||||
self.send_text(HTTPStatus.BAD_REQUEST, f"unknown profile: {profile_name}; expected one of {known}\n")
|
||||
return
|
||||
|
||||
content_length = self.headers.get("Content-Length")
|
||||
if content_length is None:
|
||||
self.send_text(HTTPStatus.LENGTH_REQUIRED, "missing content length\n")
|
||||
@@ -114,25 +193,25 @@ class UploadHandler(BaseHTTPRequestHandler):
|
||||
self.send_text(HTTPStatus.UNSUPPORTED_MEDIA_TYPE, "expected jpeg\n")
|
||||
return
|
||||
|
||||
self.server.output_dir.mkdir(parents=True, exist_ok=True)
|
||||
target = unique_path(self.server.output_dir)
|
||||
with NamedTemporaryFile(dir=self.server.output_dir, delete=False) as tmp:
|
||||
output_dir = self.server.output_dir_override or profile.output_dir
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
target = unique_path(output_dir)
|
||||
with NamedTemporaryFile(dir=output_dir, delete=False) as tmp:
|
||||
tmp.write(data)
|
||||
temp_path = Path(tmp.name)
|
||||
temp_path.replace(target)
|
||||
|
||||
if self.server.notify_on_upload:
|
||||
if notify("iPhone Photo Inbox", target.name):
|
||||
if profile.notify:
|
||||
if notify("iPhone Photo Inbox", f"{profile.name}: {target.name}"):
|
||||
print("notification sent", flush=True)
|
||||
if self.server.reveal_on_upload:
|
||||
if profile.reveal:
|
||||
if reveal_in_finder(target):
|
||||
print("revealed in Finder", flush=True)
|
||||
if self.server.copy_on_upload:
|
||||
if copy_file_to_clipboard(target):
|
||||
print("copied file to clipboard", flush=True)
|
||||
if apply_clipboard_mode(profile.clipboard, target):
|
||||
print(f"clipboard mode applied: {profile.clipboard}", flush=True)
|
||||
|
||||
self.send_text(HTTPStatus.CREATED, f"{target}\n")
|
||||
print(f"saved {target}", flush=True)
|
||||
print(f"saved {target} profile={profile.name}", flush=True)
|
||||
|
||||
def log_message(self, format: str, *args: object) -> None:
|
||||
print(f"{self.address_string()} - {format % args}", flush=True)
|
||||
@@ -151,55 +230,105 @@ class UploadServer(ThreadingHTTPServer):
|
||||
self,
|
||||
server_address: tuple[str, int],
|
||||
handler_class: type[BaseHTTPRequestHandler],
|
||||
output_dir: Path,
|
||||
profiles: dict[str, Profile],
|
||||
default_profile: str,
|
||||
output_dir_override: Path | None,
|
||||
upload_token: str,
|
||||
max_bytes: int,
|
||||
notify_on_upload: bool,
|
||||
reveal_on_upload: bool,
|
||||
copy_on_upload: bool,
|
||||
) -> None:
|
||||
super().__init__(server_address, handler_class)
|
||||
self.output_dir = output_dir
|
||||
self.profiles = profiles
|
||||
self.default_profile = default_profile
|
||||
self.output_dir_override = output_dir_override
|
||||
self.upload_token = upload_token
|
||||
self.max_bytes = max_bytes
|
||||
self.notify_on_upload = notify_on_upload
|
||||
self.reveal_on_upload = reveal_on_upload
|
||||
self.copy_on_upload = copy_on_upload
|
||||
|
||||
|
||||
def apply_legacy_clipboard_overrides(profile: Profile) -> Profile:
|
||||
clipboard = profile.clipboard
|
||||
if env_flag("IPHONE_PHOTO_COPY"):
|
||||
clipboard = CLIPBOARD_IMAGE
|
||||
if env_flag("IPHONE_PHOTO_COPY_FILE"):
|
||||
clipboard = CLIPBOARD_FILE
|
||||
if env_flag("IPHONE_PHOTO_COPY_PATH"):
|
||||
clipboard = CLIPBOARD_PATH
|
||||
if env_flag("IPHONE_PHOTO_COPY_TERMINAL_PATH"):
|
||||
clipboard = CLIPBOARD_TERMINAL_PATH
|
||||
|
||||
notify_enabled = env_flag("IPHONE_PHOTO_NOTIFY", profile.notify)
|
||||
reveal_enabled = env_flag("IPHONE_PHOTO_REVEAL", profile.reveal)
|
||||
return Profile(profile.name, profile.output_dir, clipboard, notify_enabled, reveal_enabled)
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--host", default=os.getenv("IPHONE_PHOTO_HOST", "0.0.0.0"))
|
||||
parser.add_argument("--port", type=int, default=int(os.getenv("IPHONE_PHOTO_PORT", "8787")))
|
||||
parser.add_argument("--output-dir", type=Path, default=Path(os.getenv("IPHONE_PHOTO_OUTPUT_DIR", DEFAULT_OUTPUT_DIR)))
|
||||
parser.add_argument("--profile", default=os.getenv("IPHONE_PHOTO_PROFILE", "opencode"))
|
||||
parser.add_argument("--output-dir", type=Path, default=os.getenv("IPHONE_PHOTO_OUTPUT_DIR"))
|
||||
parser.add_argument("--token", default=os.getenv("IPHONE_PHOTO_TOKEN", ""))
|
||||
parser.add_argument("--max-mb", type=int, default=int(os.getenv("IPHONE_PHOTO_MAX_MB", "30")))
|
||||
parser.add_argument("--notify", action="store_true", default=env_flag("IPHONE_PHOTO_NOTIFY"))
|
||||
parser.add_argument(
|
||||
"--clipboard",
|
||||
choices=[CLIPBOARD_NONE, CLIPBOARD_IMAGE, CLIPBOARD_PATH, CLIPBOARD_TERMINAL_PATH, CLIPBOARD_FILE],
|
||||
default=os.getenv("IPHONE_PHOTO_CLIPBOARD"),
|
||||
)
|
||||
parser.add_argument("--no-notify", action="store_true", default=env_flag("IPHONE_PHOTO_NO_NOTIFY"))
|
||||
parser.add_argument("--reveal", action="store_true", default=env_flag("IPHONE_PHOTO_REVEAL"))
|
||||
parser.add_argument("--copy", action="store_true", default=env_flag("IPHONE_PHOTO_COPY"))
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def build_profiles(args: argparse.Namespace) -> dict[str, Profile]:
|
||||
profiles = {
|
||||
name: apply_legacy_clipboard_overrides(profile)
|
||||
for name, profile in profile_defaults().items()
|
||||
}
|
||||
|
||||
if args.clipboard:
|
||||
profiles = {
|
||||
name: Profile(profile.name, profile.output_dir, args.clipboard, profile.notify, profile.reveal)
|
||||
for name, profile in profiles.items()
|
||||
}
|
||||
if args.no_notify:
|
||||
profiles = {
|
||||
name: Profile(profile.name, profile.output_dir, profile.clipboard, False, profile.reveal)
|
||||
for name, profile in profiles.items()
|
||||
}
|
||||
if args.reveal:
|
||||
profiles = {
|
||||
name: Profile(profile.name, profile.output_dir, profile.clipboard, profile.notify, True)
|
||||
for name, profile in profiles.items()
|
||||
}
|
||||
return profiles
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = parse_args()
|
||||
profiles = build_profiles(args)
|
||||
default_profile = normalize_profile(args.profile)
|
||||
if default_profile not in profiles:
|
||||
known = ", ".join(sorted(profiles))
|
||||
raise SystemExit(f"unknown default profile: {default_profile}; expected one of {known}")
|
||||
|
||||
output_dir_override = Path(args.output_dir).expanduser().resolve() if args.output_dir else None
|
||||
server = UploadServer(
|
||||
(args.host, args.port),
|
||||
UploadHandler,
|
||||
args.output_dir.expanduser().resolve(),
|
||||
profiles,
|
||||
default_profile,
|
||||
output_dir_override,
|
||||
args.token,
|
||||
args.max_mb * 1024 * 1024,
|
||||
args.notify,
|
||||
args.reveal,
|
||||
args.copy,
|
||||
)
|
||||
print(f"listening on http://{args.host}:{args.port}/upload", flush=True)
|
||||
print(f"saving JPEGs to {server.output_dir}", flush=True)
|
||||
if args.notify:
|
||||
print("notifications enabled", flush=True)
|
||||
if args.reveal:
|
||||
print("Finder reveal enabled", flush=True)
|
||||
if args.copy:
|
||||
print("clipboard copy enabled", flush=True)
|
||||
print(f"default profile: {default_profile}", flush=True)
|
||||
for profile in profiles.values():
|
||||
output_dir = output_dir_override or profile.output_dir
|
||||
print(
|
||||
f"profile {profile.name}: dir={output_dir.expanduser().resolve()} "
|
||||
f"clipboard={profile.clipboard} notify={profile.notify} reveal={profile.reveal}",
|
||||
flush=True,
|
||||
)
|
||||
if not args.token:
|
||||
print("warning: no token configured; anyone on this network can upload", flush=True)
|
||||
server.serve_forever()
|
||||
|
||||
Reference in New Issue
Block a user