diff --git a/get_token.py b/get_token.py index aefd70d..b6ecc8b 100644 --- a/get_token.py +++ b/get_token.py @@ -12,8 +12,8 @@ args = parser.parse_args() SCOPES = [ - 'read:accounts', 'read:blocks', 'read:mutes', - 'write:accounts', 'write:blocks', 'write:mutes' + 'read:accounts', 'read:blocks', 'read:mutes', 'read:search', + 'write:accounts', 'write:blocks', 'write:mutes', ] # create app @@ -64,6 +64,7 @@ r = httpx.get(f'https://{args.instance}/api/v1/accounts/verify_credentials', hea 'Authorization': f'Bearer {token["access_token"]}' }) r.raise_for_status() -print('GET /api/v1/accounts/verify_credentials', json.dumps(r.json(), indent=2), file=stderr) +trunc_json = '\n'.join(json.dumps(r.json(), indent=2).split('\n')[:11]) + '\n...' +print('GET /api/v1/accounts/verify_credentials\n' + trunc_json, file=stderr) json.dump(token, stdout) diff --git a/mute_sync.py b/mute_sync.py index 5e43869..7d4452b 100644 --- a/mute_sync.py +++ b/mute_sync.py @@ -13,6 +13,7 @@ class BearerAuth(httpx.Auth): request.headers['Authorization'] = f'Bearer {self.token}' yield request + def paginated_get(client: httpx.Client, url: str, *args, **kwargs) -> list[dict]: full_response = [] @@ -39,11 +40,43 @@ def paginated_get(client: httpx.Client, url: str, *args, **kwargs) -> list[dict] return full_response + +class Instance: + def __init__(self, domain: str) -> None: + self.domain = domain + self.account_cache = [] + + def lookup_acct(self, url: str, client: httpx.Client) -> dict | None: + cached_match = next(filter(lambda acct: acct['url'] == url, self.account_cache), None) + if cached_match is not None: + return cached_match + + r = client.get('/api/v2/search', params={ + 'q': url, + 'type': 'accounts', + 'resolve': 'true', + 'limit': '5' # should be enough...? + }) + r.raise_for_status() + matches = [acct for acct in r.json()['accounts'] if acct['url'] == url] + if len(matches) == 0: + return None + + self.account_cache.append(matches[0]) + return matches[0] + + def __repr__(self) -> str: + return f'' + + def __str__(self) -> str: + return self.domain + + class UserAccount: - def __init__(self, token: dict) -> None: - self.instance = token['instance'] + def __init__(self, token: dict, instance: Instance) -> None: + self.instance = instance self.client = httpx.Client( - base_url=f'https://{self.instance}', + base_url=f'https://{self.instance.domain}', auth=BearerAuth(token['access_token']) ) @@ -55,7 +88,12 @@ class UserAccount: @functools.cached_property def mutes(self) -> list[dict]: - return paginated_get(self.client, '/api/v1/mutes') + mutes = paginated_get(self.client, '/api/v1/mutes') + self.instance.account_cache.extend(mutes) + return mutes + + def lookup_acct(self, handle: str) -> dict | None: + return self.instance.lookup_acct(handle, self.client) def __repr__(self) -> str: return f'' @@ -64,6 +102,26 @@ class UserAccount: return f'@{self.info["acct"]} ({self.instance})' +# mehhhhhh +class UserInstanceManager: + def __init__(self): + self.instances = [] + self.accounts = [] + + def __iter__(self) -> list: + return self.accounts.__iter__() + + def create_user(self, token: dict) -> 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) + self.accounts.append(acct) + return acct + + if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument( @@ -74,9 +132,13 @@ if __name__ == '__main__': ) args = parser.parse_args() - accounts = [UserAccount(json.load(token_f)) for token_f in args.token_file] + accounts = UserInstanceManager() + for token_f in args.token_file: + accounts.create_user(json.load(token_f)) + print('accounts:', ', '.join((str(acct) for acct in accounts))) for acct in accounts: print(str(acct)) print([mute_acct["acct"] for mute_acct in acct.mutes]) + print(acct.lookup_acct('https://merping.synth.download/@chfour'))