yjernite/space-privacy
8
1import logging2import os3import re4import tempfile5 6from huggingface_hub import HfApi, hf_hub_download7from huggingface_hub.utils import EntryNotFoundError, RepositoryNotFoundError8 9# Configure logging10logging.basicConfig(11 level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"12)13 14# Files/extensions to definitely include15INCLUDE_PATTERNS = [16 ".py",17 "requirements.txt",18 "Dockerfile",19 ".js",20 ".jsx",21 ".ts",22 ".tsx",23 ".html",24 ".css",25 ".svelte",26 ".vue",27 ".json",28 ".yaml",29 ".yml",30 ".toml",31 "Procfile",32 ".sh",33]34 35# Files/extensions/folders to ignore36IGNORE_PATTERNS = [37 ".git",38 ".hfignore",39 "README.md",40 "LICENSE",41 "__pycache__",42 ".ipynb_checkpoints",43 ".png",44 ".jpg",45 ".jpeg",46 ".gif",47 ".svg",48 ".ico",49 ".mp3",50 ".wav",51 ".mp4",52 ".mov",53 ".avi",54 ".onnx",55 ".pt",56 ".pth",57 ".bin",58 ".safetensors",59 ".tflite",60 ".pickle",61 ".pkl",62 ".joblib",63 ".parquet",64 ".csv",65 ".tsv",66 ".zip",67 ".tar.gz",68 ".gz",69 ".ipynb",70 ".DS_Store",71 "node_modules",72]73 74# Regex to find potential Hugging Face model IDs (e.g., "org/model-name", "user/model-name")75# This is a simple heuristic and might catch non-model strings or miss complex cases.76HF_MODEL_ID_PATTERN = re.compile(r"([\"\'])([\w\-.]+/[\w\-\.]+)\1\'")77 78# Max length for model descriptions to keep prompts manageable79MAX_MODEL_DESC_LENGTH = 150080 81SUMMARY_FILENAME = "summary_highlights.md"82PRIVACY_FILENAME = "privacy_report.md"83TLDR_FILENAME = "tldr_summary.json"84 85 86def _is_relevant_file(filename):87 """Check if a file should be included based on patterns."""88 # Ignore files matching ignore patterns (case-insensitive check for some)89 lower_filename = filename.lower()90 if any(91 pattern in lower_filename92 for pattern in [".git", ".hfignore", "readme.md", "license"]93 ):94 return False95 if any(96 filename.endswith(ext) for ext in IGNORE_PATTERNS if ext.startswith(".")97 ): # Check extensions98 return False99 if any(100 part == pattern101 for part in filename.split("/")102 for pattern in IGNORE_PATTERNS103 if "." not in pattern and "/" not in pattern104 ): # Check directory/file names105 return False106 if filename in IGNORE_PATTERNS: # Check full filenames107 return False108 109 # Include files matching include patterns110 if any(filename.endswith(ext) for ext in INCLUDE_PATTERNS if ext.startswith(".")):111 return True112 if any(filename == pattern for pattern in INCLUDE_PATTERNS if "." not in pattern):113 return True114 115 # Default to False if not explicitly included (safer)116 # logging.debug(f"File '{filename}' excluded by default.")117 return False118 119 120def get_space_code_files(space_id: str) -> dict[str, str]:121 """122 Downloads relevant code and configuration files from a Hugging Face Space.123 124 Args:125 space_id: The ID of the Hugging Face Space (e.g., 'gradio/hello_world').126 127 Returns:128 A dictionary where keys are filenames and values are file contents as strings.129 Returns an empty dictionary if the space is not found or has no relevant files.130 """131 code_files = {}132 api = HfApi()133 134 try:135 logging.info(f"Fetching file list for Space: {space_id}")136 repo_files = api.list_repo_files(repo_id=space_id, repo_type="space")137 logging.info(f"Found {len(repo_files)} total files in {space_id}.")138 139 relevant_files = [f for f in repo_files if _is_relevant_file(f)]140 logging.info(f"Identified {len(relevant_files)} relevant files for download.")141 142 for filename in relevant_files:143 try:144 logging.debug(f"Downloading {filename} from {space_id}...")145 file_path = hf_hub_download(146 repo_id=space_id,147 filename=filename,148 repo_type="space",149 # Consider adding use_auth_token=os.getenv("HF_TOKEN") if accessing private spaces later150 )151 with open(file_path, "r", encoding="utf-8", errors="ignore") as f:152 content = f.read()153 code_files[filename] = content154 logging.debug(f"Successfully read content of {filename}")155 except EntryNotFoundError:156 logging.warning(157 f"File {filename} listed but not found in repo {space_id}."158 )159 except UnicodeDecodeError:160 logging.warning(161 f"Could not decode file {filename} from {space_id} as UTF-8. Skipping."162 )163 except OSError as e:164 logging.warning(f"OS error reading file {filename} from cache: {e}")165 except Exception as e:166 logging.error(167 f"Unexpected error downloading or reading file {filename} from {space_id}: {e}"168 )169 170 except RepositoryNotFoundError:171 logging.error(f"Space repository '{space_id}' not found.")172 return {}173 except Exception as e:174 logging.error(f"Failed to list or process files for space {space_id}: {e}")175 return {}176 177 logging.info(178 f"Successfully retrieved content for {len(code_files)} files from {space_id}."179 )180 return code_files181 182 183def extract_hf_model_ids(code_files: dict[str, str]) -> set[str]:184 """185 Extracts potential Hugging Face model IDs mentioned in code files.186 187 Args:188 code_files: Dictionary of {filename: content}.189 190 Returns:191 A set of unique potential model IDs found.192 """193 potential_ids = set()194 for filename, content in code_files.items():195 # Limit search to relevant file types196 if filename.endswith((".py", ".json", ".yaml", ".yml", ".toml", ".md")):197 try:198 matches = HF_MODEL_ID_PATTERN.findall(content)199 for _, model_id in matches:200 # Basic validation: must contain exactly one '/'201 if model_id.count("/") == 1:202 # Avoid adding common paths that look like IDs203 if not any(204 part in model_id.lower()205 for part in ["http", "www", "@", " ", ".", ":"]206 ): # Check if '/' is only separator207 if len(model_id) < 100: # Avoid overly long strings208 potential_ids.add(model_id)209 except Exception as e:210 logging.warning(f"Regex error processing file {filename}: {e}")211 212 logging.info(f"Extracted {len(potential_ids)} potential model IDs.")213 # Add simple filter for very common false positives if needed214 # potential_ids = {id for id in potential_ids if id not in ['user/repo']}215 return potential_ids216 217 218def get_model_descriptions(model_ids: set[str]) -> dict[str, str]:219 """220 Fetches the README.md content (description) for a set of model IDs.221 222 Args:223 model_ids: A set of Hugging Face model IDs.224 225 Returns:226 A dictionary mapping model_id to its description string (or an error message).227 """228 descriptions = {}229 if not model_ids:230 return descriptions231 232 logging.info(f"Fetching descriptions for {len(model_ids)} models...")233 for model_id in model_ids:234 try:235 # Check if the model exists first (optional but good practice)236 # api.model_info(model_id)237 238 # Download README.md239 readme_path = hf_hub_download(240 repo_id=model_id,241 filename="README.md",242 repo_type="model",243 # Add token if needing to access private/gated models - unlikely for Space analysis244 # use_auth_token=os.getenv("HF_TOKEN"),245 error_if_not_found=True, # Raise error if README doesn't exist246 )247 with open(readme_path, "r", encoding="utf-8", errors="ignore") as f:248 description = f.read()249 descriptions[model_id] = description[:MAX_MODEL_DESC_LENGTH] + (250 "... [truncated]" if len(description) > MAX_MODEL_DESC_LENGTH else ""251 )252 logging.debug(f"Successfully fetched description for {model_id}")253 except RepositoryNotFoundError:254 logging.warning(f"Model repository '{model_id}' not found.")255 descriptions[model_id] = "[Model repository not found]"256 except EntryNotFoundError:257 logging.warning(f"README.md not found in model repository '{model_id}'.")258 descriptions[model_id] = "[README.md not found in model repository]"259 except Exception as e:260 logging.error(f"Error fetching description for model '{model_id}': {e}")261 descriptions[model_id] = f"[Error fetching description: {e}]"262 263 logging.info(f"Finished fetching descriptions for {len(descriptions)} models.")264 return descriptions265 266 267def list_cached_spaces(dataset_id: str, hf_token: str | None) -> list[str]:268 """Lists the space IDs (owner/name) that have cached reports in the dataset repository."""269 if not hf_token:270 logging.warning("HF Token not provided, cannot list cached spaces.")271 return []272 try:273 api = HfApi(token=hf_token)274 # Get all filenames in the dataset repository275 all_files = api.list_repo_files(repo_id=dataset_id, repo_type="dataset")276 277 # Extract unique directory paths that look like owner/space_name278 # by checking if they contain our specific report files.279 space_ids = set()280 for f_path in all_files:281 # Check if the file is one of our report files282 if f_path.endswith(f"/{PRIVACY_FILENAME}") or f_path.endswith(283 f"/{SUMMARY_FILENAME}"284 ):285 # Extract the directory path part (owner/space_name)286 parts = f_path.split("/")287 if len(parts) == 3: # Expecting owner/space_name/filename.md288 owner_slash_space_name = "/".join(parts[:-1])289 # Basic validation: owner and space name shouldn't start with '.'290 if not parts[0].startswith(".") and not parts[1].startswith("."):291 space_ids.add(owner_slash_space_name)292 293 sorted_space_ids = sorted(list(space_ids))294 logging.info(295 f"Found {len(sorted_space_ids)} cached space reports in {dataset_id} via HfApi."296 )297 return sorted_space_ids298 299 except RepositoryNotFoundError:300 logging.warning(301 f"Dataset {dataset_id} not found or empty when listing cached spaces."302 )303 return []304 except Exception as e:305 logging.error(f"Error listing cached spaces in {dataset_id} via HfApi: {e}")306 return [] # Return empty list on error307 308 309def check_report_exists(space_id: str, dataset_id: str, hf_token: str | None) -> bool:310 """Checks if report files already exist in the target dataset repo using HfApi."""311 print(312 f"[Debug Cache Check] Checking for space_id: '{space_id}' in dataset: '{dataset_id}'"313 ) # DEBUG314 if not hf_token:315 logging.warning("HF Token not provided, cannot check dataset cache.")316 print("[Debug Cache Check] No HF Token, returning False.") # DEBUG317 return False318 try:319 api = HfApi(token=hf_token)320 # List ALL files in the repo321 print(f"[Debug Cache Check] Listing ALL files in repo '{dataset_id}'") # DEBUG322 all_repo_files = api.list_repo_files(repo_id=dataset_id, repo_type="dataset")323 # DEBUG: Optionally print a subset if the list is huge324 # print(f"[Debug Cache Check] First 10 files returned by API: {all_repo_files[:10]}")325 326 # Construct the exact paths we expect for the target space_id327 expected_summary_path = f"{space_id}/{SUMMARY_FILENAME}"328 expected_privacy_path = f"{space_id}/{PRIVACY_FILENAME}"329 print(330 f"[Debug Cache Check] Expecting summary file: '{expected_summary_path}'"331 ) # DEBUG332 print(333 f"[Debug Cache Check] Expecting privacy file: '{expected_privacy_path}'"334 ) # DEBUG335 336 # Check if both expected paths exist in the full list of files337 summary_exists = expected_summary_path in all_repo_files338 privacy_exists = expected_privacy_path in all_repo_files339 exists = summary_exists and privacy_exists340 print(341 f"[Debug Cache Check] Summary exists in full list: {summary_exists}"342 ) # DEBUG343 print(344 f"[Debug Cache Check] Privacy exists in full list: {privacy_exists}"345 ) # DEBUG346 print(f"[Debug Cache Check] Overall exists check result: {exists}") # DEBUG347 return exists348 349 except RepositoryNotFoundError:350 logging.warning(351 f"Dataset repository {dataset_id} not found or not accessible during check."352 )353 print(354 f"[Debug Cache Check] Repository {dataset_id} not found, returning False."355 ) # DEBUG356 except Exception as e:357 # ... (error handling remains the same) ...358 print(f"[Debug Cache Check] Exception caught: {type(e).__name__}: {e}") # DEBUG359 # Note: 404 check based on path_in_repo is no longer applicable here360 # We rely on RepositoryNotFoundError or general Exception361 logging.error(362 f"Error checking dataset {dataset_id} for {space_id} via HfApi: {e}"363 )364 print("[Debug Cache Check] Other exception, returning False.") # DEBUG365 return False # Treat errors as cache miss366 367 368def download_cached_reports(369 space_id: str, dataset_id: str, hf_token: str | None370) -> dict[str, str]:371 """Downloads cached reports (summary, privacy, tldr json) from the dataset repo.372 373 Returns:374 Dict containing report contents keyed by 'summary', 'privacy', 'tldr_json_str'.375 Keys will be missing if a specific file is not found.376 Raises error on critical download failures (repo not found, etc.).377 """378 if not hf_token:379 raise ValueError("HF Token required to download cached reports.")380 381 logging.info(382 f"Attempting to download cached reports for {space_id} from {dataset_id}..."383 )384 reports = {}385 # Define paths relative to dataset root for hf_hub_download386 summary_repo_path = f"{space_id}/{SUMMARY_FILENAME}"387 privacy_repo_path = f"{space_id}/{PRIVACY_FILENAME}"388 tldr_repo_path = f"{space_id}/{TLDR_FILENAME}" # Path for TLDR JSON389 390 try:391 # Download summary392 try:393 summary_path_local = hf_hub_download(394 repo_id=dataset_id,395 filename=summary_repo_path,396 repo_type="dataset",397 token=hf_token,398 )399 with open(summary_path_local, "r", encoding="utf-8") as f:400 reports["summary"] = f.read()401 logging.info(f"Successfully downloaded cached summary for {space_id}.")402 except EntryNotFoundError:403 logging.warning(404 f"Cached summary file {summary_repo_path} not found for {space_id}."405 )406 except Exception as e_summary:407 logging.error(408 f"Error downloading cached summary for {space_id}: {e_summary}"409 )410 # Decide if this is critical - for now, we warn and continue411 412 # Download privacy report413 try:414 privacy_path_local = hf_hub_download(415 repo_id=dataset_id,416 filename=privacy_repo_path,417 repo_type="dataset",418 token=hf_token,419 )420 with open(privacy_path_local, "r", encoding="utf-8") as f:421 reports["privacy"] = f.read()422 logging.info(423 f"Successfully downloaded cached privacy report for {space_id}."424 )425 except EntryNotFoundError:426 logging.warning(427 f"Cached privacy file {privacy_repo_path} not found for {space_id}."428 )429 except Exception as e_privacy:430 logging.error(431 f"Error downloading cached privacy report for {space_id}: {e_privacy}"432 )433 # Decide if this is critical - for now, we warn and continue434 435 # Download TLDR JSON436 try:437 tldr_path_local = hf_hub_download(438 repo_id=dataset_id,439 filename=tldr_repo_path,440 repo_type="dataset",441 token=hf_token,442 )443 with open(tldr_path_local, "r", encoding="utf-8") as f:444 reports["tldr_json_str"] = f.read() # Store raw string content445 logging.info(f"Successfully downloaded cached TLDR JSON for {space_id}.")446 except EntryNotFoundError:447 logging.warning(448 f"Cached TLDR file {tldr_repo_path} not found for {space_id}."449 )450 # Don't treat TLDR absence as an error, just won't be in the dict451 except Exception as e_tldr:452 logging.error(453 f"Error downloading cached TLDR JSON for {space_id}: {e_tldr}"454 )455 # Don't treat TLDR download error as critical, just won't be included456 457 # Check if at least one report was downloaded successfully458 if not reports.get("summary") and not reports.get("privacy"):459 raise FileNotFoundError(460 f"Failed to download *any* primary cache files (summary/privacy) for {space_id}"461 )462 463 return reports464 465 except RepositoryNotFoundError as e_repo:466 logging.error(467 f"Cache download error: Dataset repo {dataset_id} not found. {e_repo}"468 )469 raise FileNotFoundError(f"Dataset repo {dataset_id} not found") from e_repo470 except Exception as e_critical: # Catch other potential critical errors471 logging.error(472 f"Unexpected critical error downloading cached reports for {space_id} from {dataset_id}: {e_critical}"473 )474 raise IOError(475 f"Failed critically during cached report download for {space_id}"476 ) from e_critical477 478 479def upload_reports_to_dataset(480 space_id: str,481 summary_report: str,482 detailed_report: str,483 dataset_id: str,484 hf_token: str | None,485):486 """Uploads the generated reports to the specified dataset repository."""487 if not hf_token:488 logging.warning("HF Token not provided, skipping dataset report upload.")489 return490 491 logging.info(492 f"Attempting to upload reports for {space_id} to dataset {dataset_id}..."493 )494 api = HfApi(token=hf_token)495 496 # Sanitize space_id for path safety (though HF Hub usually handles this)497 safe_space_id = space_id.replace("..", "")498 499 try:500 with tempfile.TemporaryDirectory() as tmpdir:501 summary_path_local = os.path.join(tmpdir, SUMMARY_FILENAME)502 privacy_path_local = os.path.join(tmpdir, PRIVACY_FILENAME)503 504 with open(summary_path_local, "w", encoding="utf-8") as f:505 f.write(summary_report)506 with open(privacy_path_local, "w", encoding="utf-8") as f:507 f.write(detailed_report)508 509 commit_message = f"Add privacy analysis reports for Space: {safe_space_id}"510 repo_url = api.create_repo(511 repo_id=dataset_id,512 repo_type="dataset",513 exist_ok=True,514 )515 logging.info(f"Ensured dataset repo {repo_url} exists.")516 517 api.upload_file(518 path_or_fileobj=summary_path_local,519 path_in_repo=f"{safe_space_id}/{SUMMARY_FILENAME}",520 repo_id=dataset_id,521 repo_type="dataset",522 commit_message=commit_message,523 )524 logging.info(f"Successfully uploaded summary report for {safe_space_id}.")525 526 api.upload_file(527 path_or_fileobj=privacy_path_local,528 path_in_repo=f"{safe_space_id}/{PRIVACY_FILENAME}",529 repo_id=dataset_id,530 repo_type="dataset",531 commit_message=commit_message,532 )533 logging.info(534 f"Successfully uploaded detailed privacy report for {safe_space_id}."535 )536 537 except Exception as e:538 logging.error(539 f"Failed to upload reports for {safe_space_id} to dataset {dataset_id}: {e}"540 )541 542 543# Example usage (for testing)544# if __name__ == '__main__':545# # Make sure HF_TOKEN is set if accessing private spaces or for higher rate limits546# from dotenv import load_dotenv547# load_dotenv()548# # test_space = "gradio/hello_world"549# test_space = "huggingface-projects/diffusers-gallery" # A more complex example550# # test_space = "nonexistent/space" # Test not found551# files_content = get_space_code_files(test_space)552# if files_content:553# print(f"\n--- Files retrieved from {test_space} ---")554# for name in files_content.keys():555# print(f"- {name}")556# # print("\n--- Content of app.py (first 200 chars) ---")557# # print(files_content.get("app.py", "app.py not found")[:200])558# else:559# print(f"Could not retrieve files from {test_space}")560