import threading


class SingletonMeta(type):
    """自定义元类"""

    def __init__(cls, *args, **kwargs):
        cls.__instance = None
        cls.__lock = threading.RLock()
        super().__init__(*args, **kwargs)

    def __call__(cls, *args, **kwargs):
        if cls.__instance is None:
            with cls.__lock:
                if cls.__instance is None:
                    cls.__instance = super().__call__(*args, **kwargs)
        return cls.__instance


class President(metaclass=SingletonMeta):
    """总统(单例类)"""

    pass

  class StreamHasher:
    """哈希摘要生成器"""

    def __init__(self, alg='md5', size=4096):
        self.size = size
        alg = alg.lower()
        self.hasher = getattr(__import__('hashlib'), alg.lower())()

    def __call__(self, stream):
        return self.to_digest(stream)

    def to_digest(self, stream):
        """生成十六进制形式的摘要"""
        for buf in iter(lambda: stream.read(self.size), b''):
            self.hasher.update(buf)
        return self.hasher.hexdigest()

def main():
    """主函数"""
    hasher1 = StreamHasher()
    with open('Python-3.7.6.tgz', 'rb') as stream:
        print(hasher1.to_digest(stream))
    hasher2 = StreamHasher('sha1')
    with open('Python-3.7.6.tgz', 'rb') as stream:
        print(hasher2(stream))


if __name__ == '__main__':
    main()

class Fib(object):
    """迭代器"""

    def __init__(self, num):
        self.num = num
        self.a, self.b = 0, 1
        self.idx = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.idx < self.num:
            self.a, self.b = self.b, self.a + self.b
            self.idx += 1
            return self.a
        raise StopIteration()

def fib(num):
    """生成器"""
    a, b = 0, 1
    for _ in range(num):
        a, b = b, a + b
        yield a

def calc_avg():
    """流式计算平均值"""
    total, counter = 0, 0
    avg_value = None
    while True:
        value = yield avg_value
        total, counter = total + value, counter + 1
        avg_value = total / counter


gen = calc_avg()
next(gen)
print(gen.send(10))
print(gen.send(20))
print(gen.send(30))

"""
面试题：进程和线程的区别和联系？
进程 - 操作系统分配内存的基本单位 - 一个进程可以包含一个或多个线程
线程 - 操作系统分配CPU的基本单位
并发编程（concurrent programming）
1. 提升执行性能 - 让程序中没有因果关系的部分可以并发的执行
2. 改善用户体验 - 让耗时间的操作不会造成程序的假死
"""
import glob
import os
import threading

from PIL import Image

PREFIX = 'thumbnails'


def generate_thumbnail(infile, size, format='PNG'):
    """生成指定图片文件的缩略图"""
	file, ext = os.path.splitext(infile)
	file = file[file.rfind('/') + 1:]
	outfile = f'{PREFIX}/{file}_{size[0]}_{size[1]}.{ext}'
	img = Image.open(infile)
	img.thumbnail(size, Image.ANTIALIAS)
	img.save(outfile, format)


def main():
    """主函数"""
	if not os.path.exists(PREFIX):
		os.mkdir(PREFIX)
	for infile in glob.glob('images/*.png'):
		for size in (32, 64, 128):
            # 创建并启动线程
			threading.Thread(
				target=generate_thumbnail,
				args=(infile, (size, size))
			).start()


if __name__ == '__main__':
	main()

"""
多线程程序如果没有竞争资源处理起来通常也比较简单
当多个线程竞争临界资源的时候如果缺乏必要的保护措施就会导致数据错乱
说明：临界资源就是被多个线程竞争的资源
"""
import time
import threading

from concurrent.futures import ThreadPoolExecutor


class Account(object):
    """银行账户"""

    def __init__(self):
        self.balance = 0.0
        self.lock = threading.Lock()

    def deposit(self, money):
        # 通过锁保护临界资源
        with self.lock:
            new_balance = self.balance + money
            time.sleep(0.001)
            self.balance = new_balance


def main():
    """主函数"""
    account = Account()
    # 创建线程池
    pool = ThreadPoolExecutor(max_workers=10)
    futures = []
    for _ in range(100):
        future = pool.submit(account.deposit, 1)
        futures.append(future)
    # 关闭线程池
    pool.shutdown()
    for future in futures:
        future.result()
    print(account.balance)


