Scanning and Analysis

from safety.scan.main import process_files, download_policy, load_policy_file
from safety.scan.finder import FileFinder
from safety.scan.ecosystems.base import InspectableFile, Inspectable, Remediable
from safety.scan.models import ScanExport
from safety_schemas.models import (
    ConfigModel, FileType, PolicyFileModel, 
    ScanType, Stage, Ecosystem
)

FileFinder(
    max_level: int,                                    # Maximum directory depth
    ecosystems: List[Ecosystem],                       # Target ecosystems
    target: Path,                                      # Root directory to scan
    live_status=None,                                  # Progress status object
    exclude: Optional[List[str]] = None,               # Exclusion patterns
    include_files: Optional[Dict[FileType, List[Path]]] = None,  # Specific files
    handlers: Optional[Set[FileHandler]] = None        # Custom file handlers
)

def process_directory(
    self, 
    dir_path: str, 
    max_deep: Optional[int] = None
) -> Tuple[str, Dict[str, Set[Path]]]
    """
    Process directory and discover dependency files.
    
    Args:
        dir_path (str): Directory path to process
        max_deep (Optional[int]): Maximum depth override
        
    Returns:
        Tuple[str, Dict[str, Set[Path]]]: Directory info and discovered files
    """

def should_exclude(
    excludes: Set[Path], 
    to_analyze: Path
) -> bool
    """
    Determine if path should be excluded from analysis.
    
    Args:
        excludes (Set[Path]): Set of exclusion paths
        to_analyze (Path): Path to check
        
    Returns:
        bool: True if path should be excluded
    """

from safety.scan.finder import FileFinder
from safety_schemas.models import Ecosystem
from pathlib import Path

# Initialize file finder
finder = FileFinder(
    max_level=10,
    ecosystems=[Ecosystem.PYTHON],
    target=Path("."),
    exclude=["node_modules", ".git", "__pycache__"]
)

# Process directory
project_info, discovered_files = finder.process_directory("./my_project")

def process_files(
    paths: Dict[str, Set[Path]],                    # Discovered file paths by type
    config: Optional[ConfigModel] = None,           # Scan configuration
    use_server_matching: bool = False,              # Use server-side analysis
    obj=None,                                       # Context object
    target: Path = Path(".")                        # Project root path
) -> Generator[Tuple[Path, InspectableFile], None, None]
    """
    Process dependency files and yield inspectable file objects.
    
    Args:
        paths: Dictionary of file paths organized by file type
        config: Configuration model for analysis settings
        use_server_matching: Enable server-side vulnerability matching
        obj: Context object with authentication and settings
        target: Root directory path
        
    Yields:
        Tuple[Path, InspectableFile]: File path and corresponding inspectable file
    """

from safety.scan.main import process_files
from safety_schemas.models import ConfigModel

# Configure scan settings
config = ConfigModel(
    telemetry_enabled=True
)

# Process discovered files
for file_path, inspectable_file in process_files(
    paths=discovered_files,
    config=config,
    use_server_matching=True,
    target=Path(".")
):
    # Access vulnerability results
    results = inspectable_file.inspect(config=config)
    print(f"File: {file_path}")
    print(f"Dependencies: {len(results.dependencies)}")
    print(f"Vulnerabilities: {len(results.vulnerabilities)}")

class Inspectable(ABC):
    @abstractmethod
    def inspect(self, config: ConfigModel) -> DependencyResultModel:
        """Analyze file for dependencies and vulnerabilities."""
        
class Remediable(ABC):
    @abstractmethod
    def remediate(self):
        """Apply automatic fixes to detected vulnerabilities."""

class InspectableFile(Inspectable):
    file_type: FileType                    # Type of dependency file
    ecosystem: Ecosystem                   # Target ecosystem (Python, etc.)
    
    def inspect(self, config: ConfigModel) -> DependencyResultModel:
        """
        Analyze file for dependencies and vulnerabilities.
        
        Args:
            config: Configuration model with scan settings
            
        Returns:
            DependencyResultModel: Analysis results with dependencies and vulnerabilities
        """
        
    def remediate(self) -> None:
        """
        Apply automatic remediation to fix detected vulnerabilities.
        Modifies dependency specifications to use secure versions.
        """

