240 lines
7.5 KiB
Python
240 lines
7.5 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from datetime import datetime, timezone
|
|
from uuid import uuid4
|
|
|
|
from backend.app.core.security import clean_url
|
|
from backend.app.database import get_connection
|
|
|
|
MAX_TAGS = 10
|
|
|
|
|
|
def normalize_tags(tags: list[str] | None) -> list[str]:
|
|
normalized = []
|
|
normalized_keys = set()
|
|
for tag in tags or []:
|
|
value = tag.strip()
|
|
if value and not value.startswith('#'):
|
|
value = f'#{value}'
|
|
key = value.casefold()
|
|
if value and key not in normalized_keys:
|
|
normalized.append(value)
|
|
normalized_keys.add(key)
|
|
if len(normalized) > MAX_TAGS:
|
|
raise ValueError(f'A link can have at most {MAX_TAGS} tags')
|
|
return normalized
|
|
|
|
|
|
def save_link_tags(conn, link_id: str, tags: list[str]) -> None:
|
|
canonical_tags = []
|
|
for tag in tags:
|
|
tag_row = conn.execute(
|
|
'SELECT id FROM tags WHERE lower(name) = lower(?)',
|
|
(tag,),
|
|
).fetchone()
|
|
if tag_row is None:
|
|
conn.execute(
|
|
'INSERT INTO tags (id, name) VALUES (?, ?)',
|
|
(str(uuid4()), tag),
|
|
)
|
|
tag_row = conn.execute('SELECT id FROM tags WHERE name = ?', (tag,)).fetchone()
|
|
conn.execute(
|
|
'INSERT OR IGNORE INTO link_tags (link_id, tag_id) VALUES (?, ?)',
|
|
(link_id, tag_row['id']),
|
|
)
|
|
canonical_tags.append(conn.execute(
|
|
'SELECT name FROM tags WHERE id = ?',
|
|
(tag_row['id'],),
|
|
).fetchone()['name'])
|
|
return canonical_tags
|
|
|
|
|
|
def get_link_tags(conn, link_id: str) -> list[str]:
|
|
rows = conn.execute(
|
|
'''
|
|
SELECT tags.name FROM tags
|
|
JOIN link_tags ON link_tags.tag_id = tags.id
|
|
WHERE link_tags.link_id = ? ORDER BY tags.name
|
|
''',
|
|
(link_id,),
|
|
).fetchall()
|
|
return [row['name'] for row in rows]
|
|
|
|
|
|
def create_link(
|
|
user_id: str,
|
|
title: str,
|
|
url: str,
|
|
comment: str,
|
|
timestamp: str | None,
|
|
tags: list[str] | None = None,
|
|
):
|
|
cleaned_url = clean_url(url)
|
|
normalized_tags = normalize_tags(tags)
|
|
created_at = datetime.now(timezone.utc).isoformat()
|
|
record = {
|
|
'id': str(uuid4()),
|
|
'user_id': user_id,
|
|
'title': title,
|
|
'url': cleaned_url,
|
|
'comment': comment,
|
|
'timestamp': timestamp or created_at,
|
|
'created_at': created_at,
|
|
'updated_at': created_at,
|
|
'is_public': 1,
|
|
}
|
|
with get_connection() as conn:
|
|
conn.execute(
|
|
'''
|
|
INSERT INTO links (id, user_id, title, url, comment, timestamp, created_at, updated_at, is_public)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
''',
|
|
(
|
|
record['id'],
|
|
record['user_id'],
|
|
record['title'],
|
|
record['url'],
|
|
record['comment'],
|
|
record['timestamp'],
|
|
record['created_at'],
|
|
record['updated_at'],
|
|
record['is_public'],
|
|
),
|
|
)
|
|
stored_tags = save_link_tags(conn, record['id'], normalized_tags)
|
|
conn.commit()
|
|
record['tags'] = stored_tags
|
|
return record
|
|
|
|
|
|
def list_public_links(username: str | None = None):
|
|
with get_connection() as conn:
|
|
rows = conn.execute(
|
|
'''
|
|
SELECT links.*, users.username, users.avatar_url, users.bio
|
|
FROM links
|
|
JOIN users ON users.id = links.user_id
|
|
WHERE links.is_public = 1
|
|
AND (? IS NULL OR users.username = ?)
|
|
ORDER BY created_at DESC
|
|
''',
|
|
(username, username),
|
|
).fetchall()
|
|
records = [dict(row) for row in rows]
|
|
for record in records:
|
|
record['tags'] = get_link_tags(conn, record['id'])
|
|
return records
|
|
|
|
|
|
def update_link(
|
|
link_id: str,
|
|
user_id: str,
|
|
title: str,
|
|
url: str,
|
|
comment: str,
|
|
tags: list[str] | None = None,
|
|
):
|
|
cleaned_url = clean_url(url)
|
|
normalized_tags = normalize_tags(tags)
|
|
with get_connection() as conn:
|
|
cursor = conn.execute(
|
|
'''
|
|
UPDATE links
|
|
SET title = ?, url = ?, comment = ?, updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND user_id = ?
|
|
''',
|
|
(title, cleaned_url, comment, link_id, user_id),
|
|
)
|
|
if cursor.rowcount == 0:
|
|
return None
|
|
conn.execute('DELETE FROM link_tags WHERE link_id = ?', (link_id,))
|
|
stored_tags = save_link_tags(conn, link_id, normalized_tags)
|
|
conn.commit()
|
|
row = conn.execute('SELECT * FROM links WHERE id = ?', (link_id,)).fetchone()
|
|
record = dict(row)
|
|
record['tags'] = stored_tags
|
|
return record
|
|
|
|
|
|
def list_public_users():
|
|
with get_connection() as conn:
|
|
rows = conn.execute(
|
|
'SELECT username FROM users ORDER BY username'
|
|
).fetchall()
|
|
return [row['username'] for row in rows]
|
|
|
|
|
|
def list_tags():
|
|
with get_connection() as conn:
|
|
rows = conn.execute('SELECT name FROM tags ORDER BY name').fetchall()
|
|
tags = []
|
|
seen = set()
|
|
for row in rows:
|
|
if row['name'].casefold() not in seen:
|
|
tags.append(row['name'])
|
|
seen.add(row['name'].casefold())
|
|
return tags
|
|
|
|
|
|
def list_user_labels(user_id: str):
|
|
with get_connection() as conn:
|
|
rows = conn.execute(
|
|
'SELECT id, name, created_by FROM tags WHERE created_by = ? ORDER BY name',
|
|
(user_id,),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def create_label(user_id: str, name: str):
|
|
normalized = normalize_tags([name])
|
|
label = normalized[0] if normalized else ''
|
|
if not label:
|
|
raise ValueError('Label cannot be empty')
|
|
with get_connection() as conn:
|
|
existing = conn.execute(
|
|
'SELECT id FROM tags WHERE lower(name) = lower(?)', (label,)
|
|
).fetchone()
|
|
if existing:
|
|
raise ValueError('Label already exists')
|
|
label_id = str(uuid4())
|
|
conn.execute(
|
|
'INSERT INTO tags (id, name, created_by) VALUES (?, ?, ?)',
|
|
(label_id, label, user_id),
|
|
)
|
|
conn.commit()
|
|
return {'id': label_id, 'name': label, 'created_by': user_id}
|
|
|
|
|
|
def update_label(label_id: str, user_id: str, name: str):
|
|
normalized = normalize_tags([name])
|
|
if not normalized:
|
|
raise ValueError('Label cannot be empty')
|
|
with get_connection() as conn:
|
|
current = conn.execute(
|
|
'SELECT id, name, created_by FROM tags WHERE id = ?', (label_id,)
|
|
).fetchone()
|
|
if current is None or current['created_by'] != user_id:
|
|
return None
|
|
duplicate = conn.execute(
|
|
'SELECT id FROM tags WHERE lower(name) = lower(?) AND id != ?',
|
|
(normalized[0], label_id),
|
|
).fetchone()
|
|
if duplicate:
|
|
raise ValueError('Label already exists')
|
|
conn.execute('UPDATE tags SET name = ? WHERE id = ?', (normalized[0], label_id))
|
|
conn.commit()
|
|
return {'id': label_id, 'name': normalized[0], 'created_by': user_id}
|
|
|
|
|
|
def delete_label(label_id: str, user_id: str | None = None, is_admin: bool = False):
|
|
with get_connection() as conn:
|
|
current = conn.execute(
|
|
'SELECT id, created_by FROM tags WHERE id = ?', (label_id,)
|
|
).fetchone()
|
|
if current is None or (not is_admin and current['created_by'] != user_id):
|
|
return False
|
|
conn.execute('DELETE FROM tags WHERE id = ?', (label_id,))
|
|
conn.commit()
|
|
return True
|