跳到主要内容

26 篇博文 含有标签「Python」

查看所有标签

如何在 python 异步函数中 debug


import asyncio
from aredis import StrictRedis
import nest_asyncio

# pip3 install aredis[hiredis]
# pip3 install nest_asyncio

async def example():
client = StrictRedis(host='127.0.0.1', port=6379, db=0)
await client.flushdb()
# error await client.flushdb()
# error asyncio.get_event_loop().run_until_complete(client.flushdb())
# error asyncio.new_event_loop().run_until_complete(client.flushdb())
# error asyncio.run(client.flushdb())

await client.set('foo', 1)
assert await client.exists('foo') is True
await client.incr('foo', 100)

assert int(await client.get('foo')) == 101
await client.expire('foo', 1)
await asyncio.sleep(0.1)
await client.ttl('foo')
await asyncio.sleep(1)
assert not await client.exists('foo')

loop = asyncio.get_event_loop()
# success nest_asyncio.apply()
loop.run_until_complete(example())

Python阅读需 1 分钟

参考

想法

  • 接微信机器人
    • 群友对话直接 at 机器人、以指令开头、触发逻辑。
  • 接 chatgpt
    • 根据问题与获取的卦辞或者表达的象征,通过有预设的 gpt 做解释后发回给群友。

代码部分

十二地支时间换算


import datetime
from lunardate import LunarDate

Zhi = ["子", "醜", "寅", "卯", "辰", "巳", "午", "未", "申", "酉", "戌", "亥"]

def get_factor_time(time:datetime.datetime=None):
if time is None:
time = datetime.datetime.now()
# 获取农历时间
lunar_time = LunarDate.fromSolarDate(time.year, time.month, time.day)
# 输出农历时间
lunar_month = lunar_time.month
lunar_day = lunar_time.day
lunar_hour = time.hour//2
print(f"农历时间:{lunar_month}{lunar_day}{lunar_hour}({Zhi[lunar_hour]}) 时")

完整代码

#!/usr/bin/env python
# encoding: utf-8
# @Time : 2023/07/04 01:05:33
# @author : zza
# @Email : z740713651@outlook.com
# @File : xiao_liu_ren.py
from collections import defaultdict
import datetime
from lunardate import LunarDate

Zhi = ["子", "醜", "寅", "卯", "辰", "巳", "午", "未", "申", "酉", "戌", "亥"]
xiao_liu_ren = [
"大安",
"留连",
"速喜",
"赤口",
"小吉",
"空亡",
]

def get_factor_time(time: datetime.datetime = None):
if time is None:
time = datetime.datetime.now()
print(f"阳历时间:{time.isoformat()}")
# 获取农历时间
lunar_time = LunarDate.fromSolarDate(time.year, time.month, time.day)
# 输出农历时间
lunar_month = lunar_time.month
lunar_day = lunar_time.day
lunar_hour = (time.hour + 1) // 2
print(f"农历时间:{lunar_month}{lunar_day}{Zhi[lunar_hour]}({lunar_hour}) 时")

factor_month = xiao_liu_ren[(lunar_month - 1) % 6]
factor_day = xiao_liu_ren[(lunar_month - 1 + lunar_day - 1) % 6]
factor_hour = xiao_liu_ren[(lunar_month - 1 + lunar_day - 1 + lunar_hour) % 6]
print(f"小六壬月日时:{factor_month} {factor_day} {factor_hour}")

return factor_month, factor_day, factor_hour

