代码故事Python

重构微信群名历史记录 BOT

文章目录
  1. 1. 微信个人号接口
    1. 1.1. Wechaty
    2. 1.2. Itchat
  2. 2. 模块：platform
  3. 3. 模块：itchat
    1. 3.1. 不能持续在线
  4. 4. 有关 Flask
    1. 4.1. 初始化
    2. 4.2. 自定义返回头信息
    3. 4.3. 访问鉴权
    4. 4.4. 登录
    5. 4.5. 登出
  5. 5. 有关 PyMongo
    1. 5.1. 选出 key 最大的那个 item
  6. 6. 有关 SubProcess
    1. 6.1. 启动子目录下的一个脚本
  7. 7. shutil 模块下与文件拷贝有关的函数
    1. 7.1. shutil.copy2()
    2. 7.2. shutil.copyfile()
      1. 7.2.1. 注2.1：字符串内 {} 相关
      2. 7.2.2. 注2.2：Windows 平台下用到的 _copyfileobj_readinto()
        1. 7.2.2.1. MemoryView 对象
      3. 7.2.3. 注2.3：也许是更为通用但速度较慢的 copyfileobj()
      4. 7.2.4. 注2.4：_stat()

一时兴起，把 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) 来获取最新信息。

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