mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2025-12-20 11:11:10 -05:00
303 lines
11 KiB
Python
303 lines
11 KiB
Python
from django.db import models
|
|
from django.conf import settings
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
|
|
from django.db.models.fields.related import RelatedField
|
|
from django.contrib.auth import get_user_model
|
|
from simple_history.models import HistoricalRecords
|
|
from .mixins import HistoricalChangeMixin
|
|
from .historical_fields import HistoricalFieldsMixin
|
|
from typing import Any, Type, TypeVar, cast, Optional, List
|
|
from django.db.models import QuerySet
|
|
from django.core.exceptions import ValidationError
|
|
from django.utils import timezone
|
|
|
|
T = TypeVar('T', bound=models.Model)
|
|
|
|
User = get_user_model()
|
|
|
|
class HistoricalModel(models.Model, HistoricalFieldsMixin):
|
|
"""Abstract base class for models with history tracking"""
|
|
id = models.BigAutoField(primary_key=True)
|
|
|
|
@classmethod
|
|
def __init_subclass__(cls, **kwargs):
|
|
"""Initialize subclass with proper configuration."""
|
|
super().__init_subclass__(**kwargs)
|
|
# Mark historical models
|
|
if cls.__name__.startswith('Historical'):
|
|
cls._is_historical_model = True
|
|
# Remove any inherited generic relations
|
|
for field in list(cls._meta.private_fields):
|
|
if isinstance(field, GenericRelation):
|
|
cls._meta.private_fields.remove(field)
|
|
else:
|
|
cls._is_historical_model = False
|
|
history = HistoricalRecords(
|
|
inherit=True,
|
|
bases=[HistoricalChangeMixin],
|
|
excluded_fields=['comments', 'comment_threads', 'photos', 'reviews'],
|
|
use_base_model_db=True # Use base model's db
|
|
)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
@property
|
|
def _history_model(self) -> Type[T]:
|
|
"""Get the history model class"""
|
|
return cast(Type[T], self.history.model) # type: ignore
|
|
|
|
def get_history(self) -> QuerySet:
|
|
"""Get all history records for this instance"""
|
|
model = self._history_model
|
|
return model.objects.filter(id=self.pk).order_by('-history_date')
|
|
|
|
class HistoricalSlug(models.Model):
|
|
"""Track historical slugs for models"""
|
|
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
|
object_id = models.PositiveIntegerField()
|
|
content_object = GenericForeignKey('content_type', 'object_id')
|
|
slug = models.SlugField(max_length=255)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
|
|
class Meta:
|
|
unique_together = ('content_type', 'slug')
|
|
indexes = [
|
|
models.Index(fields=['content_type', 'object_id']),
|
|
models.Index(fields=['slug']),
|
|
]
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.content_type} - {self.object_id} - {self.slug}"
|
|
|
|
class VersionBranch(models.Model):
|
|
"""Represents a version control branch for tracking parallel development"""
|
|
name = models.CharField(max_length=255, unique=True)
|
|
parent = models.ForeignKey('self', null=True, blank=True, on_delete=models.CASCADE, related_name='children')
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
|
|
metadata = models.JSONField(default=dict, blank=True)
|
|
is_active = models.BooleanField(default=True)
|
|
lock_status = models.JSONField(
|
|
default=dict,
|
|
help_text="Current lock status: {user: ID, expires: datetime, reason: str}"
|
|
)
|
|
lock_history = models.JSONField(
|
|
default=list,
|
|
help_text="History of lock operations: [{user: ID, action: lock/unlock, timestamp: datetime, reason: str}]"
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ['-created_at']
|
|
indexes = [
|
|
models.Index(fields=['name']),
|
|
models.Index(fields=['parent']),
|
|
models.Index(fields=['created_at']),
|
|
]
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.name} ({'active' if self.is_active else 'inactive'})"
|
|
|
|
def clean(self) -> None:
|
|
# Prevent circular references
|
|
if self.parent and self.pk:
|
|
branch = self.parent
|
|
while branch:
|
|
if branch.pk == self.pk:
|
|
raise ValidationError("Circular branch reference detected")
|
|
branch = branch.parent
|
|
|
|
class VersionTag(models.Model):
|
|
"""Tags specific versions for reference (releases, milestones, etc)"""
|
|
name = models.CharField(max_length=255, unique=True)
|
|
branch = models.ForeignKey(VersionBranch, on_delete=models.CASCADE, related_name='tags')
|
|
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
|
object_id = models.PositiveIntegerField()
|
|
historical_instance = GenericForeignKey('content_type', 'object_id')
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
|
|
metadata = models.JSONField(default=dict, blank=True)
|
|
comparison_metadata = models.JSONField(
|
|
default=dict,
|
|
help_text="Stores diff statistics and comparison results"
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ['-created_at']
|
|
indexes = [
|
|
models.Index(fields=['name']),
|
|
models.Index(fields=['branch']),
|
|
models.Index(fields=['created_at']),
|
|
models.Index(fields=['content_type', 'object_id']),
|
|
]
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.name} ({self.branch.name})"
|
|
|
|
class HistoricalCommentThread(models.Model):
|
|
"""Represents a thread of comments specific to historical records and version control"""
|
|
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
|
object_id = models.PositiveIntegerField()
|
|
content_object = GenericForeignKey('content_type', 'object_id')
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, related_name='created_threads')
|
|
anchor = models.JSONField(
|
|
default=dict,
|
|
help_text="Anchoring information: {line_start: int, line_end: int, file_path: str}"
|
|
)
|
|
is_resolved = models.BooleanField(default=False)
|
|
resolved_by = models.ForeignKey(
|
|
User,
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
related_name='resolved_threads'
|
|
)
|
|
resolved_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ['-created_at']
|
|
indexes = [
|
|
models.Index(fields=['content_type', 'object_id']),
|
|
models.Index(fields=['created_at']),
|
|
models.Index(fields=['is_resolved']),
|
|
]
|
|
|
|
def __str__(self) -> str:
|
|
return f"Comment Thread {self.pk} on {self.content_type}"
|
|
|
|
class Comment(models.Model):
|
|
"""Individual comment within a thread"""
|
|
thread = models.ForeignKey(HistoricalCommentThread, on_delete=models.CASCADE, related_name='comments')
|
|
author = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
|
|
content = models.TextField()
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
mentioned_users = models.ManyToManyField(
|
|
User,
|
|
related_name='mentioned_in_comments',
|
|
blank=True
|
|
)
|
|
parent_comment = models.ForeignKey(
|
|
'self',
|
|
on_delete=models.CASCADE,
|
|
null=True,
|
|
blank=True,
|
|
related_name='replies'
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ['created_at']
|
|
|
|
def __str__(self) -> str:
|
|
return f"Comment {self.pk} by {self.author}"
|
|
|
|
def extract_mentions(self) -> None:
|
|
"""Extract @mentions from comment content and update mentioned_users"""
|
|
# Simple @username extraction - could be enhanced with regex
|
|
mentioned = [
|
|
word[1:] for word in self.content.split()
|
|
if word.startswith('@') and len(word) > 1
|
|
]
|
|
if mentioned:
|
|
users = User.objects.filter(username__in=mentioned)
|
|
self.mentioned_users.set(users)
|
|
|
|
class ChangeSet(models.Model):
|
|
"""Groups related changes together for atomic version control operations"""
|
|
branch = models.ForeignKey(VersionBranch, on_delete=models.CASCADE, related_name='changesets')
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
|
|
description = models.TextField(blank=True)
|
|
metadata = models.JSONField(default=dict, blank=True)
|
|
dependencies = models.JSONField(default=dict, blank=True)
|
|
status = models.CharField(
|
|
max_length=20,
|
|
choices=[
|
|
('draft', 'Draft'),
|
|
('pending_approval', 'Pending Approval'),
|
|
('approved', 'Approved'),
|
|
('rejected', 'Rejected'),
|
|
('applied', 'Applied'),
|
|
('failed', 'Failed'),
|
|
('reverted', 'Reverted')
|
|
],
|
|
default='draft'
|
|
)
|
|
approval_state = models.JSONField(
|
|
default=list,
|
|
help_text="List of approval stages and their status"
|
|
)
|
|
approval_history = models.JSONField(
|
|
default=list,
|
|
help_text="History of approval actions and decisions"
|
|
)
|
|
required_approvers = models.ManyToManyField(
|
|
User,
|
|
related_name='pending_approvals',
|
|
blank=True
|
|
)
|
|
approval_policy = models.CharField(
|
|
max_length=20,
|
|
choices=[
|
|
('sequential', 'Sequential'),
|
|
('parallel', 'Parallel')
|
|
],
|
|
default='sequential'
|
|
)
|
|
approval_deadline = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
help_text="Optional deadline for approvals"
|
|
)
|
|
|
|
# Instead of directly relating to HistoricalRecord, use GenericForeignKey
|
|
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
|
object_id = models.PositiveIntegerField()
|
|
historical_instance = GenericForeignKey('content_type', 'object_id')
|
|
|
|
class Meta:
|
|
ordering = ['-created_at']
|
|
indexes = [
|
|
models.Index(fields=['branch']),
|
|
models.Index(fields=['created_at']),
|
|
models.Index(fields=['status']),
|
|
models.Index(fields=['content_type', 'object_id']),
|
|
]
|
|
|
|
def __str__(self) -> str:
|
|
return f"ChangeSet {self.pk} ({self.branch.name} - {self.status})"
|
|
|
|
def apply(self) -> None:
|
|
"""Apply the changeset to the target branch"""
|
|
if self.status != 'pending':
|
|
raise ValidationError(f"Cannot apply changeset with status: {self.status}")
|
|
|
|
try:
|
|
# Apply changes through the historical instance
|
|
if self.historical_instance:
|
|
instance = self.historical_instance.instance
|
|
if instance:
|
|
instance.save()
|
|
self.status = 'applied'
|
|
except Exception as e:
|
|
self.status = 'failed'
|
|
self.metadata['error'] = str(e)
|
|
self.save()
|
|
|
|
def revert(self) -> None:
|
|
"""Revert the changes in this changeset"""
|
|
if self.status != 'applied':
|
|
raise ValidationError(f"Cannot revert changeset with status: {self.status}")
|
|
|
|
try:
|
|
# Revert changes through the historical instance
|
|
if self.historical_instance:
|
|
instance = self.historical_instance.instance
|
|
if instance:
|
|
instance.save()
|
|
self.status = 'reverted'
|
|
except Exception as e:
|
|
self.metadata['revert_error'] = str(e)
|
|
self.save()
|