#47 Python 定时任务的简单部署

2018-12-20

部署

# curl: (35) SSL connect error
# StackOverflow: You are using a very old version of curl.
# yum upgrade curl

# 安装 pyenv
curl -L https://github.com/pyenv/pyenv-installer/raw/master/bin/pyenv-installer | bash

# 环境变量
PYENV_ROOT="$HOME/.pyenv"
PATH="$PYENV_ROOT/bin:$PATH"
eval "$(pyenv init -)"
eval "$(pyenv virtualenv-init -)"

# 可用版本列表
pyenv install -l
# 使用国内 Python 镜像,相当于给 pyenv 加速
wget https://mirrors.sohu.com/python/3.6.7/Python-3.6.7.tar.xz -P ~/.pyenv/cache/
# 安装(优先使用缓存目录中的文件,会检查缓存校验码是否正确)
pyenv install 3.6.7

运行方式

假定:

  • 项目路径:/path/to/project/
  • 定时任务命令:python main.py

crontab 配置

早上 01:15 执行某某定时任务:

15 1 * * * /bin/bash /path/to/project/cron.sh

脚本 cron.sh

pyenv + pip 模式

#!/bin/bash

# 环境变量
PYENV_ROOT="$HOME/.pyenv"
PATH="$PYENV_ROOT/bin:$PATH"
eval "$(pyenv init -)"
eval "$(pyenv virtualenv-init -)"

cd /path/to/project/

# 创建并激活虚拟环境,安装依赖
# $PYENV_ROOT/versions/3.6.7/envs/tasks/
pyenv virtualenv 3.6.7 tasks

if [ $? -ne 0 ]; then
    pyenv activate tasks
    pip install -r requirements.txt
else
    pyenv activate tasks
fi

# 运行脚本
python main.py

pyenv + pipenv 模式

#!/bin/bash

# 环境变量
PYENV_ROOT="$HOME/.pyenv"
PATH="$PYENV_ROOT/bin:$PATH"
eval "$(pyenv init -)"
eval "$(pyenv virtualenv-init -)"

cd /path/to/project/

# 选择 Python 版本
pyenv local 3.6.7

pip show pipenv > /dev/null

if [ $? -ne 0 ]; then
    # 安装 pipenv
    pip install pipenv
    # 创建虚拟环境,安装依赖
    pipenv install
fi

# 运行脚本
pipenv run python main.py

#45 PEP

2018-11-22

主要是了解一下 PEP(Python Enhancement Proposal)的状态和类型。

#44 Python 打包上传到 PyPI

2018-05-26

首先你得有 PyPI 的账号,没有注册的话,不用搞了。
然后你肯定要先准备 setup.py 了,如果这个都没弄就不用搞了。

算了,先贴一个简单的样板吧。

setup.py

# -*- coding: utf-8 -*-

from setuptools import setup

setup(
    name='PageageName',
    version='0.0.1',
    author='YourName',
    author_email='YourEmail',
    url='PackageSite (e.g. GitHub)',
    description='....',
    long_description=open('README.md').read(),
    long_description_content_type='text/markdown',
    packages=['PackageDir'],
    install_requires=[
        'requests>=2.18.4',
        'balabala',
    ],
    extras_require={  # 可选
        'dev': [
            'balabala...',
        ],
        'test': [
            'balabala...',
        ],
    },
    classifiers=[
        'Development Status :: 3 - Alpha',
        'Programming Language :: Python',
        'Programming Language :: Python :: 2',
        'Programming Language :: Python :: 2.7',
        'Programming Language :: Python :: 3',
        'Programming Language :: Python :: 3.6',
        'Programming Language :: Python :: 3.7',
        'Programming Language :: Python :: 3.8',
        'License :: OSI Approved :: MIT License',
    ],
)

流程

python3 -m pip install twine

rm -rf dist

python3 setup.py sdist bdist_wheel
python2 setup.py sdist bdist_wheel

