一时兴起，把 JiangGua/chatroom-name-history-weixin 给重构了，下面正好记录一下新知道的一些玩意。

微信个人号接口

本项目最初采用 Itchat，在2021年9月2日因微信接口变动更换为 Itchat-UOS，代码无改动。

Wechaty

是目前用户量最大、维护最活跃的接口 SDK。

Itchat

目前借用 UOS 协议还能苟活，但原作者已经不再维护了，不知道社区的力量能撑多久。

模块：platform

检测当前的运行平台等。本例中用来区别对待 Windows 与 Linux (Other OS) 用户。

1
2
3
4
if platform.platform().lower().find('windows') == -1:
    itchat.auto_login(hotReload=True, enableCmdQR=2)
else:
    itchat.auto_login(hotReload=True)

模块：itchat

这是本项目的核心模块了，提供微信个人号的接口。

不能持续在线

有关 Flask

初始化

1
2
app = Flask(__name__)
app.config['SECRET_KEY'] = os.environ.get("SECRET_KEY", default="DEFAULT-SECRET-KEY-HERE")

这里这个 SECRET_KEY 是用来加密 Cookies 的。

自定义返回头信息

1
2
3
4
5
6
7
8
9
10
@app.after_request
def after_request(response):
    # 实现了一个「伪 Access-Control-Allow-Origin: *」
    try:
        response.headers['Access-Control-Allow-Origin'] = request.headers['Origin']
        response.headers['Access-Control-Allow-Credentials'] = 'true'
    except:
        response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Methods'] = 'PUT,GET,POST,DELETE'
    response.headers['Access-Control-Allow-Headers'] = 'Content-Type,Authorization'

因为 W3C 标准规定在 Access-Control-Allow-Credentials: true 的时候 Access-Control-Allow-Origin 就不能设置成 * ，应该是出于安全的考虑吧。不过我这个项目是自用为主，而且没有什么隐私数据，所以安全性姑且降低一些问题也不大。

访问鉴权

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@app.before_request
def before_action():
    def exclude(path):
        # 登录路由
        if path == '/login':
            return True
        # 路由以 /status/ 开头
        if path.find('/status/') == 0:
            return True
        # 路由包含 .ico
        if path.find('.ico') != -1:
            return True
        
    # 如果路径不在例外列表里面
    if not exception(request.path):
        # 如果 Cookies 里面没有 username 这个键值
        if not 'username' in session:
            return jsonify({'result': 0, 'msg': '未登录'}), 401

登录

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@app.route('/login', methods=['GET','POST'])
def login():
    if request.method == 'POST':
        username = request.json.get('username')
        password = request.json.get('password')
        remember_user = request.json.get('remember')

        if verify_user(username, password) == True:
            session['username'] = username
            # 永久记住登录状态: 其实默认是31天
            if remember_user == True:
                session.permanent = True
            else:
                session.permanent = False
            return jsonify({'result': 1, 'msg': '欢迎回来，{}'.format(username)}), 200
        else:
            return jsonify({'result': 0, 'msg': '登录失败'}), 401
    return jsonify({'result': 0, 'description': 'Please use POST method instead'}), 401

登出

1
2
3
4
5
6
7
8
@app.route('/logout')
def logout():
    session.pop('username', None)
    session.pop('password', None)
    return {
        'result': 1,
        'msg': '注销成功'
    }

有关 PyMongo

选出 key 最大的那个 item

1
cursor = db.find_one({}, sort=[("somekey", pymongo.DESCENDING)])

有关 SubProcess

启动子目录下的一个脚本

1
2
3
4
# 基础形态
subprocess.Popen(['python', './folder/script.py'])
# 命令行参数
subprocess.Popen(['python', './folder/script.py', '--arg', str(arg_value)])

shutil 模块下与文件拷贝有关的函数

纯属好奇宝宝，Ctrl点进去看了几眼。

shutil.copy2()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def copy2(src, dst, *, follow_symlinks=True):
    """复制文件本身及其元数据.
    返回值：文件粘贴到的路径.
    其中，文件元数据是用 copystat() 复制过去的. 更多信息请参阅 copystat 函数.
    粘贴目的地可能是一个文件夹.
    如果 follow_symlinks == False, copy2 将不会跟踪(follow) 符号链接 (symlinks). This
    resembles(vt. 像…, 类似于…) GNU's "cp -P src dst".
    """
    # 「粘贴目的地可能是一个文件夹」的处理
    if os.path.isdir(dst):
        dst = os.path.join(dst, os.path.basename(src))
    copyfile(src, dst, follow_symlinks=follow_symlinks)
    copystat(src, dst, follow_symlinks=follow_symlinks)
    return dst

