Python集合(set)中update方法与可迭代对象的使用方法

import time

large_list = list(range(1000000))
large_set = set(large_list)

start = time.time()
_ = 999999 in large_list  # 列表查找
list_time = time.time() - start

start = time.time()
_ = 999999 in large_set   # 集合查找
set_time = time.time() - start

print(f"列表查找耗时: {list_time:.6f}秒")
print(f"集合查找耗时: {set_time:.6f}秒")
# 典型输出: 列表0.08秒 vs 集合0.000001秒！

set.update(iterable)

# 场景：向集合添加10000个元素
s1 = set()
s2 = set()

# 使用add方法（低效）
start = time.time()
for i in range(10000):
    s1.add(i)
add_time = time.time() - start

# 使用update方法（高效）
start = time.time()
s2.update(range(10000))  # 直接传入可迭代对象
update_time = time.time() - start

print(f"add方法耗时: {add_time:.6f}秒")
print(f"update方法耗时: {update_time:.6f}秒")
print(f"速度提升: {add_time/update_time:.1f}倍")

def is_iterable(obj):
    try:
        iter(obj)
        return True
    except TypeError:
        return False

print(is_iterable([1, 2, 3]))  # True - 列表
print(is_iterable("hello"))    # True - 字符串
print(is_iterable(123))        # False - 整数不可迭代

fruits = {"apple", "banana"}
new_fruits = ["orange", "grape", "apple"]  # 包含重复项

fruits.update(new_fruits)
print(fruits) 
# 输出: {'banana', 'apple', 'grape', 'orange'} 
# 注意: 1. 顺序改变（集合无序） 2. 重复的'apple'被自动忽略

# 元组更新（高效且不可变）
codes = {100, 200}
codes.update((300, 400, 500)) 
print(codes)  # {100, 200, 300, 400, 500}

# 字符串更新（字符级拆分）
chars = {'a', 'b'}
chars.update("hello") 
print(chars)  # {'a', 'b', 'h', 'e', 'l', 'o'} 
# 注意: 'l'出现两次但集合中只保留一个

user_data = {"name": "Alice", "age": 30}
s = {1, 2}

# 默认：添加键
s.update(user_data)
print(s)  # {1, 2, 'name', 'age'}

# 添加值
s.update(user_data.values())
print(s)  # {1, 2, 'name', 'age', 'Alice', 30}

# 添加键值对（作为元组）
s.update(user_data.items())
print(s)  # {1, 2, 'name', 'age', 'Alice', 30, ('name', 'Alice'), ('age', 30)}

main_set = {1, 2}
nested_sets = [{3, 4}, {4, 5}]  # 列表包含集合

# 直接update会添加集合对象本身
main_set.update(nested_sets)
print(main_set)  # {1, 2, {3, 4}, {4, 5}} → 包含子集合

# 正确做法：扁平化处理
main_set = {1, 2}
for subset in nested_sets:
    main_set.update(subset)  # 逐个更新子集
print(main_set)  # {1, 2, 3, 4, 5}

# 假设有large_data.txt，每行一个整数
with open("large_data.txt", "r") as f:
    # 逐行读取并转换为int（生成器表达式）
    data_generator = (int(line.strip()) for line in f)
    
    unique_numbers = set()
    unique_numbers.update(data_generator)  # 流式添加，内存友好

print(f"唯一数字数量: {len(unique_numbers)}")

import requests

def fetch_users():
    # 模拟API调用（真实项目替换为实际URL）
    response = requests.get("https://jsonplaceholder.typicode.com/users")
    return [user['id'] for user in response.json()]

active_users = {101, 102}
new_users = fetch_users()  # 假设返回[1, 2, 3, ...]

active_users.update(new_users)
print(f"总活跃用户: {len(active_users)}")

class EvenNumbers:
    def __init__(self, limit):
        self.limit = limit
        self.current = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        self.current += 2
        if self.current > self.limit:
            raise StopIteration
        return self.current

evens = EvenNumbers(10)
number_set = {1, 3}
number_set.update(evens)  # 传入自定义迭代器
print(number_set)  # {1, 3, 2, 4, 6, 8, 10}

s = {1, 2}

# 错误1: 传递非可迭代对象
try:
    s.update(123)  # 整数不可迭代
except TypeError as e:
    print(f"错误类型: {type(e).__name__}")
    print(f"错误信息: {e}")
    # 修复: 包装成可迭代对象
    s.update([123])
    print(f"修复后: {s}")  # {1, 2, 123}

# 错误2: 传递不可哈希元素
try:
    s.update([[1, 2]])  # 列表不可哈希
except TypeError as e:
    print(f"错误信息: {e}")
    # 修复: 转换为元组
    s.update([(1, 2)])
    print(f"修复后: {s}")  # {1, 2, 123, (1, 2)}

from itertools import chain

set_a = {1, 2}
set_b = {2, 3}
set_c = {3, 4}

# 传统方式（低效）
combined = set_a.copy()
combined.update(set_b)
combined.update(set_c)

# 高效方式（单次update）
combined = set_a.copy()
combined.update(chain(set_b, set_c))  # 等效于set_b.update(set_c)

print(combined)  # {1, 2, 3, 4}

a = {1, 2}
b = {2, 3}

# 两种等效写法
a.update(b)
# 等价于
a |= b  # 并集赋值运算符

print(a)  # {1, 2, 3}

# 但注意: a = a | b 会创建新集合（非原地修改）

tags = {"python"}
tags.update("django")  # 期望添加"django"，实际添加了d,j,a,n,g,o
print(tags)  # {'python', 'd', 'j', 'a', 'n', 'g', 'o'} → 错误！

# 方法1: 包装成列表
tags.update(["django"])

