mirror of
https://github.com/chibicitiberiu/ytsm.git
synced 2024-02-24 05:43:31 +00:00
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
|
import logging
|
||
|
|
from typing import Callable, Union, Any, Optional
|
||
|
|
|
||
|
|
from django.contrib.auth.models import User
|
||
|
|
from django.db import models
|
||
|
|
from django.db.models.functions import Lower
|
||
|
|
|
||
|
|
|
||
|
|
class SubscriptionFolder(models.Model):
|
||
|
|
name = models.CharField(null=False, max_length=250)
|
||
|
|
parent = models.ForeignKey('self', on_delete=models.CASCADE, null=True, blank=True)
|
||
|
|
user = models.ForeignKey(User, on_delete=models.CASCADE, null=False, blank=False)
|
||
|
|
|
||
|
|
class Meta:
|
||
|
|
ordering = [Lower('parent__name'), Lower('name')]
|
||
|
|
|
||
|
|
def __str__(self):
|
||
|
|
s = ""
|
||
|
|
current = self
|
||
|
|
while current is not None:
|
||
|
|
s = current.name + " > " + s
|
||
|
|
current = current.parent
|
||
|
|
return s[:-3]
|
||
|
|
|
||
|
|
def __repr__(self):
|
||
|
|
return f'folder {self.id}, name="{self.name}"'
|
||
|
|
|
||
|
|
def delete_folder(self, keep_subscriptions: bool):
|
||
|
|
from .subscription import Subscription
|
||
|
|
if keep_subscriptions:
|
||
|
|
|
||
|
|
def visit(node: Union["SubscriptionFolder", "Subscription"]):
|
||
|
|
if isinstance(node, Subscription):
|
||
|
|
node.parent_folder = None
|
||
|
|
node.save()
|
||
|
|
|
||
|
|
SubscriptionFolder.traverse(self.id, self.user, visit)
|
||
|
|
|
||
|
|
self.delete()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def traverse(root_folder_id: Optional[int],
|
||
|
|
user: User,
|
||
|
|
visit_func: Callable[[Union["SubscriptionFolder", "Subscription"]], Any]):
|
||
|
|
from .subscription import Subscription
|
||
|
|
|
||
|
|
data_collected = []
|
||
|
|
|
||
|
|
def collect(data):
|
||
|
|
if data is not None:
|
||
|
|
data_collected.append( data)
|
||
|
|
|
||
|
|
# Visit root
|
||
|
|
if root_folder_id is not None:
|
||
|
|
root_folder = SubscriptionFolder.objects.get(id=root_folder_id)
|
||
|
|
collect(visit_func(root_folder))
|
||
|
|
|
||
|
|
queue = [root_folder_id]
|
||
|
|
visited = []
|
||
|
|
|
||
|
|
while len(queue) > 0:
|
||
|
|
folder_id = queue.pop()
|
||
|
|
|
||
|
|
if folder_id in visited:
|
||
|
|
logging.error('Found folder tree cycle for folder id %d.', folder_id)
|
||
|
|
continue
|
||
|
|
visited.append(folder_id)
|
||
|
|
|
||
|
|
for folder in SubscriptionFolder.objects.filter(parent_id=folder_id, user=user).order_by(Lower('name')):
|
||
|
|
collect(visit_func(folder))
|
||
|
|
queue.append(folder.id)
|
||
|
|
|
||
|
|
for subscription in Subscription.objects.filter(parent_folder_id=folder_id, user=user).order_by(Lower('name')):
|
||
|
|
collect(visit_func(subscription))
|
||
|
|
|
||
|
|
return data_collected
|