def lost_item(factor):
guaci = {
"大安": {
"留连": "大安加留连,失物西北去。",
"速喜": "大安加速喜,失物当日见。",
"赤口": "大安加赤口,失物不用找。",
"小吉": "大安加小吉,失物不出门。",
"空亡": "大安加空亡,失物无踪影。",
"大安": "大安加大安,失物在家里。",
},
"留连": {
"大安": "留连加大安,物在家中藏。",
"速喜": "留连加速喜,失物三天里。",
"赤口": "留连加赤口,失物准丢失。",
"小吉": "留连加小吉,失物东南去。",
"空亡": "留连加空亡,失物不见面。",
"留连": "留连加留连,失物落在南。",
},
"速喜": {
"赤口": "速喜加赤口,失物往正北。",
"小吉": "速喜加小吉,失物在家里。",
"空亡": "速喜加空亡,失物不久见。",
"大安": "速喜加大安,失物不丢失。",
"留连": "速喜加留连,失物无信息。",
"速喜": "速喜加速喜,失物落正南。",
},
"赤口": {
"小吉": "赤口加小吉,失物无信息。",
"空亡": "赤口加空亡,失物不用找。",
"大安": "赤口加大安,失物东北找。",
"留连": "赤口加留连,失物不回还。",
"速喜": "赤口加速喜,失物有着落。",
"赤口": "赤口加赤口,失物正西找。",
},
"小吉": {
"空亡": "小吉加空亡,失物正东找。",
"大安": "小吉加大安,失物自己出。",
"留连": "小吉加留连,失物上西南。",
"速喜": "小吉加速喜,失物在院里。",
"赤口": "小吉加赤口,失物丢了手。",
"小吉": "小吉加小吉,失物在西南。",
},
"空亡": {
"大安": "空亡加大安,失物反复间。",
"留连": "空亡加留连,失物永不还。",
"速喜": "空亡加速喜,失物在家里。",
"赤口": "空亡加赤口,失物往远走。",
"小吉": "空亡加小吉,失物回家里。",
"空亡": "空亡加空亡,失物寻不见。",
},
}
return guaci[factor[1]][factor[2]]

if __name__ == "__main__":
# get_factor_time()
# 日上起時。 假如三月初五日辰時:三月在速喜上;就速喜上起初一;初五在大安;大安上起子時;數至辰時即是小吉; 以類推占,餘皆倣此。
# 速喜 大安 小吉
res = get_factor_time(datetime.datetime(2023, 4, 24, 7, 15))
assert res == ("速喜", "大安", "小吉"), res
res = lost_item(res)
assert res == "大安加小吉,失物不出门。", res

res = get_factor_time(datetime.datetime(2022, 4, 30, 8, 20))
assert res == ("速喜", "留连", "空亡"), res
res = lost_item(res)
assert res == "留连加空亡,失物不见面。", res

表象属性

大安
不动 木 青色 东 吉祥
大安事事昌,求谋在东方,失物去不远

留连
慢 四方土 黑色 纠狸 拖延
留连事难成,求谋日未明,官事只宜缓

速喜
快速 火 红色 南方
吉利求财向南行,失物申未午(南或西南)

赤口
吵 金 白色 西方 凶险
赤口主口舌,官非切要防,失物急去寻

小吉
主动 缓慢 水 北方 吉利
小吉最吉昌,失物在坤方(西南)

空亡
无 随性 中央土 黄色 难以找回
空亡事不禅,失物寻不见

卦辞

大安日

  • 大安加留連,辦事不周全,失物西北去,婚姻晚幾天。
  • 大安加速喜,事事自己起,失物當日見,婚姻自己提。
  • 大安加赤口,辦事不順手,失物不用找,婚姻兩分手。
  • 大安加小吉,事事從己及,失物不出門,婚姻成就地。
  • 大安加空亡,病人要上床,失物無蹤影,事事不順情。

留連日

  • 留連加大安,辦事兩分張,婚姻有喜事,先苦後來甜。
  • 留連加速喜,事事由自己,婚姻有成意,失物三天裡。
  • 留連加赤口,病者死人口,失物準丟失,婚姻兩分手。
  • 留連加小吉,事事不用提,失物東南去,病者出人齊。
  • 留連加空亡,病人準死亡,失物不見面,婚姻兩分張。