if __name__ == '__main__':
    main()

"""
多个线程竞争一个资源 - 保护临界资源 - 锁（Lock/RLock）
多个线程竞争多个资源（线程数>资源数） - 信号量（Semaphore）
多个线程的调度 - 暂停线程执行/唤醒等待中的线程 - Condition
"""
from concurrent.futures import ThreadPoolExecutor
from random import randint
from time import sleep

import threading


class Account:
    """银行账户"""

    def __init__(self, balance=0):
        self.balance = balance
        lock = threading.RLock()
        self.condition = threading.Condition(lock)

    def withdraw(self, money):
        """取钱"""
        with self.condition:
            while money > self.balance:
                self.condition.wait()
            new_balance = self.balance - money
            sleep(0.001)
            self.balance = new_balance

    def deposit(self, money):
        """存钱"""
        with self.condition:
            new_balance = self.balance + money
            sleep(0.001)
            self.balance = new_balance
            self.condition.notify_all()


def add_money(account):
    while True:
        money = randint(5, 10)
        account.deposit(money)
        print(threading.current_thread().name,
              ':', money, '====>', account.balance)
        sleep(0.5)


def sub_money(account):
    while True:
        money = randint(10, 30)
        account.withdraw(money)
        print(threading.current_thread().name,
              ':', money, '<====', account.balance)
        sleep(1)


def main():
    account = Account()
    with ThreadPoolExecutor(max_workers=15) as pool:
        for _ in range(5):
            pool.submit(add_money, account)
        for _ in range(10):
            pool.submit(sub_money, account)


if __name__ == '__main__':
    main()

"""
多进程和进程池的使用
多线程因为GIL的存在不能够发挥CPU的多核特性
对于计算密集型任务应该考虑使用多进程
time python3 example22.py
real    0m11.512s
user    0m39.319s
sys     0m0.169s
使用多进程后实际执行时间为11.512秒，而用户时间39.319秒约为实际执行时间的4倍
这就证明我们的程序通过多进程使用了CPU的多核特性，而且这台计算机配置了4核的CPU
"""
import concurrent.futures
import math

PRIMES = [
    1116281,
    1297337,
    104395303,
    472882027,
    533000389,
    817504243,
    982451653,
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419
] * 5


def is_prime(n):
    """判断素数"""
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True


def main():
    """主函数"""
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))


if __name__ == '__main__':
    main()

"""
异步I/O - async / await
"""
import asyncio


def num_generator(m, n):
    """指定范围的数字生成器"""
    yield from range(m, n + 1)


async def prime_filter(m, n):
    """素数过滤器"""
    primes = []
    for i in num_generator(m, n):
        flag = True
        for j in range(2, int(i ** 0.5 + 1)):
            if i % j == 0:
                flag = False
                break
        if flag:
            print('Prime =>', i)
            primes.append(i)

        await asyncio.sleep(0.001)
    return tuple(primes)


async def square_mapper(m, n):
    """平方映射器"""
    squares = []
    for i in num_generator(m, n):
        print('Square =>', i * i)
        squares.append(i * i)

        await asyncio.sleep(0.001)
    return squares


def main():
    """主函数"""
    loop = asyncio.get_event_loop()
    future = asyncio.gather(prime_filter(2, 100), square_mapper(1, 100))
    future.add_done_callback(lambda x: print(x.result()))
    loop.run_until_complete(future)
    loop.close()


if __name__ == '__main__':
    main()

import asyncio
import re

import aiohttp

PATTERN = re.compile(r'\<title\>(?P<title>.*)\<\/title\>')


async def fetch_page(session, url):
    async with session.get(url, ssl=False) as resp:
        return await resp.text()


async def show_title(url):
    async with aiohttp.ClientSession() as session:
        html = await fetch_page(session, url)
        print(PATTERN.search(html).group('title'))


def main():
    urls = ('https://www.python.org/',
            'https://git-scm.com/',
            'https://www.jd.com/',
            'https://www.taobao.com/',
            'https://www.douban.com/')
    loop = asyncio.get_event_loop()
    cos = [show_title(url) for url in urls]
    loop.run_until_complete(asyncio.wait(cos))
    loop.close()


