--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/script/utils/dump_public_toots.py Mon Nov 20 18:10:58 2023 +0100
@@ -0,0 +1,150 @@
+#!/usr/bin/env python
+"""
+Dump public toots
+"""
+
+# import sqlite3
+
+# con = sqlite3.connect("registered_app.db")
+
+# cur = con.cursor()
+# cur.execute("CREATE TABLE registered_app(title, year, score)")
+
+
+import argparse
+import collections
+from sqlalchemy import Column, Integer, String, create_engine, UniqueConstraint, select
+from sqlalchemy.orm import declarative_base, Session
+from mastodon import Mastodon
+import socket
+import requests.packages.urllib3.util.connection as urllib3_cn
+import requests
+
+
+Base = declarative_base()
+
+# Disable ipv6
+
+
+def allowed_gai_family():
+ """
+ https://github.com/shazow/urllib3/blob/master/urllib3/util/connection.py
+ """
+ return socket.AF_INET
+
+urllib3_cn.allowed_gai_family = allowed_gai_family
+
+class RegisteredApp(Base):
+ """RegisteredApp"""
+
+ __tablename__ = "registered_app"
+ __table_args__ = (UniqueConstraint("url", "name", name="_url_name_uc"),)
+
+ id = Column(Integer, primary_key=True)
+ url = Column(String)
+ name = Column(String)
+ key = Column(String, unique=True, index=True)
+ secret = Column(String)
+
+ def __repr__(self):
+ return f"RegisteredApp(id={self.id!r}, url={self.url!r}, name={self.name!r})"
+
+
+AppKeySecret = collections.namedtuple(
+ "AppKeySecret", ("key", "secret", "url"), defaults=(None, None, None)
+)
+
+def get_application_secret(name, url, db_con):
+ """Get the application id and secret
+
+ Args:
+ name (String): The name of the application
+ url (String): The mastodon base url (without http(s))
+ db_con (Connection): The database connection
+
+ Returns:
+ AppKeySecret: The application secret
+ """
+ # get existing secret keys
+ stmt = select(RegisteredApp).where(
+ RegisteredApp.url == url, RegisteredApp.name == name
+ )
+ result = db_con.execute(stmt).fetchone()
+
+ if result:
+ return AppKeySecret(key=result.key, secret=result.secret, url=url)
+
+ (client_id, client_secret) = Mastodon.create_app(
+ name,
+ api_base_url = f'https://{url}'
+ )
+
+ registered_app = RegisteredApp(name=name, url=url, key=client_id, secret=client_secret)
+ with Session(db_con) as session:
+ session.add(registered_app)
+ session.commit()
+
+ return AppKeySecret(key=client_id, secret=client_secret, url=url)
+
+def get_public_statuses(appKeySecret, tag, limit=40):
+ mastodon = Mastodon(client_id=appKeySecret.key, client_secret=appKeySecret.secret, api_base_url=f'https://{appKeySecret.url}')
+ results = mastodon.timeline_hashtag(hashtag=tag, limit=limit)
+ while results:
+ for status in results:
+ print("-------------------------")
+ print(repr(status))
+ results = mastodon.fetch_next(results)
+
+
+def parse_arg():
+ """
+ parse args
+ """
+ parser = argparse.ArgumentParser(
+ prog="dump_public_toots",
+ description="Dump the latest public toots",
+ epilog="dump only public toots",
+ )
+ parser.add_argument(
+ "-d",
+ "--database",
+ dest="database",
+ default="registered_apps.sqlite",
+ help="The path to the registered apps db",
+ )
+ parser.add_argument(
+ "-u", "--url", dest="url", help="The mastodon base url", required=True
+ )
+ parser.add_argument(
+ "-n", "--name", dest="name", help="The application name", required=True
+ )
+ parser.add_argument(
+ "-t", "--tag", dest="tag", help="The tag to search", required=True
+ )
+ parser.add_argument(
+ "-l", "--limit", dest="limit", help="page size", required=False, default=40
+ )
+
+
+ return parser.parse_args()
+
+
+if __name__ == "__main__":
+
+ options = parse_arg()
+
+ # req = requests.get(f"https://{options.url}/api/v1/timelines/tag/{options.tag}/")
+ # print(req.text)
+
+
+ app_db_eng = create_engine(f"sqlite+pysqlite:///{options.database}", future=True)
+
+ with app_db_eng.connect() as app_db_con:
+ Base.metadata.create_all(app_db_con)
+
+ keySecret = get_application_secret(options.name, options.url, app_db_con)
+
+ get_public_statuses(keySecret, options.tag, options.limit)
+
+ app_db_con.commit()
+