速喜日

  • 速喜加赤口,自己往外走,失物往正北,婚姻得勤走。
  • 速喜加小吉,婚姻有人提,病人當天好,時物在家裡。
  • 速喜加空亡,婚姻有分張,病者積極治,失物不久見。
  • 速喜加大安,事事都平安,姻姻成全了,佔病都相安。
  • 速喜加留連,婚姻不可言,失物無資訊,病人有仙緣。

赤口日

  • 赤口加小吉,辦事自己提,婚姻不能成,失物無資訊。
  • 赤口加空亡,無病也上床,失物不用找,婚姻不能成。
  • 赤口加大安,辦事險和難,失物東北找,婚姻指定難。
  • 赤口加留連,辦事有困難,行人在外走,失物不回還。
  • 赤口加速喜,婚姻在自己,失物有著落,辦事官事起。

小吉日

  • 小吉加空亡,病人不妥當,失物正東找,婚姻再想想。
  • 小吉加大安,事事兩週全,婚姻當日定,失物自己損。
  • 小吉加留連,事事有反還,婚姻有人破,失物上西南。
  • 小吉加速喜,事事從頭起,婚姻能成就,失物在院裡。
  • 小吉加赤口,辦事往外走,婚姻有難處,失物丟了手。

空亡日

  • 空亡加大安,事事不周全,婚姻從和好,失物反覆間。
  • 空亡加留連,辦事處處難,婚姻重新定,失物永不還。
  • 空亡加速喜,事事怨自己,婚姻有一定,失物在家裡。
  • 空亡加赤口,辦事官非有,婚姻難定準,失物往遠走。
  • 空亡加小吉,事事有猜疑,婚姻有喜事,失物回家裡。
Python阅读需 6 分钟

老方案

save_dir = os.path.join(os.path.expanduser("~"), "Pictures", "wallpaper")
os.makedirs(save_dir, exist_ok=True)

新方案

In [16]: import ctypes
...: from ctypes.wintypes import MAX_PATH
...:
...: dll = ctypes.windll.shell32
...: buf = ctypes.create_unicode_buffer(MAX_PATH + 1)
...: for i in range(100):
...: if dll.SHGetSpecialFolderPathW(None, buf, 0x0000 +i, False):
...: print( 0x0000 +i,buf.value)
...: else:
...: pass
...: # print("Failure!")
...:
0 C:\Users\tmp_user\Desktop
2 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\Start Menu\Programs
5 E:\Users\tmp_user\Documents
6 E:\Users\tmp_user\Favorites
7 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Startup
8 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\Recent
9 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\SendTo
11 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\Start Menu
13 E:\Users\tmp_user\Music
14 E:\Users\tmp_user\Videos
16 C:\Users\tmp_user\Desktop
19 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\Network Shortcuts
20 C:\WINDOWS\Fonts
21 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\Templates
22 C:\ProgramData\Microsoft\Windows\Start Menu
23 C:\ProgramData\Microsoft\Windows\Start Menu\Programs
24 C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Startup
25 C:\Users\Public\Desktop
26 C:\Users\tmp_user\AppData\Roaming
27 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\Printer Shortcuts
28 C:\Users\tmp_user\AppData\Local
29 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Startup
30 C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Startup
31 E:\Users\tmp_user\Favorites
32 C:\Users\tmp_user\AppData\Local\Microsoft\Windows\INetCache
33 C:\Users\tmp_user\AppData\Local\Microsoft\Windows\INetCookies
34 C:\Users\tmp_user\AppData\Local\Microsoft\Windows\History
35 C:\ProgramData
36 C:\WINDOWS
37 C:\WINDOWS\system32
38 C:\Program Files
39 E:\Users\tmp_user\Pictures
40 C:\Users\tmp_user
41 C:\WINDOWS\SysWOW64
42 C:\Program Files (x86)
43 C:\Program Files\Common Files
44 C:\Program Files (x86)\Common Files
45 C:\ProgramData\Microsoft\Windows\Templates
46 C:\Users\Public\Documents
47 C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Administrative Tools
48 C:\Users\tmp_user\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Administrative Tools
53 C:\Users\Public\Music
54 C:\Users\Public\Pictures
55 C:\Users\Public\Videos
56 C:\WINDOWS\resources
59 C:\Users\tmp_user\AppData\Local\Microsoft\Windows\Burn\Burn

