Spaces:
Runtime error
Runtime error
Download persistStorage.py from Express-Analytics/QueryHelper: direct link, hf CLI and curl.
- Browser
- Download file 1.9 kB
-
https://huggingface.co/spaces/Express-Analytics/QueryHelper/resolve/e7c1910ae94c764e2c2c00bcbd736b692299135b/persistStorage.py
- Command line
-
hf download hf://spaces/Express-Analytics/QueryHelper@e7c1910ae94c764e2c2c00bcbd736b692299135b/persistStorage.py
-
curl -L -o persistStorage.py https://huggingface.co/spaces/Express-Analytics/QueryHelper/resolve/e7c1910ae94c764e2c2c00bcbd736b692299135b/persistStorage.py
1.9 kB
| from huggingface_hub import HfFileSystem | |
| from datetime import datetime | |
| import pytz | |
| import os | |
| from config import HUGGING_FACE_TOKEN | |
| import csv | |
| logsDir = os.getenv("HF_HOME", "/data") | |
| # # Create a new file | |
| # with open(os.path.join(data_dir, "my_data.txt"), "a") as f: | |
| # f.write("Hello World! From pesistent storage line 2") | |
| # # Read the data from the file | |
| # with open(os.path.join(data_dir, "my_data.txt"), "r") as f: | |
| # data = f.read() | |
| # # Print the data | |
| # print(data) | |
| # Set the time zone to Pacific Time Zone | |
| TIME_ZONE = 'US/Pacific' | |
| TIMEZONE_OBJ = pytz.timezone(TIME_ZONE) | |
| def append_dict_to_csv(file_path, row_data): | |
| fieldnames = row_data.keys() | |
| with open(file_path, 'a') as csv_file: | |
| csv_writer = csv.DictWriter(csv_file, fieldnames=fieldnames) | |
| # Check if the file is empty, and if so, write the header | |
| if csv_file.tell() == 0: | |
| csv_writer.writeheader() | |
| csv_writer.writerow(row_data) | |
| def saveLog(message, level='info') -> None: | |
| global logsDir | |
| if not os.path.isdir(logsDir): | |
| print("Log directory/Data Directory not available.") | |
| return | |
| current_time = datetime.now(TIMEZONE_OBJ) | |
| message = str(message) | |
| log_file_path = os.path.join(logsDir, f"{current_time.strftime('%Y-%m')}-log.csv") | |
| data_dict = {"time":str(current_time), "level": level, "message": message} | |
| append_dict_to_csv(log_file_path, data_dict) | |
| def getAllLogFilesPaths(): | |
| global logsDir | |
| # Save processed data to temporary file | |
| if not os.path.isdir(logsDir): | |
| print("Log directory/Data Directory not available.") | |
| return [] | |
| logFiles = [file for file in os.listdir(logsDir) if 'log' in file.lower()] | |
| print(logFiles,"avaiable logs") | |
| downloadableFilesPaths = [os.path.join(os.path.abspath(logsDir), logFilePath) for logFilePath in logFiles] | |
| return downloadableFilesPaths |