celery是一个基于Python的分布式调度系统，文档在这 ,最近有个需求,想要动态的添加任务而不用重启celery服务,找了一圈没找到什么好办法(也有可能是文档没看仔细)，所以只能自己实现囉

为celery动态添加任务，首先我想到的是传递一个函数进去，让某个特定任务去执行这个传递过去的函数，就像这样

@app.task
def execute(func, *args, **kwargs):
    return func(*args, **kwargs)

很可惜，会出现这样的错误

kombu.exceptions.EncodeError: Object of type 'function' is not JSON serializable

换一种序列化方式

@app.task(serializer='pickle')
def execute(func, *args, **kwargs):
    return func(*args, **kwargs)

结果又出现一大串错误信息

ERROR/MainProcess] Pool callback raised exception: ContentDisallowed('Refusing to deserialize untrusted content of type …

flask 貌似很少全文搜索的插件,有一个 https://github.com/gyllstromk/Flask-WhooshAlchemy , 但试了几次都用不了,所以参考 Flask-WhooshAlchemy 自己写了一个

Quickstart

插件基于 whoosh,纯 python 编写,使用上很简单

from flask_msearch import Search
[...]
search = Search()
search.init_app(app)

# models.py
class Post(db.Model):
    __tablename__ = 'post'
    __searchable__ = ['title', 'content']

# views.py
@app.route("/search")
def w_search():
    keyword = request.args.get('keyword')
    results = search.whoosh_search(Post,query …

第一版下载图片后设为壁纸,四周都是黑黑的不好看, 所以结合本地壁纸将两张图片进行合成

只要计算一下要缩减的大小与要放置的位置

我的方法是:打开 gimp,合成两张图片,记下缩减的大小(214,214)与位置(160,160)

每次只要运行一下

python himawari8.py

源码

from PIL import Image, ImageOps, ImageDraw
from io import BytesIO
from urllib.request import Request, urlopen
from datetime import datetime, timedelta
import json

SCALE = 2
WIDTH = 1368
HEIGHT = 768


def get_info():
    url = "http://himawari8-dl.nict.go.jp/himawari8/img/D531106/latest.json"
    request = Request(url …

直接给出源码

设置壁纸

feh --bg-scale /tmp/earth.png

设置桌面大小,及图片放大倍数

SCALE = 4
WIDTH = 1368
HEIGHT = 768

from PIL import Image
from io import BytesIO
from urllib.request import Request, urlopen
from datetime import datetime
import json

SCALE = 4
WIDTH = 1368
HEIGHT = 768


def get_info():
    url = "http://himawari8-dl.nict.go.jp/himawari8/img/D531106/latest.json"
    request = Request(url)
    response = urlopen(request, timeout=10)
    return json …

sqlalchemy object序列化为json

灵感来源于 Django REST framework

多个实例

posts = Post.query.all()
serializer = Seralizer(posts,many=True)
data = serializer.data

单个实例

post = Post.query.first()
serializer = Seralizer(post,many=False)
data = serializer.data

排除字段

serializer = Seralizer(post,exclude=['title'])

仅包括字段

serializer = Seralizer(post …

GitHub地址: https://github.com/honmaple/maple-file

maple-file

使用flask实现的一个简单的图片上传服务

设计初衷

对于图片的存储,有很多选择,一般采用云服务如(七牛,又拍等),但是国内的服务像七牛 自定义域名竟然需要域名备案(Excuse me,当初就是因为备案麻烦才选择国外的),而且浪费了我十块钱,

而我又想像七牛一样可以直接在本地就可以上传图片,找来找去，没有找到一个比较合适的，所以花两天时间自己写了一个

使用

由于初衷是本地脚本就可以发布,所以没有前端界面,等以后有时间了再加上

API

• /api/login
  • POST 登录 …

TIME

GMT时间格式

GMT_FORMAT = '%a, %d %b %Y %H:%M:%S GMT'

获取某个时间的时间戳

import datetime
import time

string = "2018-03-31"
date_time = datetime.datetime.strptime(string, '%Y-%m-%d')
time_time = time.mktime …

最近archlinux升级至了 python3.6 ，直接导致了我的开发环境失效,唉

archlinux 果然不适合开发

虽然后面都已经解决了

事故（一）

virtualenv不可用,原先使用python3.5创建的环境直接不能用了,只好删了原有了东西,重新使用python3.6创建

pip freeze > requestments.txt
pip install -r requestments.txt

事故（二）

py3status 报错

   Traceback (most recent call last):
  File "/usr/bin/py3status", line 11, in <module>
    load_entry_point('py3status==3.3', 'console_scripts', 'py3status')()
  File "/usr/lib/python3.6/site-packages/pkg_resources/__init__.py …

_main_

转自 这里

很多新手刚开始学习python的时候经常会看到python 中 __name__ = '_main_' 这样的代码，可能很多新手一开始学习的时候都比较疑惑，python 中_name_ = '_main_' 的作用，到底干嘛的？

有句话经典的概括了这段代码的意义：

“Make a script both importable and executable”

意思就是说让你写的脚本模块既可以导入到别的模块中用，另外该模块自己也可执行。 这句话，可能一开始听的还不是很懂。下面举例说明： 先写一个模块：

#module.py
def main():
    print("we are in %s" % __name__)

if __name__ == '__main__':
    main()

这个函数定义了一个main函数，我们执行一下该py文件发现结果是打印出

we are in __main__

说明我们的if语句中的内容被执行了，调用了 main()： 但是如果我们从另我一个模块导入该模块，并调用一次main()函数会是怎样的结果呢？

#anothermodle.py
from …

为什么需要用token验证

原因呢是因为写博客时已经在本地写好了，但是要发表到网站上还需要这么几步:

• 打开浏览器
• 打开我的网站
• 进入登陆页
• 登陆
• 进入后台页
• 进入文章发表页
• 复制粘贴
• 发表

所以使用token验证成为必然

如何使用token?

生成token

使用itsdangerous对token进行加密

class User(model):
    ......

    @property
    def token(self):
        config = current_app.config
        secret_key = config.setdefault('SECRET_KEY')
        salt = config.setdefault('SECURITY_PASSWORD_SALT')
        serializer = URLSafeTimedSerializer(secret_key)
        # column = self.(需要加密的字段)
        token = serializer.dumps(column, salt=salt)
        return token

请保管好SECRET_KEY 和 …