RFNPCs19

1from qiniu import Auth, put_file, etag, urlsafe_base64_encode, BucketManager
2from typing import List, Dict
3import os
4
5from qiniu import build_batch_delete
6
7
8class Sync:
9    """
10    同步目录至七牛云
11    """
12
13    def __init__(
14        self,
15        access_key: str,
16        secret_key: str,
17        bucket_name: str,
18        sync_dir: str,
19        exclude: List,
20        cover: bool,
21        remove_redundant: bool,
22    ):
23        self.bucket_name = bucket_name
24        self.q = Auth(access_key, secret_key)
25        self.bucket = BucketManager(self.q)
26        self.sync_dir = sync_dir
27        self.exclude = exclude
28        self.cover = cover
29        self.remove_redundant = remove_redundant
30
31        self.sync()
32
33    def sync(self):
34        """
35        同步操作
36        :return:
37        """
38        remote_files = self.list_remote()
39        local_files = self.list_local()
40
41        # 首先删除远端仓库中多余的文件
42        remove_remote_files = []
43        for remote_filename in remote_files:
44            if remote_filename not in local_files:
45                remove_remote_files.append(remote_filename)
46        self.bucket.batch(build_batch_delete(self.bucket_name, remove_remote_files))
47
48        # 上传本地文件到远端(仅上传远端不存在的以及修改过的)
49        for local_filename in local_files:
50            if (
51                local_filename not in remote_files
52                or local_files[local_filename]["hash"]
53                != remote_files[local_filename]["hash"]
54            ):
55                print("puting " + local_filename)
56                ret, info = put_file(
57                    self.q.upload_token(self.bucket_name, local_filename, 3600),
58                    local_filename,
59                    local_files[local_filename]["fullpath"],
60                )
61
62    def list_remote(self) -> Dict:
63        """
64        列出远程仓库所有的文件信息
65        :return: List
66        """
67        result = {}
68        for file in self.bucket.list(self.bucket_name)[0]["items"]:
69            result[file["key"]] = file
70
71        return result
72
73    def list_local(self) -> Dict:
74        """
75        列出本地仓库所有的文件信息
76        """
77        files = {}
78
79        def get_files(path):
80            for filename in os.listdir(path):
81                if filename in self.exclude:
82                    continue
83                fullpath = os.path.join(path, filename)
84                if os.path.isfile(fullpath):
85                    key = fullpath.split(self.sync_dir)[1]
86                    files[key] = {"fullpath": fullpath, "hash": etag(fullpath)}
87                else:
88                    get_files(fullpath)
89
90        get_files(self.sync_dir)
91        return files
92
93
94if __name__ == "__main__":
95    Sync(
96        access_key="",  # access_key
97        secret_key="",  # secret_key
98        bucket_name="blog",  # bucket_name
99        sync_dir="~/blog/public/",  # 静态文件目录(后面必须有斜杠/)
100        exclude=[".DS_Store"],
101        cover=True,
102        remove_redundant=True,
103    )