整合以后

def get_default_save_dir(dir_name:str= "Pictures"):
# windows platform
res_path = os.path.join(os.path.expanduser("~"), dir_name)
if os.name == "nt":
import ctypes
from ctypes.wintypes import MAX_PATH
...:
dll = ctypes.windll.shell32
buf = ctypes.create_unicode_buffer(MAX_PATH + 1)
for i in range(100):
if dll.SHGetSpecialFolderPathW(None, buf, 0x0000 +i, False) and dir_name in buf.value:
return buf.value
return res_path
Python阅读需 1 分钟

  • 参考 (doc just the docs 主题蛮好看的)
  • 基本不用,现场不方便改代码 debug。
python -m pdb my_script.py

cat my_script.py

import pdb

def make_bread():
pdb.set_trace()
return "I don't have time"

print(make_bread())

命令列表:

  • c(continue): 继续执行
  • w(where): 显示当前正在执行的代码行的上下文信息
  • a(args): 打印当前函数的参数列表
  • s(step): 执行当前代码行,并停在第一个能停的地方(相当于单步进入)
  • n(next): 继续执行到当前函数的下一行,或者当前行直接返回(单步跳过)

单步跳过(next)和单步进入(step)的区别在于,单步进入会进入当前行调用的函数内部并停在里面,而单步跳过会(几乎)全速执行完当前行调用的函数,并停在当前函数的下一行。

Python阅读需 1 分钟

在项目中写了 doctest,但是需要先初始化环境才方便使用。

在运行 pytest 的目录下创建conftest.py

#!/usr/bin/python3
# encoding: utf-8
# @author : zza
# @File : conftest.py
""" FOR DOCTEST """
import pytest

@pytest.fixture(autouse=True)
def init_env(doctest_namespace):
print("doctest_namespace", doctest_namespace)
import os
sql_url = "sqlite:///data.db"
os.environ['sql_uri'] = sql_url
init(sql_url)
  • pytest 启动就用 pytest 命令就好了:pytest
  • coverage 启动时使用该命令启动:coverage run --source my_pachage -m pytest

更多使用案例建议 github 直接 搜索conftest.py文件。

Python阅读需 1 分钟

fire 是一个比 click 更有使用体验的命令行工具包。

但是在 help 命令中,会进入 man 的交互模式里,不能直接打印到标准输出里。

看了下 Issue,有几种解决方案,记录一下。

修改 fire.core.Display

if __name__ == "__main__":
# Make Python Fire not use a pager when it prints a help text
fire.core.Display = lambda lines, out: print(*lines, file=out)
fire.Fire(...)

更改 PAGER 环境变量(推荐)

* windows 上设置 setx PAGER type
* linux 上设置 export PAGER=cat

在程序中修改环境变量

os.environ["PAGER"] = 'cat'

  • 一个写完后发现没必要的代码
def _fire_print_prepare() -> None:
"""help to print mode `setx PAGER cat`"""
import os
import platform
from shutil import which

if os.environ.get("PAGER"):
return
if which("cat"):
os.environ["PAGER"] = "cat"
elif platform.platform().startswith("Windows"):
os.environ["PAGER"] = "type"
Python阅读需 1 分钟

相关资料:

安装依赖

pip install pstats  
pip install snakeviz

获取程序运行数据

通过cProfile获取性能分析数据

  • cProfile自python2.5以来就是标准版Python解释器默认的性能分析器。
  • cProfile是一种确定性分析器,只测量CPU时间,并不关心内存消耗和其他与内存相关联的信息。
def run_1():  
"""your function """
pass


