File size: 5,454 Bytes
9a1712b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
"""
database/models.py — SQLAlchemy ORM models for the knowledge base.

Schema Design:
- User: Tracks all users who interact with the bot.
- Admin: Stores admins appointed by the Owner.
- Folder: A hierarchical container (can be nested via parent_id).
- Item: A piece of content (file, image, video, link, or text) inside a Folder.
- AdminLog: Audit trail of admin actions for the Owner dashboard.
- UserLog: Tracks user navigation for the Admin dashboard.
"""

from datetime import datetime
from sqlalchemy import (
    BigInteger, Boolean, DateTime, ForeignKey,
    Integer, String, Text, func
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    """Base class for all SQLAlchemy models."""
    pass


class User(Base):
    """Represents any Telegram user who has interacted with the bot."""
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)  # Telegram user ID
    username: Mapped[str | None] = mapped_column(String(64), nullable=True)
    full_name: Mapped[str] = mapped_column(String(256), nullable=False)
    is_banned: Mapped[bool] = mapped_column(Boolean, default=False)
    joined_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    # Relationships
    logs: Mapped[list["UserLog"]] = relationship("UserLog", back_populates="user")


class Admin(Base):
    """Represents a user with admin privileges, appointed by the Owner."""
    __tablename__ = "admins"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    user_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("users.id"), unique=True)
    username: Mapped[str | None] = mapped_column(String(64), nullable=True)
    full_name: Mapped[str] = mapped_column(String(256), nullable=False)
    added_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    added_by: Mapped[int] = mapped_column(BigInteger)  # Owner's user ID

    # Relationships
    action_logs: Mapped[list["AdminLog"]] = relationship("AdminLog", back_populates="admin")


class Folder(Base):
    """
    A hierarchical folder (category) for organizing content.
    parent_id=None means it's a root-level folder.
    """
    __tablename__ = "folders"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    emoji: Mapped[str] = mapped_column(String(8), default="📁")
    parent_id: Mapped[int | None] = mapped_column(
        Integer, ForeignKey("folders.id", ondelete="CASCADE"), nullable=True
    )
    created_by: Mapped[int] = mapped_column(BigInteger, ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    # Self-referential relationship for nested folders
    children: Mapped[list["Folder"]] = relationship(
        "Folder",
        back_populates="parent",
        cascade="all, delete-orphan"
    )
    parent: Mapped["Folder | None"] = relationship(
        "Folder", back_populates="children", remote_side=[id]
    )
    items: Mapped[list["Item"]] = relationship(
        "Item", back_populates="folder", cascade="all, delete-orphan"
    )


class Item(Base):
    """
    A piece of content stored inside a Folder.
    Stores ONLY file_id for media — never downloads files locally.
    """
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    folder_id: Mapped[int] = mapped_column(Integer, ForeignKey("folders.id", ondelete="CASCADE"))
    title: Mapped[str] = mapped_column(String(256), nullable=False)

    # Content type: 'text', 'link', 'photo', 'video', 'document', 'audio'
    content_type: Mapped[str] = mapped_column(String(32), nullable=False)

    # For media: Telegram's file_id (permanent reference, never download!)
    file_id: Mapped[str | None] = mapped_column(Text, nullable=True)

    # For text/link content
    text_content: Mapped[str | None] = mapped_column(Text, nullable=True)

    created_by: Mapped[int] = mapped_column(BigInteger, ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    # Relationships
    folder: Mapped["Folder"] = relationship("Folder", back_populates="items")


class AdminLog(Base):
    """Audit trail of every action taken by an admin (for Owner dashboard)."""
    __tablename__ = "admin_logs"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    admin_id: Mapped[int] = mapped_column(Integer, ForeignKey("admins.id"), nullable=True)
    admin_user_id: Mapped[int] = mapped_column(BigInteger)
    action: Mapped[str] = mapped_column(String(512), nullable=False)
    timestamp: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    admin: Mapped["Admin | None"] = relationship("Admin", back_populates="action_logs")


class UserLog(Base):
    """Tracks user navigation events for the Admin dashboard."""
    __tablename__ = "user_logs"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    user_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("users.id"))
    action: Mapped[str] = mapped_column(String(512), nullable=False)
    timestamp: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    user: Mapped["User"] = relationship("User", back_populates="logs")