分类目录归档:Python

pyenv — 多版本Python(含镜像设置加速下载以及Windows)

编译python依赖包

sudo apt install libbz2-dev libcurses-ocaml-dev libctypes-ocaml-dev libreadline-dev libssl-ocaml-dev libffi-dev libsqlite3-dev liblzma-dev

安装pyenv

curl https://pyenv.run | bash

显示已安装python版本

pyenv install --list

设置使用镜像

 export PYTHON_BUILD_MIRROR_URL_SKIP_CHECKSUM=1
 export PYTHON_BUILD_MIRROR_URL="https://npm.taobao.org/mirrors/python/"

安装指定版本Python

pyenv install 3.9.9

运行指定版本Python

pyenv global 3.9.9

查看当前已安装和正在运行的Python版本

pyenv versions

查看已安装版本位置

pyenv prefix 3.9.9

windows安装pyenv

管理员身份进入powershell

Invoke-WebRequest -UseBasicParsing -Uri "https://raw.githubusercontent.com/pyenv-win/pyenv-win/master/pyenv-win/install-pyenv-win.ps1" -OutFile "./install-pyenv-win.ps1"; &"./install-pyenv-win.ps1"

C:\Users\Administrator\.pyenv\pyenv-win\.versions_cache.xml中的所有www.python.org/ftp替换为npmmirror.com/mirrors

安装完以后重新管理员打开powershell

pyenv install 3.10.5

其他

  • pyenv的替代工具是asdf, 可以管理多个语言的版本,包含Python,node,Erlang,Ruby,PHP,Mysql,Postgres等

参考

pipx — 在隔离的环境中安装和运行Python应用程序

pipx 介绍

在隔离的环境中安装和运行Python应用程序,安装的程序都在自己隔离的环境,不用考虑依赖冲突,且可以安装多个版本,可以非root身份运行程序。官网:https://github.com/pypa/pipx

安装 pipx

python3 -m pip install --user pipx

配置自动完成

pipx completions

使用 pipx 安装

pipx install pipenv

通过 url 安装

pipx install git+https://github.com/psf/black.git
pipx install git+https://github.com/psf/black.git@branch  # branch of your choice
pipx install git+https://github.com/psf/black.git@ce14fa8b497bae2b50ec48b3bd7022573a59cdb1  # git hash
pipx install https://github.com/psf/black/archive/18.9b0.zip  # install a release

列出已安装应用

pipx list

升级

pipx upgrade pipenv

运行

pipx run pipenv

指定版本运行

pipx run APP==1.0.0

通过 url 运行

pipx run https://gist.githubusercontent.com/cs01/fa721a17a326e551ede048c5088f9e0f/raw/6bdfbb6e9c1132b1c38fdd2f195d4a24c540c324/pipx-demo.py

卸载

pipx  uninstall pipenv

卸载所有

pipx  uninstall-all

显示帮助

pipx  -h

Django admin export inline sub models

Django 导入导出大部分情况都可以使用django-import-export扩展,但是实际业务中经常需要导出子模型数据,比如导出订单同时需要导出订单商品信息。django-import-export扩展默认不支持导出inline子模型,这时候需要自己实现resource的export方法:

自定义django-import-export的export方法

class OrderResource(resources.ModelResource):
    class Meta:
        model = Order

    def export(self, queryset=None, *args, **kwargs):
        if queryset is None:
            queryset = self.get_queryset()
        ds = tablib.Dataset()
        data = []
        for order in queryset:
            for item in order.items.all():
                row = {
                    'order_id': order.id,
                    "item_id": item.idd,
                }
                data.append(row)
        ds.dict = data
        return ds

参考

django-q — 比django-celery和django-rq更简单又不失强大的django队列

django-q介绍

Django Q is a native Django task queue, scheduler and worker application using Python multiprocessing.

Features

  • Multiprocessing worker pools
  • Asynchronous tasks
  • Scheduled, cron and repeated tasks
  • Signed and compressed packages
  • Failure and success database or cache
  • Result hooks, groups and chains
  • Django Admin integration
  • PaaS compatible with multiple instances
  • Multi cluster monitor
  • Redis, Disque, IronMQ, SQS, MongoDB or ORM
  • Rollbar and Sentry support

Django Q is tested with: Python 3.7 and 3.8, Django 2.2.x and 3.1.x

安装django-q模块

pip install django-q

增加 django_q 到 INSTALLED_APPS:

