|
import argparse
|
|
import sys
|
|
from devplacepy.database import get_table
|
|
from devplacepy.utils import strip_html
|
|
|
|
|
|
def cmd_role_get(args):
|
|
users = get_table("users")
|
|
user = users.find_one(username=args.username)
|
|
if not user:
|
|
print(f"User '{args.username}' not found")
|
|
sys.exit(1)
|
|
print(user.get("role", "member").lower())
|
|
|
|
|
|
def cmd_role_set(args):
|
|
role = args.role.lower()
|
|
if role not in ("member", "admin"):
|
|
print("Role must be 'member' or 'admin'")
|
|
sys.exit(1)
|
|
|
|
users = get_table("users")
|
|
user = users.find_one(username=args.username)
|
|
if not user:
|
|
print(f"User '{args.username}' not found")
|
|
sys.exit(1)
|
|
|
|
users.update({"uid": user["uid"], "role": role.capitalize()}, ["uid"])
|
|
print(f"User '{args.username}' role set to '{role}'")
|
|
|
|
|
|
def cmd_news_clear(args):
|
|
from devplacepy.database import db
|
|
for table in ("news", "news_images", "news_sync"):
|
|
if table in db.tables:
|
|
count = len(list(db[table].all()))
|
|
db[table].delete()
|
|
print(f"Deleted {count} rows from '{table}'")
|
|
else:
|
|
print(f"Table '{table}' does not exist, skipping")
|
|
print("News data cleared")
|
|
|
|
|
|
def cmd_news_sanitize(args):
|
|
from devplacepy.database import db
|
|
if "news" not in db.tables:
|
|
print("News table does not exist")
|
|
return
|
|
news_table = db["news"]
|
|
updated = 0
|
|
for row in news_table.all():
|
|
desc = (strip_html(row.get("description", "") or ""))[:5000]
|
|
content = (strip_html(row.get("content", "") or ""))[:10000]
|
|
if desc != row.get("description", "") or content != row.get("content", ""):
|
|
news_table.update({"id": row["id"], "description": desc, "content": content}, ["id"])
|
|
updated += 1
|
|
print(f"Sanitized {updated} news article(s)")
|
|
|
|
|
|
def cmd_attachments_prune(args):
|
|
from devplacepy.database import db
|
|
from devplacepy.config import STATIC_DIR
|
|
import os
|
|
deleted_records = 0
|
|
deleted_files = 0
|
|
freed_bytes = 0
|
|
|
|
if "attachments" in db.tables:
|
|
orphans = list(db["attachments"].find(resource_uid=""))
|
|
orphans += list(db["attachments"].find(resource_type=""))
|
|
seen = set()
|
|
unique_orphans = []
|
|
for o in orphans:
|
|
if o["uid"] not in seen:
|
|
seen.add(o["uid"])
|
|
unique_orphans.append(o)
|
|
|
|
for att in unique_orphans:
|
|
sp = att.get("storage_path", "")
|
|
if sp:
|
|
fp = STATIC_DIR / "uploads" / sp
|
|
try:
|
|
if fp.exists():
|
|
freed_bytes += fp.stat().st_size
|
|
fp.unlink()
|
|
deleted_files += 1
|
|
parent = fp.parent
|
|
if parent.exists() and not any(parent.iterdir()):
|
|
parent.rmdir()
|
|
grandparent = parent.parent
|
|
if grandparent.exists() and not any(grandparent.iterdir()):
|
|
grandparent.rmdir()
|
|
except Exception as e:
|
|
print(f" Error deleting {sp}: {e}")
|
|
db["attachments"].delete(id=att["id"])
|
|
deleted_records += 1
|
|
|
|
print(f"Pruned {deleted_records} orphan records, {deleted_files} files, {freed_bytes / 1024:.1f} KB freed")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="DevPlace admin CLI")
|
|
sub = parser.add_subparsers(title="commands", dest="command")
|
|
|
|
role = sub.add_parser("role", help="Manage user roles")
|
|
role_sub = role.add_subparsers(title="action", dest="action")
|
|
|
|
role_get = role_sub.add_parser("get", help="Get a user's role")
|
|
role_get.add_argument("username")
|
|
role_get.set_defaults(func=cmd_role_get)
|
|
|
|
role_set = role_sub.add_parser("set", help="Set a user's role")
|
|
role_set.add_argument("username")
|
|
role_set.add_argument("role", choices=["member", "admin"])
|
|
role_set.set_defaults(func=cmd_role_set)
|
|
|
|
news = sub.add_parser("news", help="News management")
|
|
news_sub = news.add_subparsers(title="action", dest="action")
|
|
news_clear = news_sub.add_parser("clear", help="Delete all news from local database")
|
|
news_clear.set_defaults(func=cmd_news_clear)
|
|
news_sanitize = news_sub.add_parser("sanitize", help="Strip HTML from all existing news descriptions and content")
|
|
news_sanitize.set_defaults(func=cmd_news_sanitize)
|
|
|
|
attachments = sub.add_parser("attachments", help="Attachment management")
|
|
att_sub = attachments.add_subparsers(title="action", dest="action")
|
|
att_prune = att_sub.add_parser("prune", help="Remove orphaned attachment records and files")
|
|
att_prune.set_defaults(func=cmd_attachments_prune)
|
|
|
|
args = parser.parse_args()
|
|
if hasattr(args, "func"):
|
|
args.func(args)
|
|
else:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|