7 posts tagged with "Python"

    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(get_file_do_scan, file_tg=file, api_keys=copy.deepcopy(my_api_keys)) for file in
                   just_compress_files]
    for future in concurrent.futures.as_completed(futures):
        future_result = future.result()
        if future_result == "dangerous" or future_result is None:
            continue
        print(future_result)
        print("以上檔案是安全的")

    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(get_file_do_scan, file_tg=file, api_keys=copy.deepcopy(my_api_keys)) for file in
                   just_compress_files]

with ThreadPoolExecutor() as executor:
    # 使用 list comprehension 創建一個 Future 對象的列表
    # 每個 Future 對象代表一個即將完成的操作（在這裡，操作是 fetch_yahoo_homepage 函數的調用）
    # executor.submit 函數的參數是要調用的函數和該函數的參數
    # 当你使用executor.submit()提交任务给线程池时，任务会立即开始执行。executor.submit()会立即返回一个concurrent.futures.Future对象
    futures = [executor.submit(fetch_yahoo_homepage, url) for url in url_will_request]
for future in as_completed(futures):
    # 使用 Future 對象的 result 方法來獲取操作的結果（在這裡，結果是 HTML 字串）
    # 通过调用future.result()获取任务的执行结果
    html = future.result()

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as e:
# Executors 的相關方法，map(func, *iterables)，執行可迭代物件 ex list tuple named tuple
# 其實就是把可迭代物件的元素一個個拆開喂給Executor，提升程式執行速度
    for hostinfo in e.map(lambda x: get_certificate(x[0], x[1]), HOSTS):
        if hostinfo is None:
            continue
        print_basic_info(hostinfo)

import urllib.request
import ssl
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
ssl._create_default_https_context = ssl._create_unverified_context
url_will_request=["https://www.google.com",'https://www.yahoo.com'," https://www.youtube.com","https://www.nba.com"]

def fetch_yahoo_homepage(url):
    try:
        response = urllib.request.urlopen(url)
        html = response.read().decode('utf-8')
        return html
    except urllib.error.URLError as e:
        print(f'Error: {e}')
        return None

#single_thead
start_time_single_thead=time.time()
for i in url_will_request:
    html=fetch_yahoo_homepage(i)
    print(html[:100])
end_time_single_thead=time.time()
all_time_single_thead=end_time_single_thead-start_time_single_thead

start_time=time.time()
# 使用示例ex
with ThreadPoolExecutor() as executor:
    for html in ecutor.map(fetch_yahoo_homepage,url_will_request ):
        print(html[:100])
