Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 120 additions & 14 deletions apps/camera/photo_intelligence/nc_photo_list.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
import argparse



def get_env(key, default=None):
val = os.getenv(key)
if val is None:
Expand All @@ -34,6 +33,14 @@ def validate_env():
return url, user, password, photo_dir


def _load_processed_set(processed_log):
processed = set()
if processed_log and os.path.exists(processed_log):
with open(processed_log, "r", encoding="utf-8") as f:
processed = set(l.strip() for l in f if l.strip())
return processed


def parse_exif_pillow(data):
"""Return shooting date and location from image bytes."""
try:
Expand Down Expand Up @@ -123,6 +130,91 @@ def parse_exif_exiftool(data):
return date_taken, location


def list_local_photos(
directory,
progress_cb=None,
exif_method="exiftool",
measure_speed=False,
processed_log=None,
):
"""Walk a local directory and return metadata for JPEG files."""

files = []
processed = _load_processed_set(processed_log)

for root, _dirs, filenames in os.walk(directory):
rel_root = os.path.relpath(root, directory)
for name in filenames:
if not name.lower().endswith((".jpg", ".jpeg")):
continue

rel_path = name if rel_root == "." else os.path.join(rel_root, name)
if rel_path in processed:
continue

if progress_cb:
progress_cb(rel_path)
else:
print(f"Scanning {rel_path}")

full_path = os.path.join(root, name)
try:
with open(full_path, "rb") as f:
data = f.read()
except Exception:
data = None

date_taken = None
location = None
timing = {}
if data:
if measure_speed:
start = time.perf_counter()
dt_pillow, loc_pillow = parse_exif_pillow(data)
pillow_time = time.perf_counter() - start

start = time.perf_counter()
dt_tool, loc_tool = parse_exif_exiftool(data)
tool_time = time.perf_counter() - start

timing = {
"pillow_ms": round(pillow_time * 1000, 3),
"exiftool_ms": round(tool_time * 1000, 3),
"diff_ms": round((tool_time - pillow_time) * 1000, 3),
}
if exif_method == "exiftool":
date_taken, location = dt_tool, loc_tool
else:
date_taken, location = dt_pillow, loc_pillow
else:
if exif_method == "exiftool":
date_taken, location = parse_exif_exiftool(data)
else:
date_taken, location = parse_exif_pillow(data)

stat_info = os.stat(full_path)
last_mod = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(stat_info.st_mtime))
size = stat_info.st_size

entry = {
"path": rel_path,
"last_modified": last_mod,
"size": size,
"date_taken": date_taken,
"location": location,
}
if measure_speed and timing:
entry.update(timing)
files.append(entry)

if processed_log and (exif_method == "exiftool" or measure_speed):
if rel_path not in processed:
processed.add(rel_path)
with open(processed_log, "a", encoding="utf-8") as f:
f.write(rel_path + "\n")

return files