def profile_func(func):
import cProfile
file_name = "prof_{}_1.pstat".format(func.__name__)
cProfile.run("{}()".format(func.__name__), file_name)
import pstats
p = pstats.Stats(file_name).sort_stats("cumtime")
# p.print_stats("rqalpha_mod_ricequant_data")
p.print_stats("base_position")
return p


if __name__ == '__main__':
p = profile_func(run_1)

通过pycharm直接获取运行数据

Pycharm 右上角启动按钮旁边有性能分析按钮,
点击后会自动弹出pycharm自带的性能分析界面,
当然也可以复制.pstat文件地址,使用snakeviz进行分析(推荐)。


snakeviz 生成剖面图

运行目录下 命令行输入:snakeviz prof_run_1_1.pstat

点击生成的连接 查看柱状剖面图

image.png


gprof2dot 时间分析图

gprof2dot -f pstats mkm_run.prof | dot -Tpng -o mkm_run.png

image.png

Python阅读需 1 分钟

转载自《PyPI 打包上传实践》

1. 代码打包

要打包代码,首先需要编写自己的代码包。比如你写了一个。py 文件,里面有一些函数啥的,为了方便调用,你需要将代码打包,下次使用时直接调用就好,因此,第一步,将你写的代码打包。 创建一个文件夹,并在该文件夹下创建 __init__.py 文件,然后将你写的。py 文件放到这个文件夹下面就行。

packagename/    
|
+-- __init__.py
|
+-- myfunction.py
|
+-- mymorefunction.py
|
+-- ...
|

packagename为你创建的包名称,myxxx.py是你写的 python 代码,还有添加个__init__.py文件(文件内容可以为空). 现在你可以调用这个包了(引入包的路径)

import packagename    

2. 符合 pypi 的格式

将上面的文件的目录结构改成如下格式

packagename    
|
+-- COPYING.txt
|
+-- README.rst
|
+-- setup.py
|
+-- packagename
. |
. +-- __init__.py
. |
. +-- myscripts1.py
. |
. +-- mysscripts2.py
. |
. +-- mymorescripts.py
. |
.
|
+-- docs/
|

就是将原来的目录深移一层,文件夹的名称一样即可。在第一层目录下创建些特殊文件。 Tips

  • COPYING.txt : 可以不要(节约时间,重要的事情先说、简单说)。 就是授权文件,里面是你关于这个包的授权,比如:MIT license,那么你里面放入 MIT License 全文即可,当然,如果你不清楚这个,你完全可以不要这个文件。
  • README.rst:就是介绍,可以不要吧(不推荐,要是想让大家用的话还是好好写一写) 这个文件想必研发都应该清楚。如果有,尽量放些东西在这里了,后面如果可能我们会用到它的。
  • setup.py:核心文件 这里面的内容后面讲
  • docs/(这是个文件夹,存放一些文档的) 这个文件夹你放你的 documents 吧,不过要用心写文档真是个难事,所以这个文件夹基本是不存在的——为自己的懒惰可耻一把。

**setup.py **的样例

# coding: utf-8    
import codecs
import os
import sys

try:
from setuptools import setup
except:
from distutils.core import setup

"""
打包的用的 setup 必须引入,
"""

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
if sys.version_info < (2, 5):
sys.exit('Python 2.5 or greater is required.')

try:
from setuptools import setup
except ImportError:
from distutils.core import setup

import SendMoney

with open('README.rst', 'rb') as fp:
readme = fp.read()

# 版本号,自己随便写
VERSION = "1.0.7"

LICENSE = "MIT"

