Inna
2c30844bf7
- app 'forum': portada (categorías/foros), ver foro y tema (paginado), crear tema, responder, editar/borrar posts y moderación (fijar, bloquear, mover, borrar/restaurar) - lee/escribe la BD acore_web (5ª conexión, DB_NAME_WEB) por SQL directo, tablas forum_categories/forums/forum_topics/forum_posts/forum_reads (portado de NovaWeb-main) - identidad = cuenta de juego en sesión; permisos simplificados por gmlevel (FORUM_MOD_GMLEVEL) en vez de la matriz group_level×permission del original - saneado del HTML de posts con nh3 (evita el XSS del original); POST+CSRF en formularios - plantillas con los partials del sitio; enlace 'FOROS' del menú apunta al foro interno - sql/forum_schema.sql y docs/FORO.md Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
322 lines
12 KiB
Python
322 lines
12 KiB
Python
"""
|
|
Acceso a datos del foro (BD ``acore_web``), con SQL directo — mismo estilo que
|
|
el resto del proyecto (``connections['acore_web']``).
|
|
|
|
Tablas: forum_categories, forums, forum_topics, forum_posts, forum_reads.
|
|
"""
|
|
|
|
from django.db import connections
|
|
|
|
|
|
def _dictfetchall(cursor):
|
|
cols = [c[0] for c in cursor.description]
|
|
return [dict(zip(cols, row)) for row in cursor.fetchall()]
|
|
|
|
|
|
def _dictfetchone(cursor):
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
return None
|
|
cols = [c[0] for c in cursor.description]
|
|
return dict(zip(cols, row))
|
|
|
|
|
|
def _conn():
|
|
return connections['acore_web']
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Portada: categorías + foros (con estadísticas)
|
|
# --------------------------------------------------------------------------
|
|
|
|
def get_index(include_hidden=False):
|
|
"""
|
|
Devuelve las categorías ordenadas, cada una con sus foros (visibles y no
|
|
borrados salvo include_hidden) y estadísticas: nº de temas, nº de posts y
|
|
último tema.
|
|
"""
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT id, `order`, name FROM forum_categories ORDER BY `order`, id"
|
|
)
|
|
categories = _dictfetchall(cursor)
|
|
|
|
vis = "" if include_hidden else "AND f.visibility = 1 AND f.deleted = 0"
|
|
cursor.execute(
|
|
f"""
|
|
SELECT f.id, f.category_id, f.`order`, f.name, f.description, f.icon,
|
|
f.colortitle, f.type, f.visibility, f.deleted,
|
|
(SELECT COUNT(*) FROM forum_topics t
|
|
WHERE t.forum_id = f.id AND t.deleted = 0) AS topic_count,
|
|
(SELECT COUNT(*) FROM forum_posts p
|
|
JOIN forum_topics t ON p.topic_id = t.id
|
|
WHERE t.forum_id = f.id AND p.deleted = 0 AND t.deleted = 0) AS post_count,
|
|
(SELECT t2.id FROM forum_topics t2
|
|
WHERE t2.forum_id = f.id AND t2.deleted = 0
|
|
ORDER BY t2.updated_at DESC, t2.id DESC LIMIT 1) AS last_topic_id,
|
|
(SELECT t3.name FROM forum_topics t3
|
|
WHERE t3.forum_id = f.id AND t3.deleted = 0
|
|
ORDER BY t3.updated_at DESC, t3.id DESC LIMIT 1) AS last_topic_name,
|
|
(SELECT t4.poster FROM forum_topics t4
|
|
WHERE t4.forum_id = f.id AND t4.deleted = 0
|
|
ORDER BY t4.updated_at DESC, t4.id DESC LIMIT 1) AS last_topic_poster,
|
|
(SELECT t5.updated_at FROM forum_topics t5
|
|
WHERE t5.forum_id = f.id AND t5.deleted = 0
|
|
ORDER BY t5.updated_at DESC, t5.id DESC LIMIT 1) AS last_topic_time
|
|
FROM forums f
|
|
WHERE 1=1 {vis}
|
|
ORDER BY f.`order`, f.id
|
|
"""
|
|
)
|
|
forums = _dictfetchall(cursor)
|
|
|
|
by_cat = {}
|
|
for f in forums:
|
|
by_cat.setdefault(f['category_id'], []).append(f)
|
|
for c in categories:
|
|
c['forums'] = by_cat.get(c['id'], [])
|
|
# Solo categorías con foros visibles (salvo include_hidden)
|
|
if not include_hidden:
|
|
categories = [c for c in categories if c['forums']]
|
|
return categories
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Foro individual y sus temas (paginado)
|
|
# --------------------------------------------------------------------------
|
|
|
|
def get_forum(forum_id, include_hidden=False):
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT id, category_id, `order`, name, description, icon, colortitle, "
|
|
"type, visibility, deleted FROM forums WHERE id = %s",
|
|
[forum_id],
|
|
)
|
|
forum = _dictfetchone(cursor)
|
|
if not forum:
|
|
return None
|
|
if not include_hidden and (forum['deleted'] or not forum['visibility']):
|
|
return None
|
|
return forum
|
|
|
|
|
|
def count_topics(forum_id):
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT COUNT(*) FROM forum_topics WHERE forum_id = %s AND deleted = 0",
|
|
[forum_id],
|
|
)
|
|
return cursor.fetchone()[0]
|
|
|
|
|
|
def get_topics(forum_id, offset, limit):
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT t.id, t.name, t.poster, t.poster_id, t.created, t.sticky,
|
|
t.locked, t.moved, t.updated_at,
|
|
(SELECT COUNT(*) FROM forum_posts p
|
|
WHERE p.topic_id = t.id AND p.deleted = 0) AS post_count,
|
|
(SELECT p2.poster FROM forum_posts p2
|
|
WHERE p2.topic_id = t.id AND p2.deleted = 0
|
|
ORDER BY p2.time DESC, p2.id DESC LIMIT 1) AS last_poster,
|
|
(SELECT p3.time FROM forum_posts p3
|
|
WHERE p3.topic_id = t.id AND p3.deleted = 0
|
|
ORDER BY p3.time DESC, p3.id DESC LIMIT 1) AS last_time
|
|
FROM forum_topics t
|
|
WHERE t.forum_id = %s AND t.deleted = 0
|
|
ORDER BY t.sticky DESC, t.updated_at DESC, t.id DESC
|
|
LIMIT %s OFFSET %s
|
|
""",
|
|
[forum_id, limit, offset],
|
|
)
|
|
return _dictfetchall(cursor)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Tema individual y sus posts (paginado)
|
|
# --------------------------------------------------------------------------
|
|
|
|
def get_topic(topic_id, include_hidden=False):
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT id, forum_id, name, poster, poster_id, created, sticky, "
|
|
"locked, deleted, moved, updated_at FROM forum_topics WHERE id = %s",
|
|
[topic_id],
|
|
)
|
|
topic = _dictfetchone(cursor)
|
|
if not topic:
|
|
return None
|
|
if not include_hidden and topic['deleted']:
|
|
return None
|
|
return topic
|
|
|
|
|
|
def count_posts(topic_id):
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT COUNT(*) FROM forum_posts WHERE topic_id = %s AND deleted = 0",
|
|
[topic_id],
|
|
)
|
|
return cursor.fetchone()[0]
|
|
|
|
|
|
def get_posts(topic_id, offset, limit, include_deleted=False):
|
|
delf = "" if include_deleted else "AND deleted = 0"
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
f"""
|
|
SELECT id, topic_id, poster, poster_id, text, time, deleted, updated_at
|
|
FROM forum_posts
|
|
WHERE topic_id = %s {delf}
|
|
ORDER BY time ASC, id ASC
|
|
LIMIT %s OFFSET %s
|
|
""",
|
|
[topic_id, limit, offset],
|
|
)
|
|
return _dictfetchall(cursor)
|
|
|
|
|
|
def get_post(post_id):
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT id, topic_id, poster, poster_id, text, time, deleted, updated_at "
|
|
"FROM forum_posts WHERE id = %s",
|
|
[post_id],
|
|
)
|
|
return _dictfetchone(cursor)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Escritura: crear tema, responder, editar, borrar/restaurar
|
|
# --------------------------------------------------------------------------
|
|
|
|
def create_topic(forum_id, name, poster, poster_id, text):
|
|
"""Crea un tema y su primer post. Devuelve el id del tema."""
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"INSERT INTO forum_topics "
|
|
"(forum_id, name, poster, poster_id, created, created_at, updated_at, locked, sticky, deleted) "
|
|
"VALUES (%s, %s, %s, %s, NOW(), NOW(), NOW(), 0, 0, 0)",
|
|
[forum_id, name, poster, poster_id],
|
|
)
|
|
cursor.execute("SELECT LAST_INSERT_ID()")
|
|
topic_id = cursor.fetchone()[0]
|
|
cursor.execute(
|
|
"INSERT INTO forum_posts (topic_id, poster, poster_id, text, time, created_at, deleted) "
|
|
"VALUES (%s, %s, %s, %s, NOW(), NOW(), 0)",
|
|
[topic_id, poster, poster_id, text],
|
|
)
|
|
return topic_id
|
|
|
|
|
|
def create_post(topic_id, poster, poster_id, text):
|
|
"""Añade una respuesta y actualiza la marca de tiempo del tema."""
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"INSERT INTO forum_posts (topic_id, poster, poster_id, text, time, created_at, deleted) "
|
|
"VALUES (%s, %s, %s, %s, NOW(), NOW(), 0)",
|
|
[topic_id, poster, poster_id, text],
|
|
)
|
|
cursor.execute("SELECT LAST_INSERT_ID()")
|
|
post_id = cursor.fetchone()[0]
|
|
cursor.execute(
|
|
"UPDATE forum_topics SET updated_at = NOW() WHERE id = %s", [topic_id]
|
|
)
|
|
return post_id
|
|
|
|
|
|
def get_author_info(poster_id):
|
|
"""
|
|
Datos del autor para mostrar en los posts: nivel GM, fecha de registro y
|
|
nº de mensajes. Cruza acore_auth (account/account_access) y acore_web.
|
|
"""
|
|
info = {'gmlevel': 0, 'joindate': None, 'post_count': 0, 'status': 'Miembro'}
|
|
if not poster_id:
|
|
return info
|
|
try:
|
|
with connections['acore_auth'].cursor() as cursor:
|
|
cursor.execute("SELECT MAX(gmlevel) FROM account_access WHERE id = %s", [poster_id])
|
|
row = cursor.fetchone()
|
|
info['gmlevel'] = int(row[0]) if row and row[0] is not None else 0
|
|
cursor.execute("SELECT joindate FROM account WHERE id = %s", [poster_id])
|
|
row = cursor.fetchone()
|
|
if row:
|
|
info['joindate'] = row[0]
|
|
except Exception:
|
|
pass
|
|
try:
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT COUNT(*) FROM forum_posts WHERE poster_id = %s AND deleted = 0",
|
|
[poster_id],
|
|
)
|
|
info['post_count'] = cursor.fetchone()[0]
|
|
except Exception:
|
|
pass
|
|
if info['gmlevel'] >= 16:
|
|
info['status'] = 'Administrador'
|
|
elif info['gmlevel'] >= 1:
|
|
info['status'] = 'GM'
|
|
return info
|
|
|
|
|
|
def update_post(post_id, text):
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"UPDATE forum_posts SET text = %s, updated_at = NOW() WHERE id = %s",
|
|
[text, post_id],
|
|
)
|
|
|
|
|
|
def set_post_deleted(post_id, deleted):
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"UPDATE forum_posts SET deleted = %s WHERE id = %s",
|
|
[1 if deleted else 0, post_id],
|
|
)
|
|
|
|
|
|
def set_topic_flag(topic_id, field, value):
|
|
"""field ∈ {sticky, locked, deleted, moved}."""
|
|
if field not in ('sticky', 'locked', 'deleted', 'moved'):
|
|
raise ValueError('campo no permitido')
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
f"UPDATE forum_topics SET {field} = %s WHERE id = %s",
|
|
[1 if value else 0, topic_id],
|
|
)
|
|
|
|
|
|
def move_topic(topic_id, new_forum_id):
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"UPDATE forum_topics SET forum_id = %s, moved = 1 WHERE id = %s",
|
|
[new_forum_id, topic_id],
|
|
)
|
|
|
|
|
|
def mark_read(user_id, topic_id):
|
|
if not user_id:
|
|
return
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"INSERT INTO forum_reads (user_id, topic_id, read_at) "
|
|
"VALUES (%s, %s, NOW()) "
|
|
"ON DUPLICATE KEY UPDATE read_at = NOW()",
|
|
[user_id, topic_id],
|
|
)
|
|
|
|
|
|
def list_forums_for_move(exclude_forum_id=None):
|
|
"""Lista de foros visibles para el desplegable de 'mover tema'."""
|
|
with _conn().cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT id, name FROM forums WHERE visibility = 1 AND deleted = 0 "
|
|
"ORDER BY `order`, id"
|
|
)
|
|
forums = _dictfetchall(cursor)
|
|
if exclude_forum_id:
|
|
forums = [f for f in forums if f['id'] != exclude_forum_id]
|
|
return forums
|