script/utils/dump_public_toots.py
changeset 1557 7c67caaafdeb
--- /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()
+