跳转至

参考

Configuration

KToolBox Configuration

Attributes:

Name Type Description
api APIConfiguration

Kemono API Configuration

downloader DownloaderConfiguration

File Downloader Configuration

job JobConfiguration

Download jobs Configuration

logger LoggerConfiguration

Logger configuration

ssl_verify bool

Enable SSL certificate verification for Kemono API server and download server

json_dump_indent int

Indent of JSON file dump

use_uvloop bool

Use uvloop for asyncio (Disabled on Windows by default) uvloop will improve concurrent performance, but it is not compatible with Windows. Install uvloop by pip install ktoolbox[uvloop] or it will not work.

Source code in ktoolbox/configuration.py
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
class Configuration(BaseSettings):
    # noinspection SpellCheckingInspection,GrazieInspection
    """
    KToolBox Configuration

    :ivar api: Kemono API Configuration
    :ivar downloader: File Downloader Configuration
    :ivar job: Download jobs Configuration
    :ivar logger: Logger configuration
    :ivar ssl_verify: Enable SSL certificate verification for Kemono API server and download server
    :ivar json_dump_indent: Indent of JSON file dump
    :ivar use_uvloop: Use uvloop for asyncio (Disabled on Windows by default) \
    uvloop will improve concurrent performance, but it is not compatible with Windows. \
    Install uvloop by `pip install ktoolbox[uvloop]` or it will not work.
    """
    api: APIConfiguration = APIConfiguration()
    downloader: DownloaderConfiguration = DownloaderConfiguration()
    job: JobConfiguration = JobConfiguration()
    logger: LoggerConfiguration = LoggerConfiguration()

    ssl_verify: bool = True
    json_dump_indent: int = 4
    use_uvloop: bool = True

    # noinspection SpellCheckingInspection
    model_config = SettingsConfigDict(
        env_prefix='ktoolbox_',
        env_nested_delimiter='__',
        env_file='.env',
        env_file_encoding='utf-8',
        extra='ignore'
    )

api: APIConfiguration = APIConfiguration() class-attribute instance-attribute

downloader: DownloaderConfiguration = DownloaderConfiguration() class-attribute instance-attribute

job: JobConfiguration = JobConfiguration() class-attribute instance-attribute

logger: LoggerConfiguration = LoggerConfiguration() class-attribute instance-attribute

ssl_verify: bool = True class-attribute instance-attribute

json_dump_indent: int = 4 class-attribute instance-attribute

use_uvloop: bool = True class-attribute instance-attribute

APIConfiguration

Kemono API Configuration

Attributes:

Name Type Description
scheme Literal['http', 'https']

Kemono API URL scheme

netloc str

Kemono API URL netloc

statics_netloc str

URL netloc of Kemono server for static files (e.g. images)

files_netloc str

URL netloc of Kemono server for post files

path str

Kemono API URL root path

timeout float

API request timeout

retry_times int

API request retry times (when request failed)

retry_interval float

Seconds of API request retry interval

Source code in ktoolbox/configuration.py
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class APIConfiguration(BaseModel):
    """
    Kemono API Configuration

    :ivar scheme: Kemono API URL scheme
    :ivar netloc: Kemono API URL netloc
    :ivar statics_netloc: URL netloc of Kemono server for static files (e.g. images)
    :ivar files_netloc: URL netloc of Kemono server for post files
    :ivar path: Kemono API URL root path
    :ivar timeout: API request timeout
    :ivar retry_times: API request retry times (when request failed)
    :ivar retry_interval: Seconds of API request retry interval
    """
    scheme: Literal["http", "https"] = "https"
    netloc: str = "kemono.su"
    statics_netloc: str = "img.kemono.su"
    files_netloc: str = "kemono.su"
    path: str = "/api/v1"
    timeout: float = 5.0
    retry_times: int = 3
    retry_interval: float = 2.0

scheme: Literal['http', 'https'] = 'https' class-attribute instance-attribute

netloc: str = 'kemono.su' class-attribute instance-attribute

statics_netloc: str = 'img.kemono.su' class-attribute instance-attribute

files_netloc: str = 'kemono.su' class-attribute instance-attribute

path: str = '/api/v1' class-attribute instance-attribute

timeout: float = 5.0 class-attribute instance-attribute

retry_times: int = 3 class-attribute instance-attribute

retry_interval: float = 2.0 class-attribute instance-attribute

DownloaderConfiguration

File Downloader Configuration

Attributes:

Name Type Description
scheme Literal['http', 'https']

Downloader URL scheme

timeout float

Downloader request timeout

encoding str

Charset for filename parsing and post content text saving

buffer_size int

Number of bytes of file I/O buffer for each downloading file

chunk_size int

Number of bytes of chunk of downloader stream

temp_suffix str

Temp filename suffix of downloading files

retry_times int

Downloader retry times (when download failed)

retry_stop_never bool

Never stop downloader from retrying (when download failed) (retry_times will be ignored when enabled)

retry_interval float

Seconds of downloader retry interval