if __name__ == '__main__':
    main()

soup = BeautifulSoup(resp, "html.parser") # html.parser 是默认的解析器
title_name = soup.findAll('p', attrs = {"class": "price_color"}) # 找到第一个p标签，参数class等于price_color
for name in title_name:
  print(name.string)

pip3 install bs4

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
    "Accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
}

proxies = {
    "http": "http://10.10.1.10:3128",
    "https": "http://10.10.1.10:1080",
}
r = requests.get("http://httpbin.org/ip", proxies=proxies)

params = {
    "wd": "python",
}
response = requests.get(url, headers=headers, params=params)
url : https://www.baidu.com/s?wd=python

cookies = {
    "cookie": "name=germey",
}
response = requests.get(url, headers=headers, cookies=cookies)

from requests.auth import HTTPBasicAuth
HTTPBasicAuth = requests.auth.HTTPBasicAuth('user', 'pass')
r = requests.get('https://httpbin.org/basic-auth/user/pass', auth=HTTPBasicAuth)
print(r.status_code)

response = requests.get(url, allow_redirects=False)

r = requests.get('https://expired.badssl.com', verify=False)
print(r.status_code)

# 发送 GET 请求
response = requests.get('https://httpbin.org/get')

# 响应状态码
print("状态码：", response.status_code)

# 响应内容（文本格式）
print("响应内容：", response.text)
# 响应内容（JSON 格式）
print("响应内容（JSON）：", response.json())

# 添加请求参数和请求头
params = {'key': 'value'}
headers = {'User-Agent': 'my-app'}

response = requests.get('https://httpbin.org/get', params=params, headers=headers)
print(response.url)  # 展示最终拼接后的 URL

from DrissionPage import ChromiumPage

# 创建 ChromiumPage 对象，启动内置 Chromium 浏览器（默认非无头模式）
page = ChromiumPage()

# 访问网页
page.get('https://example.com')

# 查找元素：使用 CSS 选择器
el = page.e('h1')                      # 获取第一个 <h1> 元素
print(el.text)                         # 输出文本内容

# 查找元素：使用 XPath 选择器（前面加 @ 表示 XPath）
el2 = page.e('@//p')                   # 获取第一个 <p> 元素
print(el2.text)

# 点击元素（如按钮）
page.e('button#submit').click()        # 点击 id 为 submit 的按钮

# 输入文本
page.e('input[name="username"]').input('admin')  # 在 name 为 username 的输入框输入 'admin'

# 等待元素出现，最多等待10秒
page.wait.e('div.result', timeout=10)  # 等待 class 为 result 的 div 出现

# 获取整个页面 HTML 源码
html = page.html

# 执行 JavaScript 脚本，获取页面标题
title = page.run_js('return document.title;')
print(f'页面标题：{title}')

# 模拟截图并保存到当前目录
page.screenshot('screenshot.png')

# 保存当前 HTML 到本地文件
with open('page.html', 'w', encoding='utf-8') as f:
    f.write(html)

# 设置移动端 User-Agent（模拟手机访问）
page.set_user_agent(mobile=True)

# 关闭浏览器（一定要在结束时关闭以释放资源）
page.close()

# 启动时设置无头模式（不弹出浏览器窗口）
page = ChromiumPage(headless=True)

# 启动时设置代理（适用于需要翻墙或隐藏 IP）
page = ChromiumPage(proxy='http://127.0.0.1:1080')

with open('example.csv', newline='', encoding='utf-8', delimiter=',') as csvfile:
    reader = csv.reader(csvfile)
    for row in reader:
        print(row)

csv.reader(csvfile, delimiter=';', quotechar='"', skipinitialspace=True)

with open('example.csv', newline='', encoding='utf-8') as csvfile:
    reader = csv.DictReader(csvfile)
    for row in reader:
        print(row['name'], row['age'])

with open('output.csv', 'w', newline='', encoding='utf-8') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(['name', 'age', 'gender'])  # 写入表头
    writer.writerow(['Alice', 25, 'F'])          # 写入一行
    writer.writerows([
        ['Bob', 30, 'M'],
        ['Charlie', 35, 'M']
    ])

