Spaces:
Sleeping
Sleeping
File size: 3,697 Bytes
c8b1f17 |
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 |
from sqlalchemy import Column, Integer, String, Boolean, Text, DateTime, create_engine, UniqueConstraint
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime
import json
import os
Base = declarative_base()
class ConfigItem(Base):
__tablename__ = 'configs'
id = Column(Integer, primary_key=True, autoincrement=True)
key = Column(String(255), unique=True, nullable=False, index=True)
value = Column(Text, nullable=False) # JSON string
type = Column(String(50), nullable=False) # string, int, float, bool, json
group_name = Column(String(50), nullable=False, index=True) # api, model, data, evaluation
description = Column(Text)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
class SensitiveWord(Base):
__tablename__ = 'sensitive_words'
id = Column(Integer, primary_key=True, autoincrement=True)
word = Column(String(255), nullable=False, index=True)
category = Column(String(50), nullable=False, index=True)
subcategory = Column(String(50))
severity = Column(String(20), default='medium')
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
__table_args__ = (
UniqueConstraint('word', 'category', 'subcategory', name='uix_word_cat_sub'),
)
class WhitelistItem(Base):
__tablename__ = 'whitelist'
id = Column(Integer, primary_key=True, autoincrement=True)
word = Column(String(255), unique=True, nullable=False, index=True)
category = Column(String(50), default='general')
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
class ApiKey(Base):
__tablename__ = 'api_keys'
id = Column(Integer, primary_key=True, autoincrement=True)
access_key = Column(String(64), unique=True, nullable=False, index=True)
secret_key = Column(String(128), nullable=False) # Should be hashed in production, but plaintext for demo HMAC
name = Column(String(100), nullable=False) # e.g. "Admin Dashboard", "N8N Workflow"
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
expires_at = Column(DateTime, nullable=True) # None means never expires
class TaskQueue(Base):
__tablename__ = 'task_queue'
id = Column(Integer, primary_key=True, autoincrement=True)
task_id = Column(String(64), unique=True, nullable=False, index=True)
queue_name = Column(String(50), nullable=False, index=True)
payload = Column(Text, nullable=False) # JSON
status = Column(String(20), default='queued', index=True) # queued, processing, completed, failed
result = Column(Text, nullable=True) # JSON result or error
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Database Setup
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DB_DIR = os.path.join(BASE_DIR, "data")
if not os.path.exists(DB_DIR):
os.makedirs(DB_DIR)
DB_PATH = os.path.join(DB_DIR, "config.db")
SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_PATH}"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def init_db():
Base.metadata.create_all(bind=engine)
|