修改settings.py

INSTALLED_APPS = (
    # other apps
    'django_q',
)

创建数据库表:

运行 Django migrations

$ python manage.py migrate

配置一个broker

使用django orm数据库作为broker

Q_CLUSTER = {
    'name': 'DjangORM',
    'workers': 1,
    'timeout': 90,
    'retry': 120,
    'queue_limit': 50,
    'bulk': 10,
    'orm': 'default'
}

使用redis作为broker

Q_CLUSTER = {
    'redis': 'redis://h:asdfqwer1234asdf@ec2-111-1-1-1.compute-1.amazonaws.com:111'
}

使用django_redis作为broker

Q_CLUSTER = {
    'name': 'DJRedis',
    'workers': 4,
    'timeout': 90,
    'django_redis': 'default'
}

其他broker: https://django-q.readthedocs.io/en/latest/brokers.html

使用qcluster处理异步任务

python manage.py qcluster

增加异步任务到队列

from django_q.tasks import async_task, result

# create the task
async_task('math.copysign', 2, -2)

# or with import and storing the id
import math.copysign

task_id = async_task(copysign, 2, -2)

# get the result
task_result = result(task_id)

# result returns None if the task has not been executed yet
# you can wait for it
task_result = result(task_id, 200)

# but in most cases you will want to use a hook:

async_task('math.modf', 2.5, hook='hooks.print_result')

# hooks.py
def print_result(task):
    print(task.result)

管理后台页面

Django Q不使用自定义页面,而是默认使用Django模型管理员提供的功能。当您打开Django Q的管理页面时,您将看到三种模型:

成功的任务

意味着他们在执行过程中没有遇到任何错误。

file

失败的任务

失败的任务遇到错误,阻止其完成执行。

计划任务

在这里,您可以检查计划任务的状态,创建,编辑或删除它们。

file

排队的任务

仅当您使用Django ORM代理时才出现

参考

通过Xvfb在命令行界面运行GUI程序

wkhtmltopdf使用的qt版本比较久,不支持很多css3和html5特性,于是用qt写了一个html转PDF工具,需要常驻后台运行,采用supervisord守护,无需安装图形界面,这时候需要用到Xvfb。

Xvfb介绍

Xvfb or X virtual framebuffer is a display server implementing the X11 display server protocol. In contrast to other display servers, Xvfb performs all graphical operations in virtual memory without showing any screen output.

简单说就是Xvfb是一个虚拟的X11显示服务,在虚拟内存中执行不需要显示图像。可以用来做远程左面显示,以及将桌面程序在非桌面环境使用(如headless的图形界面测试)。

Xvfb基本使用

安装xvfb

sudo apt-get install xvfb

命令行运行虚拟桌面

Xvfb :2 -screen 0 1024x768x16 &
x11vnc -listen 0.0.0.0 -rfbport 5900 -noipv6 -passwd xxxxxx -display 2
export DISPLAY=:2

使用Python xvfbwrapper

from xvfbwrapper import Xvfb

with Xvfb(width=1920, height=1024, colordepth=24) as xvfb:
    # launch stuff inside virtual display here.
    # Xvfb will stop when this block completes

参考

Django工作流(状态管理)

本文通过简单的例子比较了django状态管理扩展的使用,涵盖django-fsm,xworkflows,django_transitions以及参考

django-fsm 使用例子

https://github.com/viewflow/django-fsm

state = FSMField(
    default=State.DRAFT,
    verbose_name='Publication State',
    choices=State.CHOICES,
    protected=True,
)
@transition(field=state, source=[State.APPROVED, State.EXPIRED],
    target=State.PUBLISHED,
    conditions=[can_display])
def publish(self):
    '''
    Publish the object.
    '''
    email_the_team()
    update_sitemap()
    busta_cache()

django-fsm-admin

https://github.com/gadventures/django-fsm-admin

from fsm_admin.mixins import FSMTransitionMixin

class YourModelAdmin(FSMTransitionMixin, admin.ModelAdmin):
    # The name of one or more FSMFields on the model to transition
    fsm_field = ['state',]
    readonly_fields = ['state', ]

admin.site.register(YourModel, YourModelAdmin)

xworkflows

https://github.com/rbarrois/xworkflows

import xworkflows

