diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index f6cdafe..2ad434e 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -47,6 +47,8 @@ jobs: run: python -m pip install -e .[all] - name: Unit tests + env: + GEOTRIBU_MASTODON_API_ACCESS_TOKEN: ${{ secrets.GEOTRIBU_MASTODON_API_ACCESS_TOKEN }} run: pytest - name: Upload coverage to Codecov diff --git a/docs/usage/examples.md b/docs/usage/examples.md index b51fd0b..84a8694 100644 --- a/docs/usage/examples.md +++ b/docs/usage/examples.md @@ -264,3 +264,21 @@ A partir d'un dossier local : ```sh geotribu images optimize ~/Images/Geotribu/images/ ``` + +---- + +## Réseaux sociaux + +### Exporter les données du compte Mastodon + +Utile pour le partage des comptes suivis et listes (voir [cet article](https://geotribu.fr/articles/2024/2024-02-16_de-twitter-a-mastodon-guide-geo-import-liste-comptes/)) + +```sh +geotribu social mastodon-export +``` + +Préciser le dossier de sortie : + +```sh +geotribu social mastodon-export -w ./export-mastodon +``` diff --git a/geotribu_cli/cli.py b/geotribu_cli/cli.py index 66f5ded..a26b0a1 100644 --- a/geotribu_cli/cli.py +++ b/geotribu_cli/cli.py @@ -31,6 +31,7 @@ parser_comments_read, parser_images_optimizer, parser_latest_content, + parser_mastodon_export, parser_new_article, parser_open_result, parser_search_content, @@ -321,6 +322,26 @@ def main(args: list[str] = None): add_common_arguments(subcmd_search_image) parser_search_image(subcmd_search_image) + # -- NESTED SUBPARSER : COMMENTS --------------------------------------------------- + subcmd_social = subparsers.add_parser( + "social", + aliases=["rezosocio", "social"], + help="Commandes liées aux réseaux sociaux.", + formatter_class=main_parser.formatter_class, + prog="social", + ) + social_subparsers = subcmd_social.add_subparsers(title="Social", dest="cmd_social") + + # Mastodon - Export + subcmd_social_matsodon_export = social_subparsers.add_parser( + "mastodon-export", + help="Exporter les données du compte Mastodon (listes, comptes suivis...).", + formatter_class=main_parser.formatter_class, + prog="mastodon-export", + ) + add_common_arguments(subcmd_social_matsodon_export) + parser_mastodon_export(subcmd_social_matsodon_export) + # -- PARSE ARGS -------------------------------------------------------------------- set_default_subparser( parser_to_update=main_parser, diff --git a/geotribu_cli/comments/comments_broadcast.py b/geotribu_cli/comments/comments_broadcast.py index f90f8c8..651e4e7 100644 --- a/geotribu_cli/comments/comments_broadcast.py +++ b/geotribu_cli/comments/comments_broadcast.py @@ -179,8 +179,3 @@ def run(args: argparse.Namespace): # open a result if args.opt_auto_open_disabled: open_uri(in_filepath=online_post.get("url")) - - -# -- Stand alone execution -if __name__ == "__main__": - pass diff --git a/geotribu_cli/comments/comments_latest.py b/geotribu_cli/comments/comments_latest.py index b0f1d93..c0c7fc6 100644 --- a/geotribu_cli/comments/comments_latest.py +++ b/geotribu_cli/comments/comments_latest.py @@ -118,8 +118,3 @@ def run(args: argparse.Namespace): else: print(":person_shrugging: Aucun commentaire trouvé") sys.exit(0) - - -# -- Stand alone execution -if __name__ == "__main__": - pass diff --git a/geotribu_cli/comments/comments_open.py b/geotribu_cli/comments/comments_open.py index bc2d494..b9b7cd8 100644 --- a/geotribu_cli/comments/comments_open.py +++ b/geotribu_cli/comments/comments_open.py @@ -150,8 +150,3 @@ def run(args: argparse.Namespace): ) else: open_uri(in_filepath=comment_obj.url_to_comment) - - -# -- Stand alone execution -if __name__ == "__main__": - pass diff --git a/geotribu_cli/content/new_article.py b/geotribu_cli/content/new_article.py index 7cce52a..448c942 100644 --- a/geotribu_cli/content/new_article.py +++ b/geotribu_cli/content/new_article.py @@ -143,8 +143,3 @@ def run(args: argparse.Namespace): # open a result if args.opt_auto_open_disabled: open_uri(in_filepath=out_filepath) - - -# -- Stand alone execution -if __name__ == "__main__": - pass diff --git a/geotribu_cli/images/images_optimizer.py b/geotribu_cli/images/images_optimizer.py index 0688b83..f13e5b6 100644 --- a/geotribu_cli/images/images_optimizer.py +++ b/geotribu_cli/images/images_optimizer.py @@ -205,8 +205,3 @@ def run(args: argparse.Namespace): "images/optim" ) ) - - -# -- Stand alone execution -if __name__ == "__main__": - pass diff --git a/geotribu_cli/rss/rss_reader.py b/geotribu_cli/rss/rss_reader.py index 76082d9..a354712 100644 --- a/geotribu_cli/rss/rss_reader.py +++ b/geotribu_cli/rss/rss_reader.py @@ -287,8 +287,3 @@ def run(args: argparse.Namespace): content_uri=feed_items[int(result_to_open)].url, application=getenv("GEOTRIBU_OPEN_WITH", "shell"), ) - - -# -- Stand alone execution -if __name__ == "__main__": - pass diff --git a/geotribu_cli/search/search_content.py b/geotribu_cli/search/search_content.py index 70e8627..9ce47d0 100644 --- a/geotribu_cli/search/search_content.py +++ b/geotribu_cli/search/search_content.py @@ -392,9 +392,9 @@ def run(args: argparse.Namespace): # crée un résultat de sortie out_result = { - "type": "Article" - if result.get("ref").startswith("articles/") - else "GeoRDP", + "type": ( + "Article" if result.get("ref").startswith("articles/") else "GeoRDP" + ), "date": rezult_date, "score": f"{result.get('score'):.3}", "url": f"{defaults_settings.site_base_url}{result.get('ref')}", @@ -460,8 +460,3 @@ def run(args: argparse.Namespace): content_uri=final_results[int(result_to_open)].get("url"), application=getenv("GEOTRIBU_OPEN_WITH", "shell"), ) - - -# -- Stand alone execution -if __name__ == "__main__": - pass diff --git a/geotribu_cli/search/search_image.py b/geotribu_cli/search/search_image.py index 9be6e83..f2f7a7a 100644 --- a/geotribu_cli/search/search_image.py +++ b/geotribu_cli/search/search_image.py @@ -271,8 +271,3 @@ def run(args: argparse.Namespace): content_uri=final_results[int(result_to_open)].get("url"), application=getenv("GEOTRIBU_OPEN_WITH", "shell"), ) - - -# -- Stand alone execution -if __name__ == "__main__": - pass diff --git a/geotribu_cli/social/cmd_mastodon_export.py b/geotribu_cli/social/cmd_mastodon_export.py new file mode 100644 index 0000000..d93f36e --- /dev/null +++ b/geotribu_cli/social/cmd_mastodon_export.py @@ -0,0 +1,88 @@ +#! python3 # noqa: E265 + +# ############################################################################ +# ########## IMPORTS ############# +# ################################ + +# standard library +import argparse +import logging +from os import getenv +from pathlib import Path + +# package +from geotribu_cli.constants import GeotribuDefaults +from geotribu_cli.social.mastodon_client import ExtendedMastodonClient + +# 3rd party + + +# ############################################################################ +# ########## GLOBALS ############# +# ################################ + +logger = logging.getLogger(__name__) +defaults_settings = GeotribuDefaults() + + +# ############################################################################ +# ########## CLI ################# +# ################################ + + +def parser_mastodon_export( + subparser: argparse.ArgumentParser, +) -> argparse.ArgumentParser: + """Set the argument parser for subcommand. + + Args: + subparser (argparse.ArgumentParser): parser to set up + + Returns: + argparse.ArgumentParser: parser ready to use + """ + + subparser.add_argument( + "-w", + "--where", + help="Dossier dans lequel exporter les fichiers.", + default=getenv( + "GEOTRIBU_MASTODON_EXPORT_DEST_FOLDER", + defaults_settings.geotribu_working_folder.joinpath("mastodon/"), + ), + type=Path, + dest="dest_export_folder", + ) + + subparser.set_defaults(func=run) + + return subparser + + +# ############################################################################ +# ########## MAIN ################ +# ################################ + + +def run(args: argparse.Namespace): + """Run the sub command logic. + + Open result of a previous command. + + Args: + args (argparse.Namespace): arguments passed to the subcommand + """ + logger.debug(f"Running {args.command} with {args}") + + mastodon_client = ExtendedMastodonClient() + mastodon_client.export_data( + dest_path_following_accounts=Path(args.dest_export_folder).joinpath( + "mastodon_comptes_suivis_geotribu.csv" + ), + dest_path_lists=Path(args.dest_export_folder).joinpath( + "mastodon_listes_geotribu.csv" + ), + dest_path_lists_only_accounts=Path(args.dest_export_folder).joinpath( + "mastodon_comptes_des_listes_geotribu.csv" + ), + ) diff --git a/geotribu_cli/social/mastodon_client.py b/geotribu_cli/social/mastodon_client.py index d8e040c..7fe60b4 100644 --- a/geotribu_cli/social/mastodon_client.py +++ b/geotribu_cli/social/mastodon_client.py @@ -4,14 +4,20 @@ # ########## IMPORTS ############# # ################################ + # standard library +import csv import json import logging from os import getenv +from pathlib import Path from textwrap import shorten +from typing import Optional from urllib import request +from urllib.parse import urlparse # 3rd party +from mastodon import Mastodon, MastodonAPIError, MastodonError from requests import Session from rich import print @@ -37,6 +43,337 @@ \n#Geotribot #commentaire comment-{id}""" +# CSV output +default_dest_path_following_accounts = ( + defaults_settings.geotribu_working_folder.joinpath( + "mastodon/export/mastodon_comptes_suivis_geotribu.csv" + ) +) +default_dest_path_lists = defaults_settings.geotribu_working_folder.joinpath( + "mastodon/export/mastodon_listes_geotribu.csv" +) +default_dest_path_lists_only_accounts = ( + defaults_settings.geotribu_working_folder.joinpath( + "mastodon/export/mastodon_comptes_des_listes_geotribu" + ) +) + +# ############################################################################ +# ########## CLASSES ############# +# ################################ + + +class ExtendedMastodonClient(Mastodon): + """Extended Mastodon client with custom methods and attributes. + + Raises: + MastodonError: if access token is not set + """ + + csv_accounts_columns_names = [ + "Account address", + "Show boosts", + "Notify on new posts", + "Languages", + ] + + def __init__( + self, + client_id: Optional[str] = None, + client_secret: Optional[str] = None, + access_token: Optional[str] = None, + api_base_url: str = defaults_settings.mastodon_base_url, + debug_requests: Optional[bool] = None, + ratelimit_method: str = "wait", + ratelimit_pacefactor: float = 1.1, + request_timeout: int = 60, + mastodon_version: Optional[str] = None, + version_check_mode: str = "created", + session: Optional[Session] = None, + feature_set: str = "mainline", + user_agent: str = f"{__title_clean__}/{__version__}", + lang: Optional[str] = "fra", + ): + # handle some attributes + if access_token is None: + access_token = getenv("GEOTRIBU_MASTODON_API_ACCESS_TOKEN") + if access_token is None: + logger.critical( + "Le jeton d'accès à l'API Mastodon n'a pas été trouvé en variable " + "d'environnement GEOTRIBU_MASTODON_API_ACCESS_TOKEN. " + "Le récupérer depuis : https://mapstodon.space/settings/applications/7909" + ) + raise MastodonError( + f"Le jeton d'accès à l'API Mastodon (instance : {api_base_url}) est requis." + ) + + if debug_requests is None: + debug_requests = getenv("GEOTRIBU_LOGS_LEVEL", "") == "DEBUG" + + # instanciate subclass + super().__init__( + client_id=client_id, + client_secret=client_secret, + access_token=access_token, + api_base_url=api_base_url, + debug_requests=debug_requests, + ratelimit_method=ratelimit_method, + ratelimit_pacefactor=ratelimit_pacefactor, + request_timeout=request_timeout, + mastodon_version=mastodon_version, + version_check_mode=version_check_mode, + session=session, + feature_set=feature_set, + user_agent=user_agent, + lang=lang, + ) + + @classmethod + def full_account_with_instance( + cls, account: dict, default_instance: str = "mapstodon.space" + ): + """Make sure the account contains instance domain. + + Args: + default_instance: default instance domain. Defaults to mapstodon.space. + account: account dictionary + + Returns: + account with default instance if not present + + Example: + + .. code-block:: python + + >>> print(ExtendedMastodonClient.full_account_with_instance(account={"acct": "datagouvfr@social.numerique.gouv.fr"})) + datagouvfr@social.numerique.gouv.fr + >>> print(ExtendedMastodonClient.full_account_with_instance(account={"acct": "leaflet"})) + leaflet@mapstodon.space + >>> print(ExtendedMastodonClient.full_account_with_instance(account={"acct": "opengisch"}, default_instance="fosstodon.org)) + opengisch@fosstodon.org + + """ + member_account_full: str = account.get("acct") + if "@" not in member_account_full: + member_account_full = f"{member_account_full}@{default_instance}" + return member_account_full + + @classmethod + def url_to_instance_domain(cls, url: str) -> str: + """Extract instance domain from URL. + + Args: + url: input URL + + Returns: + instance domain + + Example: + + .. code-block:: python + + >>> print(ExtendedMastodonClient.url_to_instance_domain(url="https://mapstodon.space/@geotribu")) + mapstodon.space + + """ + parsed_url = urlparse(url) + return parsed_url.netloc + + def export_data( + self, + dest_path_following_accounts: Optional[ + Path + ] = default_dest_path_following_accounts, + dest_path_lists: Optional[Path] = default_dest_path_lists, + dest_path_lists_only_accounts: Optional[ + Path + ] = default_dest_path_lists_only_accounts, + ) -> tuple[Optional[Path], Optional[Path], Optional[Path]]: + """Export account data. + + Args: + dest_path_following_accounts: path to the CSV file for following accounts + export. Defaults to default_dest_path_following_accounts. + dest_path_lists: path to the CSV file for lists export. Defaults to + default_dest_path_lists. + dest_path_lists_only_accounts: path to the CSV file for only accounts from + lists export. Defaults to default_dest_path_lists_only_accounts. + + Raises: + MastodonAPIError: when it's impossible to perform API request for profile + information. + + Returns: + tuple of paths of exported files + """ + # check si au moins un export est défini + if not any( + [ + dest_path_following_accounts, + dest_path_lists, + dest_path_lists_only_accounts, + ] + ): + logger.debug("Aucun format d'export spécifié. Abandon.") + return (None, None, None) + + # -- Récupération des éléments à exporter auprès de l'API -- + try: + mastodon_profile = self.me() + default_instance_domain = self.url_to_instance_domain( + url=mastodon_profile.get("url") + ) + except Exception as err: + logger.critical( + "La récupération du profil auprès de l'API a échoué. L'export est " + f"impossible. Trace: {err}" + ) + raise MastodonAPIError( + "Impossible de récupérer les informations du profil." + ) + + # récupération des listes + if dest_path_lists is not None or dest_path_lists_only_accounts is not None: + try: + dico_listes = { + liste.get("title"): self.list_accounts(id=liste.get("id")) + for liste in self.lists() + } + except Exception as err: + logger.critical( + "La récupération des listes a échoué. L'export est " + f"impossible. Trace: {err}" + ) + dest_path_lists = dest_path_lists_only_accounts = None + + # récupération des comptes suivis + if dest_path_following_accounts is not None: + try: + masto_following_accounts = self.account_following(id=self.me()) + except Exception as err: + logger.critical( + "La récupération des comptes suivis a échoué. L'export est " + f"impossible. Trace: {err}" + ) + dest_path_following_accounts = None + + # -- Export -- + if dest_path_lists: + dest_path_lists = self.export_lists( + mastodon_lists=dico_listes, + dest_csv_path=dest_path_lists, + default_instance=default_instance_domain, + ) + + if dest_path_lists_only_accounts: + dest_path_lists_only_accounts = self.export_accounts( + mastodon_accounts=[ + account_from_list + for liste in dico_listes.values() + for account_from_list in liste + ], + dest_csv_path=dest_path_lists_only_accounts, + default_instance=default_instance_domain, + ) + + if dest_path_following_accounts: + dest_path_following_accounts = self.export_accounts( + mastodon_accounts=masto_following_accounts, + dest_csv_path=dest_path_following_accounts, + default_instance=default_instance_domain, + ) + + return ( + dest_path_following_accounts, + dest_path_lists, + dest_path_lists_only_accounts, + ) + + def export_accounts( + self, + mastodon_accounts: list[dict], + dest_csv_path: Path = default_dest_path_following_accounts, + default_instance: str = "mapstodon.space", + ) -> Path: + """Export Mastodon following accounts into CSV file as web UI. + + Args: + mastodon_accounts: list of accounts + dest_csv_path: path to the CSV file to write to. Defaults to + default_dest_path_following_accounts. + default_instance: default instance domain when account is on the same. + Defaults to mapstodon.space. + + Returns: + path to the CSV file + """ + dest_csv_path.parent.mkdir(parents=True, exist_ok=True) + with dest_csv_path.open( + mode="w", newline="", encoding="utf-8" + ) as out_csv_following_accounts: + # générateurs de CSV + csv_writer_following_accounts = csv.writer(out_csv_following_accounts) + # en-tête (colonnes de la première ligne) + csv_writer_following_accounts.writerow(self.csv_accounts_columns_names) + + for following in mastodon_accounts: + member_account_full = self.full_account_with_instance( + account=following, + default_instance=default_instance, + ) + + # et zou, dans les CSV + csv_writer_following_accounts.writerow( + (member_account_full, "true", "false", "") + ) + + logger.info(f"L'export des comptes a réussi: {dest_csv_path.resolve()}.") + return dest_csv_path + + def export_lists( + self, + mastodon_lists: dict[str, list[dict]], + dest_csv_path: Path = default_dest_path_lists, + default_instance: str = "mapstodon.space", + ) -> Path: + """Export lists. + + Args: + mastodon_lists: _description_ + dest_csv_path: path to the CSV file to write to. Defaults to + default_dest_path_following_accounts. + default_instance: default instance domain when account is on the same. + Defaults to mapstodon.space. + + Returns: + path to the CSV file + """ + dest_csv_path.parent.mkdir(parents=True, exist_ok=True) + + with dest_csv_path.open( + mode="w", encoding="utf-8", newline="" + ) as out_csv_lists: + csv_writer_lists = csv.writer(out_csv_lists) + + # on parcourt les listes du compte authentifié + for liste, members in mastodon_lists.items(): + # on parcourt la liste en la triant sur le nom du compte pour faciliter + # d'éventuelles comparaisons à l'oeil nu ou autres + for member in sorted(members, key=lambda x: x["acct"]): + # Aucune info retournée par l'API ne correspond au formaslime du module + # import/export de l'application web... ainsi les comptes d'une même + # instance n'ont pas son adresse. On gère donc cela manuellement + member_account_full: str = self.full_account_with_instance( + account=member, + default_instance=default_instance, + ) + + # et zou, dans les CSV + csv_writer_lists.writerow((liste, member_account_full)) + logger.info(f"L'export des listes a réussi: {dest_csv_path.resolve()}") + return dest_csv_path + + # ############################################################################ # ########## FUNCTIONS ########### # ################################ diff --git a/geotribu_cli/subcommands/__init__.py b/geotribu_cli/subcommands/__init__.py index be04e5e..a9908e9 100644 --- a/geotribu_cli/subcommands/__init__.py +++ b/geotribu_cli/subcommands/__init__.py @@ -9,6 +9,7 @@ from geotribu_cli.rss.rss_reader import parser_latest_content # noqa: F401 from geotribu_cli.search.search_content import parser_search_content # noqa: F401 from geotribu_cli.search.search_image import parser_search_image # noqa: F401 +from geotribu_cli.social.cmd_mastodon_export import parser_mastodon_export # noqa: F401 from .open_result import parser_open_result # noqa: F401 from .upgrade import parser_upgrade # noqa: F401 diff --git a/geotribu_cli/subcommands/open_result.py b/geotribu_cli/subcommands/open_result.py index 89d54a7..558aa10 100644 --- a/geotribu_cli/subcommands/open_result.py +++ b/geotribu_cli/subcommands/open_result.py @@ -146,8 +146,3 @@ def run(args: argparse.Namespace): print(f"Ouverture du résultat précédent n°{args.result_index} : {result_uri}") open_content(content_uri=result_uri, application=args.open_with) - - -# -- Stand alone execution -if __name__ == "__main__": - pass diff --git a/requirements/base.txt b/requirements/base.txt index 4edbbbc..6a0686c 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -1,6 +1,7 @@ imagesize>=1.4,<1.5 lunr[languages]>=0.7,<0.8 markdownify>=0.11,<0.12 +Mastodon.py>=1.8.1,<1.9 orjson>=3.8,<3.10 packaging>=20,<24 rich_argparse>=0.6,<1.5 diff --git a/tests/dev/dev_mastodon_py.py b/tests/dev/dev_mastodon_py.py new file mode 100644 index 0000000..b71d4c5 --- /dev/null +++ b/tests/dev/dev_mastodon_py.py @@ -0,0 +1,124 @@ +#! python3 # noqa: E265 + +# standard lib +import csv +import logging +from os import getenv +from pathlib import Path + +# 3rd party +from mastodon import Mastodon + +# projet +from geotribu_cli.__about__ import __title_clean__, __version__ +from geotribu_cli.constants import GeotribuDefaults + +defaults_settings = GeotribuDefaults() +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +# Output CSV settings +csv_path_accounts = Path("mastodon_comptes_suivis_geotribu.csv") +csv_path_lists_only_accounts = Path("mastodon_comptes_des_listes_geotribu.csv") +csv_accounts_columns_names = [ + "Account address", + "Show boosts", + "Notify on new posts", + "Languages", +] +csv_path_lists = Path("mastodon_listes_geotribu.csv") +# csv_lists_columns_names = ["List name", "Account address"] # on this export, there is no column names + + +masto_client = Mastodon( + access_token=getenv("GEOTRIBU_MASTODON_API_ACCESS_TOKEN"), + api_base_url=f"{defaults_settings.mastodon_base_url}", + user_agent=f"{__title_clean__}-dev/{__version__}", +) + +print(masto_client.me().keys()) +print(masto_client.me().get("url")) + +# for liste in masto_client.lists(): +# print(liste.get("title")) +# print(masto_client.list_accounts(id=liste.get("id"))[0].keys()) +# print(masto_client.list_accounts(id=liste.get("id"))[0].get("uri")) +# print(masto_client.list_accounts(id=liste.get("id"))[0].get("acct")) +# print(masto_client.list_accounts(id=liste.get("id"))[0].get("id")) +# print(masto_client.list_accounts(id=liste.get("id"))[0].get("url")) + +dico_listes = { + liste.get("title"): masto_client.list_accounts(id=liste.get("id")) + for liste in masto_client.lists() +} +print( + [account_from_list for liste in dico_listes.values() for account_from_list in liste] +) + +try: + with csv_path_lists.open( + mode="w", encoding="utf-8", newline="" + ) as out_csv_lists, csv_path_lists_only_accounts.open( + mode="w", encoding="utf-8", newline="" + ) as out_csv_lists_accounts, csv_path_accounts.open( + mode="w", newline="", encoding="utf-8" + ) as out_csv_following_accounts: + # générateurs de CSV + csv_writer_following_accounts = csv.writer(out_csv_following_accounts) + csv_writer_listed_accounts_without_lists = csv.writer(out_csv_lists_accounts) + csv_writer_lists = csv.writer(out_csv_lists) + + # en-tête (colonnes de la première ligne) + csv_writer_following_accounts.writerow(csv_accounts_columns_names) + csv_writer_listed_accounts_without_lists.writerow(csv_accounts_columns_names) + + # -- Export des comptes ajoutés à des listes + + # on parcourt les listes du compte authentifié + for liste in masto_client.lists(): + # Récupérer les membres de chaque liste + members = masto_client.list_accounts(id=liste.get("id")) + + # on parcourt la liste en la triant sur le nom du compte pour faciliter + # d'éventuelles comparaisons à l'oeil nu ou autres + for member in sorted(members, key=lambda x: x["acct"]): + # Aucune info retournée par l'API ne correspond au formaslime du module + # import/export de l'application web... ainsi les comptes d'une même + # instance n'ont pas son adresse. On gère donc cela manuellement + member_account_full = member.get("acct") + if "@" not in member_account_full: + member_account_full = f"{member_account_full}@mapstodon.space" + + # et zou, dans les CSV + csv_writer_listed_accounts_without_lists.writerow( + (member_account_full, "true", "false", "") + ) + csv_writer_lists.writerow((liste.get("title"), member_account_full)) + logger.info( + "L'export des comptes ajoutés à des listes a réussi. " + f"Comptes (sans les listes) : {csv_path_lists_only_accounts.resolve()}. " + f"Comptes avec les listes : {csv_path_lists.resolve()}" + ) + + # -- Export de tous les comptes suivis + for following in masto_client.account_following(id=masto_client.me()): + + member_account_full = following.get("acct") + if "@" not in member_account_full: + member_account_full = f"{member_account_full}@mapstodon.space" + + # et zou, dans les CSV + csv_writer_following_accounts.writerow( + (member_account_full, "true", "false", "") + ) + + logger.info( + f"L'export des comptes suivis a réussi: {csv_path_accounts.resolve()}." + ) + +except IOError as err: + logger.critical( + "Un problème a empêché l'export en CSV des listes et comptes associés. " + f"Trace : {err}" + ) diff --git a/tests/dev/dev_mastodon_requests.py b/tests/dev/dev_mastodon_requests.py new file mode 100644 index 0000000..6f7f9e7 --- /dev/null +++ b/tests/dev/dev_mastodon_requests.py @@ -0,0 +1,37 @@ +import csv +from os import getenv + +import requests + +from geotribu_cli.__about__ import __title_clean__, __version__ +from geotribu_cli.constants import GeotribuDefaults + +defaults_settings = GeotribuDefaults() + + +# prepare requests session object +headers = { + "User-Agent": f"{__title_clean__}-dev/{__version__}", + "Content-Type": "application/json; charset=utf-8", + "Authorization": f"Bearer {getenv('GEOTRIBU_MASTODON_API_ACCESS_TOKEN')}", +} + +req_session = requests.Session() +req_session.headers = headers + + +# get lists +geotribu_lists = req_session.get( + url=f"{defaults_settings.mastodon_base_url}api/v1/lists", +) + +# print(f"listes : {geotribu_lists.json()}") + + +for liste in geotribu_lists.json(): + liste_accounts = req_session.get( + url=f"{defaults_settings.mastodon_base_url}api/v1/lists/{liste.get('id')}/accounts" + ) + print(f"\n'{liste.get('title')}: {liste_accounts.json()}") + +req_session.close() diff --git a/tests/test_mastodon_client.py b/tests/test_mastodon_client.py new file mode 100644 index 0000000..1ca4508 --- /dev/null +++ b/tests/test_mastodon_client.py @@ -0,0 +1,99 @@ +""" + Usage from the repo root folder: + + .. code-block:: bash + # for whole tests + python -m unittest tests.test_mastodon_client + # for specific test + python -m unittest tests.test_mastodon_client.TestCustomMastodonClient.test_export_data_all +""" + +# standard +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory + +# project +from geotribu_cli.__about__ import __title_clean__, __version__ +from geotribu_cli.social.mastodon_client import ExtendedMastodonClient + +# 3rd party + + +# ############################################################################ +# ########## Classes ############# +# ################################ + + +class TestCustomMastodonClient(unittest.TestCase): + """Test package static variables.""" + + def test_full_account_with_instance(self): + """Test full account completion using a default instance.""" + self.assertEqual( + ExtendedMastodonClient.full_account_with_instance( + account={"acct": "datagouvfr@social.numerique.gouv.fr"} + ), + "datagouvfr@social.numerique.gouv.fr", + ) + self.assertEqual( + ExtendedMastodonClient.full_account_with_instance( + account={"acct": "leaflet"} + ), + "leaflet@mapstodon.space", + ) + self.assertEqual( + ExtendedMastodonClient.full_account_with_instance( + account={"acct": "opengisch"}, default_instance="fosstodon.org" + ), + "opengisch@fosstodon.org", + ) + + def test_instance_domain_from_url(self): + """Test instance domain extraction from URL.""" + self.assertEqual( + ExtendedMastodonClient.url_to_instance_domain( + url="https://mapstodon.space/@geotribu" + ), + "mapstodon.space", + ) + + def test_export_data_all(self): + """Test export following accounts to CSV.""" + masto_client = ExtendedMastodonClient( + user_agent=f"{__title_clean__}-TESTS/{__version__}", debug_requests=False + ) + with TemporaryDirectory( + f"{__title_clean__}_{__version__}_tests_mastodon_" + ) as tempo_dir: + # export + masto_client.export_data( + dest_path_following_accounts=Path(tempo_dir).joinpath( + "following_accounts.csv" + ), + dest_path_lists=Path(tempo_dir).joinpath("lists.csv"), + dest_path_lists_only_accounts=Path(tempo_dir).joinpath( + "lists_only_accounts.csv" + ), + ) + # checks + self.assertTrue(Path(tempo_dir).joinpath("following_accounts.csv").exists()) + self.assertTrue(Path(tempo_dir).joinpath("lists.csv").exists()) + self.assertTrue( + Path(tempo_dir).joinpath("lists_only_accounts.csv").exists() + ) + + self.assertGreater( + Path(tempo_dir).joinpath("following_accounts.csv").stat().st_size, 0 + ) + self.assertGreater(Path(tempo_dir).joinpath("lists.csv").stat().st_size, 0) + self.assertGreater( + Path(tempo_dir).joinpath("lists_only_accounts.csv").stat().st_size, 0 + ) + + +# ############################################################################ +# ####### Stand-alone run ######## +# ################################ +if __name__ == "__main__": + unittest.main()