79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
import logging
|
|
|
|
from google.cloud import storage
|
|
import os
|
|
import uuid
|
|
|
|
import firebase_admin
|
|
from firebase_admin import credentials, firestore
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
def download_firebase_file(bucket_name, source_blob_name, destination_file_name):
|
|
# Downloads a file from Firebase Storage.
|
|
storage_client = storage.Client()
|
|
bucket = storage_client.bucket(bucket_name)
|
|
blob = bucket.blob(source_blob_name)
|
|
blob.download_to_filename(destination_file_name)
|
|
logging.info(f"File uploaded to {destination_file_name}")
|
|
return destination_file_name
|
|
|
|
def upload_file_firebase(bucket_name, destination_blob_name, source_file_name):
|
|
# Uploads a file to Firebase Storage.
|
|
storage_client = storage.Client()
|
|
bucket = storage_client.bucket(bucket_name)
|
|
try:
|
|
blob = bucket.blob(destination_blob_name)
|
|
blob.upload_from_filename(source_file_name)
|
|
logging.info(f"File uploaded to {destination_blob_name}")
|
|
return True
|
|
except Exception as e:
|
|
logging.error("Error uploading file to Google Cloud Storage: " + str(e))
|
|
return False
|
|
|
|
def upload_file_firebase_get_url(bucket_name, destination_blob_name, source_file_name):
|
|
# Uploads a file to Firebase Storage.
|
|
storage_client = storage.Client()
|
|
bucket = storage_client.bucket(bucket_name)
|
|
try:
|
|
blob = bucket.blob(destination_blob_name)
|
|
blob.upload_from_filename(source_file_name)
|
|
logging.info(f"File uploaded to {destination_blob_name}")
|
|
|
|
# Make the file public
|
|
blob.make_public()
|
|
|
|
# Get the public URL
|
|
url = blob.public_url
|
|
return url
|
|
except Exception as e:
|
|
logging.error("Error uploading file to Google Cloud Storage: " + str(e))
|
|
return None
|
|
|
|
def save_to_db(collection: str, item):
|
|
db = firestore.client()
|
|
collection_ref = db.collection(collection)
|
|
(update_time, document_ref) = collection_ref.add(item)
|
|
if document_ref:
|
|
logging.info(f"Document added with ID: {document_ref.id}")
|
|
return (True, document_ref.id)
|
|
else:
|
|
return (False, None)
|
|
|
|
|
|
def save_to_db_with_id(collection: str, item, id: str):
|
|
db = firestore.client()
|
|
collection_ref = db.collection(collection)
|
|
# Reference to the specific document with the desired ID
|
|
document_ref = collection_ref.document(id)
|
|
# Set the data to the document
|
|
document_ref.set(item)
|
|
if document_ref:
|
|
logging.info(f"Document added with ID: {document_ref.id}")
|
|
return (True, document_ref.id)
|
|
else:
|
|
return (False, None)
|
|
|
|
|
|
|