Add configuration handling
This commit is contained in:
commit
aa714efd53
3 changed files with 146 additions and 0 deletions
6
.gitignore
vendored
Normal file
6
.gitignore
vendored
Normal file
|
|
@ -0,0 +1,6 @@
|
|||
/config*.json
|
||||
|
||||
*.swp
|
||||
*.bak
|
||||
*.old
|
||||
~*
|
||||
4
mypy.ini
Normal file
4
mypy.ini
Normal file
|
|
@ -0,0 +1,4 @@
|
|||
[mypy]
|
||||
strict = True
|
||||
#disallow_untyped_decorators = False
|
||||
#warn_return_any = False
|
||||
136
wither.py
Executable file
136
wither.py
Executable file
|
|
@ -0,0 +1,136 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
import argparse
|
||||
import json
|
||||
|
||||
import httpx
|
||||
|
||||
from typing import Any, Self
|
||||
|
||||
VERSION = "0.1"
|
||||
|
||||
# CAUTION: this just exists for type hinting; actual config is in reality a 'Namepsace'
|
||||
class Config(argparse.Namespace):
|
||||
# run-time-only variables
|
||||
token: str
|
||||
error_count: int
|
||||
config_path: str
|
||||
# persiting variables
|
||||
api_uri: str
|
||||
user_id: str
|
||||
min_id: str
|
||||
delete_reblogs: bool
|
||||
delete_private: bool
|
||||
delete_direct: bool
|
||||
preserve_max_age: int
|
||||
preserve_contexts: list[str]
|
||||
preserve_statuses: list[str]
|
||||
cooldown_delete: int
|
||||
cooldown_fetch: int
|
||||
|
||||
@staticmethod
|
||||
def ephemerals() -> set[str]:
|
||||
return set(["token", "error_count", "config_path"])
|
||||
|
||||
# Work part
|
||||
def process_next_page(config: Config, client: httpx.Client) -> tuple[Config, bool]:
|
||||
#TODO
|
||||
return config, True
|
||||
|
||||
def purge(config: Config) -> Config:
|
||||
client = httpx.Client(
|
||||
headers={
|
||||
"Authorization": config.token,
|
||||
"User-Agent": f"wither (v{VERSION}) - cli tool deleting old fedi posts"
|
||||
},
|
||||
base_url=config.api_uri,
|
||||
http2=True
|
||||
)
|
||||
|
||||
while True:
|
||||
config, stop = process_next_page(config, client)
|
||||
if stop:
|
||||
break
|
||||
|
||||
return config
|
||||
|
||||
# Config stuff
|
||||
def prune_ephemeral_vars(config: Config) -> Config:
|
||||
del config.token
|
||||
del config.error_count
|
||||
del config.config_path
|
||||
return config
|
||||
|
||||
def store_config(config: Config) -> None:
|
||||
conf_path = config.config_path
|
||||
config = prune_ephemeral_vars(config)
|
||||
|
||||
with open(conf_path, "w", encoding="utf-8") as f:
|
||||
json.dump(config.__dict__, f, indent=2, ensure_ascii=False)
|
||||
|
||||
def assert_config(config: Config) -> None:
|
||||
def empty(v: Any) -> bool:
|
||||
return v is None or v == ""
|
||||
|
||||
if empty(config.token):
|
||||
raise ValueError("Unset token!")
|
||||
if empty(config.api_uri):
|
||||
raise ValueError("Unset api uri!")
|
||||
if empty(config.user_id):
|
||||
raise ValueError("Unset user id!")
|
||||
|
||||
def apply_stored_config(config: Config, newvals: dict[str, Any]) -> Config:
|
||||
ephemeral = Config.ephemerals()
|
||||
for k in Config.__annotations__.keys():
|
||||
if k not in ephemeral and newvals.get(k) != None:
|
||||
setattr(config, k, newvals[k])
|
||||
return config
|
||||
|
||||
def load_config(config: Config) -> Config:
|
||||
try:
|
||||
with open(config.config_path, "r") as cf:
|
||||
conf_stored = json.load(cf)
|
||||
apply_stored_config(config, conf_stored)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
return config
|
||||
|
||||
def parse_cli() -> Config:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="wither",
|
||||
description="""
|
||||
Deletes old fedi posts with additional cirteria.
|
||||
State is stored in a config file to facilitate easy reruns in the future.
|
||||
However, the access token is NEVER stored and always needs to be supplied explicitly.
|
||||
Options specified in an existing config file take precedence over values specified on the command line,
|
||||
thus it usually doesn’t make sense to specify anything but token while using a file.
|
||||
Instead edit the values in the config file directly.
|
||||
"""
|
||||
)
|
||||
parser.add_argument("-t", "--token", type=str, required=True)
|
||||
parser.add_argument("--api_uri", type=str)
|
||||
parser.add_argument("--user_id", type=str)
|
||||
parser.add_argument("--min_id", type=str, default="0")
|
||||
parser.add_argument("--delete_reblogs", type=bool, default=True)
|
||||
parser.add_argument("--delete_private", type=bool, default=True)
|
||||
parser.add_argument("--delete_direct", type=bool, default=True)
|
||||
parser.add_argument("--preserve_max_age", type=int, default=7776000)
|
||||
parser.add_argument("--preserve_contexts", action="append", default=[])
|
||||
parser.add_argument("--preserve_statuses", action="append", default=[])
|
||||
parser.add_argument("--cooldown_delete", type=int, default=7)
|
||||
parser.add_argument("--cooldown_fetch", type=int, default=20)
|
||||
parser.add_argument("config_path", type=str)
|
||||
config: Config = parser.parse_args() # type: ignore
|
||||
config.error_count = 0
|
||||
|
||||
return load_config(config)
|
||||
|
||||
# Main
|
||||
def run() -> None:
|
||||
config = parse_cli()
|
||||
assert_config(config)
|
||||
config_new = purge(config)
|
||||
store_config(config_new)
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
Loading…
Add table
Add a link
Reference in a new issue