contentdb/app/models/threads.py

278 lines
9.6 KiB
Python
Raw Normal View History

2020-12-10 12:37:15 +01:00
# ContentDB
2021-01-30 17:59:42 +01:00
# Copyright (C) 2018-21 rubenwardy
2020-12-10 12:37:15 +01:00
#
# This program is free software: you can redistribute it and/or modify
2021-01-30 17:59:42 +01:00
# it under the terms of the GNU Affero General Public License as published by
2020-12-10 12:37:15 +01:00
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
2021-01-30 17:59:42 +01:00
# GNU Affero General Public License for more details.
2020-12-10 12:37:15 +01:00
#
2021-01-30 17:59:42 +01:00
# You should have received a copy of the GNU Affero General Public License
2020-12-10 12:37:15 +01:00
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import datetime
2021-08-18 23:09:41 +02:00
from typing import Tuple, List
2020-12-10 12:37:15 +01:00
from flask import url_for
from sqlalchemy import select, func, text
from sqlalchemy.orm import column_property
2020-12-10 12:37:15 +01:00
from . import db
from .users import Permission, UserRank, User
2020-12-10 12:37:15 +01:00
from .packages import Package
watchers = db.Table("watchers",
db.Column("user_id", db.Integer, db.ForeignKey("user.id"), primary_key=True),
db.Column("thread_id", db.Integer, db.ForeignKey("thread.id"), primary_key=True)
)
2020-12-10 17:49:37 +01:00
2020-12-10 12:37:15 +01:00
class Thread(db.Model):
id = db.Column(db.Integer, primary_key=True)
package_id = db.Column(db.Integer, db.ForeignKey("package.id"), nullable=True)
package = db.relationship("Package", foreign_keys=[package_id], back_populates="threads")
is_review_thread = db.relationship("Package", foreign_keys=[Package.review_thread_id], back_populates="review_thread")
review_id = db.Column(db.Integer, db.ForeignKey("package_review.id"), nullable=True)
review = db.relationship("PackageReview", foreign_keys=[review_id], cascade="all, delete")
author_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
author = db.relationship("User", back_populates="threads", foreign_keys=[author_id])
title = db.Column(db.String(100), nullable=False)
private = db.Column(db.Boolean, server_default="0", nullable=False)
locked = db.Column(db.Boolean, server_default="0", nullable=False)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow)
replies = db.relationship("ThreadReply", back_populates="thread", lazy="dynamic",
order_by=db.asc("thread_reply_id"), cascade="all, delete, delete-orphan")
watchers = db.relationship("User", secondary=watchers, backref="watching")
2020-12-10 12:37:15 +01:00
first_reply = db.relationship("ThreadReply", uselist=False, foreign_keys="ThreadReply.thread_id",
lazy=True, order_by=db.asc("id"), viewonly=True,
primaryjoin="Thread.id==ThreadReply.thread_id")
replies_count = column_property(select(func.count(text("thread_reply.id")))
.select_from(text("thread_reply"))
.where(text("thread_reply.thread_id") == id)
.as_scalar())
2021-08-17 22:16:43 +02:00
def get_description(self):
comment = self.first_reply.comment.replace("\r\n", " ").replace("\n", " ").replace(" ", " ")
2021-08-17 22:16:43 +02:00
if len(comment) > 100:
return comment[:97] + "..."
else:
return comment
def get_view_url(self, absolute=False):
2021-08-17 22:16:43 +02:00
if absolute:
2023-06-19 20:32:36 +02:00
from app.utils import abs_url_for
2021-08-17 22:16:43 +02:00
return abs_url_for("threads.view", id=self.id)
else:
return url_for("threads.view", id=self.id, _external=False)
2020-12-10 12:37:15 +01:00
def get_subscribe_url(self):
2020-12-10 12:37:15 +01:00
return url_for("threads.subscribe", id=self.id)
def get_unsubscribe_url(self):
2020-12-10 12:37:15 +01:00
return url_for("threads.unsubscribe", id=self.id)
def check_perm(self, user, perm):
2020-12-10 12:37:15 +01:00
if type(perm) == str:
perm = Permission[perm]
elif type(perm) != Permission:
raise Exception("Unknown permission given to Thread.check_perm()")
2020-12-10 12:37:15 +01:00
2023-08-14 22:48:50 +02:00
if not user.is_authenticated:
return perm == Permission.SEE_THREAD and not self.private
2020-12-10 12:37:15 +01:00
isMaintainer = user == self.author or (self.package is not None and self.package.author == user)
if self.package:
isMaintainer = isMaintainer or user in self.package.maintainers
2023-06-19 22:27:49 +02:00
canSee = not self.private or isMaintainer or user.rank.at_least(UserRank.APPROVER) or user in self.watchers
2020-12-10 12:37:15 +01:00
if perm == Permission.SEE_THREAD:
return canSee
elif perm == Permission.COMMENT_THREAD:
2023-06-19 22:27:49 +02:00
return canSee and (not self.locked or user.rank.at_least(UserRank.MODERATOR))
2020-12-10 12:37:15 +01:00
elif perm == Permission.LOCK_THREAD:
2023-06-19 22:27:49 +02:00
return user.rank.at_least(UserRank.MODERATOR)
2020-12-10 12:37:15 +01:00
elif perm == Permission.DELETE_THREAD:
from app.utils.models import get_system_user
return (self.author == get_system_user() and self.package and
2023-06-19 22:27:49 +02:00
user in self.package.maintainers) or user.rank.at_least(UserRank.MODERATOR)
2020-12-10 12:37:15 +01:00
else:
raise Exception("Permission {} is not related to threads".format(perm.name))
def get_visible_to(self) -> list[User]:
retval = {
self.author.username: self.author
}
for user in self.watchers:
retval[user.username] = user
if self.package:
for user in self.package.maintainers:
retval[user.username] = user
return list(retval.values())
2020-12-22 13:22:52 +01:00
def get_latest_reply(self):
return ThreadReply.query.filter_by(thread_id=self.id).order_by(db.desc(ThreadReply.id)).first()
2020-12-10 17:49:37 +01:00
2020-12-10 12:37:15 +01:00
class ThreadReply(db.Model):
id = db.Column(db.Integer, primary_key=True)
thread_id = db.Column(db.Integer, db.ForeignKey("thread.id"), nullable=False)
thread = db.relationship("Thread", back_populates="replies", foreign_keys=[thread_id])
comment = db.Column(db.String(2000), nullable=False)
author_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
author = db.relationship("User", back_populates="replies", foreign_keys=[author_id])
2022-04-23 21:53:38 +02:00
is_status_update = db.Column(db.Boolean, server_default="0", nullable=False)
2020-12-10 12:37:15 +01:00
created_at = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow)
2023-01-03 13:17:01 +01:00
def get_url(self, absolute=False):
return self.thread.get_view_url(absolute) + "#reply-" + str(self.id)
2022-01-21 00:30:56 +01:00
def check_perm(self, user, perm):
2020-12-10 12:37:15 +01:00
if not user.is_authenticated:
return False
if type(perm) == str:
perm = Permission[perm]
elif type(perm) != Permission:
raise Exception("Unknown permission given to ThreadReply.check_perm()")
2020-12-10 12:37:15 +01:00
if perm == Permission.EDIT_REPLY:
2023-06-19 22:27:49 +02:00
return user.rank.at_least(UserRank.NEW_MEMBER if user == self.author else UserRank.MODERATOR) and not self.thread.locked
2020-12-10 12:37:15 +01:00
elif perm == Permission.DELETE_REPLY:
2023-06-19 22:27:49 +02:00
return user.rank.at_least(UserRank.MODERATOR) and self.thread.first_reply != self
2020-12-10 12:37:15 +01:00
else:
raise Exception("Permission {} is not related to threads".format(perm.name))
class PackageReview(db.Model):
id = db.Column(db.Integer, primary_key=True)
package_id = db.Column(db.Integer, db.ForeignKey("package.id"), nullable=True)
package = db.relationship("Package", foreign_keys=[package_id], back_populates="reviews")
created_at = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow)
author_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
author = db.relationship("User", foreign_keys=[author_id], back_populates="reviews")
2023-04-15 03:37:58 +02:00
rating = db.Column(db.Integer, nullable=False)
2020-12-10 12:37:15 +01:00
2020-12-10 17:49:37 +01:00
thread = db.relationship("Thread", uselist=False, back_populates="review")
2021-08-21 23:55:11 +02:00
votes = db.relationship("PackageReviewVote", back_populates="review", cascade="all, delete, delete-orphan")
2021-08-18 23:09:41 +02:00
score = db.Column(db.Integer, nullable=False, default=1)
def get_totals(self, current_user = None) -> Tuple[int,int,bool]:
votes: List[PackageReviewVote] = self.votes
pos = sum([ 1 for vote in votes if vote.is_positive ])
neg = sum([ 1 for vote in votes if not vote.is_positive])
user_vote = next(filter(lambda vote: vote.user == current_user, votes), None)
return pos, neg, user_vote.is_positive if user_vote else None
2020-12-10 12:37:15 +01:00
def as_dict(self, include_package=False):
2023-07-12 22:30:38 +02:00
from app.utils import abs_url_for
2021-11-24 17:33:37 +01:00
pos, neg, _user = self.get_totals()
ret = {
"is_positive": self.rating > 3,
2023-04-15 03:37:58 +02:00
"rating": self.rating,
2021-11-24 17:33:37 +01:00
"user": {
"username": self.author.username,
"display_name": self.author.display_name,
},
"created_at": self.created_at.isoformat(),
"votes": {
"helpful": pos,
"unhelpful": neg,
},
"title": self.thread.title,
"comment": self.thread.first_reply.comment,
2023-07-12 22:30:38 +02:00
"thread": {
"id": self.thread.id,
"url": abs_url_for("threads.view", id=self.thread.id),
},
2021-11-24 17:33:37 +01:00
}
if include_package:
ret["package"] = self.package.as_key_dict()
2021-11-24 17:33:37 +01:00
return ret
def as_weight(self):
2023-04-15 05:00:38 +02:00
"""
From (1, 5) to (-1 to 1)
"""
return (self.rating - 3.0) / 2.0
2020-12-10 12:37:15 +01:00
def get_edit_url(self):
return self.package.get_url("packages.review")
2020-12-10 12:37:15 +01:00
def get_delete_url(self):
2020-12-10 12:37:15 +01:00
return url_for("packages.delete_review",
author=self.package.author.username,
name=self.package.name,
reviewer=self.author.username)
2021-08-18 23:09:41 +02:00
2023-06-19 22:27:49 +02:00
def get_vote_url(self, next_url=None):
2021-08-18 23:09:41 +02:00
return url_for("packages.review_vote",
author=self.package.author.username,
name=self.package.name,
review_id=self.id,
r=next_url)
def update_score(self):
(pos, neg, _) = self.get_totals()
self.score = 3 * (pos - neg) + 1
def check_perm(self, user, perm):
if not user.is_authenticated:
return False
if type(perm) == str:
perm = Permission[perm]
elif type(perm) != Permission:
raise Exception("Unknown permission given to PackageReview.check_perm()")
if perm == Permission.DELETE_REVIEW:
2023-06-19 22:27:49 +02:00
return user == self.author or user.rank.at_least(UserRank.MODERATOR)
else:
raise Exception("Permission {} is not related to reviews".format(perm.name))
2021-08-18 23:09:41 +02:00
class PackageReviewVote(db.Model):
review_id = db.Column(db.Integer, db.ForeignKey("package_review.id"), primary_key=True)
review = db.relationship("PackageReview", foreign_keys=[review_id], back_populates="votes")
user_id = db.Column(db.Integer, db.ForeignKey("user.id"), primary_key=True)
user = db.relationship("User", foreign_keys=[user_id], back_populates="review_votes")
is_positive = db.Column(db.Boolean, nullable=False)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow)