import json import os import filelock import huggingface_hub import pandas as pd from utils import ( build_datasets_urls, build_models_urls, build_text_icon, download_favicons, get_base_url, get_domain_name, ) HF_ICON = "https://huggingface.co/front/assets/huggingface_logo.svg" CROSS_ICON = "https://upload.wikimedia.org/wikipedia/commons/4/4e/Cross.png" DISABLE_ONLINE_CACHE = False ONLINE_CACHE = "CONDA-Workshop/RequestCache" def save_cache(cache_data, cache_file, initial_timestamp): print(f"Saving cache to {cache_file}") # Acquire lock before reading and updating the file to prevent race conditions with filelock.FileLock(f"{cache_file}.lock"): # Check if the file has been modified since the initial read current_timestamp = ( os.path.getmtime(cache_file) if os.path.exists(cache_file) else None ) if current_timestamp is None or initial_timestamp != current_timestamp: # File has been modified or created since initial read, re-read the file try: with open(cache_file, "r", encoding="utf8") as f: # Update the dictionary with newly added entries cache_dict = json.load(f) # Test if cache_dict and cache_data are different if cache_dict != cache_data: cache_data.update(cache_dict) except FileNotFoundError: pass # If the file doesn't exist at this point, continue with the current dictionary # Write the updated dictionary back to the file with open(cache_file, "w", encoding="utf8") as f: json.dump(cache_data, f, ensure_ascii=False, indent=4) if not DISABLE_ONLINE_CACHE: try: huggingface_hub.upload_file( repo_id=ONLINE_CACHE, repo_type="dataset", token=os.environ.get("TOKEN") or True, path_in_repo=cache_file, path_or_fileobj=cache_file, ) except Exception as e: print(f"Unable to upload {cache_file}: {e}") return cache_data def update_favicon_cache(sources): # Load the favicon dictionary if it exists favicon_dict = {} favicon_file_path = "favicons.json" initial_timestamp = None if not DISABLE_ONLINE_CACHE: try: huggingface_hub.hf_hub_download( repo_id=ONLINE_CACHE, repo_type="dataset", token=os.environ.get("TOKEN") or True, filename=favicon_file_path, local_dir=os.getcwd(), ) except Exception as e: print(f"Unable to download favicons.json: {e}") # Attempt to load the favicon dictionary and record its last modification time if os.path.exists(favicon_file_path): initial_timestamp = os.path.getmtime(favicon_file_path) try: with open(favicon_file_path, "r", encoding="utf8") as f: favicon_dict = json.load(f) except FileNotFoundError: pass # File not found, proceed with an empty dictionary # Determine which favicons need to be downloaded missing_domains = [domain for domain in sources if domain not in favicon_dict] # Download missing favicons in batch if missing_domains: new_favicon_urls = download_favicons(missing_domains) favicon_dict.update(new_favicon_urls) favicon_dict = save_cache( cache_data=favicon_dict, cache_file=favicon_file_path, initial_timestamp=initial_timestamp, ) return favicon_dict def update_model_url_cache(models): models = [x for x in models if x is not None] models = list(set(models)) # Load the model url dictionary if it exists model_url_dict = {} model_url_file_path = "model_urls.json" initial_timestamp = None if not DISABLE_ONLINE_CACHE: try: huggingface_hub.hf_hub_download( repo_id=ONLINE_CACHE, repo_type="dataset", token=os.environ.get("TOKEN") or True, filename=model_url_file_path, local_dir=os.getcwd(), ) except Exception as e: print(f"Unable to download model_urls.json: {e}") # Attempt to load the model url dictionary and record its last modification time if os.path.exists(model_url_file_path): initial_timestamp = os.path.getmtime(model_url_file_path) try: with open(model_url_file_path, "r", encoding="utf8") as f: model_url_dict = json.load(f) except FileNotFoundError: pass # File not found, proceed with an empty dictionary # Determine which model urls need to be downloaded missing_model_urls = [model for model in models if model not in model_url_dict] # Download missing model urls in batch if missing_model_urls: new_model_urls = build_models_urls(missing_model_urls) model_url_dict.update(new_model_urls) model_url_dict = save_cache( cache_data=model_url_dict, cache_file=model_url_file_path, initial_timestamp=initial_timestamp, ) return model_url_dict def update_dataset_url_cache(datasets): datasets = [x for x in datasets if x is not None] datasets = list(set(datasets)) # Load the dataset url dictionary if it exists dataset_url_dict = {} dataset_url_file_path = "dataset_urls.json" initial_timestamp = None if not DISABLE_ONLINE_CACHE: try: huggingface_hub.hf_hub_download( repo_id=ONLINE_CACHE, repo_type="dataset", token=os.environ.get("TOKEN") or True, filename=dataset_url_file_path, local_dir=os.getcwd(), ) except Exception as e: print(f"Unable to download dataset_urls.json: {e}") # Attempt to load the dataset url dictionary and record its last modification time if os.path.exists(dataset_url_file_path): initial_timestamp = os.path.getmtime(dataset_url_file_path) try: with open(dataset_url_file_path, "r", encoding="utf8") as f: dataset_url_dict = json.load(f) except FileNotFoundError: pass # File not found, proceed with an empty dictionary # Determine which dataset urls need to be downloaded missing_dataset_urls = [ dataset for dataset in datasets if dataset not in dataset_url_dict ] # Download missing dataset urls in batch if missing_dataset_urls: new_dataset_urls = build_datasets_urls(missing_dataset_urls) dataset_url_dict.update(new_dataset_urls) dataset_url_dict = save_cache( cache_data=dataset_url_dict, cache_file=dataset_url_file_path, initial_timestamp=initial_timestamp, ) return dataset_url_dict def get_dataframe(): # Load the contamination_report.csv file data = pd.read_csv("contamination_report.csv", delimiter=";", header=0) # Load the favicon dictionary if it exists favicon_dict = {} # Update the favicon dictionary favicon_dict = update_favicon_cache([get_base_url(x) for x in data["Reference"]]) # Update the model url dictionary model_url_dict = update_model_url_cache( data[data["Model or corpus"] == "model"]["Contaminated Source"] ) # Update the dataset url dictionary dataset_url_dict = update_dataset_url_cache( list(data["Evaluation Dataset"]) + list(data[data["Model or corpus"] == "corpus"]["Contaminated Source"]) ) # Add favicons URLs to the dataframe in a vectorized manner data["Reference"] = data["Reference"].apply( lambda x: build_text_icon( text=get_domain_name(x), url=x, icon_url=favicon_dict.get(get_base_url(x), ""), ) ) PR_URL_FORMAT = "https://huggingface.co/spaces/CONDA-Workshop/Data-Contamination-Report/discussions/{}" data["PR"] = data["PR"].apply( lambda x: build_text_icon( text="", url=PR_URL_FORMAT.format(int(x)) if not pd.isna(x) else "no link", icon_url=HF_ICON if x == x else CROSS_ICON, ) ) data["Evaluation Dataset"] = data["Evaluation Dataset"].apply( lambda x: build_text_icon( text=x, url=dataset_url_dict.get(x, ""), icon_url=HF_ICON, ) ) data["Evaluation Dataset"] = data.apply( lambda x: x["Evaluation Dataset"] + f" ({x['Subset']})" if pd.notna(x["Subset"]) else x["Evaluation Dataset"], axis=1, ) del data["Subset"] # For "Contaminated Source" use build_dataset_url if "Model or corpus" is "corpus" and build_model_url if "Model or corpus" is "model" data["Contaminated Source"] = data.apply( lambda x: build_text_icon( text=x["Contaminated Source"], url=dataset_url_dict.get(x["Contaminated Source"], "") if x["Model or corpus"] == "corpus" else model_url_dict.get(x["Contaminated Source"], ""), icon_url=HF_ICON, ), axis=1, ) data["Train Split"] = data["Train Split"].apply(lambda x: x/100 if x else x) data["Development Split"] = data["Development Split"].apply(lambda x: x/100 if x else x) data["Test Split"] = data["Test Split"].apply(lambda x: x/100 if x else x) return data