def list_photos(
nc_url,
username,
Expand All @@ -132,7 +224,6 @@ def list_photos(
exif_method="pillow",
measure_speed=False,
processed_log=None,

):
"""Iteratively walk the photo directory and return metadata for JPEG files.

Expand Down Expand Up @@ -266,7 +357,6 @@ def list_photos(
with open(processed_log, "a", encoding="utf-8") as f:
f.write(relative + "\n")


return files


Expand All @@ -291,28 +381,43 @@ def main():
"--processed-log",
help="file to track processed JPEGs when using exiftool",
)

parser.add_argument(
"--local-dir",
help="process JPEGs from a local directory instead of Nextcloud",
)
args = parser.parse_args()

url, user, password, photo_dir = validate_env()
exif_method = (
"exiftool" if args.use_exiftool else get_env("EXIF_METHOD", "pillow").lower()
)
measure_speed = args.compare_speed or get_env("COMPARE_SPEED", "0") == "1"
processed_log = args.processed_log or get_env("PROCESSED_LOG")

local_dir = args.local_dir or get_env("LOCAL_PHOTO_DIR")
if local_dir and exif_method == "exiftool":
url = user = password = photo_dir = None
else:
url, user, password, photo_dir = validate_env()

try:
photos = list_photos(
url,
user,
password,
photo_dir,
exif_method=exif_method,
measure_speed=measure_speed,
processed_log=processed_log,
if local_dir and exif_method == "exiftool":
photos = list_local_photos(
local_dir,
exif_method=exif_method,
measure_speed=measure_speed,
processed_log=processed_log,
)
else:
photos = list_photos(
url,
user,
password,
photo_dir,
exif_method=exif_method,
measure_speed=measure_speed,
processed_log=processed_log,
)

)
result = json.dumps(photos, indent=2, ensure_ascii=False)
if args.output:
with open(args.output, "w", encoding="utf-8") as f:
Expand All @@ -322,5 +427,6 @@ def main():
except Exception:
traceback.print_exc()


if __name__ == '__main__':
main()
40 changes: 24 additions & 16 deletions apps/camera/photo_intelligence/nc_photo_streamlit.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
import json
import streamlit as st
import traceback
from nc_photo_list import list_photos, list_local_photos

from nc_photo_list import list_photos


def main():
Expand All @@ -14,19 +14,28 @@ def main():
method = st.selectbox("EXIF method", ["pillow", "exiftool"], index=0)
measure_speed = st.checkbox("측정 모드 (두 방법 속도 비교)")
processed_log = st.text_input("Processed log (optional)")
local_dir = st.text_input("Local directory (exiftool only)")

if st.button("Fetch"):
if not (url and username and password):
st.warning("모든 정보를 입력하세요.")
else:
progress_text = st.empty()
def cb(path):
progress_text.write(f"Scanning {path}")

with st.spinner("이미지 정보를 가져오는 중..."):
try:
progress_text = st.empty()
def cb(path):
progress_text.write(f"Scanning {path}")

with st.spinner("이미지 정보를 가져오는 중..."):
try:
if local_dir and method == "exiftool":
photos = list_local_photos(
local_dir,
progress_cb=cb,
exif_method=method,
measure_speed=measure_speed,
processed_log=processed_log if processed_log else None,
)
else:
if not (url and username and password):
st.warning("모든 정보를 입력하세요.")
return
photos = list_photos(

url,
username,
password,
Expand All @@ -35,12 +44,11 @@ def cb(path):
exif_method=method,
measure_speed=measure_speed,
processed_log=processed_log if processed_log else None,

)
progress_text.write("완료")
st.json(photos)
except Exception:
st.error(traceback.format_exc())
progress_text.write("완료")
st.json(photos)
except Exception:
st.error(traceback.format_exc())


if __name__ == "__main__":
Expand Down
13 changes: 11 additions & 2 deletions apps/camera/photo_intelligence/readme.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,9 @@ options can override the EXIF parser and enable speed measurement:
- `COMPARE_SPEED` (optional) - set to `1` to measure both methods
- `PROCESSED_LOG` (optional) - path to a file that tracks processed JPEGs when
using exiftool
- `LOCAL_PHOTO_DIR` (optional) - when using exiftool you may provide a local
directory instead of connecting to Nextcloud. In this mode, Nextcloud
credentials are not required.


Install the [Pillow](https://python-pillow.org/) package to enable EXIF processing or
Expand Down Expand Up @@ -47,12 +50,18 @@ python3 nc_photo_list.py -o result.json

# record processed files so reruns skip them
python3 nc_photo_list.py --use-exiftool --processed-log processed.txt

# use exiftool on a local directory
python3 nc_photo_list.py --use-exiftool --local-dir ./my_photos

```

For an interactive interface using [Streamlit](https://streamlit.io/) install the
extra dependency and run the Streamlit app. The UI lets you choose the EXIF
parsing method and optionally measure both to compare speeds. The current
directory being scanned is displayed as the script runs:
parsing method and optionally measure both to compare speeds. You can also
specify a local directory when using exiftool so Nextcloud credentials are not
needed. The current directory being scanned is displayed as the script runs:


```bash
pip install streamlit pillow
Expand Down
Loading