rucki
1180 lines
57 KiB
Python
1180 lines
57 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 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
|
|
from .models import ApiKey, Attachment, Collection, CollectionSection, Item, Kanban, Tag, Team, TeamInvite, TeamMembership, UserInvite, UserPreference, without_markdown_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+$')
|
|
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))
|
|
if type_command == 'todo':
|
|
kind = Item.Kind.TODO
|
|
elif type_command == 'journal':
|
|
kind = Item.Kind.JOURNAL
|
|
elif type_command == 'link' or only_url:
|
|
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 = URL_RE.search(content)
|
|
url = url_match.group(0) if kind == Item.Kind.LINK and url_match else ''
|
|
if 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 attach_files_and_replace_tokens(item, files):
|
|
replacements = {}
|
|
for file in files:
|
|
attachment = Attachment.objects.create(item=item, file=file)
|
|
replacements[f'![[paste:{file.name}]]'] = f'![[file:{attachment.id}]]'
|
|
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 user_team_ids(user):
|
|
if not user.is_authenticated:
|
|
return []
|
|
return list(user.team_memberships.values_list('team_id', flat=True))
|
|
|
|
|
|
def visible_items(user, api_key=None):
|
|
qs = Item.objects.select_related('owner', 'team').prefetch_related('tags', 'attachments', 'linked_items')
|
|
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)) | Q(visibility=Item.Visibility.PUBLIC)).distinct()
|
|
return qs.filter(visibility=Item.Visibility.PUBLIC)
|
|
|
|
|
|
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 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__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__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').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(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, prefs.default_item_kind)
|
|
active_tag = request.GET.get('tag')
|
|
if active_tag and f'#{active_tag.lower()}' not in content.lower():
|
|
content = f'{content} #{active_tag}'
|
|
item = Item.objects.create(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'):
|
|
response = render(request, 'core/_create_response.html', {'item': item, 'tags': Tag.objects.all(), 'active_tag': request.GET.get('tag')})
|
|
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)
|
|
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)
|
|
if status == 'open':
|
|
items = items.filter(kind=Item.Kind.TODO, is_done=False)
|
|
elif status == 'due':
|
|
today_end = timezone.make_aware(datetime.combine(timezone.localdate(), time(23, 59, 59)))
|
|
items = items.filter(kind=Item.Kind.TODO, is_done=False, due_at__lte=today_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()
|
|
today_end = timezone.make_aware(datetime.combine(timezone.localdate(), time(23, 59, 59)))
|
|
due_count = items.filter(kind=Item.Kind.TODO, is_done=False, due_at__lte=today_end).count()
|
|
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': Tag.objects.all(), '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')
|
|
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, prefs.default_item_kind)
|
|
item = Item.objects.create(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)))
|
|
return redirect(next_url)
|
|
return render(request, 'core/new_item.html', {
|
|
'form': QuickItemForm(), 'tags': Tag.objects.all(), 'requested_kind': requested_kind, 'next_url': next_url,
|
|
})
|
|
|
|
|
|
@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(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)
|
|
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': Tag.objects.all(), '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
|
|
def settings_view(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') == 'apikey':
|
|
key_form = ApiKeyForm(request.POST)
|
|
if key_form.is_valid():
|
|
api_key = key_form.save(commit=False)
|
|
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),
|
|
'key_form': ApiKeyForm(),
|
|
'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)
|
|
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(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)
|
|
teams = Team.objects.filter(memberships__user=request.user).distinct().prefetch_related('memberships__user')
|
|
stats = []
|
|
base_items = visible_items(request.user)
|
|
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(), '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 = 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).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)
|
|
TeamMembership.objects.get_or_create(team=invite.team, user=request.user, defaults={'role': TeamMembership.Role.MEMBER})
|
|
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):
|
|
qs = Collection.objects.select_related('owner', 'team').prefetch_related('filter_tags', 'sections__item', 'sections__sub_collection')
|
|
if user.is_authenticated:
|
|
return qs.filter(Q(owner=user) | Q(team_id__in=user_team_ids(user)) | Q(visibility=Item.Visibility.PUBLIC)).distinct()
|
|
return qs.filter(visibility=Item.Visibility.PUBLIC)
|
|
|
|
|
|
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(kind=Item.Kind.NOTE).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):
|
|
slugs = Team.objects.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.content = remove_team_tags(item.content)
|
|
if collection.team:
|
|
item.content = f'{item.content} #{collection.team.slug}'.strip()
|
|
item.visibility = collection.visibility
|
|
item.team = collection.team
|
|
item.save(update_fields=['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(
|
|
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):
|
|
return Item.objects.select_related('owner', 'team').prefetch_related('tags', 'attachments').filter(
|
|
kind=Item.Kind.NOTE,
|
|
).filter(Q(owner=user) | Q(team_id__in=user_team_ids(user))).distinct()
|
|
|
|
|
|
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).annotate(collection_use_count=Count('collection_sections', distinct=True))
|
|
collection_candidates = visible_collections(request.user).filter(mode=Collection.Mode.MANUAL).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):
|
|
entries = []
|
|
for collection in visible_collections(request.user).prefetch_related('filter_tags'):
|
|
count = collection_tag_items(collection).count() if collection.mode == Collection.Mode.TAGS else collection.sections.count()
|
|
entries.append({'collection': collection, 'section_count': count})
|
|
return render(request, 'core/collection_list.html', {'entries': entries})
|
|
|
|
|
|
@login_required
|
|
def collection_create(request):
|
|
if request.method == 'POST':
|
|
form = CollectionForm(request.POST, user=request.user)
|
|
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.owner = request.user
|
|
collection.visibility = visibility
|
|
collection.team = team
|
|
collection.save()
|
|
form.save_m2m()
|
|
if collection.mode == Collection.Mode.MANUAL:
|
|
collection.filter_tags.clear()
|
|
messages.success(request, _('Collection created.'))
|
|
return redirect(collection)
|
|
else:
|
|
form = CollectionForm(user=request.user)
|
|
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), 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.mode == Collection.Mode.MANUAL and collection.sections.exists()
|
|
incompatible = target_mode == Collection.Mode.MANUAL 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.MANUAL:
|
|
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), pk=pk)
|
|
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.mode == Collection.Mode.MANUAL else [],
|
|
'tag_items': collection_tag_items(collection) if collection.mode == Collection.Mode.TAGS else [],
|
|
'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), 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), 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 == 'add_collection':
|
|
sub_collection = get_object_or_404(visible_collections(request.user), pk=request.POST.get('collection_id'), mode=Collection.Mode.MANUAL)
|
|
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)
|
|
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), 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), 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), 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'])
|
|
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), 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), 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': Tag.objects.all()}, status=409)
|
|
item.visibility, item.team = target_visibility, target_team
|
|
item.save()
|
|
form.save_m2m()
|
|
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': Tag.objects.all()})
|
|
|
|
|
|
@login_required
|
|
def item_editor(request, pk):
|
|
item = get_object_or_404(visible_items(request.user), 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': Tag.objects.all(), 'next_url': next_url}, status=409)
|
|
item.visibility, item.team = target_visibility, target_team
|
|
item.save()
|
|
form.save_m2m()
|
|
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': Tag.objects.all(), 'next_url': next_url})
|
|
|
|
|
|
@login_required
|
|
@require_POST
|
|
def delete_item(request, pk):
|
|
item = get_object_or_404(visible_items(request.user), 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)
|
|
if form.is_valid():
|
|
kanban = form.save(commit=False)
|
|
kanban.owner = request.user
|
|
kanban.save()
|
|
return redirect('kanban_detail', pk=kanban.pk)
|
|
kanbans = Kanban.objects.filter(owner=request.user).select_related('tag')
|
|
stats = []
|
|
for kanban in kanbans:
|
|
qs = Item.objects.filter(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(), '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(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, 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):
|
|
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': Tag.objects.all(), 'active_tag': active_tags[0] if active_tags else None, 'active_tags': active_tags, 'active_tag_mode': active_tag_mode})
|
|
|
|
|
|
@login_required
|
|
def api_tags(request):
|
|
return JsonResponse({'tags': list(Tag.objects.order_by('name').values_list('name', flat=True))})
|
|
|
|
|
|
@login_required
|
|
def api_attachment_suggestions(request):
|
|
item = get_object_or_404(visible_items(request.user), 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)
|
|
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(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))
|