231 lines
9.8 KiB
Python
231 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Build chronological, per-day editing manifests for home videos."""
|
|
|
|
import argparse
|
|
from collections import defaultdict
|
|
from datetime import datetime, timezone
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
VIDEO_EXTENSIONS = {".mp4", ".mov", ".m4v", ".mkv", ".avi", ".webm", ".mts", ".m2ts", ".3gp"}
|
|
NAME_DATE = re.compile(r"(?<!\d)(20\d{2})[-_.]?([01]\d)[-_.]?([0-3]\d)(?:[T _.-]?([0-2]\d)[-_.]?([0-5]\d)[-_.]?([0-5]\d))?(?!\d)")
|
|
|
|
|
|
def probe(path):
|
|
result = subprocess.run(
|
|
["ffprobe", "-v", "error", "-show_format", "-show_streams", "-of", "json", str(path)],
|
|
capture_output=True, text=True, check=True,
|
|
)
|
|
data = json.loads(result.stdout)
|
|
if not any(stream.get("codec_type") == "video" for stream in data.get("streams", [])):
|
|
raise ValueError("no video stream")
|
|
durations = [data.get("format", {}).get("duration")]
|
|
durations += [s.get("duration") for s in data.get("streams", []) if s.get("codec_type") == "video"]
|
|
duration = next((float(x) for x in durations if x is not None and float(x) > 0), None)
|
|
if duration is None:
|
|
raise ValueError("no positive duration")
|
|
return data, duration
|
|
|
|
|
|
def recorded_at(path, data, local_zone, fallback):
|
|
# QuickTime metadata is typically UTC. An embedded offset takes precedence.
|
|
tags = [data.get("format", {}).get("tags", {})]
|
|
tags += [s.get("tags", {}) for s in data.get("streams", [])]
|
|
for key in ("com.apple.quicktime.creationdate", "creation_time", "date"):
|
|
for group in tags:
|
|
value = next((v for k, v in group.items() if k.lower() == key), None)
|
|
if not value:
|
|
continue
|
|
try:
|
|
stamp = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
if stamp.tzinfo is None:
|
|
stamp = stamp.replace(tzinfo=local_zone if key != "creation_time" else timezone.utc)
|
|
return stamp.astimezone(local_zone), "metadata"
|
|
except ValueError:
|
|
pass
|
|
match = NAME_DATE.search(path.stem)
|
|
if match:
|
|
try:
|
|
y, m, d, h, minute, second = match.groups()
|
|
return datetime(int(y), int(m), int(d), int(h or 0), int(minute or 0), int(second or 0), tzinfo=local_zone), "filename"
|
|
except ValueError:
|
|
pass
|
|
if fallback == "mtime":
|
|
return datetime.fromtimestamp(path.stat().st_mtime, local_zone), "mtime"
|
|
raise ValueError("no recording timestamp; add a dated filename or use --fallback mtime")
|
|
|
|
|
|
def write_json(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
|
|
|
|
def existing_manifest_title(path):
|
|
"""Return a title already set in a daily manifest, if one exists."""
|
|
try:
|
|
manifest = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return ""
|
|
if isinstance(manifest, dict) and isinstance(manifest.get("title"), str):
|
|
return manifest["title"]
|
|
return ""
|
|
|
|
|
|
def normalized_path(value):
|
|
return os.path.normcase(os.path.abspath(str(value)))
|
|
|
|
|
|
def find_existing_project(project_root, source):
|
|
"""Find a project from an earlier layout, preferring a direct path match."""
|
|
wanted = normalized_path(source)
|
|
for candidate in project_root.rglob("*.llc"):
|
|
try:
|
|
data = json.loads(candidate.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
continue
|
|
media_name = data.get("mediaFileName", "")
|
|
if media_name and normalized_path(media_name) == wanted:
|
|
return candidate, data
|
|
metadata_path = candidate.parent / "source.json"
|
|
try:
|
|
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
metadata = {}
|
|
if metadata.get("source_path") and normalized_path(metadata["source_path"]) == wanted:
|
|
return candidate, data
|
|
return None, None
|
|
|
|
|
|
def remove_old_project_artifacts(old_project, source, project_root):
|
|
if old_project is None or not old_project.exists():
|
|
return
|
|
old_dir = old_project.parent
|
|
alias = old_dir / source.name
|
|
if alias.exists():
|
|
try:
|
|
if alias.samefile(source):
|
|
alias.unlink()
|
|
print(f"Removed obsolete hard-link entry: {alias}")
|
|
except OSError as exc:
|
|
print(f"Warning: could not remove old media link {alias}: {exc}", file=sys.stderr)
|
|
metadata_path = old_dir / "source.json"
|
|
try:
|
|
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
|
|
if metadata.get("source_path") and normalized_path(metadata["source_path"]) == normalized_path(source):
|
|
metadata_path.unlink()
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
old_project.unlink(missing_ok=True)
|
|
current = old_dir
|
|
while current != project_root:
|
|
try:
|
|
current.rmdir()
|
|
except OSError:
|
|
break
|
|
current = current.parent
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("input", type=Path, help="folder containing videos (searched recursively)")
|
|
parser.add_argument("output", type=Path, help="folder for daily JSON manifests and .llc files")
|
|
parser.add_argument("--timezone", default="UTC", help="IANA filming timezone, e.g. America/New_York (default: UTC)")
|
|
parser.add_argument("--fallback", choices=("skip", "mtime"), default="skip", help="date source when metadata and filename lack a date")
|
|
parser.add_argument("--date", help="only produce one filming day, YYYY-MM-DD")
|
|
args = parser.parse_args()
|
|
try:
|
|
local_zone = ZoneInfo(args.timezone)
|
|
except ZoneInfoNotFoundError:
|
|
parser.error("unknown timezone; install tzdata if your system lacks zone information")
|
|
if args.date:
|
|
try:
|
|
datetime.strptime(args.date, "%Y-%m-%d")
|
|
except ValueError:
|
|
parser.error("--date must be YYYY-MM-DD")
|
|
source = args.input.expanduser().resolve()
|
|
dest = args.output.expanduser().resolve()
|
|
if not source.is_dir():
|
|
parser.error("input must be an existing directory")
|
|
if source == dest or source in dest.parents:
|
|
parser.error("output must be outside input so generated media is not rescanned")
|
|
|
|
days = defaultdict(list)
|
|
excluded = 0
|
|
failed = 0
|
|
for path in sorted(source.rglob("*")):
|
|
if not path.is_file() or path.suffix.lower() not in VIDEO_EXTENSIONS:
|
|
continue
|
|
try:
|
|
data, duration = probe(path)
|
|
stamp, date_source = recorded_at(path, data, local_zone, args.fallback)
|
|
day = stamp.date().isoformat()
|
|
if args.date and day != args.date:
|
|
continue
|
|
if duration < 5:
|
|
excluded += 1
|
|
print(f"Excluded (<5s): {path}")
|
|
continue
|
|
days[day].append((stamp, path.resolve(), duration, date_source))
|
|
except (subprocess.CalledProcessError, json.JSONDecodeError, ValueError, OSError) as exc:
|
|
failed += 1
|
|
print(f"Skipped {path}: {exc}", file=sys.stderr)
|
|
|
|
project_root = dest
|
|
for day, videos in sorted(days.items()):
|
|
entries = []
|
|
for stamp, path, duration, date_source in sorted(videos, key=lambda row: (row[0], str(row[1]))):
|
|
if duration > 120:
|
|
base_name = f"{day}__{path.stem}"
|
|
old_project, existing = find_existing_project(project_root, path)
|
|
suffix = 1
|
|
while True:
|
|
numbered_name = base_name if suffix == 1 else f"{base_name}__{suffix}"
|
|
llc = project_root / f"{numbered_name}.llc"
|
|
if not llc.exists():
|
|
break
|
|
try:
|
|
current = json.loads(llc.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
current = {}
|
|
if current.get("mediaFileName") and normalized_path(current["mediaFileName"]) == normalized_path(path):
|
|
old_project, existing = llc, current
|
|
break
|
|
suffix += 1
|
|
try:
|
|
if existing is None:
|
|
existing = {"version": 2, "mediaFileName": str(path),
|
|
"cutSegments": [{"start": 0, "end": duration, "name": "", "selected": True}]}
|
|
|
|
# This LosslessCut build supports an absolute mediaFileName.
|
|
existing["version"] = 2
|
|
existing["mediaFileName"] = str(path)
|
|
segments = existing.setdefault("cutSegments", [])
|
|
for segment in segments:
|
|
segment.setdefault("name", "")
|
|
segment.setdefault("selected", True)
|
|
write_json(llc, existing)
|
|
if old_project != llc:
|
|
remove_old_project_artifacts(old_project, path, project_root)
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise ValueError(f"cannot reuse existing project {llc}: {exc}") from exc
|
|
target = llc
|
|
else:
|
|
target = path
|
|
entries.append(str(target))
|
|
if date_source == "mtime":
|
|
print(f"Warning: filesystem modified time used for {path}", file=sys.stderr)
|
|
manifest = dest / f"{day}.json"
|
|
write_json(manifest, {"title": existing_manifest_title(manifest), "videos": entries})
|
|
print(f"{manifest}: {len(entries)} videos")
|
|
print(f"Finished: {len(days)} days, {excluded} short clips excluded, {failed} files skipped")
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|