class MyWorkflow(xworkflows.Workflow):
    # A list of state names
    states = (
        ('foo', "Foo"),
        ('bar', "Bar"),
        ('baz', "Baz"),
    )
    # A list of transition definitions; items are (name, source states, target).
    transitions = (
        ('foobar', 'foo', 'bar'),
        ('gobaz', ('foo', 'bar'), 'baz'),
        ('bazbar', 'baz', 'bar'),
    )
    initial_state = 'foo'

class MyObject(xworkflows.WorkflowEnabled):
    state = MyWorkflow()

    @xworkflows.transition()
    def foobar(self):
        return 42

    # It is possible to use another method for a given transition.
    @xworkflows.transition('gobaz')
    def blah(self):
        return 13

django_transitions

LiveStatus Status

class LiveStatus(StatusBase):
    """Workflow for Lifecycle."""

    # Define the states as constants
    DEVELOP = 'develop'
    LIVE = 'live'
    MAINTENANCE = 'maintenance'
    DELETED = 'deleted'

    # Give the states a human readable label
    STATE_CHOICES = (
        (DEVELOP, 'Under Development'),
        (LIVE, 'Live'),
        (MAINTENANCE, 'Under Maintenance'),
        (DELETED, 'Deleted'),
    )

    # Define the transitions as constants
    PUBLISH = 'publish'
    MAKE_PRIVATE = 'make_private'
    MARK_DELETED = 'mark_deleted'
    REVERT_DELETED = 'revert_delete'

    # Give the transitions a human readable label and css class
    # which will be used in the django admin
    TRANSITION_LABELS = {
        PUBLISH : {'label': 'Make live', 'cssclass': 'default'},
        MAKE_PRIVATE: {'label': 'Under maintenance'},
        MARK_DELETED: {'label': 'Mark as deleted', 'cssclass': 'deletelink'},
        REVERT_DELETED: {'label': 'Revert Delete', 'cssclass': 'default'},
    }

    # Construct the values to pass to the state machine constructor

    # The states of the machine
    SM_STATES = [
        DEVELOP, LIVE, MAINTENANCE, DELETED,
    ]

    # The machines initial state
    SM_INITIAL_STATE = DEVELOP

    # The transititions as a list of dictionaries
    SM_TRANSITIONS = [
        # trigger, source, destination
        {
            'trigger': PUBLISH,
            'source': [DEVELOP, MAINTENANCE],
            'dest': LIVE,
        },
        {
            'trigger': MAKE_PRIVATE,
            'source': LIVE,
            'dest': MAINTENANCE,
        },
        {
            'trigger': MARK_DELETED,
            'source': [
                DEVELOP, LIVE, MAINTENANCE,
            ],
            'dest': DELETED,
        },
        {
            'trigger': REVERT_DELETED,
            'source':  DELETED,
            'dest': MAINTENANCE,
        },
    ]

Model:

class Lifecycle(LifecycleStateMachineMixin, models.Model):
    """
    A model that provides workflow state and workflow date fields.

    This is a minimal example implementation.
    """

    class Meta:  # noqa: D106
        abstract = False

    wf_state = models.CharField(
        verbose_name = 'Workflow Status',
        null=False,
        blank=False,
        default=LiveStatus.SM_INITIAL_STATE,
        choices=LiveStatus.STATE_CHOICES,
        max_length=32,
        help_text='Workflow state',
    )

    wf_date =  models.DateTimeField(
        verbose_name = 'Workflow Date',
        null=False,
        blank=False,
        default=timezone.now,
        help_text='Indicates when this workflowstate was entered.',
    )

Admin

# -*- coding: utf-8 -*-
"""Example django admin."""

from django_transitions.admin import WorkflowAdminMixin
from django.contrib import admin

from .models import Lifecycle

class LifecycleAdmin(WorkflowAdminMixin, admin.ModelAdmin):
    """
    Minimal Admin for Lifecycles Example.

    You probably want to make the workflow fields
    read only so yo can not change these values
    manually.

    readonly_fields = ['wf_state', 'wf_date']
    """

    list_display = ['wf_date', 'wf_state']
    list_filter = ['wf_state']

admin.site.register(Lifecycle, LifecycleAdmin)

参考

著作权代码收集工具

使用Python编写,可以设置源码目录,排除目录,收集文件类型,输出目标文件,是否允许空白,编码等。

使用示例:

python zhuzuoquan.py -s D:\pythoncode\projectname \
-e debug,release \
-t .h,.cpp \
-en gb18030 \
-o result.txt

zhuzuoquan.py 代码

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

import argparse
import os

