84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
from datetime import datetime, timezone
|
|
from uuid import uuid4
|
|
|
|
from backend.app.core.security import clean_url
|
|
from backend.app.database import get_connection
|
|
|
|
|
|
def create_link(user_id: str, title: str, url: str, comment: str, timestamp: str | None):
|
|
cleaned_url = clean_url(url)
|
|
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'],
|
|
),
|
|
)
|
|
conn.commit()
|
|
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()
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def update_link(link_id: str, user_id: str, title: str, url: str, comment: str):
|
|
cleaned_url = clean_url(url)
|
|
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.commit()
|
|
row = conn.execute('SELECT * FROM links WHERE id = ?', (link_id,)).fetchone()
|
|
return dict(row)
|
|
|
|
|
|
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]
|