setup(
name='<项目的名称>',
version=VERSION,
description=(
'<项目的简单描述>'
),
long_description=readme,
author='<你的名字>',
author_email='<你的邮件地址>',
maintainer='<维护人员的名字>',
maintainer_email='<维护人员的邮件地址',
license=LICENSE,
packages=find_packages(),
platforms=["all"],
url='<项目的网址,我一般都是 github 的 url>',
install_requires=[
"beautifulsoup4",
lxml_requirement
],
classifiers=[
'Development Status :: 4 - Beta',
'Operating System :: OS Independent',
'Intended Audience :: Developers',
'License :: OSI Approved :: BSD License',
'Programming Language :: Python',
'Programming Language :: Python :: Implementation',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'Topic :: Software Development :: Libraries'
],
)

# URL 你这个包的项目地址,如果有,给一个吧,没有你直接填写在 PyPI 你这个包的地址也是可以的
# INSTALL_REQUIRES 模块所依赖的 python 模块
# 以上字段不需要都包含

文中的 classifiers 的内容并不是随便填写的,你需要参照本文参考文档中的 PyPI Classifiers 来写

3、开始使用 Distutils 进行打包

为了保证效果,在打包之前我们可以验证 setup.py 的正确性,执行下面的代码

python setup.py check

输出一般是 running check 如果有错误或者警告,就会在此之后显示 没有任何显示表示 Distutils 认可你这个 setup.py 文件。

如果没有问题,那么就可以正式打包,执行下面的代码:

python setup.py sdist

执行完成后,会在顶层目录下生成 dist 目录和 egg 目录

打包完成后就可以准备将打包好的模块上传到 pypi 了,首先你需要在 pypi 上进行注册 注册完成后,你需要在本地创建好 pypi 的配置文件,不然有可能会出现使用 http 无法上传到 pypi 的问题 在用户目录下创建。pypirc 文件,文件的内容如下 window 用户创建.pypirc可以命名为.pypirc. 位置示例:C:\Users\admin\.pypirc.

[distutils]    
index-servers=pypi

[pypi]
repository = https://upload.pypi.org/legacy/
username = <username>
password = <password>

完成后运行:

python setup.py register sdist upload

最后出现Server response (200): OK就是成功了,可以去 pypi 上查看自己发布的包

包到这里,就完成了上传 PyPI 的工作了。你如果要用,安装下就好:

pip install packagename

这个过程还是很顺利的,以后多尝试,出现问题再补充!

作者:snowy_sunny 链接:https://www.jianshu.com/p/be91c70adb27

Python阅读需 5 分钟

相关参考 python 中文写入 CSV 乱码解决

  • 原因:csv 文件开头缺少 BOM_UTF8 字符
  • 解决:在 Response 开头加上 BOM_UTF8 字符
  • 核心代码
titles[0] = codecs.BOM_UTF8.decode("utf8")+codecs.BOM_UTF8.decode()+titles[0]    
  • 相关引入
import codecs    
import csv

from flask_admin._compat import csv_encode
from flask import request, redirect, flash, current_app, Response, stream_with_context
  • 逻辑代码(复制 放到对应的 ModelView 类里就好了)
    def _export_csv(self, return_url):    
"""
Export a CSV of records as a stream.
"""
count, data = self._export_data()

# https://docs.djangoproject.com/en/1.8/howto/outputting-csv/
class Echo(object):
"""
An object that implements just the write method of the file-like
interface.
"""
def write(self, value):
"""
Write the value by returning it, instead of storing
in a buffer.
"""
return value

#
writer = csv.writer(Echo())

def generate():
# Append the column titles at the beginning
titles = [csv_encode(c[1]) for c in self._export_columns]
titles[0] = codecs.BOM_UTF8.decode("utf8")+codecs.BOM_UTF8.decode()+titles[0]
yield writer.writerow(titles)

for row in data:
vals = [csv_encode(self.get_export_value(row, c[0]))
for c in self._export_columns]
yield writer.writerow(vals)

filename = self.get_export_name(export_type='csv')

disposition = 'attachment;filename=%s' % (secure_filename(filename),)

return Response(
stream_with_context(generate()),
headers={'Content-Disposition': disposition},
mimetype='text/csv'
)
Python阅读需 1 分钟