June 13, 2024
异步调用时自动关闭 # 当事件循环正在运行时，会创建一个任务来关闭 redis 连接，而不会阻塞事件循环。如果事件循环没有运行，就可以像之前一样调用 run_until_complete。
async def close_redis(self): if self.redis: await self.redis.close() def __del__(self): loop = asyncio.get_event_loop() if loop.is_running(): asyncio.create_task(self.close_redis()) else: loop.run_until_complete(self.close_redis()) 数据迁移(hash数据) # 脚本 # import os import json import redis import argparse from loguru import logger home_path = os.environ.get('HOME') parser = argparse.ArgumentParser(description='Copy Reten Config') parser.add_argument('--mode', help='Pull or Push data from Redis') parser.add_argument('-ds', '--datasource', help='retens or events config data from Redis') args = parser.parse_args() source_redis = redis.StrictRedis(host='redis-product.com', port=6379, db=0) def scan_keys_with_pattern(pattern): # 初始游标 cursor = '0' count_group = 0 while True: # 使用 SCAN 命令进行迭代 cursor, keys = source_redis.
...
一个协程示例： import asyncio async def my_coroutine(): # 这里是协程的代码 await asyncio.sleep(1) print("Coroutine is done") async def main(): # 创建一个任务对象 task = asyncio.create_task(my_coroutine()) # 等待任务完成 await task # 运行主协程 asyncio.run(main()) 一个线程示例： import threading # 定义一个函数，用于在线程中执行 def thread_function(name): print(f"Thread {name}: starting") # 在这里执行一些操作 print(f"Thread {name}: finishing") # 创建线程的实例 thread = threading.Thread(target=thread_function, args=(1,)) # 启动线程 thread.start() # 等待线程结束 thread.join() print("Main thread is done") 一个loop tip： self.loop = asyncio.get_event_loop() self.queue.set_loop(loop=self.loop) await self.af.set_loop(loop=self.loop) await self.adjust.set_loop(loop=self.loop) 为什么这里使用同一个loop ?
...
pydantic # 来自类 # from pydantic import BaseModel class User(BaseModel): name: str age: int class Config: from_attributes = True # 假设我们有一个普通的 Python 对象，它具有与模型字段相同的属性名 class OrdinaryObject: def __init__(self, name, age): self.name = name self.age = age # 创建一个 OrdinaryObject 实例 ordinary_object = OrdinaryObject(name='Alice', age=30) # 现在我们可以直接将 ordinary_object 传递给 User 模型的构造函数 user_model = User(ordinary_object) # 输出模型的字段值 print(user_model.dict()) # 输出: {'name': 'Alice', 'age': 30} 自定义字段 # class User(BaseModel): id: int is_active: bool wx_infos: List[str] = [] class Config: arbitrary_types_allowed = True # 允许任意类型 Tuple定义问题 # from typing import List, Tuple class Task(BaseModel): launch_count_today: int launch_groups: List[Tuple[str, .
...
特殊函数调用 property # 使用字段的方式调用函数
from datetime import datetime from pydantic import BaseModel class InstallCountInfo(BaseModel): count: int total: int @property def done_rate(self): return round(self.count / self.total, 4) class InstallCount(BaseModel): pass_day: int install_day: datetime info: InstallCountInfo 彻底摆脱to_dict和from_dict # 使用 pydantic # BaseModel类型支持:
b = BattleAxiePositionInfo.parse_obj(DICT_DATA) b.json() b.dict() parse_file parse_raw from pydantic import BaseModel class PositionInfo(BaseModel): error: int = -1 # 收集错误 none: int = 0 # 还没开始 clicked: int = 1 # 在client 赋此值 done: int = 2 # 在server 赋此值 xy: List[int] = [0, 0] status: int = 0 # clicked or done or none or error class BattleAxiePositionInfo(BaseModel): our: List[PositionInfo] = [PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo()] enemy: List[PositionInfo] = [PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo(), PositionInfo()] pp = BattleAxiePositionInfo() print(f"pp json: {pp.
...
author:Ian
邮件 # 发送邮件时携带附件 中文名称附件
def send_email_week_report(htm, recipients, copys, file_name, file_path): port = 465 # str_today = datetime.now().strftime("%Y-%m-%d %H:%M:%S") str_today = datetime.now().strftime("%Y-%m-%d") mail_host = "mail.{demo}.com" mail_user = "{user name}" mail_pass = "{passwd}" sender = '{name}@{demo}.com' mail_from = "{name} <{name}@{demo}.com>" msg_Bcc = '{name}@{demo}.com' subject = f"{title}-{str_today}" # 标题 msg = MIMEMultipart() msg.attach(MIMEText(htm, 'html', 'utf-8')) # 这里可以传html内容 也可以传普通文字 # 创建附件 with open(file_path, 'rb') as attachment: part = MIMEBase('application', 'octet-stream') part.set_payload(attachment.read()) encoders.encode_base64(part) part.add_header( 'Content-Disposition', 'attachment', filename=("utf-8", "", file_name), ) msg.
...
author:Ian
Python GUI 💽 # pynput # 在 pynput 模块中，Win键被称为“特殊键”（Special keys），需要使用特殊的名称来表示。
以下是可以使用的特殊键名称列表：
https://pynput.readthedocs.io/en/latest/keyboard.html?highlight=%3Ccmd%3E#controlling-the-keyboard 因此，如果你想要在热键设置中使用 Win键+空格 这个热键，可以将它们分别替换为 cmd 和 space，如下所示：
from pynput import keyboard def on_activate(): print('Hotkey activated') def on_exit(): print('Hotkey exited') return False with keyboard.GlobalHotKeys({'<cmd>+<space>': on_activate}) as h: h.join(on_exit)``` 在这个例子中，我们使用 <cmd>+<space> 来表示 Win键+空格 热键，因为在Mac中，Command键（cmd）可以起到类似于Win键的作用。 ## PyQt  官方说明文档：<http://pyqt.sourceforge.net/Docs/PyQt4/index.html> 照例，先贴网址： <http://www.qaulau.com/books/PyQt4_Tutorial/index.html> ## 画界面 #PyQt4使用designer.exe import os for root, dirs, files in os.walk('.'): for file in files: if file.endswith('.ui'): os.system('pyuic4 -o ui_%s.
...
Python 机器学习库 👽 # Plotly # 与matplotlib 都是绘图工具，不过效果炫一些，我也没画过，所以只放链接，不放实例了 Plotly Python Library : https://plot.ly/python/
matplotlib # import matplotlib.pyplot as plt 参数等太多，链接最可靠 # pyplot参数
还是粘一些常用的： marker 属性（下面写在分号里呦） o . v ^ < > 1 2 3 4 8 s p * h H + x D d | _ 之类
画出一些“花儿”
绘图 # plt.plot(x, y) # 在y之后可添加参数，例如常用的label = ‘IamLabel’之类 # 线的样式、颜色 ：b: blue g: green r: red c: cyan m: magenta y: yellow k: black w: white '-' : solid , '--' : dashed, '-.
...
PiP Not Found Issue # 使用pip安装某包时， 提示让更新， 按提示操作更新没效果没反应再用就提示ModuleNotFoundError: No module named 'pip' (ˉ▽ˉ；)…
ModuleNotFoundError: No module named ‘pip’ # 升级PiP时出现问题可由下方命令修复 # python -m ensurepip python -m pip install --upgrade pip SSL校验 # 安装EasyOCR时, reader = easyocr.Reader(['ch_sim','en']) 下载到接近90,结果报错了…. 估计是SSL问题加入以下两条,不知如何.
import ssl ssl._create_default_https_context = ssl._create_unverified_context from http.client To revert to the previous, unverified, behavior ssl._create_unverified_context() can be passed to the context parameter.
使用国内镜像下载Python # pip install --index https://pypi.mirrors.ustc.edu.cn/simple/ dlib(numpy等包名) # 一键更新pip 包资源(利用管道grep传输查询到的需要更新包名，传输到install命令) pip3 freeze --local | grep -v '^\-e' | cut -d = -f 1 | xargs -n1 pip3 install -U # 权限不够的话就在`pip3 install` 之前加`sudo`反正我不习惯用`root` 安装错误？ # 居然有pip3 install XX的错误…这也是因为有旧版pip3存留。需要
...