84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""Offline, explicit local-to-Photos-API source conversion; defaults to dry run."""
|
|
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from pigway_frame_flow.providers.photo_sources import PhotoSourceInput
|
|
|
|
|
|
def convert(database, source_ids, *, apply=False):
|
|
connection = sqlite3.connect(f"file:{database}?mode=rw", uri=True)
|
|
try:
|
|
updates = []
|
|
for source_id in source_ids:
|
|
key = "photo_source:" + source_id
|
|
row = connection.execute(
|
|
"SELECT payload FROM app_config WHERE key=?", (key,)
|
|
).fetchone()
|
|
if not row:
|
|
raise ValueError("Source not found: " + source_id)
|
|
source = PhotoSourceInput.model_validate_json(row[0])
|
|
if source.kind != "local":
|
|
raise ValueError(
|
|
"Only explicitly selected local sources can be converted"
|
|
)
|
|
location = source.location.strip("/")
|
|
if (
|
|
source.location.startswith("/")
|
|
or "\\" in location
|
|
or ".." in location.split("/")
|
|
):
|
|
raise ValueError(
|
|
"Review the source's relative Photos folder path first"
|
|
)
|
|
payload = source.model_copy(
|
|
update={"kind": "synology-photos", "synology_previews": False}
|
|
)
|
|
updates.append((payload.model_dump_json(), key))
|
|
backup = None
|
|
if apply:
|
|
backup = database.with_name(
|
|
database.name
|
|
+ ".before-photos-api-"
|
|
+ datetime.now(UTC).strftime("%Y%m%dT%H%M%S%fZ")
|
|
)
|
|
with sqlite3.connect(backup) as snapshot:
|
|
connection.backup(snapshot)
|
|
backup.chmod(0o600)
|
|
with connection:
|
|
connection.executemany(
|
|
"UPDATE app_config SET payload=? WHERE key=?", updates
|
|
)
|
|
return {
|
|
"applied": apply,
|
|
"sources": source_ids,
|
|
"backup": str(backup) if backup else None,
|
|
}
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--database", type=Path, required=True)
|
|
parser.add_argument("--source", action="append", required=True)
|
|
parser.add_argument("--apply", action="store_true")
|
|
parser.add_argument(
|
|
"--service-stopped",
|
|
action="store_true",
|
|
help="Confirm the target service is stopped before changing its database",
|
|
)
|
|
args = parser.parse_args()
|
|
if args.apply and not args.service_stopped:
|
|
parser.error(
|
|
"Stop the target service and pass --service-stopped before applying"
|
|
)
|
|
print(json.dumps(convert(args.database, args.source, apply=args.apply), indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|