music on atproto
plyr.fm
zzstoatzz.io
1#!/usr/bin/env -S uv run --script --quiet
2"""migrate images from audio-* buckets to images-* buckets."""
3
4from pathlib import Path
5
6import boto3
7from pydantic import Field
8from pydantic_settings import BaseSettings, SettingsConfigDict
9
10BASE_DIR = Path(__file__).resolve().parents[1]
11
12
13class R2Settings(BaseSettings):
14 """R2 credentials from environment."""
15
16 model_config = SettingsConfigDict(
17 env_file=str(BASE_DIR / ".env"),
18 env_file_encoding="utf-8",
19 extra="ignore",
20 case_sensitive=False,
21 )
22
23 aws_access_key_id: str = Field(validation_alias="AWS_ACCESS_KEY_ID")
24 aws_secret_access_key: str = Field(validation_alias="AWS_SECRET_ACCESS_KEY")
25 r2_endpoint_url: str = Field(validation_alias="R2_ENDPOINT_URL")
26
27
28def migrate_images(env: str):
29 """migrate images for a specific environment.
30
31 args:
32 env: environment name (dev, staging, prod)
33 """
34 settings = R2Settings()
35
36 s3 = boto3.client(
37 "s3",
38 endpoint_url=settings.r2_endpoint_url,
39 aws_access_key_id=settings.aws_access_key_id,
40 aws_secret_access_key=settings.aws_secret_access_key,
41 region_name="auto",
42 )
43
44 source_bucket = f"audio-{env}"
45 dest_bucket = f"images-{env}"
46 prefix = "images/"
47
48 print(f"\nmigrating {env}:")
49 print(f" source: {source_bucket}/{prefix}")
50 print(f" dest: {dest_bucket}/")
51
52 # list all objects in source bucket with images/ prefix
53 paginator = s3.get_paginator("list_objects_v2")
54 pages = paginator.paginate(Bucket=source_bucket, Prefix=prefix)
55
56 copied_count = 0
57 for page in pages:
58 if "Contents" not in page:
59 continue
60
61 for obj in page["Contents"]:
62 source_key = obj["Key"]
63 # remove images/ prefix for destination
64 dest_key = source_key.replace("images/", "", 1)
65
66 # copy object
67 copy_source = {"Bucket": source_bucket, "Key": source_key}
68 s3.copy_object(
69 CopySource=copy_source,
70 Bucket=dest_bucket,
71 Key=dest_key,
72 )
73
74 copied_count += 1
75 print(f" ✓ copied {source_key} -> {dest_key}")
76
77 print(f" total: {copied_count} files")
78
79
80def main():
81 """migrate images for all environments."""
82 for env in ["dev", "staging", "prod"]:
83 migrate_images(env)
84
85 print("\n✅ migration complete!")
86
87
88if __name__ == "__main__":
89 main()