devplacepy/devplacepy/database/moderation.py

# retoor <retoor@molodetz.nl>

from datetime import date, datetime, timezone

from .core import _in_clause, _now_iso, db, get_table
from .settings import get_int_setting


REPORTABLE_TARGETS: dict[str, str] = {
    "post": "posts",
    "comment": "comments",
    "gist": "gists",
    "project": "projects",
    "project_file": "project_files",
    "news": "news",
    "attachment": "attachments",
    "message": "messages",
    "quiz": "quizzes",
    "poll": "polls",
    "award": "awards",
    "user": "users",
    "issue": "issue_tickets",
    "workspace": "instances",
    "devii_output": "devii_conversations",
}


MATURITY_TARGETS: set[str] = {
    "post",
    "comment",
    "gist",
    "project",
    "news",
    "attachment",
    "quiz",
}


UNREPORTABLE_TABLES: dict[str, str] = {
    "news_images": "child rows of a reportable news article",
    "poll_options": "child rows of a reportable poll",
    "quiz_questions": "child rows of a reportable quiz",
    "quiz_options": "child rows of a reportable quiz",
    "tunnels": "child rows of a reportable workspace instance",
    "issue_comment_authors": "authorship index for reportable issue comments",
    "votes": "engagement counters, carry no authored content",
    "reactions": "engagement counters, carry no authored content",
    "bookmarks": "private to the owner",
    "follows": "relationship rows, carry no authored content",
    "poll_votes": "private ballots",
    "quiz_attempts": "private to the participant",
    "quiz_answers": "private to the participant",
    "sessions": "authentication state",
    "access_tokens": "authentication state",
    "devrant_tokens": "authentication state",
    "user_relations": "private block and mute lists",
    "notification_preferences": "private to the owner",
    "user_customizations": "runs only in the owner's own browser",
    "devii_tasks": "private to the owner",
    "devii_lessons": "private to the owner",
    "devii_virtual_tools": "private to the owner",
    "deepsearch_sessions": "private to the owner",
    "deepsearch_messages": "private to the owner",
    "isslop_analyses": "generated from a public URL, not authored content",
    "email_accounts": "private mailbox credentials",
    "instance_schedules": "child rows of a reportable workspace instance",
    "workspace_flags": "moderation records, not authored content",
    "workspace_quota_rules": "administrator-set limits, not authored content",
    "workspace_editor_prefs": (
        "private per-user editor configuration, never shown to another member"
    ),
    "content_reports": "moderation records, readable only by the reporter and moderators",
    "moderation_actions": "moderation records, not authored content",
    "content_maturity": "moderation labels, not authored content",
    "user_consents": "private consent history of the account holder",
    "backup_schedules": "operator configuration",
    "project_forks": "lineage index for reportable projects",
    "seo_metadata": "generated metadata for reportable content",
}


REPORT_REASONS: dict[str, str] = {
    "hate": "Hate speech or discriminatory content",
    "violence": "Realistic violence or threats",
    "weapons": "Weapons or dangerous instructions",
    "sexual": "Sexual or pornographic content",
    "religious": "Content targeting religion or belief",
    "misinformation": "False or misleading information",
    "exploitative": "Content exploiting a person",
    "harassment": "Harassment or bullying",
    "spam": "Spam or unwanted promotion",
    "intellectual_property": "Copyright or trademark infringement",
    "self_harm": "Self-harm or suicide",
    "illegal": "Illegal activity",
    "other": "Something else",
}


def report_reason_options() -> list[dict[str, str]]:
    return [{"key": key, "label": label} for key, label in REPORT_REASONS.items()]


REPORT_STATUSES: tuple[str, ...] = ("open", "acknowledged", "actioned", "dismissed")
REPORT_OPEN_STATUSES: tuple[str, ...] = ("open", "acknowledged")
REPORT_SEVERITIES: tuple[str, ...] = ("info", "warn", "critical")
REPORT_ORIGINS: tuple[str, ...] = ("member", "filter")

MODERATION_ACTIONS: tuple[str, ...] = (
    "remove_content",
    "restore_content",
    "warn",
    "suspend",
    "ban",
    "lift",
    "dismiss",
    "escalate",
)

MATURITY_LEVELS: tuple[str, ...] = ("general", "mature", "restricted")
MATURITY_SOURCES: tuple[str, ...] = ("author", "filter", "moderator")

CONSENT_KINDS: dict[str, str] = {
    "terms": "Terms of Service and Community Guidelines",
    "privacy": "Privacy Policy",
    "ai_third_party": "Processing of your content by a third-party AI provider",
    "activity_recording": "Recording of your presence and session activity",
    "container_credentials": (
        "Sharing your DevPlace credentials with software another member runs "
        "in a container"
    ),
}
CONSENT_STATES: tuple[str, ...] = ("granted", "withdrawn")

AGE_BANDS: tuple[str, ...] = ("under_min", "13_15", "16_17", "adult")

ADULT_AGE = 18
TEEN_AGE = 16
YOUNG_TEEN_AGE = 13

MINIMUM_AGE_FLOOR = YOUNG_TEEN_AGE
DEFAULT_MINIMUM_AGE = TEEN_AGE

SYSTEM_ACTOR = "system"