use_bucket bool

Enable local storage bucket mode

bucket_path Path

Path of local storage bucket

Source code in ktoolbox/configuration.py
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class DownloaderConfiguration(BaseModel):
    """
    File Downloader Configuration

    :ivar scheme: Downloader URL scheme
    :ivar timeout: Downloader request timeout
    :ivar encoding: Charset for filename parsing and post content text saving
    :ivar buffer_size: Number of bytes of file I/O buffer for each downloading file
    :ivar chunk_size: Number of bytes of chunk of downloader stream
    :ivar temp_suffix: Temp filename suffix of downloading files
    :ivar retry_times: Downloader retry times (when download failed)
    :ivar retry_stop_never: Never stop downloader from retrying (when download failed) \
    (``retry_times`` will be ignored when enabled)
    :ivar retry_interval: Seconds of downloader retry interval
    :ivar use_bucket: Enable local storage bucket mode
    :ivar bucket_path: Path of local storage bucket
    """
    scheme: Literal["http", "https"] = "https"
    timeout: float = 30.0
    encoding: str = "utf-8"
    buffer_size: int = 20480
    chunk_size: int = 1024
    temp_suffix: str = "tmp"
    retry_times: int = 10
    retry_stop_never: bool = False
    retry_interval: float = 3.0
    use_bucket: bool = False
    bucket_path: Path = Path("./.ktoolbox/bucket_storage")

    @model_validator(mode="after")
    def check_bucket_path(self) -> "DownloaderConfiguration":
        if self.use_bucket:
            # noinspection PyBroadException
            try:
                bucket_path = Path(self.bucket_path)
                bucket_path.mkdir(parents=True, exist_ok=True)
                with tempfile.TemporaryFile(dir=bucket_path) as temp_file:
                    temp_link_file_path = f"{bucket_path / temp_file.name}.hlink"
                    os.link(temp_file.name, temp_link_file_path)
                    os.remove(temp_link_file_path)
            except Exception:
                self.use_bucket = False
                logger.exception(f"`DownloaderConfiguration.bucket_path` is not available, "
                                 f"`DownloaderConfiguration.use_bucket` has been disabled.")
        return self

scheme: Literal['http', 'https'] = 'https' class-attribute instance-attribute

timeout: float = 30.0 class-attribute instance-attribute

encoding: str = 'utf-8' class-attribute instance-attribute

buffer_size: int = 20480 class-attribute instance-attribute

chunk_size: int = 1024 class-attribute instance-attribute

temp_suffix: str = 'tmp' class-attribute instance-attribute

retry_times: int = 10 class-attribute instance-attribute

retry_stop_never: bool = False class-attribute instance-attribute

retry_interval: float = 3.0 class-attribute instance-attribute

use_bucket: bool = False class-attribute instance-attribute

bucket_path: Path = Path('./.ktoolbox/bucket_storage') class-attribute instance-attribute

check_bucket_path()

Source code in ktoolbox/configuration.py
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@model_validator(mode="after")
def check_bucket_path(self) -> "DownloaderConfiguration":
    if self.use_bucket:
        # noinspection PyBroadException
        try:
            bucket_path = Path(self.bucket_path)
            bucket_path.mkdir(parents=True, exist_ok=True)
            with tempfile.TemporaryFile(dir=bucket_path) as temp_file:
                temp_link_file_path = f"{bucket_path / temp_file.name}.hlink"
                os.link(temp_file.name, temp_link_file_path)
                os.remove(temp_link_file_path)
        except Exception:
            self.use_bucket = False
            logger.exception(f"`DownloaderConfiguration.bucket_path` is not available, "
                             f"`DownloaderConfiguration.use_bucket` has been disabled.")
    return self

PostStructureConfiguration

Post path structure model

  • Default:
    ..
    ├─ content.txt
    ├─ <Post file>
    ├─ <Post data (post.json)>
    └─ attachments
       ├─ 1.png
       └─ 2.png
    

Attributes:

Name Type Description
attachments Path

Sub path of attachment directory

content_filepath Path

Sub path of post content HTML file

Source code in ktoolbox/configuration.py
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class PostStructureConfiguration(BaseModel):
    # noinspection SpellCheckingInspection
    """
    Post path structure model

    - Default:
    ```
    ..
    ├─ content.txt
    ├─ <Post file>
    ├─ <Post data (post.json)>
    └─ attachments
       ├─ 1.png
       └─ 2.png
    ```

    :ivar attachments: Sub path of attachment directory
    :ivar content_filepath: Sub path of post content HTML file
    """
    attachments: Path = Path("attachments")
    content_filepath: Path = Path("index.html")

attachments: Path = Path('attachments') class-attribute instance-attribute

content_filepath: Path = Path('index.html') class-attribute instance-attribute

JobConfiguration

Download jobs Configuration

  • Available properties for post_dirname_format

    Property Type
    id String
    user String
    service String
    title String
    added Date
    published Date
    edited Date

Attributes:

Name Type Description
count int