# 方法2: 使用add添加单个字符串
tags.add("django")

config = {"theme": "dark", "font": "Arial"}
s = set()

s.update(config)        # 添加键 → {'theme', 'font'}
s.update(config.values()) # 添加值 → {'theme', 'font', 'dark', 'Arial'}
s.update(config.items())  # 添加元组 → 包含('theme','dark')等

s = set()
s.update([[1, 2]])  # 列表不可哈希 → TypeError

# 转换为元组
s.update([(1, 2)])  # 成功添加元组(1,2)

# 或使用frozenset
s.update([frozenset([1, 2])])

# 错误：先创建大列表再update
big_list = list(range(1000000))
s = set()
s.update(big_list)  # 额外占用列表内存

# 正确：直接使用生成器
s = set()
s.update(range(1000000))  # range是轻量级迭代器

import sys
print(sys.getsizeof(list(range(1000000))))  # 约8,000,056字节
print(sys.getsizeof(range(1000000)))        # 仅48字节！

s = {1, 2, 3}
for item in s:
    if item % 2 == 0:
        s.update([item * 10])  # 危险！可能跳过元素

for item in set(s):  # 迭代副本
    if item % 2 == 0:
        s.update([item * 10])

from collections import deque
import requests

class Crawler:
    def __init__(self):
        self.visited = set()  # 已访问URL
        self.queue = deque()  # 待爬取队列
    
    def crawl(self, start_url, max_pages=100):
        self.queue.append(start_url)
        
        while self.queue and len(self.visited) < max_pages:
            url = self.queue.popleft()
            
            # 跳过已访问
            if url in self.visited:
                continue
                
            try:
                response = requests.get(url, timeout=5)
                self.visited.add(url)  # 标记为已访问
                
                # 提取新链接（简化版）
                new_links = self.extract_links(response.text)
                
                # 批量添加新链接（高效去重）
                self.queue.extend(new_links)
                self.visited.update(new_links)  # 关键：批量更新
                
            except Exception as e:
                print(f"爬取{url}失败: {str(e)}")
    
    def extract_links(self, html):
        # 实际项目用BeautifulSoup解析
        return ["https://example.com/page1", "https://example.com/page2"]

crawler = Crawler()
crawler.crawl("https://example.com")
print(f"成功爬取 {len(crawler.visited)} 个页面")

error_codes = set()

with open("server.log", "r") as log_file:
    # 生成器表达式：仅提取错误行的错误码
    error_gen = (
        line.split()[5]  # 假设错误码在第6列
        for line in log_file 
        if "ERROR" in line
    )
    
    error_codes.update(error_gen)  # 流式添加

print(f"发现 {len(error_codes)} 种唯一错误码:")
print(", ".join(sorted(error_codes)))

class Inventory:
    def __init__(self):
        self.items = set()
    
    def sync_warehouse(self, warehouse_id):
        """从API同步指定仓库库存"""
        api_url = f"https://api.warehouse.com/{warehouse_id}/stock"
        response = requests.get(api_url)
        stock_data = response.json()
        
        # 提取商品ID（假设API返回列表）
        item_ids = [item['id'] for item in stock_data]
        
        # 原子化更新：避免部分更新问题
        new_set = self.items.copy()
        new_set.update(item_ids)
        self.items = new_set  # 替换为新集合（线程安全）
    
    def sync_all(self, warehouse_ids):
        """同步所有仓库"""
        for wid in warehouse_ids:
            self.sync_warehouse(wid)
        print(f"总库存商品: {len(self.items)}")

inv = Inventory()
inv.sync_all([101, 102, 103])

# 从多个列表创建集合
sources = [
    ["apple", "banana"],
    ("orange", "grape"),
    "hello"
]

# 传统方式
result = set()
for src in sources:
    result.update(src)

# 更Pythonic的方式
result = {item for src in sources for item in src}
print(result)  # {'a', 'p', 'l', 'e', 'b', 'n', 'o', 'r', 'g', 'h'}

def batch_add(collection, *iterables):
    """向集合批量添加多个可迭代对象"""
    for it in iterables:
        collection.update(it)
    return collection

# 使用示例
users = {"admin"}
batch_add(users, ["user1", "user2"], ("user3",))
print(users)  # {'admin', 'user1', 'user2', 'user3'}

a = {1, 2, 3}
b = {3, 4}
c = {4, 5}

# 传统：多次update
a.update(b)
a.update(c)

# 链式运算符（更简洁）
a |= b | c  # 等价于 a = a | b | c

print(a)  # {1, 2, 3, 4, 5}

def safe_update(target_set, iterable, skip_errors=True):
    """安全添加元素，跳过不可哈希项"""
    for item in iterable:
        try:
            # 尝试添加（触发哈希计算）
            target_set.add(item)
        except TypeError:
            if not skip_errors:
                raise

# 使用示例
s = {1, 2}
safe_update(s, [3, [4,5], 6])  # 跳过列表[4,5]
print(s)  # {1, 2, 3, 6}

import timeit

setup = """
s = set(range(1000))
new_data = range(1000, 2000)
"""

# 测试1: update
time_update = timeit.timeit("s.update(new_data)", setup, number=1000)

# 测试2: union
time_union = timeit.timeit("s.union(new_data)", setup, number=1000)

print(f"update 1000次耗时: {time_update:.4f}s")
print(f"union 1000次耗时: {time_union:.4f}s")
# 典型输出: update 0.003s vs union 0.12s → 快40倍！

# 场景: 添加多个数据源
final_set = set()
final_set.update(source1)  # 列表/元组
final_set.update(source2)  # 生成器
final_set.update(source3)  # 字典.values()

# 替代低效写法
# final_set = set(source1) | set(source2) | set(source3)  # 创建3个临时集合