mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2025-12-20 09:31:09 -05:00
284 lines
10 KiB
Python
284 lines
10 KiB
Python
from django.db import models
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
|
|
from django.contrib.auth import get_user_model
|
|
from simple_history.models import HistoricalRecords
|
|
from .mixins import HistoricalChangeMixin
|
|
from typing import Any, Type, TypeVar, cast, Optional
|
|
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):
|
|
"""Abstract base class for models with history tracking"""
|
|
id = models.BigAutoField(primary_key=True)
|
|
history: HistoricalRecords = HistoricalRecords(
|
|
inherit=True,
|
|
bases=(HistoricalChangeMixin,)
|
|
)
|
|
|
|
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 CommentThread(models.Model):
|
|
"""Represents a thread of comments on a historical record"""
|
|
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(CommentThread, 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()
|