REPORTS_TABLE = "content_reports"
ACTIONS_TABLE = "moderation_actions"
MATURITY_TABLE = "content_maturity"
CONSENTS_TABLE = "user_consents"

MODERATION_TABLES: tuple[str, ...] = (
    REPORTS_TABLE,
    ACTIONS_TABLE,
    MATURITY_TABLE,
    CONSENTS_TABLE,
)


def years_between(born: date, today: date) -> int:
    years = today.year - born.year
    if (today.month, today.day) < (born.month, born.day):
        years -= 1
    return years


def minimum_age() -> int:
    return max(
        MINIMUM_AGE_FLOOR,
        get_int_setting("moderation_minimum_age", DEFAULT_MINIMUM_AGE),
    )


def age_band_for(age: int) -> str:
    if age >= ADULT_AGE:
        return "adult"
    if age >= TEEN_AGE:
        return "16_17"
    if age >= YOUNG_TEEN_AGE:
        return "13_15"
    return "under_min"


def band_allows_mature(band: str) -> bool:
    return band == "adult"


def band_allows_restricted(band: str) -> bool:
    return band == "adult"


def get_maturity_by_targets(target_type: str, uids: list[str]) -> dict[str, dict]:
    uids = [uid for uid in (uids or []) if uid]
    if not uids or MATURITY_TABLE not in db.tables:
        return {}
    placeholders, params = _in_clause(uids)
    params["tt"] = target_type
    rows = db.query(
        f"SELECT target_uid, level, source FROM {MATURITY_TABLE} "
        f"WHERE target_type = :tt AND target_uid IN ({placeholders}) "
        f"AND deleted_at IS NULL",
        **params,
    )
    return {
        row["target_uid"]: {"level": row["level"], "source": row["source"]}
        for row in rows
    }


def get_maturity(target_type: str, target_uid: str) -> dict:
    found = get_maturity_by_targets(target_type, [target_uid])
    return found.get(target_uid, {"level": "general", "source": ""})


def set_maturity(
    target_type: str, target_uid: str, level: str, source: str, set_by: str
) -> dict | None:
    if target_type not in MATURITY_TARGETS or level not in MATURITY_LEVELS:
        return None
    from devplacepy.utils import generate_uid

    table = get_table(MATURITY_TABLE)
    existing = table.find_one(target_type=target_type, target_uid=target_uid)
    now = _now_iso()
    if existing:
        table.update(
            {
                "id": existing["id"],
                "level": level,
                "source": source,
                "set_by": set_by,
                "updated_at": now,
                "deleted_at": None,
                "deleted_by": None,
            },
            ["id"],
        )
        return table.find_one(id=existing["id"])
    uid = generate_uid()
    table.insert(
        {
            "uid": uid,
            "target_type": target_type,
            "target_uid": target_uid,
            "level": level,
            "source": source,
            "set_by": set_by,
            "created_at": now,
            "updated_at": now,
            "deleted_at": None,
            "deleted_by": None,
        }
    )
    return table.find_one(uid=uid)


def list_consents(owner_kind: str, owner_id: str) -> list[dict]:
    if not owner_id or CONSENTS_TABLE not in db.tables:
        return []
    return list(
        get_table(CONSENTS_TABLE).find(
            owner_kind=owner_kind,
            owner_id=owner_id,
            deleted_at=None,
            order_by=["-created_at"],
        )
    )


def consent_state(owner_kind: str, owner_id: str, kind: str) -> dict | None:
    if not owner_id or CONSENTS_TABLE not in db.tables:
        return None
    rows = list(
        get_table(CONSENTS_TABLE).find(
            owner_kind=owner_kind,
            owner_id=owner_id,
            kind=kind,
            deleted_at=None,
            order_by=["-created_at", "-id"],
            _limit=1,
        )
    )
    return rows[0] if rows else None


def consent_granted(owner_kind: str, owner_id: str, kind: str) -> bool:
    row = consent_state(owner_kind, owner_id, kind)
    return bool(row and row.get("state") == "granted")


def set_consent(
    owner_kind: str, owner_id: str, kind: str, granted: bool, version: str = "1"
) -> dict | None:
    if kind not in CONSENT_KINDS or not owner_id:
        return None
    from devplacepy.utils import generate_uid

    table = get_table(CONSENTS_TABLE)
    now = _now_iso()
    current = consent_state(owner_kind, owner_id, kind)
    if current and not granted and current.get("state") == "granted":
        table.update({"id": current["id"], "withdrawn_at": now}, ["id"])
    uid = generate_uid()
    table.insert(
        {
            "uid": uid,
            "owner_kind": owner_kind,
            "owner_id": owner_id,
            "kind": kind,
            "version": version,
            "state": "granted" if granted else "withdrawn",
            "granted_at": now if granted else "",
            "withdrawn_at": "" if granted else now,
            "created_at": now,
            "deleted_at": None,
            "deleted_by": None,
        }
    )
    return table.find_one(uid=uid)


def suspension_active(user: dict | None) -> bool:
    if not user:
        return False
    until = (user.get("suspended_until") or "").strip()
    if not until:
        return False
    try:
        expiry = datetime.fromisoformat(until)
    except ValueError:
        return False
    if expiry.tzinfo is None:
        expiry = expiry.replace(tzinfo=timezone.utc)
    return expiry > datetime.now(timezone.utc)