| 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!!!!" |
| |
| |
