diff --git a/mute_sync.py b/mute_sync.py index d826b4a..c9cd62a 100644 --- a/mute_sync.py +++ b/mute_sync.py @@ -73,11 +73,12 @@ class Instance: class UserAccount: - def __init__(self, token: dict, instance: Instance) -> None: + def __init__(self, token: dict, instance: Instance, *args, **kwargs) -> None: self.instance = instance self.client = httpx.Client( base_url=f'https://{self.instance.domain}', - auth=BearerAuth(token['access_token']) + auth=BearerAuth(token['access_token']), + *args, **kwargs ) @functools.cached_property @@ -92,6 +93,12 @@ class UserAccount: self.instance.account_cache.extend(mutes) return set((mute['url'] for mute in mutes)) + @functools.cached_property + def blocks(self) -> set[str]: + blocks = paginated_get(self.client, '/api/v1/blocks') + self.instance.account_cache.extend(blocks) + return set((block['url'] for block in blocks)) + def lookup_acct(self, handle: str) -> dict | None: return self.instance.lookup_acct(handle, self.client) @@ -111,19 +118,25 @@ class UserInstanceManager: def __iter__(self) -> list: return self.accounts.__iter__() - def create_user(self, token: dict) -> UserAccount: + def create_user(self, token: dict, *args, **kwargs) -> UserAccount: instance = next(filter(lambda inst: inst.domain == token['instance'], self.instances), None) if instance is None: instance = Instance(token['instance']) self.instances.append(instance) - acct = UserAccount(token, instance) + acct = UserAccount(token, instance, *args, **kwargs) self.accounts.append(acct) return acct if __name__ == '__main__': parser = argparse.ArgumentParser() + parser.add_argument( + '-t', '--timeout', + type=int, + help='http request timeout', + default=10 + ) parser.add_argument( 'token_file', nargs='+', @@ -134,7 +147,7 @@ if __name__ == '__main__': accounts = UserInstanceManager() for token_f in args.token_file: - accounts.create_user(json.load(token_f)) + accounts.create_user(json.load(token_f), timeout=args.timeout) print('accounts:', ', '.join((str(acct) for acct in accounts))) @@ -146,26 +159,46 @@ if __name__ == '__main__': print(', '.join(all_mutes)) print() + all_blocks = set() for acct in accounts: - not_here = all_mutes.difference(acct.mutes) + all_blocks.update(acct.blocks) - print(f'mutes to add to {acct}:') - count = 0 - for url in not_here: - print(f'looking up {url}...', end='') + print(f'all blocks ({len(all_blocks)} total):') + print(', '.join(all_blocks)) + print() + + def flag(label: str, condition: bool) -> str: + return label if condition else ' ' * len(label) + + for acct in accounts: + to_mute = all_mutes.difference(acct.mutes) + to_block = all_blocks.difference(acct.blocks) + + print(f'will add to {acct}:') + to_mute_count = 0 + to_block_count = 0 + for url in sorted(to_mute | to_block, key=lambda u: (u in to_block, u in to_mute, u)): + print(f'looking up {url}...', end='', flush=True) found = acct.lookup_acct(url) - print(f'\x1b[2K\r', end='') + print(f'\x1b[2K\r', end='', flush=True) if found is None: - print(f' {url} not found') + print(f' {url} not found') continue - print(f'+ @{found["acct"]} ({url})') - count += 1 + in_mutes = url in to_mute + in_blocks = url in to_block + + if in_mutes: to_mute_count += 1 + if in_blocks: to_block_count += 1 - print(f'{count} total\n') + print(f' {flag("+block", in_blocks)} {flag("+mute", in_mutes)} @{found["acct"]} ({url})') - choice = input('proceed? (type "yes" to continue) ') - if choice.lower() != 'yes': + print(f'{to_mute_count} mutes, {to_block_count} blocks, {to_mute_count + to_block_count} total for {acct}\n') + try: + choice = input('proceed? (type "yes" to continue) ') + if choice.lower() != 'yes': + raise KeyboardInterrupt + except KeyboardInterrupt: print('operation cancelled') exit()