rucki
1536 lines
77 KiB
Python
1536 lines
77 KiB
Python
import json
|
||
import re
|
||
from functools import wraps
|
||
from calendar import Calendar
|
||
from datetime import date, datetime, time, timedelta
|
||
from html import unescape
|
||
from urllib.request import Request, urlopen
|
||
from django.conf import settings
|
||
from django.contrib import messages
|
||
from django.contrib.auth import login
|
||
from django.contrib.auth.decorators import login_required
|
||
from django.db import transaction
|
||
from django.db.models import Case, Count, IntegerField, ProtectedError, Q, Value, When
|
||
from django.http import HttpResponse, JsonResponse
|
||
from django.urls import reverse
|
||
from django.utils.http import url_has_allowed_host_and_scheme
|
||
from django.utils import timezone
|
||
from django.utils import translation
|
||
from django.utils.translation import gettext as _
|
||
from django.shortcuts import get_object_or_404, redirect, render
|
||
from django.views.decorators.csrf import csrf_exempt
|
||
from django.views.decorators.http import require_http_methods, require_POST
|
||
from .forms import ApiKeyForm, CollectionForm, EditItemForm, InviteRegistrationForm, KanbanForm, QuickItemForm, TeamForm, TeamInviteForm, UserInviteForm, UserPreferenceForm, UserSettingsForm, WorkspaceForm
|
||
from .models import ApiKey, Attachment, Collection, CollectionSection, Item, Kanban, Tag, Team, TeamInvite, TeamMembership, UserInvite, UserPreference, Workspace, WorkspaceMembership, TAG_RE, get_or_create_personal_workspace, without_markdown_code
|
||
from .templatetags.markdown_extras import strip_tags_outside_code
|
||
|
||
TYPE_COMMAND_RE = re.compile(r'^/(todo|note|link|journal)\b', re.I)
|
||
VISIBILITY_COMMAND_RE = re.compile(r'/(public|private)\b', re.I)
|
||
URL_RE = re.compile(r'https?://\S+')
|
||
ONLY_URL_RE = re.compile(r'^https?://\S+$')
|
||
FIRST_LINE_URL_RE = re.compile(r'^(https?://\S+)[ \t]*(?:\r?\n|$)')
|
||
TITLE_RE = re.compile(r'<title[^>]*>(.*?)</title>', re.I | re.S)
|
||
DATE_COMMAND_RE = re.compile(r'/datum\s+(\d{1,2}\.\d{1,2}\.(?:\d{2}|\d{4}))(?:\s+(\d{1,2}:\d{2}))?\b', re.I)
|
||
RELATIVE_DATE_COMMAND_RE = re.compile(r'/(morgen|übermorgen|uebermorgen|nächstewoche|naechstewoche)\b', re.I)
|
||
CODE_FENCE_RE = re.compile(r'^\s*(`{3,}|~{3,})')
|
||
INLINE_CODE_RE = re.compile(r'(`+).*?\1', re.S)
|
||
|
||
|
||
def mask_markdown_code(text):
|
||
"""Replace Markdown code with spaces while preserving string positions."""
|
||
masked_lines = []
|
||
fence = None
|
||
for line in (text or '').splitlines(keepends=True):
|
||
line_without_newline = line.rstrip('\r\n')
|
||
newline = line[len(line_without_newline):]
|
||
match = CODE_FENCE_RE.match(line_without_newline)
|
||
if fence:
|
||
masked_lines.append(' ' * len(line_without_newline) + newline)
|
||
if match and match.group(1)[0] == fence[0] and len(match.group(1)) >= len(fence):
|
||
fence = None
|
||
elif match:
|
||
fence = match.group(1)
|
||
masked_lines.append(' ' * len(line_without_newline) + newline)
|
||
else:
|
||
masked_lines.append(INLINE_CODE_RE.sub(lambda m: ' ' * (m.end() - m.start()), line_without_newline) + newline)
|
||
return ''.join(masked_lines)
|
||
|
||
|
||
def remove_spans(text, spans):
|
||
for start, end in sorted(spans, reverse=True):
|
||
text = f'{text[:start]} {text[end:]}'
|
||
return re.sub(r'[ \t]{2,}', ' ', text).strip()
|
||
|
||
|
||
def fetch_page_title(url):
|
||
try:
|
||
request = Request(url, headers={'User-Agent': 'my2dos/1.0'})
|
||
with urlopen(request, timeout=4) as response:
|
||
html = response.read(200_000).decode(response.headers.get_content_charset() or 'utf-8', errors='ignore')
|
||
match = TITLE_RE.search(html)
|
||
if match:
|
||
return re.sub(r'\s+', ' ', unescape(match.group(1))).strip()
|
||
except Exception:
|
||
pass
|
||
return ''
|
||
|
||
|
||
def parse_quick_content(raw, default_kind=Item.Kind.NOTE):
|
||
content = raw.strip()
|
||
type_match = TYPE_COMMAND_RE.match(content)
|
||
type_command = type_match.group(1).lower() if type_match else None
|
||
masked_content = mask_markdown_code(content)
|
||
visibility_matches = list(VISIBILITY_COMMAND_RE.finditer(masked_content))
|
||
visibility_commands = {match.group(1).lower() for match in visibility_matches}
|
||
offset = 0
|
||
if type_match:
|
||
remainder = content[type_match.end():]
|
||
leading_space = len(remainder) - len(remainder.lstrip())
|
||
offset = type_match.end() + leading_space
|
||
content = remainder.strip()
|
||
visibility_spans = [(match.start() - offset, match.end() - offset) for match in visibility_matches if match.start() >= offset]
|
||
content = remove_spans(content, visibility_spans)
|
||
only_url = bool(ONLY_URL_RE.match(content))
|
||
first_line_url_match = FIRST_LINE_URL_RE.match(content)
|
||
if type_command == 'todo':
|
||
kind = Item.Kind.TODO
|
||
elif type_command == 'journal':
|
||
kind = Item.Kind.JOURNAL
|
||
elif type_command == 'link' or only_url or first_line_url_match:
|
||
kind = Item.Kind.LINK
|
||
elif type_command == 'note':
|
||
kind = Item.Kind.NOTE
|
||
else:
|
||
kind = default_kind if default_kind in Item.Kind.values else Item.Kind.NOTE
|
||
visibility = Item.Visibility.PUBLIC if 'public' in visibility_commands else Item.Visibility.PRIVATE
|
||
url_match = first_line_url_match or URL_RE.search(content)
|
||
url = url_match.group(1) if kind == Item.Kind.LINK and first_line_url_match else (url_match.group(0) if kind == Item.Kind.LINK and url_match else '')
|
||
if first_line_url_match:
|
||
title = fetch_page_title(url) or url
|
||
extra_content = content[first_line_url_match.end():].strip()
|
||
content = f'{title}\n{extra_content}' if extra_content else title
|
||
elif only_url:
|
||
content = fetch_page_title(url) or url
|
||
return kind, visibility, content, url
|
||
|
||
|
||
def parse_due_command(raw):
|
||
"""Extract a due date slash command and return cleaned text, date and match state."""
|
||
masked_raw = mask_markdown_code(raw)
|
||
date_match = DATE_COMMAND_RE.search(masked_raw)
|
||
relative_match = RELATIVE_DATE_COMMAND_RE.search(masked_raw)
|
||
match = date_match or relative_match
|
||
if not match:
|
||
return raw, None, False
|
||
|
||
due_time = time(23, 59)
|
||
if date_match:
|
||
date_format = '%d.%m.%Y' if len(date_match.group(1).rsplit('.', 1)[-1]) == 4 else '%d.%m.%y'
|
||
try:
|
||
due_date = datetime.strptime(date_match.group(1), date_format).date()
|
||
if date_match.group(2):
|
||
due_time = time.fromisoformat(date_match.group(2))
|
||
except ValueError:
|
||
return raw, None, False
|
||
else:
|
||
today = timezone.localdate()
|
||
command = relative_match.group(1).lower()
|
||
if command == 'morgen':
|
||
due_date = today + timedelta(days=1)
|
||
elif command in {'übermorgen', 'uebermorgen'}:
|
||
due_date = today + timedelta(days=2)
|
||
else:
|
||
due_date = today + timedelta(days=(7 - today.weekday()))
|
||
|
||
cleaned = remove_spans(raw, [(match.start(), match.end())])
|
||
return cleaned, timezone.make_aware(datetime.combine(due_date, due_time)), True
|
||
|
||
|
||
def parse_due_at_from_post(post):
|
||
due_date = (post.get('due_date') or '').strip()
|
||
due_time = (post.get('due_time') or '').strip()
|
||
if not due_date:
|
||
return None
|
||
try:
|
||
parsed_date = date.fromisoformat(due_date)
|
||
parsed_time = time.fromisoformat(due_time) if due_time else time(23, 59)
|
||
return timezone.make_aware(datetime.combine(parsed_date, parsed_time))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def delete_selected_attachments(item, attachment_ids):
|
||
for attachment in item.attachments.filter(pk__in=attachment_ids):
|
||
attachment.file.delete(save=False)
|
||
attachment.delete()
|
||
if hasattr(item, '_prefetched_objects_cache'):
|
||
item._prefetched_objects_cache.pop('attachments', None)
|
||
|
||
|
||
def attach_files_and_replace_tokens(item, files):
|
||
replacements = {}
|
||
attached = False
|
||
for file in files:
|
||
attachment = Attachment.objects.create(item=item, file=file)
|
||
attached = True
|
||
replacements[f'![[paste:{file.name}]]'] = f'![[file:{attachment.id}]]'
|
||
if attached and hasattr(item, '_prefetched_objects_cache'):
|
||
item._prefetched_objects_cache.pop('attachments', None)
|
||
if replacements:
|
||
content = item.content
|
||
comment = item.comment
|
||
for old, new in replacements.items():
|
||
content = content.replace(old, new)
|
||
comment = comment.replace(old, new)
|
||
if content != item.content or comment != item.comment:
|
||
item.content = content
|
||
item.comment = comment
|
||
item.save(update_fields=['content', 'comment', 'updated_at'])
|
||
|
||
|
||
def current_workspace(user, api_key=None, workspace=None):
|
||
if api_key:
|
||
return api_key.workspace
|
||
if workspace:
|
||
return workspace
|
||
if user.is_authenticated:
|
||
return get_or_create_personal_workspace(user)
|
||
return None
|
||
|
||
|
||
def user_workspaces(user):
|
||
if not user.is_authenticated:
|
||
return []
|
||
return WorkspaceMembership.objects.filter(user=user).select_related('workspace').order_by('workspace__name')
|
||
|
||
|
||
def active_workspace(request):
|
||
if not request.user.is_authenticated:
|
||
return None
|
||
workspace_id = request.session.get('workspace_id')
|
||
if workspace_id:
|
||
membership = WorkspaceMembership.objects.select_related('workspace').filter(user=request.user, workspace_id=workspace_id).first()
|
||
if membership:
|
||
return membership.workspace
|
||
workspace = get_or_create_personal_workspace(request.user)
|
||
request.session['workspace_id'] = workspace.id
|
||
return workspace
|
||
|
||
|
||
def user_team_ids(user, workspace=None):
|
||
if not user.is_authenticated:
|
||
return []
|
||
qs = user.team_memberships
|
||
if workspace:
|
||
qs = qs.filter(team__workspace=workspace)
|
||
return list(qs.values_list('team_id', flat=True))
|
||
|
||
|
||
def visible_items(user, api_key=None, workspace=None):
|
||
workspace = current_workspace(user, api_key, workspace)
|
||
qs = Item.objects.select_related('owner', 'team', 'workspace').prefetch_related('tags', 'attachments', 'linked_items')
|
||
if workspace:
|
||
qs = qs.filter(workspace=workspace)
|
||
elif not user.is_authenticated:
|
||
return qs.none()
|
||
if api_key:
|
||
qs = qs.filter(owner=api_key.owner)
|
||
allowed_tags = api_key.tags.all()
|
||
if allowed_tags.exists():
|
||
qs = qs.filter(tags__in=allowed_tags).distinct()
|
||
return qs
|
||
if user.is_authenticated:
|
||
return qs.filter(Q(owner=user) | Q(team_id__in=user_team_ids(user, workspace)) | Q(visibility=Item.Visibility.PUBLIC)).distinct()
|
||
return qs.none()
|
||
|
||
|
||
def user_can_edit_item(user, item):
|
||
return item.owner_id == user.id or (item.team_id and TeamMembership.objects.filter(team=item.team, user=user).exists())
|
||
|
||
|
||
def visible_tags(user, workspace=None):
|
||
"""Tags from items the user may see; avoids leaking private tags from other users."""
|
||
workspace = current_workspace(user, workspace=workspace)
|
||
if not user.is_authenticated:
|
||
return Tag.objects.none()
|
||
return Tag.objects.filter(workspace=workspace, items__in=visible_items(user, workspace=workspace)).distinct().order_by('name')
|
||
|
||
|
||
def user_can_delete_item(user, item):
|
||
return item.owner_id == user.id or (item.team_id and TeamMembership.objects.filter(team=item.team, user=user, role__in=[TeamMembership.Role.OWNER, TeamMembership.Role.ADMIN]).exists())
|
||
|
||
|
||
def desired_item_scope(item, user, force_private=False):
|
||
if force_private:
|
||
return Item.Visibility.PRIVATE, None
|
||
if item.visibility == Item.Visibility.PUBLIC:
|
||
return Item.Visibility.PUBLIC, None
|
||
tag_names = {match.group(1).lower() for match in re.finditer(r'(?<!\w)#([\wäöüÄÖÜß-]+)', without_markdown_code(item.content))}
|
||
membership = TeamMembership.objects.select_related('team').filter(user=user, team__workspace=item.workspace, team__slug__in=tag_names).first()
|
||
if membership:
|
||
return Item.Visibility.TEAM, membership.team
|
||
visibility = Item.Visibility.PRIVATE if item.visibility == Item.Visibility.TEAM else item.visibility
|
||
return visibility, None
|
||
|
||
|
||
def item_scope_fits_collections(item, visibility, team):
|
||
sections = item.collection_sections.select_related('collection') if item.pk else CollectionSection.objects.none()
|
||
team_id = team.id if team else None
|
||
return all(collection_scope_allows_item(
|
||
section.collection.visibility, section.collection.team_id, section.collection.owner_id,
|
||
visibility, team_id, item.owner_id,
|
||
) for section in sections)
|
||
|
||
|
||
def apply_team_from_tags(item, user, force_private=False):
|
||
if force_private:
|
||
item.team = None
|
||
if item.visibility == Item.Visibility.TEAM:
|
||
item.visibility = Item.Visibility.PRIVATE
|
||
item.save(update_fields=['team', 'visibility', 'updated_at'])
|
||
return
|
||
if item.visibility == Item.Visibility.PUBLIC:
|
||
item.team = None
|
||
item.save(update_fields=['team', 'updated_at'])
|
||
return
|
||
tag_names = list(item.tags.values_list('name', flat=True))
|
||
membership = TeamMembership.objects.select_related('team').filter(user=user, team__workspace=item.workspace, team__slug__in=tag_names).first()
|
||
if membership:
|
||
item.team = membership.team
|
||
if item.visibility == Item.Visibility.PRIVATE:
|
||
item.visibility = Item.Visibility.TEAM
|
||
else:
|
||
item.team = None
|
||
if item.visibility == Item.Visibility.TEAM:
|
||
item.visibility = Item.Visibility.PRIVATE
|
||
item.save(update_fields=['team', 'visibility', 'updated_at'])
|
||
|
||
|
||
def get_api_key(request):
|
||
token = request.headers.get('X-Api-Key') or request.headers.get('Authorization', '').removeprefix('Bearer ').strip()
|
||
if not token:
|
||
return None
|
||
return ApiKey.objects.filter(token=token, is_active=True).select_related('owner', 'workspace').prefetch_related('tags').first()
|
||
|
||
|
||
def api_auth_required(view):
|
||
@wraps(view)
|
||
def wrapper(request, *args, **kwargs):
|
||
request.api_key = get_api_key(request)
|
||
if request.user.is_authenticated or request.api_key:
|
||
return view(request, *args, **kwargs)
|
||
return JsonResponse({'error': 'authentication required'}, status=401)
|
||
return wrapper
|
||
|
||
|
||
def index_filter_tags(request, user, prefs, active_tags=None, workspace=None):
|
||
active_tags = active_tags or []
|
||
kind = request.GET.get('kind')
|
||
status = request.GET.get('status')
|
||
day = request.GET.get('day')
|
||
q = request.GET.get('q', '').strip()
|
||
items = visible_items(user, workspace=workspace)
|
||
applied_preferences = False
|
||
if kind in dict(Item.Kind.choices):
|
||
items = items.filter(kind=kind)
|
||
elif not any([active_tags, status, q, day]):
|
||
applied_preferences = True
|
||
type_filter = Q()
|
||
if prefs.show_notes:
|
||
type_filter |= Q(kind=Item.Kind.NOTE)
|
||
if prefs.show_links:
|
||
type_filter |= Q(kind=Item.Kind.LINK)
|
||
if prefs.show_journal:
|
||
type_filter |= Q(kind=Item.Kind.JOURNAL)
|
||
if prefs.show_open_todos:
|
||
type_filter |= Q(kind=Item.Kind.TODO, is_done=False)
|
||
if prefs.show_done_todos:
|
||
type_filter |= Q(kind=Item.Kind.TODO, is_done=True)
|
||
items = items.filter(type_filter) if type_filter else items.none()
|
||
if day:
|
||
items = items.filter(created_at__date=day)
|
||
if status == 'open':
|
||
items = items.filter(kind=Item.Kind.TODO, is_done=False)
|
||
elif status == 'due':
|
||
upcoming_end = timezone.make_aware(datetime.combine(timezone.localdate() + timedelta(days=7), time(23, 59, 59)))
|
||
items = items.filter(kind=Item.Kind.TODO, is_done=False, due_at__isnull=False, due_at__lte=upcoming_end)
|
||
elif status == 'archive':
|
||
items = items.filter(kind=Item.Kind.TODO, is_done=True)
|
||
elif not applied_preferences:
|
||
items = items.exclude(kind=Item.Kind.TODO, is_done=True)
|
||
if q:
|
||
items = items.filter(Q(content__icontains=q) | Q(url__icontains=q) | Q(tags__name__icontains=q)).distinct()
|
||
tags = Tag.objects.filter(Q(items__in=items) | Q(name__in=active_tags))
|
||
workspace = current_workspace(user, workspace=workspace)
|
||
if workspace:
|
||
tags = tags.filter(workspace=workspace)
|
||
return tags.distinct().order_by('name')
|
||
|
||
|
||
def index_filtered_items(request, prefs):
|
||
active_tags = [t.strip().lower() for t in request.GET.getlist('tag') if t.strip()]
|
||
active_tag_mode = 'all' if request.GET.get('tag_mode') == 'all' else 'any'
|
||
tag = active_tags[0] if active_tags else None
|
||
kind = request.GET.get('kind')
|
||
status = request.GET.get('status')
|
||
day = request.GET.get('day')
|
||
q = request.GET.get('q', '').strip()
|
||
items = visible_items(request.user, workspace=active_workspace(request))
|
||
applied_preferences = False
|
||
if active_tags:
|
||
if active_tag_mode == 'all':
|
||
for tag_name in active_tags:
|
||
items = items.filter(tags__name=tag_name)
|
||
else:
|
||
items = items.filter(tags__name__in=active_tags).distinct()
|
||
if kind in dict(Item.Kind.choices):
|
||
items = items.filter(kind=kind)
|
||
elif not any([tag, status, q, day]):
|
||
applied_preferences = True
|
||
type_filter = Q()
|
||
if prefs.show_notes:
|
||
type_filter |= Q(kind=Item.Kind.NOTE)
|
||
if prefs.show_links:
|
||
type_filter |= Q(kind=Item.Kind.LINK)
|
||
if prefs.show_journal:
|
||
type_filter |= Q(kind=Item.Kind.JOURNAL)
|
||
if prefs.show_open_todos:
|
||
type_filter |= Q(kind=Item.Kind.TODO, is_done=False)
|
||
if prefs.show_done_todos:
|
||
type_filter |= Q(kind=Item.Kind.TODO, is_done=True)
|
||
items = items.filter(type_filter) if type_filter else items.none()
|
||
if day:
|
||
items = items.filter(created_at__date=day)
|
||
upcoming_end = timezone.make_aware(datetime.combine(timezone.localdate() + timedelta(days=7), time(23, 59, 59)))
|
||
if status == 'open':
|
||
items = items.filter(kind=Item.Kind.TODO, is_done=False)
|
||
elif status == 'due':
|
||
items = items.filter(kind=Item.Kind.TODO, is_done=False, due_at__isnull=False, due_at__lte=upcoming_end).order_by('due_at', '-created_at')
|
||
elif status == 'archive':
|
||
items = items.filter(kind=Item.Kind.TODO, is_done=True)
|
||
elif not applied_preferences:
|
||
items = items.exclude(kind=Item.Kind.TODO, is_done=True)
|
||
if q:
|
||
items = items.filter(Q(content__icontains=q) | Q(url__icontains=q) | Q(tags__name__icontains=q)).distinct()
|
||
return items, active_tags, active_tag_mode, tag, kind, status, day, q
|
||
|
||
|
||
def export_item_line(item, hide_tags=False):
|
||
content = strip_tags_outside_code(item.content) if hide_tags else (item.content or '')
|
||
lines = [line.rstrip() for line in content.strip().splitlines()]
|
||
first = lines[0] if lines else _('Untitled')
|
||
prefix = '- [x] ' if item.kind == Item.Kind.TODO and item.is_done else '- [ ] ' if item.kind == Item.Kind.TODO else '- '
|
||
output = [f'{prefix}{first}']
|
||
output.extend(f' {line}' if line else '' for line in lines[1:])
|
||
if item.comment:
|
||
output.append('')
|
||
output.extend(f' > {line}' if line else ' >' for line in item.comment.strip().splitlines())
|
||
return '\n'.join(output)
|
||
|
||
|
||
def markdown_response(filename, markdown_text):
|
||
response = HttpResponse(markdown_text, content_type='text/markdown; charset=utf-8')
|
||
response['Content-Disposition'] = f'attachment; filename="{filename}"'
|
||
return response
|
||
|
||
|
||
@login_required
|
||
def export_items(request):
|
||
prefs, created = UserPreference.objects.get_or_create(user=request.user)
|
||
items, active_tags, active_tag_mode, tag, kind, status, day, q = index_filtered_items(request, prefs)
|
||
title_parts = [_('my2dos export')]
|
||
if kind:
|
||
title_parts.append(kind)
|
||
if active_tags:
|
||
title_parts.append(' '.join(f'#{tag_name}' for tag_name in active_tags))
|
||
if status:
|
||
title_parts.append(status)
|
||
if q:
|
||
title_parts.append(f'“{q}”')
|
||
lines = [f'# {" – ".join(title_parts)}', '']
|
||
lines.extend(export_item_line(item) for item in items)
|
||
return markdown_response('my2dos-export.md', '\n\n'.join(lines).strip() + '\n')
|
||
|
||
|
||
def collection_item_export(item, level=2):
|
||
content = strip_tags_outside_code(item.content)
|
||
if item.kind == Item.Kind.TODO:
|
||
return export_item_line(item, hide_tags=True)
|
||
if re.match(r'^#{1,6}\s+', content.lstrip()):
|
||
body = content
|
||
else:
|
||
body = f'{"#" * level} {content}'
|
||
if item.comment:
|
||
body = f'{body}\n\n> ' + item.comment.strip().replace('\n', '\n> ')
|
||
return body
|
||
|
||
|
||
def collection_sections_export(sections, level=2):
|
||
lines = []
|
||
for section in sections:
|
||
if section.item:
|
||
heading = section.title.strip() if section.title else ''
|
||
if heading:
|
||
lines.append(f'{"#" * level} {heading}')
|
||
lines.append(collection_item_export(section.item, level + 1))
|
||
else:
|
||
lines.append(collection_item_export(section.item, level))
|
||
elif section.sub_collection:
|
||
title = section.title or section.sub_collection.title
|
||
lines.append(f'{"#" * level} {title}')
|
||
if section.sub_collection.description:
|
||
lines.append(section.sub_collection.description.strip())
|
||
lines.append(collection_sections_export(section.sub_collection.sections.select_related('item', 'sub_collection').prefetch_related('item__tags', 'item__attachments'), level + 1))
|
||
return '\n\n'.join(line for line in lines if line).strip()
|
||
|
||
|
||
def collection_export(request, pk):
|
||
collection = get_object_or_404(visible_collections(request.user, workspace=active_workspace(request)), pk=pk)
|
||
lines = [f'# {collection.title}', '']
|
||
if collection.description:
|
||
lines.extend([collection.description.strip(), ''])
|
||
if collection.mode == Collection.Mode.TAGS:
|
||
lines.extend(export_item_line(item, hide_tags=True) for item in collection_tag_items(collection))
|
||
else:
|
||
body = collection_sections_export(collection.sections.select_related('item', 'sub_collection').prefetch_related('item__tags', 'item__attachments', 'sub_collection__sections__item', 'sub_collection__sections__sub_collection'))
|
||
if body:
|
||
lines.append(body)
|
||
return markdown_response(f'collection-{collection.pk}.md', '\n\n'.join(line for line in lines if line).strip() + '\n')
|
||
|
||
|
||
def index(request):
|
||
if not request.user.is_authenticated:
|
||
if request.method == 'POST':
|
||
return redirect('login')
|
||
return render(request, 'core/landing.html')
|
||
prefs, created = UserPreference.objects.get_or_create(user=request.user)
|
||
if request.method == 'POST':
|
||
form = QuickItemForm(request.POST, request.FILES)
|
||
if form.is_valid():
|
||
raw = form.cleaned_data['content']
|
||
raw, due_at, _ = parse_due_command(raw)
|
||
kind, visibility, content, url = parse_quick_content(raw, requested_default_item_kind(request.POST, prefs, save=True))
|
||
active_tag = request.GET.get('tag')
|
||
if active_tag and not TAG_RE.search(without_markdown_code(content)):
|
||
content = f'{content} #{active_tag.strip().lower()}'
|
||
item = Item.objects.create(workspace=active_workspace(request), owner=request.user, kind=kind, visibility=visibility, content=content, url=url, due_at=due_at)
|
||
attach_files_and_replace_tokens(item, request.FILES.getlist('files'))
|
||
item.sync_metadata()
|
||
apply_team_from_tags(item, request.user, force_private=bool(re.search(r'/private\b', raw, re.I)))
|
||
if request.headers.get('HX-Request'):
|
||
active_tags = [t.strip().lower() for t in request.GET.getlist('tag') if t.strip()]
|
||
response = render(request, 'core/_create_response.html', {'item': item, 'tags': index_filter_tags(request, request.user, prefs, active_tags, workspace=active_workspace(request)), 'active_tag': active_tags[0] if active_tags else None, 'active_tags': active_tags})
|
||
response['HX-Trigger'] = 'tagsChanged'
|
||
return response
|
||
return redirect('index')
|
||
active_tags = [t.strip().lower() for t in request.GET.getlist('tag') if t.strip()]
|
||
active_tag_mode = 'all' if request.GET.get('tag_mode') == 'all' else 'any'
|
||
tag = active_tags[0] if active_tags else None
|
||
kind = request.GET.get('kind')
|
||
status = request.GET.get('status')
|
||
day = request.GET.get('day')
|
||
q = request.GET.get('q', '').strip()
|
||
if q.startswith('#') and len(q) > 1:
|
||
query = request.GET.copy()
|
||
query['tag'] = q[1:].strip().lower()
|
||
query.pop('q', None)
|
||
return redirect(f'{request.path}?{query.urlencode()}')
|
||
items = visible_items(request.user, workspace=active_workspace(request))
|
||
applied_preferences = False
|
||
if active_tags:
|
||
if active_tag_mode == 'all':
|
||
for tag_name in active_tags:
|
||
items = items.filter(tags__name=tag_name)
|
||
else:
|
||
items = items.filter(tags__name__in=active_tags).distinct()
|
||
if kind in dict(Item.Kind.choices):
|
||
items = items.filter(kind=kind)
|
||
elif not any([tag, status, q, day]):
|
||
applied_preferences = True
|
||
type_filter = Q()
|
||
if prefs.show_notes:
|
||
type_filter |= Q(kind=Item.Kind.NOTE)
|
||
if prefs.show_links:
|
||
type_filter |= Q(kind=Item.Kind.LINK)
|
||
if prefs.show_journal:
|
||
type_filter |= Q(kind=Item.Kind.JOURNAL)
|
||
if prefs.show_open_todos:
|
||
type_filter |= Q(kind=Item.Kind.TODO, is_done=False)
|
||
if prefs.show_done_todos:
|
||
type_filter |= Q(kind=Item.Kind.TODO, is_done=True)
|
||
items = items.filter(type_filter) if type_filter else items.none()
|
||
if day:
|
||
items = items.filter(created_at__date=day)
|
||
today_end = timezone.make_aware(datetime.combine(timezone.localdate(), time(23, 59, 59)))
|
||
upcoming_end = timezone.make_aware(datetime.combine(timezone.localdate() + timedelta(days=7), time(23, 59, 59)))
|
||
if status == 'open':
|
||
items = items.filter(kind=Item.Kind.TODO, is_done=False)
|
||
elif status == 'due':
|
||
items = items.filter(kind=Item.Kind.TODO, is_done=False, due_at__isnull=False, due_at__lte=upcoming_end).order_by('due_at', '-created_at')
|
||
elif status == 'archive':
|
||
items = items.filter(kind=Item.Kind.TODO, is_done=True)
|
||
elif not applied_preferences:
|
||
items = items.exclude(kind=Item.Kind.TODO, is_done=True)
|
||
if q:
|
||
items = items.filter(Q(content__icontains=q) | Q(url__icontains=q) | Q(tags__name__icontains=q)).distinct()
|
||
due_count = items.filter(kind=Item.Kind.TODO, is_done=False, due_at__lte=today_end).count()
|
||
due_now_items = items.filter(kind=Item.Kind.TODO, is_done=False, due_at__lte=today_end).order_by('due_at', '-created_at') if status == 'due' else Item.objects.none()
|
||
upcoming_due_items = items.filter(kind=Item.Kind.TODO, is_done=False, due_at__gt=today_end, due_at__lte=upcoming_end).order_by('due_at', '-created_at') if status == 'due' else Item.objects.none()
|
||
if prefs.due_first and status != 'archive':
|
||
items = items.annotate(due_rank=Case(
|
||
When(kind=Item.Kind.TODO, is_done=False, due_at__lte=today_end, then=Value(0)),
|
||
default=Value(1), output_field=IntegerField(),
|
||
)).order_by('due_rank', 'due_at', '-created_at')
|
||
return render(request, 'core/index.html', {
|
||
'form': QuickItemForm(), 'items': items, 'tags': index_filter_tags(request, request.user, prefs, active_tags, workspace=active_workspace(request)), 'composer_tags': visible_tags(request.user, workspace=active_workspace(request)), 'item_kind_choices': Item.Kind.choices, 'selected_item_kind': prefs.default_item_kind, 'due_now_items': due_now_items, 'upcoming_due_items': upcoming_due_items, 'active_tag': tag, 'active_tags': active_tags, 'active_tag_mode': active_tag_mode, 'active_kind': kind, 'active_status': status, 'active_day': day, 'q': q, 'due_count': due_count, 'overview_lines': prefs.overview_lines,
|
||
})
|
||
|
||
|
||
@login_required
|
||
def new_item(request):
|
||
prefs, created = UserPreference.objects.get_or_create(user=request.user)
|
||
requested_kind = request.GET.get('kind') or prefs.default_item_kind
|
||
next_url = request.GET.get('next') or request.POST.get('next') or reverse('index')
|
||
if next_url == 'index':
|
||
next_url = reverse('index')
|
||
collection = None
|
||
collection_id = request.GET.get('collection') or request.POST.get('collection')
|
||
if collection_id:
|
||
collection = get_object_or_404(visible_collections(request.user, workspace=active_workspace(request)).filter(mode__in=Collection.STRUCTURED_MODES), pk=collection_id)
|
||
if not user_can_edit_collection(request.user, collection):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
requested_kind = Item.Kind.NOTE
|
||
if request.method == 'POST':
|
||
form = QuickItemForm(request.POST, request.FILES)
|
||
if form.is_valid():
|
||
raw = form.cleaned_data['content']
|
||
if requested_kind in {'todo', 'link', 'journal'} and not TYPE_COMMAND_RE.match(raw.strip()):
|
||
raw = f'/{requested_kind} {raw}'
|
||
raw, due_at, _ = parse_due_command(raw)
|
||
kind, visibility, content, url = parse_quick_content(raw, requested_default_item_kind(request.POST, prefs, save=True))
|
||
item = Item.objects.create(workspace=active_workspace(request), owner=request.user, kind=kind, visibility=visibility, content=content, url=url, due_at=due_at)
|
||
attach_files_and_replace_tokens(item, request.FILES.getlist('files'))
|
||
item.sync_metadata()
|
||
apply_team_from_tags(item, request.user, force_private=bool(re.search(r'/private\b', raw, re.I)))
|
||
if collection:
|
||
item = adapt_item_for_collection(item, collection)
|
||
add_collection_section(collection, item)
|
||
return redirect(next_url)
|
||
return render(request, 'core/new_item.html', {
|
||
'form': QuickItemForm(), 'tags': visible_tags(request.user, workspace=active_workspace(request)), 'requested_kind': requested_kind, 'next_url': next_url, 'collection': collection,
|
||
})
|
||
|
||
|
||
@login_required
|
||
def journal(request):
|
||
if request.method == 'POST':
|
||
form = QuickItemForm(request.POST, request.FILES)
|
||
if form.is_valid():
|
||
raw = form.cleaned_data['content']
|
||
if not TYPE_COMMAND_RE.match(raw.strip()):
|
||
raw = f'/journal {raw}'
|
||
raw, due_at, _ = parse_due_command(raw)
|
||
kind, visibility, content, url = parse_quick_content(raw)
|
||
item = Item.objects.create(workspace=active_workspace(request), owner=request.user, kind=kind, visibility=visibility, content=content, url=url, due_at=due_at)
|
||
attach_files_and_replace_tokens(item, request.FILES.getlist('files'))
|
||
item.sync_metadata()
|
||
apply_team_from_tags(item, request.user, force_private=bool(re.search(r'/private\b', raw, re.I)))
|
||
if request.headers.get('HX-Request'):
|
||
return render(request, 'core/_item.html', {'item': item, 'overview_lines': request.user.preferences.overview_lines if hasattr(request.user, 'preferences') else 4})
|
||
return redirect('journal')
|
||
|
||
today = date.today()
|
||
selected_day = request.GET.get('day') or today.isoformat()
|
||
selected = date.fromisoformat(selected_day)
|
||
year = int(request.GET.get('year', selected.year))
|
||
month = int(request.GET.get('month', selected.month))
|
||
base = visible_items(request.user, workspace=active_workspace(request))
|
||
journal_entries = base.filter(kind=Item.Kind.JOURNAL, created_at__date=selected)
|
||
created_entries = base.filter(kind__in=[Item.Kind.NOTE, Item.Kind.LINK], created_at__date=selected)
|
||
completed_todos = base.filter(kind=Item.Kind.TODO, completed_at__date=selected)
|
||
items = (journal_entries | created_entries | completed_todos).distinct().order_by('-created_at')
|
||
activity_dates = set(base.filter(kind__in=[Item.Kind.JOURNAL, Item.Kind.NOTE, Item.Kind.LINK], created_at__year=year, created_at__month=month).dates('created_at', 'day'))
|
||
activity_dates |= set(base.filter(kind=Item.Kind.TODO, completed_at__year=year, completed_at__month=month).dates('completed_at', 'day'))
|
||
weeks = [[{'day': d, 'date': date(year, month, d).isoformat() if d else '', 'has_entries': d and date(year, month, d) in activity_dates} for d in week] for week in Calendar(firstweekday=0).monthdayscalendar(year, month)]
|
||
prev_month, prev_year = (12, year - 1) if month == 1 else (month - 1, year)
|
||
next_month, next_year = (1, year + 1) if month == 12 else (month + 1, year)
|
||
prefs, created = UserPreference.objects.get_or_create(user=request.user)
|
||
return render(request, 'core/journal.html', {
|
||
'form': QuickItemForm(), 'items': items, 'tags': visible_tags(request.user, workspace=active_workspace(request)), 'weeks': weeks, 'year': year, 'month': month, 'selected_day': selected_day,
|
||
'prev_year': prev_year, 'prev_month': prev_month, 'next_year': next_year, 'next_month': next_month, 'overview_lines': prefs.overview_lines,
|
||
})
|
||
|
||
|
||
@require_POST
|
||
def set_user_language(request):
|
||
language = request.POST.get('language')
|
||
next_url = request.POST.get('next') or '/'
|
||
if language not in {'de', 'en', ''}:
|
||
language = ''
|
||
if request.user.is_authenticated:
|
||
prefs, created = UserPreference.objects.get_or_create(user=request.user)
|
||
prefs.language = language
|
||
prefs.save(update_fields=['language'])
|
||
response = redirect(next_url)
|
||
if language:
|
||
translation.activate(language)
|
||
response.set_cookie(settings.LANGUAGE_COOKIE_NAME, language)
|
||
else:
|
||
response.delete_cookie(settings.LANGUAGE_COOKIE_NAME)
|
||
return response
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def workspace_switch(request, pk):
|
||
membership = get_object_or_404(WorkspaceMembership, workspace_id=pk, user=request.user)
|
||
request.session['workspace_id'] = membership.workspace_id
|
||
next_url = request.POST.get('next') or reverse('index')
|
||
if not url_has_allowed_host_and_scheme(next_url, allowed_hosts={request.get_host()}):
|
||
next_url = reverse('index')
|
||
return redirect(next_url)
|
||
|
||
|
||
@login_required
|
||
def settings_view(request):
|
||
workspace = active_workspace(request)
|
||
if request.method == 'POST':
|
||
if request.POST.get('action') == 'profile':
|
||
form = UserSettingsForm(request.POST, instance=request.user)
|
||
if form.is_valid():
|
||
form.save()
|
||
messages.success(request, _('Settings saved.'))
|
||
return redirect('settings')
|
||
elif request.POST.get('action') == 'preferences':
|
||
prefs, created = UserPreference.objects.get_or_create(user=request.user)
|
||
pref_form = UserPreferenceForm(request.POST, instance=prefs)
|
||
if pref_form.is_valid():
|
||
pref_form.save()
|
||
messages.success(request, _('Default view saved.'))
|
||
return redirect('settings')
|
||
elif request.POST.get('action') == 'workspace':
|
||
workspace_form = WorkspaceForm(request.POST)
|
||
if workspace_form.is_valid():
|
||
workspace = workspace_form.save(commit=False)
|
||
workspace.owner = request.user
|
||
workspace.save()
|
||
WorkspaceMembership.objects.create(workspace=workspace, user=request.user, role=WorkspaceMembership.Role.OWNER)
|
||
request.session['workspace_id'] = workspace.id
|
||
messages.success(request, _('Workspace created.'))
|
||
return redirect('settings')
|
||
elif request.POST.get('action') == 'apikey':
|
||
key_form = ApiKeyForm(request.POST, user=request.user, workspace=workspace)
|
||
if key_form.is_valid():
|
||
api_key = key_form.save(commit=False)
|
||
api_key.workspace = workspace
|
||
api_key.owner = request.user
|
||
api_key.save()
|
||
key_form.save_m2m()
|
||
messages.success(request, _('API key created: %(token)s') % {'token': api_key.token})
|
||
return redirect('settings')
|
||
elif request.POST.get('action') == 'userinvite':
|
||
invite_form = UserInviteForm(request.POST)
|
||
if invite_form.is_valid():
|
||
invite = invite_form.save(commit=False)
|
||
invite.invited_by = request.user
|
||
invite.save()
|
||
messages.success(request, _('Invitation link created: %(url)s') % {'url': request.build_absolute_uri(reverse("accept_user_invite", args=[invite.token]))})
|
||
return redirect('settings')
|
||
prefs, created = UserPreference.objects.get_or_create(user=request.user)
|
||
return render(request, 'core/settings.html', {
|
||
'profile_form': UserSettingsForm(instance=request.user),
|
||
'preference_form': UserPreferenceForm(instance=prefs),
|
||
'workspace_form': WorkspaceForm(),
|
||
'workspace_memberships': user_workspaces(request.user),
|
||
'key_form': ApiKeyForm(user=request.user, workspace=workspace),
|
||
'user_invite_form': UserInviteForm(),
|
||
'user_invites': request.user.sent_user_invites.filter(accepted_at__isnull=True),
|
||
'api_keys': request.user.api_keys.prefetch_related('tags'),
|
||
})
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def delete_api_key(request, pk):
|
||
api_key = get_object_or_404(ApiKey, pk=pk, owner=request.user)
|
||
api_key.delete()
|
||
return redirect('settings')
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def delete_user_invite(request, pk):
|
||
invite = get_object_or_404(UserInvite, pk=pk, invited_by=request.user, accepted_at__isnull=True)
|
||
invite.delete()
|
||
if request.headers.get('HX-Request'):
|
||
return HttpResponse('')
|
||
messages.success(request, _('Invitation deleted.'))
|
||
return redirect('settings')
|
||
|
||
|
||
def accept_user_invite(request, token):
|
||
invite = get_object_or_404(UserInvite, token=token, accepted_at__isnull=True)
|
||
if request.user.is_authenticated:
|
||
invite.accepted_at = timezone.now()
|
||
invite.save(update_fields=['accepted_at'])
|
||
messages.success(request, _('Invitation accepted.'))
|
||
return redirect('index')
|
||
if request.method == 'POST':
|
||
form = InviteRegistrationForm(request.POST)
|
||
if form.is_valid():
|
||
user = form.save(commit=False)
|
||
if invite.email and not user.email:
|
||
user.email = invite.email
|
||
user.save()
|
||
invite.accepted_at = timezone.now()
|
||
invite.save(update_fields=['accepted_at'])
|
||
login(request, user)
|
||
messages.success(request, _('Welcome to my2dos.'))
|
||
return redirect('index')
|
||
else:
|
||
form = InviteRegistrationForm(initial={'email': invite.email})
|
||
return render(request, 'core/accept_user_invite.html', {'invite': invite, 'form': form})
|
||
|
||
|
||
@login_required
|
||
def team_list(request):
|
||
if request.method == 'POST':
|
||
team_form = TeamForm(request.POST, user=request.user, workspace=active_workspace(request))
|
||
if team_form.is_valid():
|
||
team = team_form.save(commit=False)
|
||
team.owner = request.user
|
||
team.save()
|
||
TeamMembership.objects.create(team=team, user=request.user, role=TeamMembership.Role.OWNER)
|
||
Tag.objects.get_or_create(workspace=team.workspace, name=team.slug)
|
||
messages.success(request, _('Team %(team)s created. Use #%(slug)s for team items.') % {'team': team.name, 'slug': team.slug})
|
||
return redirect('team_detail', pk=team.pk)
|
||
workspace = active_workspace(request)
|
||
teams = Team.objects.filter(workspace=workspace, memberships__user=request.user).distinct().prefetch_related('memberships__user')
|
||
stats = []
|
||
base_items = visible_items(request.user, workspace=active_workspace(request))
|
||
for team in teams:
|
||
team_items = base_items.filter(team=team)
|
||
stats.append({
|
||
'team': team,
|
||
'members_count': team.memberships.count(),
|
||
'items_count': team_items.count(),
|
||
'open_todos_count': team_items.filter(kind=Item.Kind.TODO, is_done=False).count(),
|
||
'recent_items': team_items[:3],
|
||
})
|
||
return render(request, 'core/team_list.html', {'stats': stats, 'team_form': TeamForm(user=request.user, workspace=active_workspace(request)), 'overview_lines': request.user.preferences.overview_lines if hasattr(request.user, 'preferences') else 4})
|
||
|
||
|
||
@login_required
|
||
def team_detail(request, pk):
|
||
membership = get_object_or_404(TeamMembership.objects.select_related('team'), team_id=pk, user=request.user, team__workspace=active_workspace(request))
|
||
team = membership.team
|
||
if request.method == 'POST' and membership.role in [TeamMembership.Role.OWNER, TeamMembership.Role.ADMIN]:
|
||
invite_form = TeamInviteForm(request.POST)
|
||
if invite_form.is_valid():
|
||
invite = invite_form.save(commit=False)
|
||
invite.team = team
|
||
invite.invited_by = request.user
|
||
invite.save()
|
||
messages.success(request, _('Invitation link created: %(url)s') % {'url': request.build_absolute_uri(reverse("team_accept_invite", args=[invite.token]))})
|
||
return redirect('team_detail', pk=team.pk)
|
||
items = visible_items(request.user, workspace=active_workspace(request)).filter(team=team)
|
||
return render(request, 'core/team_detail.html', {
|
||
'team': team,
|
||
'membership': membership,
|
||
'memberships': team.memberships.select_related('user'),
|
||
'invites': team.invites.filter(accepted_at__isnull=True),
|
||
'invite_form': TeamInviteForm(),
|
||
'items': items,
|
||
'overview_lines': request.user.preferences.overview_lines if hasattr(request.user, 'preferences') else 4,
|
||
})
|
||
|
||
|
||
@login_required
|
||
def team_accept_invite(request, token):
|
||
invite = get_object_or_404(TeamInvite.objects.select_related('team'), token=token, accepted_at__isnull=True)
|
||
WorkspaceMembership.objects.get_or_create(workspace=invite.team.workspace, user=request.user, defaults={'role': WorkspaceMembership.Role.MEMBER})
|
||
TeamMembership.objects.get_or_create(team=invite.team, user=request.user, defaults={'role': TeamMembership.Role.MEMBER})
|
||
request.session['workspace_id'] = invite.team.workspace_id
|
||
invite.accepted_at = timezone.now()
|
||
invite.save(update_fields=['accepted_at'])
|
||
messages.success(request, _('You are now a member of team %(team)s.') % {'team': invite.team.name})
|
||
return redirect('team_detail', pk=invite.team.pk)
|
||
|
||
|
||
def visible_collections(user, workspace=None):
|
||
workspace = current_workspace(user, workspace=workspace)
|
||
qs = Collection.objects.select_related('owner', 'team', 'workspace').prefetch_related('filter_tags', 'sections__item', 'sections__sub_collection')
|
||
if workspace:
|
||
qs = qs.filter(workspace=workspace)
|
||
if user.is_authenticated:
|
||
return qs.filter(Q(owner=user) | Q(team_id__in=user_team_ids(user, workspace)) | Q(visibility=Item.Visibility.PUBLIC)).distinct()
|
||
return qs.none()
|
||
|
||
|
||
def user_can_edit_collection(user, collection):
|
||
return user.is_authenticated and (collection.owner_id == user.id or (collection.team_id and TeamMembership.objects.filter(team=collection.team, user=user).exists()))
|
||
|
||
|
||
def user_can_delete_collection(user, collection):
|
||
return user.is_authenticated and (collection.owner_id == user.id or (collection.team_id and TeamMembership.objects.filter(team=collection.team, user=user, role__in=[TeamMembership.Role.OWNER, TeamMembership.Role.ADMIN]).exists()))
|
||
|
||
|
||
def collection_scope_allows_item(collection_visibility, collection_team_id, collection_owner_id, item_visibility, item_team_id, item_owner_id):
|
||
# A collection may reference a note that is already visible to a wider audience.
|
||
if item_visibility == Item.Visibility.PUBLIC:
|
||
return True
|
||
if collection_visibility == Item.Visibility.PUBLIC:
|
||
return False
|
||
if collection_visibility == Item.Visibility.TEAM:
|
||
return item_visibility == Item.Visibility.TEAM and item_team_id == collection_team_id
|
||
return item_visibility == Item.Visibility.PRIVATE and item_owner_id == collection_owner_id
|
||
|
||
|
||
def collection_item_is_compatible(collection, item):
|
||
return collection_scope_allows_item(
|
||
collection.visibility, collection.team_id, collection.owner_id,
|
||
item.visibility, item.team_id, item.owner_id,
|
||
)
|
||
|
||
|
||
def collection_subcollection_is_compatible(collection, sub_collection):
|
||
return collection_scope_allows_item(
|
||
collection.visibility, collection.team_id, collection.owner_id,
|
||
sub_collection.visibility, sub_collection.team_id, sub_collection.owner_id,
|
||
)
|
||
|
||
|
||
def collection_contains_collection(collection, target):
|
||
if collection.pk == target.pk:
|
||
return True
|
||
for section in collection.sections.select_related('sub_collection'):
|
||
if section.sub_collection and collection_contains_collection(section.sub_collection, target):
|
||
return True
|
||
return False
|
||
|
||
|
||
def collection_target_from_form(form, user):
|
||
team = form.cleaned_data.get('team')
|
||
if team:
|
||
return Item.Visibility.TEAM, team
|
||
return form.cleaned_data['visibility'], None
|
||
|
||
|
||
def collection_tag_items(collection):
|
||
items = Item.objects.filter(workspace=collection.workspace, kind__in=collection.included_item_kinds()).select_related('owner', 'team').prefetch_related('tags', 'attachments')
|
||
if collection.visibility == Item.Visibility.PUBLIC:
|
||
items = items.filter(visibility=Item.Visibility.PUBLIC)
|
||
elif collection.visibility == Item.Visibility.TEAM:
|
||
items = items.filter(Q(visibility=Item.Visibility.PUBLIC) | Q(visibility=Item.Visibility.TEAM, team=collection.team))
|
||
else:
|
||
items = items.filter(Q(visibility=Item.Visibility.PUBLIC) | Q(visibility=Item.Visibility.PRIVATE, owner=collection.owner))
|
||
for tag in collection.filter_tags.all():
|
||
items = items.filter(tags=tag)
|
||
return items.distinct().order_by('created_at', 'id')
|
||
|
||
|
||
def item_can_change_for_collection(user, item, collection):
|
||
if not user_can_edit_item(user, item):
|
||
return False
|
||
target_team_id = collection.team_id
|
||
for section in item.collection_sections.select_related('collection').exclude(collection=collection):
|
||
other = section.collection
|
||
if not collection_scope_allows_item(
|
||
other.visibility, other.team_id, other.owner_id,
|
||
collection.visibility, target_team_id, item.owner_id,
|
||
):
|
||
return False
|
||
return True
|
||
|
||
|
||
def remove_team_tags(content, workspace):
|
||
slugs = Team.objects.filter(workspace=workspace).values_list('slug', flat=True)
|
||
for slug in slugs:
|
||
content = re.sub(rf'(?<!\w)#{re.escape(slug)}\b', '', content, flags=re.I)
|
||
return re.sub(r'[ \t]{2,}', ' ', content).strip()
|
||
|
||
|
||
def adapt_item_for_collection(item, collection):
|
||
item.workspace = collection.workspace
|
||
item.content = remove_team_tags(item.content, collection.workspace)
|
||
if collection.team:
|
||
item.content = f'{item.content} #{collection.team.slug}'.strip()
|
||
item.visibility = collection.visibility
|
||
item.team = collection.team
|
||
item.save(update_fields=['workspace', 'content', 'visibility', 'team', 'updated_at'])
|
||
item.sync_metadata()
|
||
item.team = collection.team
|
||
item.visibility = collection.visibility
|
||
item.save(update_fields=['team', 'visibility', 'updated_at'])
|
||
return item
|
||
|
||
|
||
def copy_item_for_collection(item, collection, owner):
|
||
copied = Item.objects.create(
|
||
workspace=collection.workspace, owner=owner, team=collection.team, kind=Item.Kind.NOTE, content=item.content,
|
||
comment=item.comment, visibility=collection.visibility, url=item.url,
|
||
)
|
||
for attachment in item.attachments.all():
|
||
Attachment.objects.create(item=copied, file=attachment.file.name)
|
||
return adapt_item_for_collection(copied, collection)
|
||
|
||
|
||
def next_collection_section_position(collection):
|
||
last_position = collection.sections.order_by('-position').values_list('position', flat=True).first()
|
||
return (last_position + 1) if last_position is not None else 0
|
||
|
||
|
||
def add_collection_section(collection, item):
|
||
return CollectionSection.objects.get_or_create(
|
||
collection=collection, item=item,
|
||
defaults={'position': next_collection_section_position(collection)},
|
||
)
|
||
|
||
|
||
def add_collection_subcollection(collection, sub_collection):
|
||
return CollectionSection.objects.get_or_create(
|
||
collection=collection, sub_collection=sub_collection,
|
||
defaults={'position': next_collection_section_position(collection)},
|
||
)
|
||
|
||
|
||
def editable_collection_notes(user, workspace=None):
|
||
workspace = current_workspace(user, workspace=workspace)
|
||
return Item.objects.select_related('owner', 'team').prefetch_related('tags', 'attachments').filter(
|
||
workspace=workspace, kind=Item.Kind.NOTE,
|
||
).filter(Q(owner=user) | Q(team_id__in=user_team_ids(user, workspace))).distinct()
|
||
|
||
|
||
def collection_is_structured_mode(mode):
|
||
return mode in Collection.STRUCTURED_MODES
|
||
|
||
|
||
def collection_detail_context(request, collection, pending_item=None):
|
||
can_edit = user_can_edit_collection(request.user, collection)
|
||
q = request.GET.get('q', '').strip()
|
||
candidates = Item.objects.none()
|
||
collection_candidates = Collection.objects.none()
|
||
if can_edit:
|
||
used_item_ids = set(collection.sections.filter(item__isnull=False).values_list('item_id', flat=True))
|
||
used_collection_ids = set(collection.sections.filter(sub_collection__isnull=False).values_list('sub_collection_id', flat=True))
|
||
candidates = editable_collection_notes(request.user, workspace=collection.workspace).annotate(collection_use_count=Count('collection_sections', distinct=True))
|
||
collection_candidates = visible_collections(request.user, workspace=active_workspace(request)).filter(mode__in=Collection.STRUCTURED_MODES).exclude(pk=collection.pk).annotate(collection_use_count=Count('containing_sections', distinct=True))
|
||
collection_candidates = [candidate for candidate in collection_candidates if collection_subcollection_is_compatible(collection, candidate) and not collection_contains_collection(candidate, collection)]
|
||
if q:
|
||
candidates = candidates.filter(Q(content__icontains=q) | Q(tags__name__icontains=q)).distinct()
|
||
collection_candidates = [candidate for candidate in collection_candidates if q.lower() in candidate.title.lower()]
|
||
candidates = list(candidates[:30])
|
||
for candidate in candidates:
|
||
candidate.used_in_current_collection = candidate.id in used_item_ids
|
||
collection_candidates = collection_candidates[:30]
|
||
for candidate in collection_candidates:
|
||
candidate.used_in_current_collection = candidate.id in used_collection_ids
|
||
return {
|
||
'collection': collection,
|
||
'sections': collection.sections.select_related('item', 'sub_collection').prefetch_related('item__tags', 'item__attachments', 'sub_collection__sections__item', 'sub_collection__sections__sub_collection'),
|
||
'can_edit': can_edit,
|
||
'can_delete': user_can_delete_collection(request.user, collection),
|
||
'candidates': candidates,
|
||
'collection_candidates': collection_candidates,
|
||
'q': q,
|
||
'pending_item': pending_item,
|
||
'pending_can_convert': pending_item and item_can_change_for_collection(request.user, pending_item, collection),
|
||
}
|
||
|
||
|
||
@login_required
|
||
def collection_list(request):
|
||
active_mode = request.GET.get('mode', 'listed')
|
||
collections = visible_collections(request.user, workspace=active_workspace(request)).prefetch_related('filter_tags')
|
||
if active_mode == 'manual':
|
||
collections = collections.filter(mode=Collection.Mode.MANUAL)
|
||
elif active_mode == 'tutorial':
|
||
collections = collections.filter(mode=Collection.Mode.TUTORIAL)
|
||
elif active_mode == 'tags':
|
||
collections = collections.filter(mode=Collection.Mode.TAGS)
|
||
elif active_mode == 'chapter':
|
||
collections = collections.filter(mode=Collection.Mode.CHAPTER)
|
||
else:
|
||
active_mode = 'listed'
|
||
collections = collections.exclude(mode=Collection.Mode.CHAPTER)
|
||
entries = []
|
||
for collection in collections:
|
||
count = collection_tag_items(collection).count() if collection.mode == Collection.Mode.TAGS else collection.sections.count()
|
||
entries.append({'collection': collection, 'section_count': count})
|
||
orphan_chapters = visible_collections(request.user, workspace=active_workspace(request)).filter(mode=Collection.Mode.CHAPTER, containing_sections__isnull=True) if active_mode != 'chapter' else Collection.objects.none()
|
||
return render(request, 'core/collection_list.html', {'entries': entries, 'active_mode': active_mode, 'orphan_chapters': orphan_chapters})
|
||
|
||
|
||
@login_required
|
||
def collection_create(request):
|
||
if request.method == 'POST':
|
||
form = CollectionForm(request.POST, user=request.user, workspace=active_workspace(request))
|
||
if form.is_valid():
|
||
visibility, team = collection_target_from_form(form, request.user)
|
||
if visibility == Item.Visibility.TEAM and not team:
|
||
form.add_error('team', _('Select a team for a team collection.'))
|
||
else:
|
||
collection = form.save(commit=False)
|
||
collection.workspace = active_workspace(request)
|
||
collection.owner = request.user
|
||
collection.visibility = visibility
|
||
collection.team = team
|
||
collection.save()
|
||
form.save_m2m()
|
||
if collection.mode != Collection.Mode.TAGS:
|
||
collection.filter_tags.clear()
|
||
messages.success(request, _('Collection created.'))
|
||
return redirect(collection)
|
||
else:
|
||
form = CollectionForm(user=request.user, workspace=active_workspace(request))
|
||
return render(request, 'core/collection_form.html', {'form': form, 'heading': _('Create collection')})
|
||
|
||
|
||
@login_required
|
||
def collection_edit(request, pk):
|
||
collection = get_object_or_404(visible_collections(request.user, workspace=active_workspace(request)), pk=pk)
|
||
if not user_can_edit_collection(request.user, collection):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
if request.method == 'POST':
|
||
form = CollectionForm(request.POST, instance=collection, user=request.user)
|
||
if form.is_valid():
|
||
visibility, team = collection_target_from_form(form, request.user)
|
||
target_mode = form.cleaned_data['mode']
|
||
has_manual_sections = collection_is_structured_mode(collection.mode) and collection.sections.exists()
|
||
incompatible = collection_is_structured_mode(target_mode) and any(
|
||
not collection_scope_allows_item(
|
||
visibility, team.id if team else None, collection.owner_id,
|
||
section.item.visibility, section.item.team_id, section.item.owner_id,
|
||
) if section.item else not collection_scope_allows_item(
|
||
visibility, team.id if team else None, collection.owner_id,
|
||
section.sub_collection.visibility, section.sub_collection.team_id, section.sub_collection.owner_id,
|
||
)
|
||
for section in collection.sections.select_related('item', 'sub_collection')
|
||
)
|
||
if target_mode == Collection.Mode.TAGS and has_manual_sections:
|
||
form.add_error('mode', _('Remove all manual sections before changing to a tag collection.'))
|
||
elif incompatible:
|
||
form.add_error(None, _('The collection visibility cannot change while it contains incompatible notes.'))
|
||
elif visibility == Item.Visibility.TEAM and not team:
|
||
form.add_error('team', _('Select a team for a team collection.'))
|
||
else:
|
||
collection = form.save(commit=False)
|
||
collection.visibility = visibility
|
||
collection.team = team
|
||
collection.save()
|
||
form.save_m2m()
|
||
if collection.mode != Collection.Mode.TAGS:
|
||
collection.filter_tags.clear()
|
||
messages.success(request, _('Collection saved.'))
|
||
return redirect(collection)
|
||
else:
|
||
form = CollectionForm(instance=collection, user=request.user)
|
||
return render(request, 'core/collection_form.html', {'form': form, 'collection': collection, 'heading': _('Edit collection')})
|
||
|
||
|
||
def collection_detail(request, pk):
|
||
collection = get_object_or_404(visible_collections(request.user, workspace=active_workspace(request)), pk=pk)
|
||
tag_items = list(collection_tag_items(collection)) if collection.mode == Collection.Mode.TAGS else []
|
||
for item in tag_items:
|
||
item.can_toggle_done = item.kind == Item.Kind.TODO and user_can_edit_item(request.user, item)
|
||
return render(request, 'core/collection_detail.html', {
|
||
'collection': collection,
|
||
'sections': collection.sections.select_related('item', 'sub_collection').prefetch_related('item__tags', 'item__attachments', 'sub_collection__sections__item', 'sub_collection__sections__sub_collection') if collection.is_structured else [],
|
||
'tag_items': tag_items,
|
||
'can_edit': user_can_edit_collection(request.user, collection),
|
||
})
|
||
|
||
|
||
@login_required
|
||
def collection_manage(request, pk):
|
||
collection = get_object_or_404(visible_collections(request.user, workspace=active_workspace(request)), pk=pk)
|
||
if not user_can_edit_collection(request.user, collection):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
if collection.mode == Collection.Mode.TAGS:
|
||
return redirect('collection_edit', pk=collection.pk)
|
||
manage_url = reverse('collection_manage', args=[collection.pk])
|
||
q = request.GET.get('q', '').strip()
|
||
redirect_url = f'{manage_url}?q={q}' if q else manage_url
|
||
if request.method == 'POST':
|
||
action = request.POST.get('action')
|
||
if action in {'add', 'confirm_add'}:
|
||
item = get_object_or_404(editable_collection_notes(request.user, workspace=collection.workspace), pk=request.POST.get('item_id'))
|
||
if collection_item_is_compatible(collection, item):
|
||
add_collection_section(collection, item)
|
||
return redirect(redirect_url)
|
||
if action == 'add':
|
||
return render(request, 'core/collection_manage.html', collection_detail_context(request, collection, item))
|
||
mode = request.POST.get('mode')
|
||
if mode == 'convert':
|
||
if not item_can_change_for_collection(request.user, item, collection):
|
||
messages.error(request, _('This note is used by an incompatible collection or cannot be edited. Create a copy instead.'))
|
||
return redirect(redirect_url)
|
||
item = adapt_item_for_collection(item, collection)
|
||
elif mode == 'copy':
|
||
item = copy_item_for_collection(item, collection, request.user)
|
||
else:
|
||
return redirect(redirect_url)
|
||
add_collection_section(collection, item)
|
||
return redirect(redirect_url)
|
||
if action == 'create_chapter':
|
||
title = request.POST.get('title', '').strip()[:200]
|
||
if title:
|
||
chapter = Collection.objects.create(
|
||
workspace=collection.workspace, owner=request.user, title=title, mode=Collection.Mode.CHAPTER,
|
||
visibility=collection.visibility, team=collection.team,
|
||
)
|
||
add_collection_subcollection(collection, chapter)
|
||
return redirect(reverse('collection_manage', args=[chapter.pk]))
|
||
messages.error(request, _('Enter a chapter title.'))
|
||
return redirect(redirect_url)
|
||
if action == 'add_collection':
|
||
sub_collection = get_object_or_404(visible_collections(request.user, workspace=active_workspace(request)).filter(mode__in=Collection.STRUCTURED_MODES), pk=request.POST.get('collection_id'))
|
||
if sub_collection.pk == collection.pk or collection_contains_collection(sub_collection, collection):
|
||
messages.error(request, _('This would create a collection loop.'))
|
||
elif not collection_subcollection_is_compatible(collection, sub_collection):
|
||
messages.error(request, _('This collection has an incompatible visibility.'))
|
||
else:
|
||
add_collection_subcollection(collection, sub_collection)
|
||
return redirect(redirect_url)
|
||
section = get_object_or_404(CollectionSection, pk=request.POST.get('section_id'), collection=collection)
|
||
if action == 'remove':
|
||
section.delete()
|
||
for index, remaining in enumerate(collection.sections.order_by('position', 'id')):
|
||
if remaining.position != index:
|
||
CollectionSection.objects.filter(pk=remaining.pk).update(position=index)
|
||
elif action == 'title':
|
||
section.title = request.POST.get('title', '').strip()[:200]
|
||
section.save(update_fields=['title'])
|
||
elif action in {'up', 'down'}:
|
||
direction = -1 if action == 'up' else 1
|
||
other = collection.sections.filter(position=section.position + direction).first()
|
||
if other:
|
||
with transaction.atomic():
|
||
temporary = collection.sections.order_by('-position').values_list('position', flat=True).first() + 1000
|
||
CollectionSection.objects.filter(pk=section.pk).update(position=temporary)
|
||
CollectionSection.objects.filter(pk=other.pk).update(position=section.position)
|
||
CollectionSection.objects.filter(pk=section.pk).update(position=other.position)
|
||
elif action == 'move_to':
|
||
try:
|
||
new_position = int(request.POST.get('position', section.position))
|
||
except (TypeError, ValueError):
|
||
new_position = section.position
|
||
ordered = list(collection.sections.order_by('position', 'id'))
|
||
ordered.remove(section)
|
||
ordered.insert(max(0, min(new_position, len(ordered))), section)
|
||
with transaction.atomic():
|
||
for index, current in enumerate(ordered):
|
||
CollectionSection.objects.filter(pk=current.pk).update(position=1000 + index)
|
||
for index, current in enumerate(ordered):
|
||
CollectionSection.objects.filter(pk=current.pk).update(position=index)
|
||
return redirect(redirect_url)
|
||
return render(request, 'core/collection_manage.html', collection_detail_context(request, collection))
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def collection_delete(request, pk):
|
||
collection = get_object_or_404(visible_collections(request.user, workspace=active_workspace(request)), pk=pk)
|
||
if not user_can_delete_collection(request.user, collection):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
collection.delete()
|
||
messages.success(request, _('Collection deleted.'))
|
||
return redirect('collection_list')
|
||
|
||
|
||
@login_required
|
||
def item_detail(request, pk):
|
||
item = get_object_or_404(visible_items(request.user, workspace=active_workspace(request)), pk=pk)
|
||
return render(request, 'core/detail.html', {'item': item})
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def toggle_done(request, pk):
|
||
item = get_object_or_404(visible_items(request.user, workspace=active_workspace(request)), pk=pk, kind=Item.Kind.TODO)
|
||
if not user_can_edit_item(request.user, item):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
item.is_done = not item.is_done
|
||
item.completed_at = timezone.now() if item.is_done else None
|
||
item.save(update_fields=['is_done', 'completed_at', 'updated_at'])
|
||
next_url = request.POST.get('next') or request.GET.get('next')
|
||
if next_url and url_has_allowed_host_and_scheme(next_url, allowed_hosts={request.get_host()}):
|
||
return redirect(next_url)
|
||
if request.headers.get('HX-Request') and item.is_done and request.GET.get('status') != 'archive':
|
||
return HttpResponse('')
|
||
return render(request, 'core/_item.html', {'item': item})
|
||
|
||
|
||
@login_required
|
||
def item_card(request, pk):
|
||
item = get_object_or_404(visible_items(request.user, workspace=active_workspace(request)), pk=pk)
|
||
return render(request, 'core/_item.html', {'item': item})
|
||
|
||
|
||
@login_required
|
||
def edit_item(request, pk):
|
||
item = get_object_or_404(visible_items(request.user, workspace=active_workspace(request)), pk=pk)
|
||
if not user_can_edit_item(request.user, item):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
if request.method == 'POST':
|
||
form = EditItemForm(request.POST, request.FILES, instance=item)
|
||
if form.is_valid():
|
||
item = form.save(commit=False)
|
||
item.content, command_due_at, has_due_command = parse_due_command(item.content)
|
||
if item.kind == Item.Kind.TODO:
|
||
item.due_at = command_due_at if has_due_command else parse_due_at_from_post(request.POST)
|
||
target_visibility, target_team = desired_item_scope(item, request.user, force_private=item.visibility == Item.Visibility.PRIVATE)
|
||
if not item_scope_fits_collections(item, target_visibility, target_team):
|
||
form.add_error('visibility', _('This visibility is incompatible with a collection that uses the note.'))
|
||
return render(request, 'core/_edit_form.html', {'item': item, 'form': form, 'tags': visible_tags(request.user, workspace=active_workspace(request))}, status=409)
|
||
item.visibility, item.team = target_visibility, target_team
|
||
item.save()
|
||
form.save_m2m()
|
||
delete_selected_attachments(item, request.POST.getlist('delete_attachments'))
|
||
attach_files_and_replace_tokens(item, request.FILES.getlist('files'))
|
||
item.sync_metadata()
|
||
apply_team_from_tags(item, request.user, force_private=item.visibility == Item.Visibility.PRIVATE)
|
||
response = render(request, 'core/_item.html', {'item': item, 'overview_lines': request.user.preferences.overview_lines if hasattr(request.user, 'preferences') else 4})
|
||
response['HX-Trigger'] = 'tagsChanged'
|
||
return response
|
||
return render(request, 'core/_edit_form.html', {'item': item, 'tags': visible_tags(request.user, workspace=active_workspace(request))})
|
||
|
||
|
||
@login_required
|
||
def item_editor(request, pk):
|
||
item = get_object_or_404(visible_items(request.user, workspace=active_workspace(request)), pk=pk)
|
||
if not user_can_edit_item(request.user, item):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
next_url = request.GET.get('next') or item.get_absolute_url()
|
||
if request.method == 'POST':
|
||
form = EditItemForm(request.POST, request.FILES, instance=item)
|
||
if form.is_valid():
|
||
item = form.save(commit=False)
|
||
item.content, command_due_at, has_due_command = parse_due_command(item.content)
|
||
if item.kind == Item.Kind.TODO:
|
||
item.due_at = command_due_at if has_due_command else parse_due_at_from_post(request.POST)
|
||
target_visibility, target_team = desired_item_scope(item, request.user, force_private=item.visibility == Item.Visibility.PRIVATE)
|
||
if not item_scope_fits_collections(item, target_visibility, target_team):
|
||
form.add_error('visibility', _('This visibility is incompatible with a collection that uses the note.'))
|
||
return render(request, 'core/item_editor.html', {'item': item, 'form': form, 'tags': visible_tags(request.user, workspace=active_workspace(request)), 'next_url': next_url}, status=409)
|
||
item.visibility, item.team = target_visibility, target_team
|
||
item.save()
|
||
form.save_m2m()
|
||
delete_selected_attachments(item, request.POST.getlist('delete_attachments'))
|
||
attach_files_and_replace_tokens(item, request.FILES.getlist('files'))
|
||
item.sync_metadata()
|
||
apply_team_from_tags(item, request.user, force_private=item.visibility == Item.Visibility.PRIVATE)
|
||
return redirect(next_url)
|
||
return render(request, 'core/item_editor.html', {'item': item, 'tags': visible_tags(request.user, workspace=active_workspace(request)), 'next_url': next_url})
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def delete_item(request, pk):
|
||
item = get_object_or_404(visible_items(request.user, workspace=active_workspace(request)), pk=pk)
|
||
if not user_can_delete_item(request.user, item):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
try:
|
||
item.delete()
|
||
except ProtectedError:
|
||
collections = list(item.collection_sections.select_related('collection').values_list('collection__title', flat=True))
|
||
return JsonResponse({'error': _('This note is used in collections and cannot be deleted.'), 'collections': collections}, status=409)
|
||
return JsonResponse({'ok': True})
|
||
|
||
|
||
@login_required
|
||
def kanban_list(request):
|
||
if request.method == 'POST':
|
||
form = KanbanForm(request.POST, user=request.user, workspace=active_workspace(request))
|
||
if form.is_valid():
|
||
kanban = form.save(commit=False)
|
||
kanban.owner = request.user
|
||
kanban.save()
|
||
return redirect('kanban_detail', pk=kanban.pk)
|
||
workspace = active_workspace(request)
|
||
kanbans = Kanban.objects.filter(workspace=workspace, owner=request.user).select_related('tag')
|
||
stats = []
|
||
for kanban in kanbans:
|
||
qs = Item.objects.filter(workspace=workspace, owner=request.user, kind=Item.Kind.TODO, tags=kanban.tag)
|
||
stats.append({'kanban': kanban, 'open': qs.filter(is_done=False).count(), 'done': qs.filter(is_done=True).count()})
|
||
return render(request, 'core/kanban_list.html', {'form': KanbanForm(user=request.user, workspace=active_workspace(request)), 'stats': stats})
|
||
|
||
|
||
@login_required
|
||
def kanban_detail(request, pk):
|
||
kanban = get_object_or_404(Kanban.objects.select_related('tag'), pk=pk, owner=request.user)
|
||
todos = Item.objects.filter(workspace=kanban.workspace, owner=request.user, kind=Item.Kind.TODO, tags=kanban.tag).prefetch_related('tags')
|
||
return render(request, 'core/kanban_detail.html', {
|
||
'kanban': kanban,
|
||
'inbox_items': todos.filter(is_done=False, kanban_status='inbox'),
|
||
'todo_items': todos.filter(is_done=False, kanban_status='todo'),
|
||
'progress_items': todos.filter(is_done=False, kanban_status='progress'),
|
||
'done_items': todos.filter(is_done=True),
|
||
})
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def kanban_move(request, pk, item_pk):
|
||
kanban = get_object_or_404(Kanban, pk=pk, owner=request.user)
|
||
item = get_object_or_404(Item, workspace=kanban.workspace, pk=item_pk, owner=request.user, kind=Item.Kind.TODO, tags=kanban.tag)
|
||
status = request.POST.get('status')
|
||
if status not in {'inbox', 'todo', 'progress', 'done'}:
|
||
return JsonResponse({'error': 'invalid status'}, status=400)
|
||
item.kanban_status = status
|
||
item.is_done = status == 'done'
|
||
item.completed_at = timezone.now() if item.is_done else None
|
||
item.save(update_fields=['kanban_status', 'is_done', 'completed_at', 'updated_at'])
|
||
return JsonResponse({'ok': True})
|
||
|
||
|
||
@login_required
|
||
def tag_list(request):
|
||
prefs, created = UserPreference.objects.get_or_create(user=request.user)
|
||
active_tags = [t.strip().lower() for t in request.GET.getlist('tag') if t.strip()]
|
||
active_tag_mode = 'all' if request.GET.get('tag_mode') == 'all' else 'any'
|
||
return render(request, 'core/_tags.html', {'tags': index_filter_tags(request, request.user, prefs, active_tags, workspace=active_workspace(request)), 'active_tag': active_tags[0] if active_tags else None, 'active_tags': active_tags, 'active_tag_mode': active_tag_mode})
|
||
|
||
|
||
def requested_default_item_kind(post, prefs, save=False):
|
||
kind = post.get('default_item_kind') or prefs.default_item_kind
|
||
kind = kind if kind in Item.Kind.values else prefs.default_item_kind
|
||
if save and kind != prefs.default_item_kind:
|
||
prefs.default_item_kind = kind
|
||
prefs.save(update_fields=['default_item_kind'])
|
||
return kind
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def set_default_item_kind(request):
|
||
prefs, created = UserPreference.objects.get_or_create(user=request.user)
|
||
requested_default_item_kind(request.POST, prefs, save=True)
|
||
return HttpResponse(status=204)
|
||
|
||
|
||
@login_required
|
||
def api_tags(request):
|
||
return JsonResponse({'tags': list(visible_tags(request.user, workspace=active_workspace(request)).values_list('name', flat=True))})
|
||
|
||
|
||
@login_required
|
||
def api_attachment_suggestions(request):
|
||
item = get_object_or_404(visible_items(request.user, workspace=active_workspace(request)), pk=request.GET.get('item'))
|
||
if not user_can_edit_item(request.user, item):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
data = [{'id': a.id, 'name': a.file.name.split('/')[-1], 'url': a.file.url} for a in item.attachments.all()]
|
||
return JsonResponse({'attachments': data})
|
||
|
||
|
||
@login_required
|
||
def api_item_suggestions(request):
|
||
q = request.GET.get('q', '').strip()
|
||
items = visible_items(request.user, workspace=active_workspace(request))
|
||
if q:
|
||
items = items.filter(Q(content__icontains=q) | Q(tags__name__icontains=q)).distinct()
|
||
data = [{'id': i.id, 'kind': i.kind, 'label': i.content[:80]} for i in items[:10]]
|
||
return JsonResponse({'items': data})
|
||
|
||
|
||
def api_items_queryset(request, user, api_key=None):
|
||
items = visible_items(user, api_key)
|
||
tags = [tag.strip().lstrip('#').lower() for tag in request.GET.getlist('tag') if tag.strip()]
|
||
if tags:
|
||
if request.GET.get('tag_mode') == 'all':
|
||
for tag in tags:
|
||
items = items.filter(tags__name=tag)
|
||
else:
|
||
items = items.filter(tags__name__in=tags).distinct()
|
||
q = request.GET.get('q', '').strip()
|
||
if q:
|
||
items = items.filter(Q(content__icontains=q) | Q(comment__icontains=q) | Q(url__icontains=q) | Q(tags__name__icontains=q)).distinct()
|
||
return items
|
||
|
||
|
||
def api_items_limit(request):
|
||
try:
|
||
limit = int(request.GET.get('limit', 200))
|
||
except (TypeError, ValueError):
|
||
limit = 200
|
||
return max(1, min(limit, 500))
|
||
|
||
|
||
def serialize_item(item):
|
||
return {
|
||
'id': item.id,
|
||
'kind': item.kind,
|
||
'content': item.content,
|
||
'comment': item.comment,
|
||
'visibility': item.visibility,
|
||
'team': item.team.slug if item.team else None,
|
||
'is_done': item.is_done,
|
||
'completed_at': item.completed_at.isoformat() if item.completed_at else None,
|
||
'due_at': item.due_at.isoformat() if item.due_at else None,
|
||
'url': item.url,
|
||
'tags': [t.name for t in item.tags.all()],
|
||
'linked_items': [i.id for i in item.linked_items.all()],
|
||
'attachments': [a.file.url for a in item.attachments.all()],
|
||
'created_at': item.created_at.isoformat(),
|
||
'updated_at': item.updated_at.isoformat(),
|
||
}
|
||
|
||
|
||
@csrf_exempt
|
||
@api_auth_required
|
||
@require_http_methods(['GET', 'POST'])
|
||
def api_items(request):
|
||
user = request.api_key.owner if request.api_key else request.user
|
||
if request.method == 'GET':
|
||
items = api_items_queryset(request, user, request.api_key)[:api_items_limit(request)]
|
||
return JsonResponse({'items': [serialize_item(i) for i in items]})
|
||
data = json.loads(request.body or '{}') if request.content_type == 'application/json' else request.POST
|
||
raw = data.get('content', '')
|
||
raw, command_due_at, _ = parse_due_command(raw)
|
||
prefs, created = UserPreference.objects.get_or_create(user=user)
|
||
kind, visibility, content, url = parse_quick_content(raw, prefs.default_item_kind)
|
||
due_at = data['due_at'] if 'due_at' in data else command_due_at
|
||
item = Item.objects.create(workspace=current_workspace(user, request.api_key), owner=user, kind=data.get('kind') or kind, visibility=data.get('visibility') or visibility, content=content, url=data.get('url') or url, due_at=due_at)
|
||
item.sync_metadata()
|
||
apply_team_from_tags(item, user, force_private=bool(re.search(r'/private\b', raw, re.I)))
|
||
return JsonResponse(serialize_item(item), status=201)
|
||
|
||
|
||
@csrf_exempt
|
||
@api_auth_required
|
||
@require_http_methods(['GET', 'PATCH', 'DELETE'])
|
||
def api_item_detail(request, pk):
|
||
user = request.api_key.owner if request.api_key else request.user
|
||
item = get_object_or_404(visible_items(user, request.api_key), pk=pk)
|
||
if request.method == 'GET':
|
||
return JsonResponse(serialize_item(item))
|
||
if request.method == 'DELETE':
|
||
if not user_can_delete_item(user, item):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
try:
|
||
item.delete()
|
||
except ProtectedError:
|
||
collections = list(item.collection_sections.select_related('collection').values_list('collection__title', flat=True))
|
||
return JsonResponse({'error': _('This note is used in collections and cannot be deleted.'), 'collections': collections}, status=409)
|
||
return JsonResponse({'ok': True})
|
||
if not user_can_edit_item(user, item):
|
||
return JsonResponse({'error': 'forbidden'}, status=403)
|
||
data = json.loads(request.body or '{}')
|
||
for field in ['content', 'comment', 'kind', 'visibility', 'is_done', 'due_at', 'url']:
|
||
if field in data:
|
||
setattr(item, field, data[field])
|
||
if 'content' in data and 'due_at' not in data:
|
||
item.content, command_due_at, has_due_command = parse_due_command(item.content)
|
||
if has_due_command:
|
||
item.due_at = command_due_at
|
||
if 'is_done' in data and item.kind == Item.Kind.TODO:
|
||
item.completed_at = timezone.now() if item.is_done else None
|
||
target_visibility, target_team = desired_item_scope(item, user, force_private=item.visibility == Item.Visibility.PRIVATE)
|
||
if not item_scope_fits_collections(item, target_visibility, target_team):
|
||
return JsonResponse({'error': _('This visibility is incompatible with a collection that uses the note.')}, status=409)
|
||
item.visibility, item.team = target_visibility, target_team
|
||
item.save()
|
||
item.sync_metadata()
|
||
apply_team_from_tags(item, user, force_private=item.visibility == Item.Visibility.PRIVATE)
|
||
return JsonResponse(serialize_item(item))
|