with open('output.csv', 'w', newline='', encoding='utf-8') as csvfile:
    fieldnames = ['name', 'age', 'gender']
    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
    
    writer.writeheader()  # 写入表头
    writer.writerow({'name': 'Alice', 'age': 25, 'gender': 'F'})
    writer.writerows([
        {'name': 'Bob', 'age': 30, 'gender': 'M'},
        {'name': 'Charlie', 'age': 35, 'gender': 'M'}
    ])

with open('file.csv', 'w', newline='', encoding='utf-8') as f:
    ...

csv.writer(f, delimiter=';', quotechar="'", quoting=csv.QUOTE_MINIMAL)

import csv

with open('input.csv', newline='', encoding='utf-8') as infile, \
     open('output.csv', 'w', newline='', encoding='utf-8') as outfile:
    
    reader = csv.DictReader(infile)
    fieldnames = reader.fieldnames
    writer = csv.DictWriter(outfile, fieldnames=fieldnames)
    
    writer.writeheader()
    for row in reader:
        # 可以对数据进行处理
        writer.writerow(row)

import re

re.search(r"\d+", "编号123")  # 每次都重新解析正则
re.search(r"\d+", "订单456")

import re

pattern = re.compile(r"\d+")  # 编译成正则对象

print(pattern.search("编号123"))  # 匹配成功
print(pattern.search("订单456"))  # 复用 pattern，不用重新编译

pattern = re.compile(r"hello", re.IGNORECASE)

match = pattern.search("HeLLo World")
print(match.group())  # 输出：HeLLo

import re

text = "Hello, my email is example@example.com"
pattern = r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"

match = re.search(pattern, text)  # 返回一个 Match 对象

if match:
    print(f"找到匹配: {match.group()}")  # 输出匹配的字符串
    print(f"匹配的起始位置: {match.start()}")  # 输出匹配的起始索引
    print(f"匹配的结束位置: {match.end()}")  # 输出匹配的结束索引
    print(f"匹配的字符串片段: {text[match.start():match.end()]}")  # 输出匹配的字符串片段
else:
    print("未找到匹配")

result = re.match(r'\d+', '123abc')  # 匹配开头数字
if result:
    print(result.group())  # 输出: 123
else:
    print("No match")

result = re.search(r'\d+', '123abc')  # 匹配任意位置数字
if result:
    print(result.group())  # 输出: 123
else:
    print("No match")

result = re.findall(r'\d+', '123abc')  # 匹配所有数字
if result:
    print(result)  # 输出: ['123']
else:
    print("No match")

result = re.split(r'\d+', '123abc')  # 以数字分割字符串
if result:
    print(result)  # 输出: ['abc']
else:
    print("No match")

result = re.sub(r'\d+', 'abc', '123abc')  # 用abc替换所有数字
if result:
    print(result)  # 输出: abcabc
else:
    print("No match")

扇形面积计算公式:圆心角度数除以360乘以pi乘以半径的平方
central_angle_1 = 160
radius_1 = 30
sector_area_1 = central_angle / 360 * 3.14 * radius_1 ** 2

central_angle_2 = 160
radius_2 = 30
sector_area_2 = central_angle_2 / 360 * 3.14 * radius_2 ** 2
print(sector_area_1, sector_area_2)

dry
don't repeat yourself

def calculator_sector_1(central_angle, radius):
    sector_area = central_angle / 360 * 3.14 * radius ** 2
    print(sector_area)

calculator_sector_1(160, 30)
#  通过参数,让函数变得通用

def favority_book(title):
    print(f"One of my favority books is {title}")

favority_book("stray birds ")

# /前面的参数是强制位置参数
def make_judgement(a, b, c, /):
    """判断三条边的长度能否构成三角形"""
    return a + b > c and b + c > a and a + c > b


# 下面的代码会产生TypeError错误，错误信息提示“强制位置参数是不允许给出参数名的”
# TypeError: make_judgement() got some positional-only arguments passed as keyword arguments
# print(make_judgement(b=2, c=3, a=1))