twine upload dist/*

twine 会询问账号密码,如果不想每次这么麻烦,可以创建 ~/.pypirc 文件。

.pypirc 文件

[distutils]
index-servers =
    pypi

[pypi]
repository = https://upload.pypi.org/legacy/
username = your_username
password = your_password

参考资料

#43 Python 应用: 简易 SMTP 服务器

2018-05-07

Python2

python2 -m smtpd -h
# An RFC 2821 smtp proxy.
# Usage: /usr/lib/python2.7/smtpd.py [options] [localhost:localport [remotehost:remoteport]]
#     --nosetuid, -n
#     --version, -V
#     --class classname, -c classname
#     --debug, -d
#     --help, -h

Python3

python -m smtpd -h
# An RFC 5321 smtp proxy with optional RFC 1870 and RFC 6531 extensions.
# Usage: /usr/lib/python3.9/smtpd.py [options] [localhost:localport [remotehost:remoteport]]
#     --nosetuid, -n  默认会设置用户为 nobody,如果不是 root 会因权限不足失败
#     --version, -V
#     --class classname, -c classname 默认: PureProxy
#     --size limit, -s limit 消息大小限制(RFC 1870 SIZE extension),默认是 33554432 字节,即 32MB
#     --smtputf8, -u 启用 SMTPUTF8 扩展(RFC 6531)
#     --debug, -d
#     --help, -h

# 如果不指定主机,就使用 localhost
# 如果主机是 localhost,端口使用 8025
# 如果是其他主机,端口使用 25
python3 -m smtpd -n

# 默认的 PureProxy 会给转信出去,正常情况会被服务器拒绝
python3 -m smtpd -n -c smtpd.DebuggingServer

Python 3.9 的 PureProxy 有 BUG,会报 process_message() got an unexpected keyword argument 'mail_options'

自定义黑洞服务器

blackhole.py

import smtpd
import time
class BlackHoleServer(smtpd.SMTPServer):
    def process_message(self, peer, mailfrom, rcpttos, data, **kwargs):
        print('%s %s %s -> %s' % (time.strftime('%Y-%m-%d %H:%M:%S'), peer, mailfrom, rcpttos))

setup.py

import setuptools
setuptools.setup(name="blackhole", py_modules=["blackhole"])

附件下载:blackhole.zip

python setup.py install --user
python -m smtpd -n -c blackhole.BlackHoleServer

测试

import smtplib
smtp = smtplib.SMTP('localhost', 8025)
from_addr = 'admin@markjour.com'
to_addr = 'you@markjour.com'
smtp.sendmail(from_addr, to_addr,
    f"""From: {from_addr}\nTo: {to_addr}\nSubject: just4fun\n\nhello, world!""")

#41 Python 应用: 简易 HTTP 服务器

2018-05-07

Python2

在当前目录起 HTTP 服务,可以用于测试和临时性的文件下载服务。

# Default bind to 0.0.0.0:8000
python -m SimpleHTTPServer

# Maybe you want to use port 8080
python -m SimpleHTTPServer 8080

Python3

除了可以指定端口,还可以指定绑定地址、工作目录。

# Also bind to 0.0.0.0:8000
python -m http.server

python -m http.server -h
# usage: server.py [-h] [--cgi] [--bind ADDRESS] [--directory DIRECTORY] [port]
#
# positional arguments:
#   port                  Specify alternate port [default: 8000]
#
# optional arguments:
#   -h, --help            show this help message and exit
#   --cgi                 Run as CGI Server
#   --bind ADDRESS, -b ADDRESS
#                         Specify alternate bind address [default: all interfaces]
#   --directory DIRECTORY, -d DIRECTORY
#                         Specify alternative directory [default:current directory]

python -m http.server 9999
python -m http.server --bind=127.0.0.1
python -m http.server --bind=127.0.0.1 9999
python -m http.server -d ~/Pictures

#40 Python Redis Debug

2018-05-06

如果可以的话,使用 redis-cli monitor 命令来输出所有 Redis 命令也很方便。

有时,条件不允许,或者 Redis 需要处理其他的连接,我希望将自己代码调用的 Redis 命令输出到日志中,方便调试。

#39 Python SMTP

2018-04-13
import logging
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

logging.basicConfig(level=logging.DEBUG)

SMTP_HOST = 'smtp.example.com'
SMTP_PORT = 587
SMTP_USERNAME = 'your_username'
SMTP_PASSWORD = 'your_password'
SMTP_STARTTLS = True

sender = 'sender@example.com'
recipients = ['rcpt01@example.com', '中国 <rcpt02@example.com>']
subject = 'Test Email'
content_text = 'This is a test email.'
content_html = '<html><h1>Hello</h1></html>'


def encode_recipient(name, addr):
    pass


msg = MIMEMultipart()
msg['From'] = sender
msg['To'] = ', '.join(recipients)
msg['Subject'] = subject
msg.attach(MIMEText(content_text))
msg.attach(MIMEText(content_html))
print(msg.as_string())


def smtp_debug(self, *args):
    msg = ' '.join(map(str, args))
    logging.debug(msg)


smtp = smtplib.SMTP(SMTP_HOST, SMTP_PORT)
smtp._print_debug = smtp_debug
smtp.set_debuglevel(1)
smtp.ehlo()
if SMTP_STARTTLS:
    smtp.starttls()
    smtp.ehlo()
if SMTP_USERNAME and SMTP_USERNAME:
    smtp.login(SMTP_USERNAME, SMTP_PASSWORD)

smtp.sendmail(sender, recipients, msg.as_string())
smtp.quit()