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
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
93
94
95
96
97
98
99
100
101
102
103
104
105
|
import sys
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Callable, Dict, List
try:
import progressbar2 as progressbar
except ImportError:
import progressbar
class Visibility(Enum):
public = 1
unlisted = 2
followers = 3
direct = 4
@classmethod
def from_db(cls, raw: str) -> "Visibility":
match raw:
case "public": return cls.public
case "home": return cls.unlisted
case "followers": return cls.followers
case "specified": return cls.direct
case _: raise ValueError(f"unknown visibility `{raw}`")
def code(self) -> str:
match self:
case self.public: return "p"
case self.unlisted: return "u"
case self.followers: return "f"
case self.direct: return "d"
@dataclass
class FilterableNote:
id: str
mine: bool
replies: List["FilterableNote"]
quotes: List["FilterableNote"]
when: datetime
reactions: int
renotes: int
visibility: Visibility
cw: str
def thread(self) -> List["FilterableNote"]:
acc = []
for reply in self.replies:
acc += reply.thread()
for quote in self.quotes:
acc += quote.thread()
acc.append(self)
return acc
def thread_self(self) -> List["FilterableNote"]:
acc = []
for reply in self.replies:
acc += reply.thread_self()
for quote in self.quotes:
acc += quote.thread_self()
if self.mine:
acc.append(self)
return acc
def to_dict(self):
return {
"id": self.id,
"mine": self.mine,
"replies": [note.to_dict() for note in self.replies],
"quotes": [note.to_dict() for note in self.quotes],
"when": self.when.isoformat(),
"reactions": self.reactions,
"renotes": self.renotes,
"cw": self.cw,
}
class FilterAction(Enum):
Ignore = 'ignore'
Delete = 'delete'
Obliterate = 'obliterate'
def eval_config() -> dict:
print("configuring")
config = {}
exec(Path(sys.argv[1]).read_text(), config)
return config
def parse_graph() -> Dict[str, dict]:
print("parsing graph")
graph = {}
for line in Path("graph.db").read_text().splitlines():
id, replies, quotes, flags = line.split("\t")
graph[id] = {
"id": id,
"replies": replies.split(",") if len(replies) > 0 else [],
"quotes": quotes.split(",") if len(quotes) > 0 else [],
"flags": flags.split(",") if len(flags) > 0 else [],
}
return graph
|
