130 lines
3.4 KiB
Python
130 lines
3.4 KiB
Python
import pathlib
|
|
import shutil
|
|
import sqlite3
|
|
import asyncio
|
|
import click
|
|
from aiohttp import web
|
|
from IPython import start_ipython
|
|
from snek.shell import Shell
|
|
from snek.app import Application
|
|
|
|
|
|
@click.group()
|
|
def cli():
|
|
pass
|
|
|
|
@cli.command()
|
|
def export():
|
|
|
|
app = Application(db_path="sqlite:///snek.db")
|
|
|
|
async def fix_message(message):
|
|
message = {
|
|
"uid": message["uid"],
|
|
"user_uid": message["user_uid"],
|
|
"text": message["message"],
|
|
"sent": message["created_at"],
|
|
}
|
|
user = await app.services.user.get(uid=message["user_uid"])
|
|
message["user"] = user and user["username"] or None
|
|
return (message["user"] or "") + ": " + (message["text"] or "")
|
|
async def run():
|
|
|
|
result = []
|
|
for channel in app.db["channel"].find(
|
|
is_private=False, is_listed=True, tag="public"
|
|
):
|
|
print(f"Dumping channel: {channel['label']}.")
|
|
result += [
|
|
await fix_message(record)
|
|
for record in app.db["channel_message"].find(
|
|
channel_uid=channel["uid"], order_by="created_at"
|
|
)
|
|
]
|
|
print("Dump succesfull!")
|
|
print("Converting to json.")
|
|
print("Converting succesful, now writing to dump.txt")
|
|
with open("dump.txt", "w") as f:
|
|
f.write("\n\n".join(result))
|
|
print("Dump written to dump.json")
|
|
asyncio.run(run())
|
|
|
|
|
|
@cli.command()
|
|
def statistics():
|
|
async def run():
|
|
app = Application(db_path="sqlite:///snek.db")
|
|
app.services.statistics.database()
|
|
asyncio.run(run())
|
|
|
|
@cli.command()
|
|
def maintenance():
|
|
async def run():
|
|
app = Application(db_path="sqlite:///snek.db")
|
|
await app.services.container.maintenance()
|
|
await app.services.channel_message.maintenance()
|
|
asyncio.run(run())
|
|
|
|
@cli.command()
|
|
@click.option(
|
|
"--db_path", default="snek.db", help="Database to initialize if not exists."
|
|
)
|
|
@click.option("--source", default=None, help="Database to initialize if not exists.")
|
|
def init(db_path, source):
|
|
if source and pathlib.Path(source).exists():
|
|
print(f"Copying {source} to {db_path}")
|
|
shutil.copy2(source, db_path)
|
|
print("Database initialized.")
|
|
return
|
|
|
|
if pathlib.Path(db_path).exists():
|
|
return
|
|
print(f"Initializing database at {db_path}")
|
|
db = sqlite3.connect(db_path)
|
|
db.cursor().executescript(
|
|
pathlib.Path(__file__).parent.joinpath("schema.sql").read_text()
|
|
)
|
|
db.commit()
|
|
db.close()
|
|
print("Database initialized.")
|
|
|
|
|
|
@cli.command()
|
|
@click.option(
|
|
"--port", default=8081, show_default=True, help="Port to run the application on"
|
|
)
|
|
@click.option(
|
|
"--host",
|
|
default="0.0.0.0",
|
|
show_default=True,
|
|
help="Host to run the application on",
|
|
)
|
|
@click.option(
|
|
"--db_path",
|
|
default="snek.db",
|
|
show_default=True,
|
|
help="Database path for the application",
|
|
)
|
|
def serve(port, host, db_path):
|
|
# init(db_path)
|
|
# asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
|
|
web.run_app(Application(db_path=f"sqlite:///{db_path}"), port=port, host=host)
|
|
|
|
|
|
@cli.command()
|
|
@click.option(
|
|
"--db_path",
|
|
default="snek.db",
|
|
show_default=True,
|
|
help="Database path for the application",
|
|
)
|
|
def shell(db_path):
|
|
Shell(db_path).run()
|
|
|
|
def main():
|
|
cli()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|