Spaces:
Sleeping
Sleeping
File size: 2,861 Bytes
d8535a4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 |
import streamlit as st
import subprocess
import os
import json
from google.oauth2 import service_account
from html_templates import fynder_title, tooltip_message_fynd, logo, button_styles_fynder
# Function to load GCP credentials
def load_gcp_credentials():
try:
# Retrieve GCP credentials from the environment variable
gcp_credentials_str = os.getenv('GCP_CREDENTIALS')
if not gcp_credentials_str:
raise ValueError("GCP_CREDENTIALS environment variable not defined")
# Parse the secret (assuming it's a JSON string)
gcp_credentials = json.loads(gcp_credentials_str)
# Save to a temporary file (Google Cloud uses a JSON file for authentication)
with open("gcp_credentials.json", "w") as f:
json.dump(gcp_credentials, f)
# Authenticate using Google Cloud SDK
credentials_from_file = service_account.Credentials.from_service_account_file("gcp_credentials.json")
# Return the credentials to be used later
return credentials_from_file
except Exception as e:
print(f"Error retrieving or loading GCP credentials: {str(e)}")
return None
# Function to authenticate BigQuery
def authenticate_bigquery():
creds = load_gcp_credentials()
if not creds:
st.error("Unable to load GCP credentials for BigQuery authentication.")
return None
return creds
# Logo
st.markdown(logo, unsafe_allow_html=True)
st.logo("alerter_4.jpeg")
st.markdown(fynder_title, unsafe_allow_html=True)
st.write("")
st.markdown(tooltip_message_fynd, unsafe_allow_html=True)
# Text input for the month and year in dd-yyyy format
selected_month_year = st.text_input("Enter Month and Year (dd-yyyy): E.g. 01-2024")
# Define the base download path
base_download_path = "fynder_dump"
# Ensure the directory exists
if not os.path.exists(base_download_path):
os.makedirs(base_download_path)
# Button to initiate the search and download
st.markdown(button_styles_fynder, unsafe_allow_html=True)
if st.button("Search"):
# Authenticate BigQuery (use this for Google Cloud Storage access too)
creds = authenticate_bigquery()
if creds is None:
st.error("Authentication failed. Please check your credentials.")
else:
# Construct the gsutil command
gsutil_command = f"gsutil -m cp -r 'gs://fynd-assets-private/documents/daytrader/PDFs/**/{selected_month_year}/*' {base_download_path}"
# Run the command using subprocess
try:
# Execute the gsutil command
subprocess.run(gsutil_command, shell=True, check=True)
st.success(f"Invoices from {selected_month_year} have been successfully downloaded to Desktop/{base_download_path}")
except subprocess.CalledProcessError as e:
st.error(f"Error downloading invoices: {e}")
|