Set Algebra & Venn Operations

def verify_rbac_access(user_perms: set[str], required_perms: set[str]) -> dict:
    """
    Verifies user permissions using O(1) set algebra operations.
    """
    # 1. Check if user possesses all required permissions
    is_authorized = required_perms.issubset(user_perms)
    
    # 2. Compute missing permissions using set difference
    missing = required_perms - user_perms
    
    # 3. Compute granted permissions using set intersection
    granted = user_perms & required_perms
    
    # 4. Compute excess permissions using set difference
    excess = user_perms - required_perms
    
    return {
        "authorized": is_authorized,
        "missing": sorted(missing),
        "granted": sorted(granted),
        "excess": sorted(excess)
    }

# Test scenarios:
admin_tokens = {"read", "write", "delete", "audit"}
editor_tokens = {"read", "write"}
viewer_tokens = {"read"}

endpoint_requirement = {"read", "write", "delete"}

print("=" * 60)
print(f"{'RBAC GATEWAY ACCESS EVALUATOR':^60}")
print("=" * 60)

for role_name, token_set in [("Admin", admin_tokens), ("Editor", editor_tokens), ("Viewer", viewer_tokens)]:
    result = verify_rbac_access(token_set, endpoint_requirement)
    status = "AUTHORIZED" if result["authorized"] else "FORBIDDEN"
    print(f"Role: {role_name:<8} -> Status: {status}")
    if not result["authorized"]:
        print(f"   Missing Scopes: {result['missing']}")
    else:
        print(f"   Excess Scopes:  {result['excess']}")
    print("-" * 60)

def sanitize_user_profile(raw_payload: dict) -> dict:
    """
    Sanitizes raw user dictionaries with safe key retrieval and tag deduplication.
    """
    # 1. Safe extraction with get() fallbacks
    user_id = raw_payload.get("user_id", "anonymous")
    
    raw_name = raw_payload.get("display_name", "Guest User")
    clean_name = raw_name.strip().title() if isinstance(raw_name, str) else "Guest User"
    
    tier = raw_payload.get("tier", "free").strip().lower()
    
    # 2. Extract and deduplicate interest tags using a set
    raw_tags = raw_payload.get("tags", [])
    # Case-normalize and strip tags, then deduplicate in a set
    clean_tag_set = {t.strip().lower() for t in raw_tags if isinstance(t, str) and t.strip()}
    
    # 3. Construct clean record
    return {
        "user_id": user_id,
        "display_name": clean_name,
        "tier": tier,
        "tags": sorted(clean_tag_set),  # Store sorted for deterministic output
        "verified": raw_payload.get("verified", False)
    }

incoming_json = {
    "user_id": "usr_8831",
    "display_name": "   dr. alexander v. cross   ",
    "tier": "ENTERPRISE",
    "tags": ["Machine Learning", "ai", "CLOUD", "ai", "Machine Learning ", "python"]
    # 'verified' key is intentionally omitted from payload
}

clean_record = sanitize_user_profile(incoming_json)
print("=" * 60)
print(f"{'INGESTED USER PROFILE NORMALIZATION':^60}")
print("=" * 60)
for k, v in clean_record.items():
    print(f"{k:<15}: {v}")
print("=" * 60)

def process_crawler_batch(
    incoming_urls: list[str],
    visited_cache: set[str]
) -> tuple[dict[str, list[str]], set[str]]:
    """
    Filters unvisited URLs and partitions them by domain destination.
    """
    # 1. Convert incoming URLs to set (deduplicates within the batch)
    incoming_set = set(incoming_urls)
    
    # 2. Filter out already visited URLs using set difference: O(N)
    new_urls = incoming_set - visited_cache
    
    # 3. Partition by domain into a routing dictionary
    domain_routes = {}
    for url in sorted(new_urls):
        # Extract domain: 'https://docs.pylabs.org/page' -> 'docs.pylabs.org'
        if "://" in url:
            domain = url.split("://")[1].split("/")[0]
        else:
            domain = url.split("/")[0]
            
        domain_routes.setdefault(domain, []).append(url)
        
    # 4. In-place update of visited cache using union assignment |=
    visited_cache |= new_urls
    
    return domain_routes, visited_cache

# Initial state: URLs visited in previous crawl cycles
persistent_visited_cache = {
    "https://pylabs.org/",
    "https://docs.pylabs.org/intro",
    "https://blog.pylabs.org/v1-release"
}

# Inbound crawler stream containing duplicates and already-visited links
fresh_batch = [
    "https://pylabs.org/",              # Visited
    "https://pylabs.org/pricing",        # New
    "https://docs.pylabs.org/python",    # New
    "https://docs.pylabs.org/intro",     # Visited
    "https://api.pylabs.org/health",     # New
    "https://pylabs.org/pricing"         # Duplicate in current batch
]

print("=" * 65)
print(f"{'CRAWLER URL DEDUPLICATION & ROUTING GATEWAY':^65}")
print("=" * 65)
print(f"Initial Cache Size:  {len(persistent_visited_cache)} URLs")
print(f"Incoming Batch Size: {len(fresh_batch)} URLs")
print("-" * 65)

routes, updated_cache = process_crawler_batch(fresh_batch, persistent_visited_cache)

print("Dispatched Domain Queue Routes:")
for domain, queue in routes.items():
    print(f"  [{domain}] ({len(queue)} items):")
    for u in queue:
        print(f"     -> {u}")

print("-" * 65)
print(f"Updated Cache Size:  {len(updated_cache)} URLs")
print("=" * 65)