|
import os |
|
import json |
|
import shutil |
|
import requests |
|
import re as r |
|
from urllib.request import urlopen |
|
from datetime import datetime |
|
from huggingface_hub import Repository, upload_file |
|
|
|
HF_TOKEN = os.environ.get("HF_TOKEN") |
|
DATASET_NAME = "audio_emotion_recognition_dataset" |
|
DATASET_REPO_URL = f"https://huggingface.co/datasets/pragnakalp/{DATASET_NAME}" |
|
DATASET_REPO_ID = "pragnakalp/audio_emotion_recognition_dataset" |
|
print("is none?", HF_TOKEN is None) |
|
REPOSITORY_DIR = "audio_logs" |
|
LOCAL_DIR = 'data_local' |
|
os.makedirs(LOCAL_DIR,exist_ok=True) |
|
|
|
repo = Repository( |
|
local_dir="audio_data", clone_from=DATASET_REPO_URL, use_auth_token=HF_TOKEN |
|
) |
|
repo.git_pull() |
|
|
|
def getIP(): |
|
ip_address = '' |
|
try: |
|
d = str(urlopen('http://checkip.dyndns.com/') |
|
.read()) |
|
|
|
return r.compile(r'Address: (\d+\.\d+\.\d+\.\d+)').search(d).group(1) |
|
except Exception as e: |
|
print("Error while getting IP address -->",e) |
|
return ip_address |
|
|
|
def get_location(ip_addr): |
|
location = {} |
|
try: |
|
ip=ip_addr |
|
|
|
req_data={ |
|
"ip":ip, |
|
"token":"pkml123" |
|
} |
|
url = "https://demos.pragnakalp.com/get-ip-location" |
|
|
|
|
|
|
|
headers = {'Content-Type': 'application/json'} |
|
|
|
response = requests.request("POST", url, headers=headers, data=json.dumps(req_data)) |
|
response = response.json() |
|
print("response======>>",response) |
|
return response |
|
except Exception as e: |
|
print("Error while getting location -->",e) |
|
return location |
|
|
|
""" |
|
Save generated details |
|
""" |
|
def dump_json(thing,file): |
|
with open(file,'w+',encoding="utf8") as f: |
|
json.dump(thing,f) |
|
|
|
def flag(audio_file_path,get_audio_name,final_output): |
|
|
|
print("saving data------------------------") |
|
|
|
ip_address= getIP() |
|
print(ip_address) |
|
location = get_location(ip_address) |
|
print(location) |
|
|
|
adversarial_number = 0 |
|
adversarial_number = 0 if None else adversarial_number |
|
|
|
metadata_name = datetime.now().strftime('%Y-%m-%d %H-%M-%S') |
|
SAVE_FILE_DIR = os.path.join(LOCAL_DIR,metadata_name) |
|
os.makedirs(SAVE_FILE_DIR,exist_ok=True) |
|
audio_save_path = os.path.join(SAVE_FILE_DIR,get_audio_name) |
|
print("old audio_file_path :",audio_file_path) |
|
print("new audio_path :",audio_save_path) |
|
try: |
|
shutil.copyfile(audio_file_path, audio_save_path) |
|
except Exception: |
|
raise Exception(f"Had issues saving PIL image to file") |
|
|
|
|
|
json_file_path = os.path.join(SAVE_FILE_DIR,'metadata.jsonl') |
|
metadata= {'id':metadata_name,'file_name':get_audio_name,'output':final_output,'ip_address':ip_address, 'location':location} |
|
|
|
dump_json(metadata,json_file_path) |
|
|
|
|
|
|
|
repo_audio_path = os.path.join(REPOSITORY_DIR,os.path.join(metadata_name,get_audio_name)) |
|
|
|
_ = upload_file(path_or_fileobj = audio_save_path, |
|
path_in_repo=repo_audio_path, |
|
repo_id=DATASET_REPO_ID, |
|
repo_type='dataset', |
|
token=HF_TOKEN |
|
) |
|
|
|
|
|
repo_json_path = os.path.join(REPOSITORY_DIR,os.path.join(metadata_name,'metadata.jsonl')) |
|
_ = upload_file(path_or_fileobj = json_file_path, |
|
path_in_repo =repo_json_path, |
|
repo_id= DATASET_REPO_ID, |
|
repo_type='dataset', |
|
token=HF_TOKEN |
|
) |
|
adversarial_number+=1 |
|
repo.git_pull() |
|
|
|
url = 'http://pragnakalpdev35.pythonanywhere.com/HF_space_audio_emotion' |
|
myobj = {'audio_path': repo_audio_path,'final_output':final_output,'ip_addr':ip_address, 'loc':location} |
|
|
|
x = requests.post(url, json = myobj) |
|
print("mail status code",x.status_code) |
|
return "*****Logs save successfully!!!!" |
|
|
|
|
