257 lines
10 KiB
Python
257 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
scripts/migrate_photos_to_r2.py — Verified photo sync to Cloudflare R2 (Phase 3)
|
|
==================================================================================
|
|
|
|
Copies every photo file under ``app/static/uploads/`` to the R2 bucket and
|
|
VERIFIES each one (MD5 + size). It is:
|
|
|
|
- COPY-ONLY : never deletes or modifies local files, never touches the DB.
|
|
- IDEMPOTENT : re-running skips objects already present and verified.
|
|
- RESUMABLE : safe to interrupt and re-run; picks up where it left off.
|
|
- GATED : exits non-zero if ANY file fails verification, so you never
|
|
flip STORAGE_BACKEND=s3 on an incomplete/corrupt copy.
|
|
|
|
The object KEY equals the file's path relative to ``static/`` — i.e. the exact
|
|
string stored in the DB (``uploads/<subfolder>/<file>``). No DB rows change.
|
|
|
|
Why put_object (not upload_file): a single-part PUT makes the R2 ETag equal the
|
|
object's MD5, so verification is a direct hash comparison. Photos are <= 50 MB
|
|
(MAX_CONTENT_LENGTH), well within a single PUT.
|
|
|
|
R2 credentials are read from the app config (config.py / .env): R2_ENDPOINT_URL,
|
|
R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY, R2_BUCKET. You can run this while the app
|
|
is still serving on STORAGE_BACKEND=local — it only writes to R2.
|
|
|
|
Usage
|
|
-----
|
|
cd /home/jqc/janitorial_qc
|
|
source venv/bin/activate
|
|
pip install boto3 # if not already
|
|
|
|
python scripts/migrate_photos_to_r2.py --dry-run # show plan, upload nothing
|
|
python scripts/migrate_photos_to_r2.py # bulk sync + verify
|
|
python scripts/migrate_photos_to_r2.py # run again right before
|
|
# cutover (uploads the delta)
|
|
python scripts/migrate_photos_to_r2.py --force # re-upload even if present
|
|
python scripts/migrate_photos_to_r2.py --log /home/jqc/r2_sync.json
|
|
|
|
Exit code: 0 only when every local file is verified in R2 with zero mismatches
|
|
and zero errors (i.e. safe to cut over). Non-zero otherwise.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import hashlib
|
|
import argparse
|
|
from datetime import datetime
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
|
|
|
from app import create_app # noqa: E402
|
|
|
|
_CONTENT_TYPES = {
|
|
'jpg': 'image/jpeg', 'jpeg': 'image/jpeg',
|
|
'png': 'image/png', 'gif': 'image/gif',
|
|
}
|
|
_CHUNK = 1024 * 1024 # 1 MB
|
|
|
|
|
|
def _md5_and_size(path):
|
|
h = hashlib.md5()
|
|
size = 0
|
|
with open(path, 'rb') as fh:
|
|
while True:
|
|
chunk = fh.read(_CHUNK)
|
|
if not chunk:
|
|
break
|
|
h.update(chunk)
|
|
size += len(chunk)
|
|
return h.hexdigest(), size
|
|
|
|
|
|
def _content_type(key):
|
|
ext = key.rsplit('.', 1)[-1].lower() if '.' in key else ''
|
|
return _CONTENT_TYPES.get(ext, 'application/octet-stream')
|
|
|
|
|
|
def _iter_files(uploads_root, base):
|
|
"""Yield (key, abs_path) for every file under uploads_root; key is relative to base."""
|
|
for dirpath, _dirs, files in os.walk(uploads_root):
|
|
for name in files:
|
|
abs_path = os.path.join(dirpath, name)
|
|
key = os.path.relpath(abs_path, base).replace(os.sep, '/')
|
|
yield key, abs_path
|
|
|
|
|
|
def _remote_state(client, bucket, key):
|
|
"""Return (exists, etag_no_quotes, size) for a key in the bucket."""
|
|
try:
|
|
resp = client.head_object(Bucket=bucket, Key=key)
|
|
return True, resp['ETag'].strip('"'), resp['ContentLength']
|
|
except Exception:
|
|
return False, None, None
|
|
|
|
|
|
def human(n):
|
|
n = float(n)
|
|
for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
|
|
if abs(n) < 1024.0:
|
|
return f'{n:.1f} {unit}'
|
|
n /= 1024.0
|
|
return f'{n:.1f} PB'
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='Verified photo sync to R2 (Phase 3).')
|
|
parser.add_argument('--dry-run', action='store_true',
|
|
help='Report what would happen; upload nothing.')
|
|
parser.add_argument('--force', action='store_true',
|
|
help='Re-upload even if the object already verifies.')
|
|
parser.add_argument('--log', default=None,
|
|
help='Path to write the JSON result log (default: /tmp/...).')
|
|
args = parser.parse_args()
|
|
|
|
app = create_app(os.getenv('FLASK_ENV') or 'production')
|
|
with app.app_context():
|
|
from flask import current_app
|
|
cfg = current_app.config
|
|
|
|
missing = [k for k in ('R2_ENDPOINT_URL', 'R2_ACCESS_KEY_ID',
|
|
'R2_SECRET_ACCESS_KEY', 'R2_BUCKET') if not cfg.get(k)]
|
|
if missing:
|
|
print(f'ERROR: missing R2 config: {", ".join(missing)}')
|
|
print('Set them in .env before running the sync.')
|
|
sys.exit(2)
|
|
|
|
try:
|
|
import boto3
|
|
from botocore.config import Config as BotoConfig
|
|
except ImportError:
|
|
print('ERROR: boto3 not installed. Run: pip install boto3')
|
|
sys.exit(2)
|
|
|
|
bucket = cfg['R2_BUCKET']
|
|
client = boto3.client(
|
|
's3',
|
|
endpoint_url = cfg['R2_ENDPOINT_URL'],
|
|
aws_access_key_id = cfg['R2_ACCESS_KEY_ID'],
|
|
aws_secret_access_key = cfg['R2_SECRET_ACCESS_KEY'],
|
|
region_name = 'auto',
|
|
config = BotoConfig(signature_version='s3v4'),
|
|
)
|
|
|
|
uploads_root = cfg['UPLOAD_FOLDER'] # .../app/static/uploads
|
|
base = os.path.dirname(uploads_root) # .../app/static (keys start 'uploads/')
|
|
|
|
if not os.path.isdir(uploads_root):
|
|
print(f'ERROR: uploads folder not found: {uploads_root}')
|
|
sys.exit(2)
|
|
|
|
counts = {'total': 0, 'skipped_verified': 0, 'uploaded': 0,
|
|
'verify_failed': 0, 'errors': 0, 'bytes_uploaded': 0}
|
|
failures = []
|
|
|
|
print('=' * 70)
|
|
print(f' R2 PHOTO SYNC{" (DRY RUN)" if args.dry_run else ""}')
|
|
print(f' bucket : {bucket}')
|
|
print(f' source : {uploads_root}')
|
|
print('=' * 70)
|
|
|
|
for key, abs_path in _iter_files(uploads_root, base):
|
|
counts['total'] += 1
|
|
try:
|
|
local_md5, local_size = _md5_and_size(abs_path)
|
|
except OSError as e:
|
|
counts['errors'] += 1
|
|
failures.append({'key': key, 'stage': 'read', 'error': str(e)})
|
|
print(f' ERROR read {key}: {e}')
|
|
continue
|
|
|
|
# Resumable skip: already present and matching?
|
|
if not args.force:
|
|
exists, r_etag, r_size = _remote_state(client, bucket, key)
|
|
if exists and r_etag == local_md5 and r_size == local_size:
|
|
counts['skipped_verified'] += 1
|
|
continue
|
|
|
|
if args.dry_run:
|
|
counts['uploaded'] += 1 # would upload
|
|
print(f' WOULD PUT {key} ({human(local_size)})')
|
|
continue
|
|
|
|
# Upload (single-part PUT so ETag == MD5)
|
|
try:
|
|
with open(abs_path, 'rb') as body:
|
|
client.put_object(
|
|
Bucket=bucket, Key=key, Body=body,
|
|
ContentType=_content_type(key),
|
|
)
|
|
except Exception as e:
|
|
counts['errors'] += 1
|
|
failures.append({'key': key, 'stage': 'upload', 'error': str(e)})
|
|
print(f' ERROR upload {key}: {e}')
|
|
continue
|
|
|
|
# Verify: re-HEAD and compare ETag(MD5) + size
|
|
exists, r_etag, r_size = _remote_state(client, bucket, key)
|
|
if exists and r_etag == local_md5 and r_size == local_size:
|
|
counts['uploaded'] += 1
|
|
counts['bytes_uploaded'] += local_size
|
|
else:
|
|
counts['verify_failed'] += 1
|
|
failures.append({
|
|
'key': key, 'stage': 'verify',
|
|
'local_md5': local_md5, 'local_size': local_size,
|
|
'remote_etag': r_etag, 'remote_size': r_size,
|
|
})
|
|
print(f' VERIFY FAIL {key} local_md5={local_md5} '
|
|
f'remote_etag={r_etag} local_size={local_size} remote_size={r_size}')
|
|
|
|
# ── summary ──
|
|
if args.dry_run:
|
|
action_line = f' would upload : {counts["uploaded"]}'
|
|
else:
|
|
action_line = (f' uploaded + verified : {counts["uploaded"]}'
|
|
f' ({human(counts["bytes_uploaded"])})')
|
|
print('-' * 70)
|
|
print(f' total files under uploads : {counts["total"]}')
|
|
print(f' already verified (skipped): {counts["skipped_verified"]}')
|
|
print(action_line)
|
|
print(f' verify failures : {counts["verify_failed"]}')
|
|
print(f' errors : {counts["errors"]}')
|
|
print('=' * 70)
|
|
|
|
verified_total = counts['skipped_verified'] + (0 if args.dry_run else counts['uploaded'])
|
|
safe = (counts['verify_failed'] == 0 and counts['errors'] == 0)
|
|
|
|
if args.dry_run:
|
|
print(' DRY RUN — nothing uploaded. Re-run without --dry-run to sync.')
|
|
elif safe:
|
|
print(f' GATE: {verified_total}/{counts["total"]} local files verified in R2, '
|
|
f'0 mismatches, 0 errors.')
|
|
print(' ✅ SAFE to proceed to cutover (Phase 4). Local files are untouched.')
|
|
else:
|
|
print(f' GATE: {counts["verify_failed"]} verify failure(s), {counts["errors"]} error(s).')
|
|
print(' ❌ NOT safe to cut over. Investigate failures (see log), then re-run.')
|
|
|
|
log_path = args.log or os.path.join(
|
|
'/tmp', f'jqc_r2_sync_{datetime.utcnow().strftime("%Y%m%d_%H%M%S")}.json')
|
|
try:
|
|
with open(log_path, 'w') as fh:
|
|
json.dump({
|
|
'generated_at': datetime.utcnow().isoformat() + 'Z',
|
|
'bucket': bucket, 'dry_run': args.dry_run,
|
|
'counts': counts, 'failures': failures,
|
|
}, fh, indent=2)
|
|
print(f'\n Log written to: {log_path}')
|
|
except OSError as e:
|
|
print(f'\n Could not write log to {log_path}: {e}')
|
|
|
|
sys.exit(0 if safe else 1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|