shutil.copyfile()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def copyfile(src, dst, *, follow_symlinks=True):
    """尝试用最快的方式复制文件.
    如果 follow_symlinks 为假值且 src 为符号链接，则将创建一个新的符号链接而不是拷贝 src 所指向的文件
    (If follow_symlinks is not set and src is a symbolic link, a new
    symlink will be created instead of copying the file it points to.)
    """
    
    # 如果 src 和 dst 指定了同一个文件，则将引发 SameFileError。
    if _samefile(src, dst):
        					# 见注2.1
        raise SameFileError("{!r} and {!r} are the same file".format(src, dst))

    file_size = 0
    for i, fn in enumerate([src, dst]):
        try:
            # 注2.4
            st = _stat(fn)
        except OSError:
            # File most likely does not exist
            pass
        else:
            # XXX What about other special files? (sockets, devices...)
            if stat.S_ISFIFO(st.st_mode):
                fn = fn.path if isinstance(fn, os.DirEntry) else fn
                raise SpecialFileError("`%s` is a named pipe" % fn)
            if _WINDOWS and i == 0:
                file_size = st.st_size

    # 源是个符号链接但设置了 follow_symlinks == False
    if not follow_symlinks and _islink(src):
        # 在 dst 建立一个指向“src 这个符号链接指向的实际路径”的符号链接，保证即便 src 没了 dst 也能正常指过去
        os.symlink(os.readlink(src), dst)
    else:
        with open(src, 'rb') as fsrc, open(dst, 'wb') as fdst:
            # macOS
            if _HAS_FCOPYFILE:
                try:
                    _fastcopy_fcopyfile(fsrc, fdst, posix._COPYFILE_DATA)
                    return dst
                except _GiveupOnFastCopy:
                    pass
            # Linux
            elif _USE_CP_SENDFILE:
                try:
                    _fastcopy_sendfile(fsrc, fdst)
                    return dst
                except _GiveupOnFastCopy:
                    pass
            # Windows, see:
            # https://github.com/python/cpython/pull/7160#discussion_r195405230
            elif _WINDOWS and file_size > 0:
                # 注2.2
                _copyfileobj_readinto(fsrc, fdst, min(file_size, COPY_BUFSIZE))
                return dst

            # 注2.3
            copyfileobj(fsrc, fdst)

    return dst

注2.1：字符串内 {} 相关

……

一些简单的格式字符串示例

1
2
3
4
5
6
"First, thou shalt count to {0}"  # References first positional argument
"Bring me a {}"                   # Implicitly references the first positional argument
"From {} to {}"                   # Same as "From {0} to {1}"
"My quest is {name}"              # References keyword argument 'name'
"Weight in tons {0.weight}"       # 'weight' attribute of first positional arg
"Units destroyed: {players[0]}"   # First element of keyword argument 'players'.

使用 conversion 字段在格式化之前进行类型强制转换。 通常，格式化值的工作由值本身的 __format__() 方法来完成。 但是，在某些情况下最好强制将类型格式化为一个字符串，覆盖其本身的格式化定义。 通过在调用 __format__() 之前将值转换为字符串，可以绕过正常的格式化逻辑。

目前支持的转换旗标有三种: '!s' 会对值调用 str()，'!r' 调用 repr() 而 '!a' 则调用 ascii()。

几个例子：

1
2
3
"Harold's a clever {0!s}"        # Calls str() on the argument first
"Bring out the holy {name!r}"    # Calls repr() on the argument first
"More {!a}"                      # Calls ascii() on the argument first

注2.2：Windows 平台下用到的 _copyfileobj_readinto()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def _copyfileobj_readinto(fsrc, fdst, length=COPY_BUFSIZE):
    """
    基于 readinto()/memoryview() 的 copyfileobj() 的变体.
    readinto()/memoryview() based variant of copyfileobj().
    
    *fsrc* must support readinto() method and both files must be
    open in binary mode.
    """
    # Localize variable access to minimize overhead.
    fsrc_readinto = fsrc.readinto
    fdst_write = fdst.write
    with memoryview(bytearray(length)) as mv:
        while True:
            n = fsrc_readinto(mv)
            if not n:
                break
            elif n < length:
                with mv[:n] as smv:
                    fdst.write(smv)
            else:
                fdst_write(mv)
MemoryView 对象

注2.3：也许是更为通用但速度较慢的 copyfileobj()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def copyfileobj(fsrc, fdst, length=0):
    """copy data from file-like object fsrc to file-like object fdst"""
    # 将方法本地化以减少性能开销 (Q:这咋减少开销的，是避免class追溯方法时候的性能开销吗)
    # Localize variable access to minimize overhead.
    if not length:
        length = COPY_BUFSIZE
    fsrc_read = fsrc.read
    fdst_write = fdst.write
    # 缓冲区机制：读一点写一点
    while True:
        buf = fsrc_read(length)
        if not buf:
            break
        fdst_write(buf)

这个函数稍慢应该是因为运行于CPython抽象层之上，所以多了这个解释器的开销。其它的基本都是直接进行类C的内存操作，因此快些。

  • 将对象的方法本地化以减少开销

注2.4：_stat()

1
2
def _stat(fn):
    return fn.stat() if isinstance(fn, os.DirEntry) else os.stat(fn)

……每次进行 stat() 或 lstat() 系统调用时，os.DirEntry 对象会将结果缓存下来。

os.DirEntry 实例不适合存储在长期存在的数据结构中，如果你知道文件元数据已更改，或者自调用 scandir() 以来已经经过了很长时间，请调用 os.stat(entry.path) 来获取最新信息。

所以这一个小函数的意思就是：有缓存的数据就用缓存的，没有就去读。

许可协议

All Rights Reserved.