| """Firebase Storage uploads with local fallback. |
| |
| Env: |
| FIREBASE_STORAGE_BUCKET e.g. myproj.appspot.com |
| FIREBASE_CREDENTIALS_JSON service-account JSON (raw), or |
| FIREBASE_CREDENTIALS_BASE64 base64 of that JSON (preferred for HF Secrets) |
| |
| Behavior: |
| save_upload(file_storage, prefix) -> (stored_ref, filename) |
| - Always saves a local copy under UPLOAD_FOLDER (for immediate AI use). |
| - If Firebase is configured, also uploads to <prefix>/<uuid>_<filename> |
| and returns the public URL as stored_ref. |
| - If not configured (local dev), returns the local relative path. |
| resolve_to_local(ref, upload_folder, root_path) -> local path |
| - http(s) URL: downloads to temp, returns temp path. |
| - gs://... : downloads via bucket to temp. |
| - otherwise: legacy local relative path join. |
| """ |
| import base64 |
| import json |
| import os |
| import tempfile |
| import uuid |
|
|
| _bucket = None |
| _init_tried = False |
|
|
|
|
| def is_configured() -> bool: |
| return bool( |
| os.environ.get("FIREBASE_STORAGE_BUCKET") |
| and ( |
| os.environ.get("FIREBASE_CREDENTIALS_JSON") |
| or os.environ.get("FIREBASE_CREDENTIALS_BASE64") |
| ) |
| ) |
|
|
|
|
| def _bucket(): |
| global _bucket, _init_tried |
| if _bucket is not None: |
| return _bucket |
| if _init_tried: |
| return None |
| _init_tried = True |
| try: |
| import firebase_admin |
| from firebase_admin import credentials, storage |
| except ImportError: |
| print("firebase-admin not installed; using local storage.") |
| return None |
|
|
| bucket_name = os.environ.get("FIREBASE_STORAGE_BUCKET", "") |
| cred_json = os.environ.get("FIREBASE_CREDENTIALS_JSON", "") |
| cred_b64 = os.environ.get("FIREBASE_CREDENTIALS_BASE64", "") |
| try: |
| if cred_b64 and not cred_json: |
| cred_json = base64.b64decode(cred_b64).decode("utf-8") |
| info = json.loads(cred_json) |
| cred = credentials.Certificate(info) |
| try: |
| firebase_admin.get_app() |
| except ValueError: |
| firebase_admin.initialize_app(cred, {"storageBucket": bucket_name}) |
| _bucket = storage.bucket() |
| print(f"Firebase Storage ready (bucket={_bucket.name}).") |
| except Exception as e: |
| print(f"Firebase init failed, using local storage: {e}") |
| _bucket = None |
| return _bucket |
|
|
|
|
| def upload_path(local_path: str, dest: str, content_type: str = None) -> str: |
| """Upload an existing local file; returns public URL or gs:// ref (or '' if skipped).""" |
| bucket = _bucket() if is_configured() else None |
| if bucket is None: |
| return "" |
| try: |
| blob = bucket.blob(dest.strip("/")) |
| blob.upload_from_filename(local_path, content_type=content_type) |
| try: |
| blob.make_public() |
| return blob.public_url |
| except Exception: |
| return f"gs://{bucket.name}/{dest.strip('/')}" |
| except Exception as e: |
| print(f"Firebase upload failed: {e}") |
| return "" |
|
|
|
|
| def save_upload(file_storage, prefix: str, upload_folder: str): |
| """Save an uploaded FileStorage locally + to Firebase. Returns (ref, filename).""" |
| from werkzeug.utils import secure_filename |
|
|
| filename = secure_filename(file_storage.filename or "upload") |
| unique = f"{uuid.uuid4().hex}_{filename}" |
| local_path = os.path.join(upload_folder, unique) |
| os.makedirs(upload_folder, exist_ok=True) |
| file_storage.save(local_path) |
| rel = os.path.join(os.path.basename(upload_folder), unique).replace("\\", "/") |
| |
| stored = os.path.join("static", rel).replace("\\", "/") |
|
|
| bucket = _bucket() if is_configured() else None |
| if bucket is None: |
| try: |
| file_storage.seek(0) |
| except Exception: |
| pass |
| return stored, filename |
|
|
| dest = f"{prefix.strip('/')}/{unique}" |
| try: |
| try: |
| file_storage.seek(0) |
| except Exception: |
| pass |
| blob = bucket.blob(dest) |
| blob.upload_from_file( |
| file_storage.stream, content_type=file_storage.content_type |
| ) |
| try: |
| blob.make_public() |
| return blob.public_url, filename |
| except Exception: |
| |
| return f"gs://{bucket.name}/{dest}", filename |
| except Exception as e: |
| print(f"Firebase upload failed, keeping local: {e}") |
| return stored, filename |
|
|
|
|
| def resolve_to_local(ref: str, upload_folder: str, root_path: str) -> str: |
| """Turn a stored ref (URL, gs://, or legacy local path) into a local file path.""" |
| ref = ref or "" |
| if ref.startswith("http://") or ref.startswith("https://"): |
| import requests |
| suffix = os.path.splitext(ref.split("?")[0])[1][:8] or ".bin" |
| tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) |
| r = requests.get(ref, timeout=60) |
| r.raise_for_status() |
| tmp.write(r.content) |
| tmp.close() |
| return tmp.name |
| if ref.startswith("gs://"): |
| bucket = _bucket() |
| if bucket is None: |
| raise FileNotFoundError("Firebase not configured for gs:// download.") |
| parts = ref[5:].split("/", 1) |
| blob = bucket.blob(parts[1] if len(parts) > 1 else "") |
| suffix = os.path.splitext(parts[-1])[1][:8] or ".bin" |
| tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) |
| tmp.close() |
| blob.download_to_filename(tmp.name) |
| return tmp.name |
| |
| if os.path.isabs(ref) and os.path.exists(ref): |
| return ref |
| cand = os.path.join(root_path, ref) |
| if os.path.exists(cand): |
| return cand |
| return os.path.join(root_path, "static", "uploads", os.path.basename(ref)) |
|
|
