File size: 3,544 Bytes
b30f068 | 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 | """
Database Schema for CDMS (Document Management System)
"""
import sys
from pathlib import Path
# Add project root to path
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from sqlalchemy import create_engine, Column, Integer, String, Text, JSON, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
import hashlib
from src.config.paths import DB_PATH
Base = declarative_base()
class Document(Base):
"""PDF Document metadata"""
__tablename__ = 'documents'
id = Column(String, primary_key=True) # Hash of filepath
filename = Column(String, unique=True)
filepath = Column(String)
file_size = Column(Integer) # Size in bytes
num_pages = Column(Integer)
num_chunks = Column(Integer, default=0)
upload_date = Column(DateTime, default=datetime.utcnow)
last_processed = Column(DateTime)
processed = Column(Integer, default=0) # 0=No, 1=Yes
doc_metadata = Column(JSON) # Additional metadata (renamed from 'metadata' - SQLAlchemy reserved)
@staticmethod
def generate_id(filepath: str) -> str:
"""Generate document ID from filepath"""
return hashlib.md5(filepath.encode()).hexdigest()
class DocumentChunk(Base):
"""Text chunks from documents"""
__tablename__ = 'document_chunks'
id = Column(String, primary_key=True) # unique chunk id
document_id = Column(String) # FK to documents.id
chunk_index = Column(Integer) # Order in document
content = Column(Text) # The actual text content
page_number = Column(Integer) # Which page this chunk is from
char_count = Column(Integer)
token_count = Column(Integer) # Estimated tokens
chunk_metadata = Column(JSON) # Additional metadata (renamed from 'metadata' - SQLAlchemy reserved)
created_at = Column(DateTime, default=datetime.utcnow)
@staticmethod
def generate_id(document_id: str, chunk_index: int) -> str:
"""Generate chunk ID"""
return f"{document_id}_{chunk_index}"
class DatabaseManager:
"""Manages the CDMS database"""
def __init__(self, db_path: str = None):
"""
Initialize database manager
Args:
db_path: Path to SQLite database file (defaults to the project-root
anchored DB_PATH so the committed store loads regardless of CWD)
"""
if db_path is None:
db_path = str(DB_PATH)
# Create data directory if it doesn't exist
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
self.db_path = db_path
self.engine = create_engine(f"sqlite:///{db_path}")
# Create tables
Base.metadata.create_all(self.engine)
def get_session(self):
"""Get database session"""
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=self.engine)
return Session()
# Test function
if __name__ == "__main__":
print("Testing CDMS Database Schema...")
print("-" * 70)
# Create database
db = DatabaseManager()
print("✅ Database initialized")
print(f" Location: {db.db_path}")
print(f" Tables created: documents, document_chunks")
# Test ID generation
doc_id = Document.generate_id("/path/to/test.pdf")
chunk_id = DocumentChunk.generate_id(doc_id, 0)
print(f"\n✅ ID generation works:")
print(f" Document ID: {doc_id[:16]}...")
print(f" Chunk ID: {chunk_id}")
|