def make_judgement(a, b, c, /):
  # / 表示位置参数，只能使用位置参数传递参数，不能使用关键字参数传递参数
    
    """判断三条边的长度能否构成三角形"""
    return a + b > c and b + c > a and a + c > b

print(make_judgement(5, 3, 4))
print(make_judgement(a=5, b=3, c=4))  # 报错，不能使用关键字参数传递参数
# print(make_judgement(1, 2, 3, 4))  # 报错，不能使用关键字参数传递参数

# *后面的参数是命名关键字参数
def make_judgement(*, a, b, c):
  # * 表示命名关键字参数，必须使用关键字参数传递参数，不能使用位置参数传递参数
    """判断三条边的长度能否构成三角形"""
    return a + b > c and b + c > a and a + c > b
print(make_judgement(a=5, b=3, c=4))
# print(make_judgement(1, 2, 3))  # 报错，必须使用关键字参数传递参数

import random
import string # 用于生成随机字符串

ALL_CHARS = string.digits + string.ascii_letters # 生成所有字符
print(type(ALL_CHARS))
def generate_code(*, code_len=4):
    """生成指定长度的验证码"""
    return ''.join(random.sample(ALL_CHARS, code_len)) # 随机抽取指定长度的字符串
  # ''.join() 将列表中的字符连接成一个字符串 '' 表示连接的字符不用任何分隔符
print(generate_code()) # 生成4位验证码
# print(generate_code(code_len=6)) # 生成6位验证码

import random

RED_BALLS = [i for i in range(1, 34)] # 红球号码范围
BLUE_BALLS = [i for i in range(1, 17)] # 蓝球号码范围


def choose():
    """
    生成一组随机号码
    :return: 保存随机号码的列表
    """
    selected_balls = random.sample(RED_BALLS, 6)
    selected_balls.sort()
    selected_balls.append(random.choice(BLUE_BALLS))
    return selected_balls


def display(balls):
    """
    格式输出一组号码
    :param balls: 保存随机号码的列表
    """
    for ball in balls[:-1]: # 遍历前6个红球号码
        print(f'\033[91m{ball:0>2d}\033[0m', end=' ')
    print(f'\033[034m{balls[-1]:0>2d}\033[0m')


n = int(input('生成几注号码: '))
for _ in range(n):
    display(choose())

import functools

int2 = functools.partial(int, base=2)
int8 = functools.partial(int, base=8)
int16 = functools.partial(int, base=16)

print(int('1001'))    # 1001

print(int2('1001'))   # 9
print(int8('1001'))   # 513
print(int16('1001'))  # 4097

a = bin(9)[2:] # 去掉前缀0b
b = oct(513)[2:] # 513是八进制数对应的十进制数为513，所以输出的是513的八进制表示形式，即1001
c = hex(4097)[2:]
print(a, b, c)  # 1001 1001 1001

lambda 参数1, 参数2, ... : 表达式

def 函数名(参数1, 参数2, ...):
    return 表达式

add = lambda x, y: x + y
print(add(2, 3))  # 输出 5

def add(x, y):
    return x + y

data = [(1, 3), (2, 1), (4, 2)]
sorted_data = sorted(data, key=lambda x: x[1])
print(sorted_data)  # 输出 [(2, 1), (4, 2), (1, 3)]

sorted(iterable, key=None, reverse=False)

nums = [5, 2, 9, 1]
sorted_nums = sorted(nums)
print(sorted_nums)  # 输出 [1, 2, 5, 9]

words = ['banana', 'apple', 'cherry']
print(sorted(words))  # 输出 ['apple', 'banana', 'cherry']

nums = [3, 1, 4]
print(sorted(nums, reverse=True))  # 输出 [4, 3, 1]

words = ['banana', 'apple', 'kiwi']
print(sorted(words, key=len))  # 输出 ['kiwi', 'apple', 'banana']

people = [
    {'name': 'Tom', 'age': 25},
    {'name': 'Jerry', 'age': 20},
    {'name': 'Bob', 'age': 30}
]

sorted_people = sorted(people, key=lambda x: x['age'])
for person in sorted_people:
    print(person['name'], person['age'])
# 输出：
# Jerry 20
# Tom 25
# Bob 30