58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
"""
|
|
S3 Sync - Upload scraped data to MinIO periodically.
|
|
"""
|
|
import os, time, json, gzip
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
|
|
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "/data/output"))
|
|
S3_ENDPOINT = os.getenv("S3_ENDPOINT", "https://s3.siax.io")
|
|
S3_ACCESS_KEY = os.getenv("S3_ACCESS_KEY", "")
|
|
S3_SECRET_KEY = os.getenv("S3_SECRET_KEY", "")
|
|
S3_BUCKET = os.getenv("S3_BUCKET", "opt-scraped-data")
|
|
SYNC_INTERVAL = int(os.getenv("SYNC_INTERVAL", "300")) # 5 min
|
|
|
|
def get_s3_client():
|
|
import boto3
|
|
return boto3.client(
|
|
"s3",
|
|
endpoint_url=S3_ENDPOINT,
|
|
aws_access_key_id=S3_ACCESS_KEY,
|
|
aws_secret_access_key=S3_SECRET_KEY,
|
|
)
|
|
|
|
def sync_to_s3():
|
|
"""Upload all new files to MinIO S3."""
|
|
s3 = get_s3_client()
|
|
# Ensure bucket exists
|
|
try:
|
|
s3.head_bucket(Bucket=S3_BUCKET)
|
|
except:
|
|
s3.create_bucket(Bucket=S3_BUCKET)
|
|
|
|
uploaded = 0
|
|
for fpath in OUTPUT_DIR.glob("*.json.gz"):
|
|
key = f"raw/{fpath.name}"
|
|
try:
|
|
s3.upload_file(str(fpath), S3_BUCKET, key)
|
|
uploaded += 1
|
|
fpath.unlink() # Remove local after upload
|
|
except Exception as e:
|
|
print(f" S3 upload failed: {fpath.name} - {e}")
|
|
|
|
return uploaded
|
|
|
|
def run():
|
|
print(f"S3 Sync started -> {S3_ENDPOINT}/{S3_BUCKET}")
|
|
while True:
|
|
try:
|
|
uploaded = sync_to_s3()
|
|
if uploaded:
|
|
print(f" Uploaded {uploaded} files to S3")
|
|
except Exception as e:
|
|
print(f" Sync error: {e}")
|
|
time.sleep(SYNC_INTERVAL)
|
|
|
|
if __name__ == "__main__":
|
|
run()
|