Number of coroutines for concurrent download

post_id_as_path bool

(Deprecated) Use post ID as post directory name

post_dirname_format str

Customize the post directory name format, you can use some of the properties in Post. e.g. [{published}]{id} > [2024-1-1]123123, {user}_{published}_{title} > 234234_2024-1-1_HelloWorld

post_structure PostStructureConfiguration

Post path structure

mix_posts bool

Save all files from different posts at same path in creator directory. It would not create any post directory, and CreatorIndices would not been recorded.

sequential_filename bool

Rename attachments in numerical order, e.g. 1.png, 2.png, ...

allow_list Set[str]

Download files which match these patterns (Unix shell-style), e.g. ["*.png"]

block_list Set[str]

Not to download files which match these patterns (Unix shell-style), e.g. ["*.psd","*.zip"]

Source code in ktoolbox/configuration.py
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class JobConfiguration(BaseModel):
    """
    Download jobs Configuration

    - Available properties for ``post_dirname_format``

        | Property      | Type   |
        |---------------|--------|
        | ``id``        | String |
        | ``user``      | String |
        | ``service``   | String |
        | ``title``     | String |
        | ``added``     | Date   |
        | ``published`` | Date   |
        | ``edited``    | Date   |

    :ivar count: Number of coroutines for concurrent download
    :ivar post_id_as_path: (**Deprecated**) Use post ID as post directory name
    :ivar post_dirname_format: Customize the post directory name format, you can use some of the \
    [properties](/configuration/reference/#ktoolbox.configuration.JobConfiguration) in ``Post``. \
    e.g. ``[{published}]{id}`` > ``[2024-1-1]123123``, ``{user}_{published}_{title}`` > ``234234_2024-1-1_HelloWorld``
    :ivar post_structure: Post path structure
    :ivar mix_posts: Save all files from different posts at same path in creator directory. \
    It would not create any post directory, and ``CreatorIndices`` would not been recorded.
    :ivar sequential_filename: Rename attachments in numerical order, e.g. ``1.png``, ``2.png``, ...
    :ivar allow_list: Download files which match these patterns (Unix shell-style), e.g. ``["*.png"]``
    :ivar block_list: Not to download files which match these patterns (Unix shell-style), e.g. ``["*.psd","*.zip"]``
    """
    count: int = 4
    post_id_as_path: bool = False
    post_dirname_format: str = "{title}"
    post_structure: PostStructureConfiguration = PostStructureConfiguration()
    mix_posts: bool = False
    sequential_filename: bool = False
    allow_list: Set[str] = set()
    block_list: Set[str] = set()

    # job_list_filepath: Optional[Path] = None
    # """Filepath for job list data saving, ``None`` for disable job list saving"""

    @field_validator("post_id_as_path")
    def post_id_as_path_validator(cls, v):
        if v != cls.model_fields["post_id_as_path"].default:
            warnings.warn(
                "`JobConfiguration.post_id_as_path` is deprecated and is scheduled for removal in further version. "
                "Use `JobConfiguration.post_dirname_format` instead",
                FutureWarning
            )

count: int = 4 class-attribute instance-attribute

post_id_as_path: bool = False class-attribute instance-attribute

post_dirname_format: str = '{title}' class-attribute instance-attribute

post_structure: PostStructureConfiguration = PostStructureConfiguration() class-attribute instance-attribute

mix_posts: bool = False class-attribute instance-attribute

sequential_filename: bool = False class-attribute instance-attribute

allow_list: Set[str] = set() class-attribute instance-attribute

block_list: Set[str] = set() class-attribute instance-attribute

post_id_as_path_validator(v)

Source code in ktoolbox/configuration.py
158
159
160
161
162
163
164
165
@field_validator("post_id_as_path")
def post_id_as_path_validator(cls, v):
    if v != cls.model_fields["post_id_as_path"].default:
        warnings.warn(
            "`JobConfiguration.post_id_as_path` is deprecated and is scheduled for removal in further version. "
            "Use `JobConfiguration.post_dirname_format` instead",
            FutureWarning
        )

LoggerConfiguration

Logger configuration

Attributes:

Name Type Description
path Optional[Path]

Path to save logs, None for disable log file output

level Union[str, int]

Log filter level

rotation Union[str, int, time, timedelta]

Log rotation

Source code in ktoolbox/configuration.py
168
169
170
171
172
173
174
175
176
177
178
class LoggerConfiguration(BaseModel):
    """
    Logger configuration

    :ivar path: Path to save logs, ``None`` for disable log file output
    :ivar level: Log filter level
    :ivar rotation: Log rotation
    """
    path: Optional[Path] = None
    level: Union[str, int] = logging.DEBUG
    rotation: Union[str, int, datetime.time, datetime.timedelta] = "1 week"

path: Optional[Path] = None class-attribute instance-attribute

level: Union[str, int] = logging.DEBUG class-attribute instance-attribute

rotation: Union[str, int, datetime.time, datetime.timedelta] = '1 week' class-attribute instance-attribute