parser = argparse.ArgumentParser(
    description='著作权代码收集工具', usage='''
    zhuzuoquan.py -s 源码目录 -e 排除目录 -t 文件类型 -o 输出文件
    ''')
parser.add_argument('-s', help='源文件目录', required=True)
parser.add_argument('-e', help='排除目录,多个用半角逗号分割', )
parser.add_argument('-t', help='扩展名,多个用半角逗号分割', required=True)
parser.add_argument('-o', help='输出文件')
parser.add_argument('-el', help='允许空白行')
parser.add_argument('-en', help='编码', default='utf-8')

def filter_files(root_dir, excluded_dirs, allow_types):
    valid_files = []
    for root, dirs, files in os.walk(root_dir):

        for f in files:
            excluded = False
            p = os.path.join(root,f)
            d = p[root_dir.__len__():]
            for ex in excluded_dirs:
                if d.startswith(ex):
                    excluded = True

            allow = False
            for t in allow_types:
                if f.endswith(t):
                    allow = True
            if allow and not excluded:
                filepath = os.path.join(root, f)
                valid_files.append(filepath)
    return valid_files

if __name__ == '__main__':
    args = parser.parse_args()

    if args.e:
        excluded_dirs = args.e.split(',')
    else:
        excluded_dirs = []

    allow_types = args.t.split(',')

    root_dir = args.s

    output_file = args.o

    allow_empty_line = args.el

    encoding = args.en

    files = filter_files(root_dir, excluded_dirs, allow_types)
    total_files = total_lines = 0
    if output_file:
        with open(output_file, encoding=encoding, mode='w') as op:
            op.write("")

    for f in files:
        total_files = total_files + 1
        try:
            with open(f, encoding=encoding) as fh:
                content = fh.read()
                if not allow_empty_line:
                    content = "\n".join([s for s in content.splitlines() if s.strip()])
                lines = len(content.splitlines())
                relative_path = f[root_dir.__len__():]
                print(relative_path, lines)
                total_lines = total_lines + lines
                if output_file:
                    fh.seek(0)
                    with open(output_file, encoding=encoding, mode='a+') as op:
                        op.write(relative_path + ":\n\n" + content + "\n\n\n")
        except:
            pass

    print("Total: files %d, lines %d" % (total_files, total_lines))

Matplotlib使用中文字体(linux)

查看系统已有中文字体

fc-list :lang=zh

查询matplotlib默认配置示例

import matplotlib
matplotlib.matplotlib_fname()

查看当前系统matplotlib配置文件路径

import matplotlib
matplotlib.get_configdir()

全局修改

在系统如今下新增matplotlibrc文件,然后修改font.sans-serif,添加中文字体

font.sans-serif : WenQuanYi Micro Hei, Bitstream Vera Sans, ...

只修改当前程序

import matplotlib
matplotlib.rcParams['font.sans-serif'] = 'WenQuanYi Micro Hei'

参考

Python进入交互模式4种方法

使用python -i参数

添加-i参数

python -i myapp.py

使用code.interact()

    import code
    # before
    code.interact()
    # after

使用pdb lib

for thing in set_of_things:
    import pdb;
    pdb.set_trace()
    do_stuff_to(thing)

使用IPython embed

from IPython import embed

for thing in set_of_things:
  embed()
  do_stuff_to(thing)

Python 修改PDF文档尺寸以及去除水印图片

# -*- coding: UTF-8 -*-

import sys
import os

from pdfrw import PageMerge, PdfReader, PdfWriter, IndirectPdfDict
import fitz


# resize
def adjust(page):
    info = PageMerge().add(page)
    x1, y1, x2, y2 = info.xobj_box
    viewrect = ((x2 - 421) / 2, (y2 - 595) / 2, 421, 595)
    page = PageMerge().add(page, viewrect=viewrect)
    return page.render()

fin, = sys.argv[1:]
fout = 'mid.' + os.path.basename(fin)
reader = PdfReader(fin)
writer = PdfWriter(fout)
for p in reader.pages:
    writer.addpage(adjust(p))
writer.trailer.Info = IndirectPdfDict(reader.Info or {})
writer.write()

# trip backgroud images
doc = fitz.open(fout)
for i in range(len(doc)):
    imglist = doc.getPageImageList(i)
    for img in imglist:
        xref = img[0]
        if xref==51:
            doc._deleteObject(xref)
        print(img)
doc.save('new.' + os.path.basename(fin))