Spaces:
Paused
Paused
| import os | |
| import re | |
| import json | |
| import time | |
| import logging | |
| import requests | |
| from typing import Dict, List, Optional, Tuple, Any | |
| from dataclasses import dataclass, field | |
| # Configure logging | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' | |
| ) | |
| logger = logging.getLogger(__name__) | |
| class ContactInfo: | |
| """User contact information needed for opt-out requests.""" | |
| first_name: str | |
| last_name: str | |
| email: str | |
| phone: Optional[str] = None | |
| address: Optional[str] = None | |
| city: Optional[str] = None | |
| state: Optional[str] = None | |
| zip_code: Optional[str] = None | |
| class DataBroker: | |
| """Information about a data broker and its opt-out process.""" | |
| name: str | |
| url: str | |
| opt_out_url: str | |
| email: Optional[str] = None | |
| requires_email: bool = False | |
| requires_form: bool = False | |
| requires_mail: bool = False | |
| requires_id: bool = False | |
| automated_removal_supported: bool = False | |
| opt_out_steps: List[str] = field(default_factory=list) | |
| notes: Optional[str] = None | |
| class SmolAgent: | |
| """ | |
| An agent for automating data broker opt-outs and personal information removal. | |
| """ | |
| def __init__(self): | |
| self.data_brokers = self._load_data_brokers() | |
| self.user_info = None | |
| self.completed_opt_outs = [] | |
| self.failed_opt_outs = [] | |
| self.pending_opt_outs = [] | |
| def _load_data_brokers(self) -> List[DataBroker]: | |
| """Load the database of data brokers.""" | |
| try: | |
| # In a production environment, this would load from a more comprehensive database | |
| # For demo purposes, we'll include a few common data brokers | |
| brokers = [ | |
| DataBroker( | |
| name="Acxiom", | |
| url="https://www.acxiom.com", | |
| opt_out_url="https://www.acxiom.com/opt-out-choices/", | |
| requires_form=True, | |
| automated_removal_supported=False, | |
| opt_out_steps=[ | |
| "Visit opt-out page", | |
| "Fill out form with personal information", | |
| "Submit request", | |
| "Check email for confirmation" | |
| ] | |
| ), | |
| DataBroker( | |
| name="Spokeo", | |
| url="https://www.spokeo.com", | |
| opt_out_url="https://www.spokeo.com/opt-out", | |
| requires_form=True, | |
| automated_removal_supported=True, | |
| opt_out_steps=[ | |
| "Search for your profile", | |
| "Copy the URL of your profile", | |
| "Visit opt-out page", | |
| "Enter profile URL and email", | |
| "Complete CAPTCHA", | |
| "Check email for confirmation link" | |
| ] | |
| ), | |
| DataBroker( | |
| name="BeenVerified", | |
| url="https://www.beenverified.com", | |
| opt_out_url="https://www.beenverified.com/app/optout/search", | |
| requires_form=True, | |
| automated_removal_supported=True, | |
| opt_out_steps=[ | |
| "Search for your records", | |
| "Select all records that match", | |
| "Provide email address", | |
| "Complete verification steps", | |
| "Check email for confirmation" | |
| ] | |
| ), | |
| DataBroker( | |
| name="PeopleFinders", | |
| url="https://www.peoplefinders.com", | |
| opt_out_url="https://www.peoplefinders.com/opt-out", | |
| requires_form=True, | |
| automated_removal_supported=False, | |
| opt_out_steps=[ | |
| "Search for your listing", | |
| "Select your record", | |
| "Enter your email address", | |
| "Complete verification", | |
| "Confirm via email" | |
| ] | |
| ), | |
| DataBroker( | |
| name="WhitePages", | |
| url="https://www.whitepages.com", | |
| opt_out_url="https://www.whitepages.com/suppression-requests", | |
| requires_form=True, | |
| automated_removal_supported=True, | |
| opt_out_steps=[ | |
| "Find your listing", | |
| "Copy the URL", | |
| "Go to opt-out page", | |
| "Enter profile URL", | |
| "Provide reason for removal", | |
| "Verify email" | |
| ] | |
| ) | |
| ] | |
| return brokers | |
| except Exception as e: | |
| logger.error(f"Error loading data brokers: {e}") | |
| return [] | |
| def set_user_info(self, contact_info: ContactInfo) -> None: | |
| """Set the user's contact information.""" | |
| self.user_info = contact_info | |
| logger.info(f"User information set for {contact_info.first_name} {contact_info.last_name}") | |
| def search_online_presence(self) -> Dict[str, List[str]]: | |
| """ | |
| Search for the user's information across various websites. | |
| Returns a dictionary of sites and URLs where the user's information was found. | |
| """ | |
| if not self.user_info: | |
| raise ValueError("User information must be set before searching") | |
| results = {} | |
| # In a real implementation, this would use search APIs or web scraping | |
| # For demo purposes, we'll simulate finding information | |
| logger.info(f"Searching for online presence of {self.user_info.first_name} {self.user_info.last_name}") | |
| for broker in self.data_brokers: | |
| # Simulate searching each data broker | |
| # In reality, this would involve more sophisticated searching | |
| results[broker.name] = [f"{broker.url}/simulated-profile/{self.user_info.first_name.lower()}-{self.user_info.last_name.lower()}"] | |
| return results | |
| def generate_opt_out_plan(self) -> List[Dict[str, Any]]: | |
| """Generate a plan for opting out of data brokers.""" | |
| if not self.user_info: | |
| raise ValueError("User information must be set before generating a plan") | |
| plan = [] | |
| for broker in self.data_brokers: | |
| plan.append({ | |
| "broker": broker.name, | |
| "url": broker.opt_out_url, | |
| "automated": broker.automated_removal_supported, | |
| "steps": broker.opt_out_steps, | |
| "status": "pending" | |
| }) | |
| self.pending_opt_outs.append(broker.name) | |
| logger.info(f"Generated opt-out plan for {len(plan)} data brokers") | |
| return plan | |
| def execute_automated_opt_outs(self) -> Tuple[List[str], List[str]]: | |
| """ | |
| Execute automated opt-out requests for supported data brokers. | |
| Returns lists of successful and failed opt-outs. | |
| """ | |
| if not self.user_info: | |
| raise ValueError("User information must be set before executing opt-outs") | |
| success = [] | |
| failures = [] | |
| for broker in self.data_brokers: | |
| if not broker.automated_removal_supported: | |
| logger.info(f"Skipping {broker.name} - automated removal not supported") | |
| continue | |
| try: | |
| logger.info(f"Attempting automated opt-out for {broker.name}") | |
| # Simulate the opt-out process | |
| # In a real implementation, this would use web automation or APIs | |
| time.sleep(2) # Simulate request time | |
| # Simulate success (80% of the time for demo) | |
| if hash(broker.name + self.user_info.email) % 10 < 8: | |
| success.append(broker.name) | |
| self.completed_opt_outs.append(broker.name) | |
| self.pending_opt_outs.remove(broker.name) | |
| logger.info(f"Successfully completed opt-out for {broker.name}") | |
| else: | |
| failures.append(broker.name) | |
| self.failed_opt_outs.append(broker.name) | |
| self.pending_opt_outs.remove(broker.name) | |
| logger.error(f"Failed to complete opt-out for {broker.name}") | |
| except Exception as e: | |
| logger.error(f"Error during opt-out for {broker.name}: {e}") | |
| failures.append(broker.name) | |
| self.failed_opt_outs.append(broker.name) | |
| if broker.name in self.pending_opt_outs: | |
| self.pending_opt_outs.remove(broker.name) | |
| return success, failures | |
| def generate_manual_instructions(self) -> Dict[str, List[str]]: | |
| """Generate manual instructions for data brokers that don't support automation.""" | |
| if not self.user_info: | |
| raise ValueError("User information must be set before generating instructions") | |
| instructions = {} | |
| for broker in self.data_brokers: | |
| if not broker.automated_removal_supported and broker.name not in self.completed_opt_outs: | |
| instructions[broker.name] = [ | |
| f"Visit: {broker.opt_out_url}", | |
| *broker.opt_out_steps | |
| ] | |
| logger.info(f"Generated manual instructions for {len(instructions)} data brokers") | |
| return instructions | |
| def generate_progress_report(self) -> Dict[str, Any]: | |
| """Generate a report on the progress of data removal.""" | |
| completed_percentage = len(self.completed_opt_outs) / len(self.data_brokers) * 100 if self.data_brokers else 0 | |
| report = { | |
| "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), | |
| "total_brokers": len(self.data_brokers), | |
| "completed": { | |
| "count": len(self.completed_opt_outs), | |
| "percentage": completed_percentage, | |
| "brokers": self.completed_opt_outs | |
| }, | |
| "pending": { | |
| "count": len(self.pending_opt_outs), | |
| "percentage": len(self.pending_opt_outs) / len(self.data_brokers) * 100 if self.data_brokers else 0, | |
| "brokers": self.pending_opt_outs | |
| }, | |
| "failed": { | |
| "count": len(self.failed_opt_outs), | |
| "percentage": len(self.failed_opt_outs) / len(self.data_brokers) * 100 if self.data_brokers else 0, | |
| "brokers": self.failed_opt_outs | |
| } | |
| } | |
| logger.info(f"Generated progress report: {completed_percentage:.1f}% complete") | |
| return report | |
| def save_state(self, filename: str = "agent_state.json") -> None: | |
| """Save the agent's state to a file.""" | |
| state = { | |
| "user_info": vars(self.user_info) if self.user_info else None, | |
| "completed_opt_outs": self.completed_opt_outs, | |
| "failed_opt_outs": self.failed_opt_outs, | |
| "pending_opt_outs": self.pending_opt_outs | |
| } | |
| with open(filename, 'w') as f: | |
| json.dump(state, f, indent=2) | |
| logger.info(f"Saved agent state to {filename}") | |
| def load_state(self, filename: str = "agent_state.json") -> None: | |
| """Load the agent's state from a file.""" | |
| try: | |
| with open(filename, 'r') as f: | |
| state = json.load(f) | |
| if state.get("user_info"): | |
| self.user_info = ContactInfo(**state["user_info"]) | |
| self.completed_opt_outs = state.get("completed_opt_outs", []) | |
| self.failed_opt_outs = state.get("failed_opt_outs", []) | |
| self.pending_opt_outs = state.get("pending_opt_outs", []) | |
| logger.info(f"Loaded agent state from {filename}") | |
| except FileNotFoundError: | |
| logger.warning(f"State file {filename} not found") | |
| except Exception as e: | |
| logger.error(f"Error loading state: {e}") | |
| def monitor_removals(self, check_interval_days: int = 30) -> None: | |
| """ | |
| Set up monitoring to check if information has been successfully removed. | |
| For demonstration purposes only. | |
| """ | |
| # In a real implementation, this would set up scheduled tasks | |
| # For now, just log the intent | |
| logger.info(f"Set up monitoring to check removals every {check_interval_days} days") | |
| # Hugging Face integration code | |
| from fastapi import FastAPI, HTTPException | |
| from pydantic import BaseModel | |
| app = FastAPI(title="SMOLAGENT Data Removal Assistant") | |
| agent = SmolAgent() | |
| class UserInfoRequest(BaseModel): | |
| first_name: str | |
| last_name: str | |
| email: str | |
| phone: Optional[str] = None | |
| address: Optional[str] = None | |
| city: Optional[str] = None | |
| state: Optional[str] = None | |
| zip_code: Optional[str] = None | |
| class OptOutRequest(BaseModel): | |
| automated_only: bool = False | |
| def set_user_info(user_info: UserInfoRequest): | |
| """Set the user's contact information.""" | |
| contact_info = ContactInfo( | |
| first_name=user_info.first_name, | |
| last_name=user_info.last_name, | |
| email=user_info.email, | |
| phone=user_info.phone, | |
| address=user_info.address, | |
| city=user_info.city, | |
| state=user_info.state, | |
| zip_code=user_info.zip_code | |
| ) | |
| agent.set_user_info(contact_info) | |
| return {"status": "success", "message": "User information set successfully"} | |
| def search_presence(): | |
| """Search for the user's online presence.""" | |
| try: | |
| results = agent.search_online_presence() | |
| return {"status": "success", "results": results} | |
| except ValueError as e: | |
| raise HTTPException(status_code=400, detail=str(e)) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Search failed: {str(e)}") | |
| def generate_plan(): | |
| """Generate an opt-out plan.""" | |
| try: | |
| plan = agent.generate_opt_out_plan() | |
| return {"status": "success", "plan": plan} | |
| except ValueError as e: | |
| raise HTTPException(status_code=400, detail=str(e)) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Plan generation failed: {str(e)}") | |
| def execute_opt_outs(request: OptOutRequest): | |
| """Execute automated opt-out requests.""" | |
| try: | |
| success, failures = agent.execute_automated_opt_outs() | |
| return { | |
| "status": "completed", | |
| "successful": success, | |
| "failed": failures | |
| } | |
| except ValueError as e: | |
| raise HTTPException(status_code=400, detail=str(e)) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Execution failed: {str(e)}") | |
| def get_manual_instructions(): | |
| """Get manual instructions for brokers that don't support automation.""" | |
| try: | |
| instructions = agent.generate_manual_instructions() | |
| return {"status": "success", "instructions": instructions} | |
| except ValueError as e: | |
| raise HTTPException(status_code=400, detail=str(e)) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Failed to generate instructions: {str(e)}") | |
| def get_progress_report(): | |
| """Get a progress report on data removal.""" | |
| try: | |
| report = agent.generate_progress_report() | |
| return {"status": "success", "report": report} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Failed to generate report: {str(e)}") | |
| def save_state(): | |
| """Save the current state of the agent.""" | |
| try: | |
| agent.save_state() | |
| return {"status": "success", "message": "State saved successfully"} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Failed to save state: {str(e)}") | |
| def load_state(): | |
| """Load the saved state of the agent.""" | |
| try: | |
| agent.load_state() | |
| return {"status": "success", "message": "State loaded successfully"} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Failed to load state: {str(e)}") | |
| # For Hugging Face Spaces deployment | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run(app, host="0.0.0.0", port=7860) | |