1 #!/usr/bin/env python

     2 """

     3 Dump public toots

     4 """

     5 

     6 # import sqlite3

     7 

     8 # con = sqlite3.connect("registered_app.db")

     9 

    10 # cur = con.cursor()

    11 # cur.execute("CREATE TABLE registered_app(title, year, score)")

    12 

    13 

    14 import argparse

    15 import collections

    16 from sqlalchemy import Column, Integer, String, create_engine, UniqueConstraint, select

    17 from sqlalchemy.orm import declarative_base, Session

    18 from mastodon import Mastodon

    19 import socket

    20 import requests.packages.urllib3.util.connection as urllib3_cn

    21 import requests

    22 

    23 

    24 Base = declarative_base()

    25 

    26 # Disable ipv6

    27 

    28 

    29 def allowed_gai_family():

    30     """

    31      https://github.com/shazow/urllib3/blob/master/urllib3/util/connection.py

    32     """

    33     return socket.AF_INET

    34 

    35 urllib3_cn.allowed_gai_family = allowed_gai_family

    36 

    37 class RegisteredApp(Base):

    38     """RegisteredApp"""

    39 

    40     __tablename__ = "registered_app"

    41     __table_args__ = (UniqueConstraint("url", "name", name="_url_name_uc"),)

    42 

    43     id = Column(Integer, primary_key=True)

    44     url = Column(String)

    45     name = Column(String)

    46     key = Column(String, unique=True, index=True)

    47     secret = Column(String)

    48 

    49     def __repr__(self):

    50         return f"RegisteredApp(id={self.id!r}, url={self.url!r}, name={self.name!r})"

    51 

    52 

    53 AppKeySecret = collections.namedtuple(

    54     "AppKeySecret", ("key", "secret", "url"), defaults=(None, None, None)

    55 )

    56 

    57 def get_application_secret(name, url, db_con):

    58     """Get the application id and secret

    59 

    60     Args:

    61         name (String): The name of the application

    62         url (String): The mastodon base url (without http(s))

    63         db_con (Connection): The database connection

    64 

    65     Returns:

    66         AppKeySecret: The application secret

    67     """

    68     # get existing secret keys

    69     stmt = select(RegisteredApp).where(

    70         RegisteredApp.url == url, RegisteredApp.name == name

    71     )

    72     result = db_con.execute(stmt).fetchone()

    73 

    74     if result:

    75         return AppKeySecret(key=result.key, secret=result.secret, url=url)

    76 

    77     (client_id, client_secret) = Mastodon.create_app(

    78         name,

    79         api_base_url = f'https://{url}'

    80     )

    81 

    82     registered_app = RegisteredApp(name=name, url=url, key=client_id, secret=client_secret)

    83     with Session(db_con) as session:

    84         session.add(registered_app)

    85         session.commit()

    86 

    87     return AppKeySecret(key=client_id, secret=client_secret, url=url)

    88 

    89 def get_public_statuses(appKeySecret, tag, limit=40):

    90     mastodon = Mastodon(client_id=appKeySecret.key, client_secret=appKeySecret.secret, api_base_url=f'https://{appKeySecret.url}')

    91     results = mastodon.timeline_hashtag(hashtag=tag, limit=limit)

    92     while results:

    93         for status in results:

    94             print("-------------------------")

    95             print(repr(status))

    96         results = mastodon.fetch_next(results)

    97 

    98 

    99 def parse_arg():

   100     """

   101     parse args

   102     """

   103     parser = argparse.ArgumentParser(

   104         prog="dump_public_toots",

   105         description="Dump the latest public toots",

   106         epilog="dump only public toots",

   107     )

   108     parser.add_argument(

   109         "-d",

   110         "--database",

   111         dest="database",

   112         default="registered_apps.sqlite",

   113         help="The path to the registered apps db",

   114     )

   115     parser.add_argument(

   116         "-u", "--url", dest="url", help="The mastodon base url", required=True

   117     )

   118     parser.add_argument(

   119         "-n", "--name", dest="name", help="The application name", required=True

   120     )

   121     parser.add_argument(

   122         "-t", "--tag", dest="tag", help="The tag to search", required=True

   123     )

   124     parser.add_argument(

   125         "-l", "--limit", dest="limit", help="page size", required=False, default=40

   126     )

   127 

   128 

   129     return parser.parse_args()

   130 

   131 

   132 if __name__ == "__main__":

   133 

   134     options = parse_arg()

   135 

   136     # req = requests.get(f"https://{options.url}/api/v1/timelines/tag/{options.tag}/")

   137     # print(req.text)

   138 

   139 

   140     app_db_eng = create_engine(f"sqlite+pysqlite:///{options.database}", future=True)

   141 

   142     with app_db_eng.connect() as app_db_con:

   143         Base.metadata.create_all(app_db_con)

   144 

   145         keySecret = get_application_secret(options.name, options.url, app_db_con)

   146 

   147         get_public_statuses(keySecret, options.tag, options.limit)

   148 

   149         app_db_con.commit()

   150