def load_policy_file(path: Path) -> Optional[PolicyFileModel]:
    """
    Load policy file from specified path.
    
    Args:
        path (Path): Path to policy file (.safety-policy.yml)
        
    Returns:
        Optional[PolicyFileModel]: Parsed policy model or None if invalid
        
    Raises:
        SafetyError: If policy file is invalid or unsupported version
    """

def download_policy(
    session: SafetyAuthSession,                     # Authenticated session
    project_id: str,                               # Project identifier
    stage: Stage,                                  # Environment stage
    branch: Optional[str]                          # Git branch name
) -> Optional[PolicyFileModel]:
    """
    Download policy file from Safety platform.
    
    Args:
        session: Authenticated Safety session
        project_id: Unique project identifier
        stage: Development stage (development, production, etc.)
        branch: Git branch for branch-specific policies
        
    Returns:
        Optional[PolicyFileModel]: Downloaded policy model
        
    Raises:
        SafetyError: If download fails or policy is invalid
    """

from safety_schemas.models import PolicyFileModel, PolicySource, Stage

class PolicyFileModel:
    id: str                                        # Policy identifier
    source: PolicySource                           # Source (local, cloud)
    location: Optional[Path]                       # Local file path
    config: ConfigModel                            # Policy configuration

class PolicySource(Enum):
    LOCAL = "local"                               # Local policy file
    CLOUD = "cloud"                               # Platform policy

class Stage(Enum):
    DEVELOPMENT = "development"
    TESTING = "testing"
    STAGING = "staging"
    PRODUCTION = "production"

from safety_schemas.models import ConfigModel

class ConfigModel:
    telemetry_enabled: bool = True                 # Enable usage telemetry
    
    # Vulnerability filtering
    ignore_vulnerabilities: List[str] = []         # Vulnerability IDs to ignore
    ignore_unpinned_requirements: bool = False     # Ignore unpinned dependencies
    
    # Severity filtering  
    minimum_severity: Optional[str] = None         # Minimum severity level
    
    # Output configuration
    output_format: str = "screen"                  # Output format preference
    detailed_output: bool = False                  # Include detailed information
    
    # Auto-remediation settings
    auto_remediation_limit: Optional[int] = None   # Max automatic fixes
    continue_on_error: bool = False                # Continue despite errors
    
    @classmethod
    def parse_policy_file(cls, raw_report: Path) -> "ConfigModel":
        """
        Parse policy file and create configuration model.
        
        Args:
            raw_report (Path): Path to policy YAML file
            
        Returns:
            ConfigModel: Parsed configuration
            
        Raises:
            ValidationError: If policy file is invalid
        """
        
    @classmethod  
    def from_v30(cls, obj) -> "ConfigModel":
        """
        Create ConfigModel from v3.0 policy format.
        
        Args:
            obj: v3.0 format policy object
            
        Returns:
            ConfigModel: Converted configuration
        """

from safety_schemas.models import Ecosystem, FileType

class Ecosystem(Enum):
    PYTHON = "python"                             # Python packages and requirements
    # Additional ecosystems may be added in future versions

class FileType(Enum):
    # Python ecosystem file types
    REQUIREMENTS_TXT = "requirements.txt"         # pip requirements files
    PIPFILE = "Pipfile"                          # Pipenv files
    PIPFILE_LOCK = "Pipfile.lock"                # Pipenv lock files
    PYPROJECT_TOML = "pyproject.toml"           # PEP 518 project files
    POETRY_LOCK = "poetry.lock"                 # Poetry lock files
    SETUP_PY = "setup.py"                       # setuptools setup files
    SETUP_CFG = "setup.cfg"                     # setuptools configuration
    
    @property
    def ecosystem(self) -> Optional[Ecosystem]:
        """Get the ecosystem associated with this file type."""

from safety.scan.finder.handlers import FileHandler, ECOSYSTEM_HANDLER_MAPPING

class FileHandler(ABC):
    """Abstract base class for ecosystem-specific file handlers."""
    
    @abstractmethod
    def can_handle(self, file_path: Path) -> bool:
        """Determine if handler can process the given file."""
        
    @abstractmethod
    def process(self, file_path: Path) -> Set[FileType]:
        """Process file and return supported file types."""