end_time=time.time()
all_time=end_time-start_time`

start_time_multi_thread_submit=time.time()

# 使用 ThreadPoolExecutor 作為上下文管理器
# 當離開這個區塊時，將自動清理並關閉所有線程
with ThreadPoolExecutor() as executor:
    # 使用 list comprehension 創建一個 Future 對象的列表
    # 每個 Future 對象代表一個即將完成的操作（在這裡，操作是 fetch_yahoo_homepage 函數的調用）
    # executor.submit 函數的參數是要調用的函數和該函數的參數
    # 当你使用executor.submit()提交任务给线程池时，任务会立即开始执行。executor.submit()会立即返回一个concurrent.futures.Future对象
    futures = [executor.submit(fetch_yahoo_homepage, url) for url in url_will_request]

    # 使用as_completed()函数来迭代这些Future对象，并在future中有程式完成时获取结果到future。
    for future in as_completed(futures):
        # 使用 Future 對象的 result 方法來獲取操作的結果（在這裡，結果是 HTML 字串）
        # 通过调用future.result()获取任务的执行结果
        html = future.result()

        # 如果 HTML 不為空（也就是說，操作成功返回了 HTML），則打印 HTML 的前 100 個字符
        if html:
            print(html[:100])
end_time_multi_thread_submit=time.time()
all_time_multi_thread_submit=end_time_multi_thread_submit-start_time_multi_thread_submit

print(f"all_time_single_thead is {all_time_single_thead}")
print(f"all map time is {all_time}")
print(f"all_time_single_thead_submit is {all_time_multi_thread_submit}")

from functools import partial

# 原始函式
def add(x, y):
    return x + y

# 創建偏函式
add_five = partial(add, y=5)

# 使用偏函式
result = add_five(10)
print(result)  # 輸出: 15

from functools import partial

# 定義一個具有多個參數的函式
def multiply(x, y, z):
    return x * y * z

# 使用偏函數將其中一個參數的預設值固定為 2
new_func = partial(multiply, y=2)

# 調用新函式
result = new_func(3, z=4)
print(result)  # 輸出: 24

from functools import partial, wraps
import vt

def partial_decorator1(func=None, *, apikey=api_key_global):
    def actual_decorator(func, apikey):
        @wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            sha256_hash = result
            client = vt.Client(apikey)
            file = client.get_object(f"/files/{sha256_hash}")
            print(file.last_analysis_stats)
            client.close()
            return result

        return wrapper

    if func:
        return actual_decorator(func, apikey)
    else:
        return partial(actual_decorator, apikey=apikey)

# 使用新的装饰器
@partial_decorator1(apikey=api_key_global)
def my_function(file_path):
    # 在这里执行你的逻辑并返回一个哈希值
    return file_path

from functools import partial, wraps

def partial_decorator1(generator_func=None, *, api_key="api_key_global"):
    def actual_decorator(generator_func, api_key):
        @wraps(generator_func)
        def wrapper(*args, **kwargs):
            # 打印字串
            print("Before generator execution")
            # 並不會執行,只是產生生成器而已
            result = generator_func(*args, **kwargs)
            print(api_key)
            # 打印字串
            print("After generator execution")
            return result

        return wrapper

    if generator_func is None:
        return partial(actual_decorator, api_key=api_key)
    else:
        return actual_decorator(generator_func, api_key)


# @partial_decorator1(api_key="my_api_key"),這是代表裝飾器被當作函式調用情況,
# 因為調用時沒有帶generator_func,所以generator_func會是預設的None.會走if那邊,
# 用partial(actual_decorator, api_key=api_key),產生新裝飾器,這個裝飾器會接受一個函數作為參數，並返回該函數的裝飾版本。
@partial_decorator1(api_key="my_api_key")
def my_generator():
    yield 1
    yield 2
    yield 3

# 使用装饰器修饰生成器函数,這邊是在修改帶入的既有餐數,api_key="my_api_key"
# 注意這邊的my_generator()是返回<generator object my_generator at 0x10041ce00>
# yield是實現generator的一種方法！
for value in my_generator():
    print(value)

# 使用 partial_decorator1 创建装饰器,@partial_decorator1不帶參數,就會直接應用到function
# ,因此參數generator_func就會有帶東西,而走else.else做的事情是修飾並return新function
# actual_decorator(generator_func, api_key)會修飾function,return新function.
@partial_decorator1
def my_generator1():
    yield 1
    yield 2
    yield 3

# 使用装饰器修饰生成器函数,這邊是如果什麼都不帶,就會用默認的api_key="api_key_global"
for value in my_generator1():
    print(value)

# 使用 None 表示缺少值或變量尚未被賦予任何值
name = None
if name is None:
    print("名字尚未被賦值")

# None 作為函式的返回值
def divide(a, b):
    if b == 0:
        return None
    return a / b

result = divide(10, 0)
if result is None:
    print("無效的操作")

# 使用 False 進行條件判斷
flag = False
if flag:
    print("這個條件不成立")
else:
    print("這個條件成立")

max_retries = 3
retries = 0
while retries<max_retries:
    try:
        request....
        return sth
    except Exception as err:
        retries += 1
        time.sleep(1)# 延遲一秒後再重新執行
        if retries==max_retries:
            telegram_bot_sendtext(f"sth wrong, go cheeck")
            print(err)
    return  None

API < 開發者工具的xhr/fetch解析 < bs4解析html中css跟tag < selenium解析html跟點擊

format=datetime.now() #<class 'datetime.datetime'>

today = datetime.now().date() #<class 'datetime.date'>

yesterday = today - timedelta(days=1) #<class 'datetime.date'>

seven_days_ago = today - timedelta(days=7) #<class 'datetime.date'>

date_format = "%Y-%m-%d %H:%M:%S" #把字串為2023-04-17 02:36:55依據格式換成datetime.datetime物件

# 使用datetime.strptime方法把字串依據格式換成 datetime.datetime物件,使用date方法變成datetime.date物件進行比較
if seven_days_ago<= datetime.strptime("2023-04-17 02:36:55",date_format).date() <= yesterday:
    print("yes")

# 將到期日期字串轉換為 datetime 格式
vaild_date = datetime.strptime("2023-04-17 02:36:55", date_format)

# 取得當前時間
now_time = datetime.now() #2023-04-17 20:33:01.577635

# 計算到期日期與當前時間的時間差,注意這邊是timedelta物件囉！
vaild_days = vaild_date - now_time

# 印出時間差和時間差的資料型態
print(vaild_days) #-1 day, 6:11:15.206948
print(type(vaild_days)) #<class 'datetime.timedelta'>

# 判斷時間差是否在 當下到 30 天之間到期
if vaild_days > timedelta(days=0) and vaild_days < timedelta(days=30):
    print(vaild_days)
    print("ok")

vaild_date=datetime.strptime("2023-04-17 02:36:55",date_format).date()
print(type(vaild_date-timedelta(days=1))) # <class 'datetime.date'>

vaild_date=datetime.strptime("2023-04-17 02:36:55",date_format).date()
print(type(datetime.now().date()-vaild_date)) #<class 'datetime.timedelta'>