# Handler mapping
ECOSYSTEM_HANDLER_MAPPING: Dict[Ecosystem, Type[FileHandler]] = {
    Ecosystem.PYTHON: PythonHandler,
    # Additional handlers for future ecosystems
}

from safety.scan.models import ScanExport

class ScanExport(Enum):
    JSON = "json"                                # JSON format export
    HTML = "html"                               # HTML report export  
    TEXT = "text"                               # Plain text export
    SPDX = "spdx"                              # SPDX format (with spdx extra)

from safety_schemas.models import DependencyResultModel

class DependencyResultModel:
    """Results from dependency analysis and vulnerability scanning."""
    
    dependencies: List[DependencyModel]           # Discovered dependencies
    vulnerabilities: List[VulnerabilityModel]    # Found vulnerabilities
    metadata: Dict[str, Any]                     # Additional metadata
    
class DependencyModel:
    """Individual dependency information."""
    
    name: str                                    # Package name
    version: str                                # Package version
    ecosystem: Ecosystem                         # Package ecosystem
    file_path: Path                             # Source file path
    
class VulnerabilityModel:
    """Vulnerability information for dependencies."""
    
    vulnerability_id: str                        # Unique vulnerability ID
    package_name: str                           # Affected package
    severity: str                               # Vulnerability severity
    advisory: str                               # Vulnerability description
    fixed_versions: List[str]                   # Versions with fixes

from safety.scan.util import GIT

# Git metadata extraction for enhanced reporting
def build_meta(target: Path) -> Dict[str, Any]:
    """
    Build metadata for scan context including Git information.
    
    Args:
        target (Path): Project root directory
        
    Returns:
        Dict[str, Any]: Metadata including Git branch, commit, etc.
    """

from safety.encoding import detect_encoding

def detect_encoding(file_path: Path) -> str:
    """
    Detect file encoding for proper text processing.
    
    Args:
        file_path (Path): Path to file for encoding detection
        
    Returns:
        str: Detected encoding (e.g., 'utf-8', 'latin-1')
    """

from safety.scan.finder import FileFinder
from safety.scan.main import process_files
from safety_schemas.models import Ecosystem, ConfigModel
from pathlib import Path

# Configure scan
config = ConfigModel(
    telemetry_enabled=False,
    detailed_output=True,
    auto_remediation_limit=5
)

# Discover files
finder = FileFinder(
    max_level=5,
    ecosystems=[Ecosystem.PYTHON],
    target=Path("./my_project"),
    exclude=[".venv", ".git", "node_modules"]
)

_, discovered_files = finder.process_directory("./my_project")

# Process and analyze
vulnerabilities_found = []
for file_path, inspectable_file in process_files(
    paths=discovered_files,
    config=config,
    target=Path("./my_project")
):
    results = inspectable_file.inspect(config=config)
    if results.vulnerabilities:
        vulnerabilities_found.extend(results.vulnerabilities)
        
    # Apply automatic remediation
    inspectable_file.remediate()

print(f"Found {len(vulnerabilities_found)} vulnerabilities")

from safety.scan.main import load_policy_file, download_policy
from safety.auth.utils import SafetyAuthSession

# Load local policy
policy = load_policy_file(Path("./.safety-policy.yml"))

if policy:
    config = policy.config
else:
    # Download from platform (requires authentication)
    session = SafetyAuthSession()
    policy = download_policy(
        session=session,
        project_id="my-project-id", 
        stage=Stage.PRODUCTION,
        branch="main"
    )
    config = policy.config if policy else ConfigModel()

# Use policy configuration for scanning
for file_path, inspectable_file in process_files(
    paths=discovered_files,
    config=config
):
    results = inspectable_file.inspect(config=config)
    # Process results according to policy

from safety_schemas.models import FileType

# Include specific files in scan
include_files = {
    FileType.REQUIREMENTS_TXT: [
        Path("./requirements.txt"),
        Path("./requirements-dev.txt")
    ],
    FileType.PYPROJECT_TOML: [
        Path("./pyproject.toml")
    ]
}

finder = FileFinder(
    max_level=1,
    ecosystems=[Ecosystem.PYTHON],
    target=Path("."),
    include_files=include_files
)