launchpad-reviewers team mailing list archive

[Alvaro Crespo Serrano <mp+493034@xxxxxxxxxxxxxxxxxx>]

Alvaro Crespo Serrano has proposed merging ~alvarocs/launchpad:spam-flags-spamobject into ~ilkeremrekoc/launchpad:spam-flags.

Commit message:
Draft for spamobject model

Requested reviews:
  Launchpad code reviewers (launchpad-reviewers)

For more details, see:
https://code.launchpad.net/~alvarocs/launchpad/+git/launchpad/+merge/493034
-- 
Your team Launchpad code reviewers is requested to review the proposed merge of ~alvarocs/launchpad:spam-flags-spamobject into ~ilkeremrekoc/launchpad:spam-flags.

diff --git a/lib/lp/services/spam/interface.py b/lib/lp/services/spam/interface.py
index e69de29..a65b51a 100644
--- a/lib/lp/services/spam/interface.py
+++ b/lib/lp/services/spam/interface.py
@@ -0,0 +1,88 @@
+# Copyright 2015-2025 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from lazr.restful.fields import Reference
+from zope.interface import Interface
+from zope.schema import Datetime, Int
+
+from lp import _
+from lp.answers.interfaces.question import IQuestion, IQuestionMessage
+from lp.bugs.interfaces.bug import IBug
+from lp.bugs.interfaces.bugmessage import IBugMessage
+from lp.registry.interfaces.person import IPerson
+from lp.registry.interfaces.product import IProduct
+
+"""Interfaces for spam flagging."""
+
+__all__ = [
+    "ISpamObject",
+    "ISpamObjectSet",
+]
+
+
+class ISpamObject(Interface):
+    """A single item flagged as spam."""
+
+    id = Int(title=_("Primary key"))
+
+    flagged_at = Datetime(
+        title=_("Time when the ISpamObject was flagged as spam.")
+    )
+    flagged_by = Reference(
+        IPerson,
+        title=_("The person who flagged the ISpamObject as spam"),
+        required=True,
+        readonly=True,
+    )
+
+    # Exactly one of the following targets will be non-null.
+    project = Reference(
+        IProduct,
+        title=_("Project flagged as spam."),
+        required=False,
+    )
+    bug = Reference(
+        IBug,
+        title=_("Bug flagged as spam."),
+        required=False,
+    )
+    bug_comment = Reference(
+        IBugMessage,
+        title=_("Bug comment flagged as spam."),
+        required=False,
+    )
+    question = Reference(
+        IQuestion,
+        title=_("Question flagged as spam."),
+        required=False,
+    )
+    question_comment = Reference(
+        IQuestionMessage,
+        title=_("Question comment flagged as spam."),
+        required=False,
+    )
+    team_or_user = Reference(
+        IPerson,
+        title=_("Team or user flagged as spam."),
+        required=False,
+    )
+    spam_owner = Reference(
+        IPerson,
+        title=_("The suspected spammer (canonical Person/Team)."),
+        required=True,
+    )
+
+    def show():
+        """Return a structured view of the spamtarget for review."""
+
+    def handleSpam():
+        """Execute the spam handling action for this target."""
+
+
+class ISpamObjectSet(Interface):
+
+    def get(spam_id):
+        """Return the `ISpamObject` with its id or raise NotFoundError."""
+
+    def new(flagged_by, spam_owner, **target_ids):
+        """Create a new `ISpamObject` and return it."""
diff --git a/lib/lp/services/spam/model.py b/lib/lp/services/spam/model.py
index e69de29..afc2101 100644
--- a/lib/lp/services/spam/model.py
+++ b/lib/lp/services/spam/model.py
@@ -0,0 +1,159 @@
+# Copyright 2015-2025 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from datetime import timezone
+
+from storm.locals import DateTime, Int, Reference
+from zope.interface import implementer
+
+from lp.app.errors import NotFoundError
+from lp.registry.interfaces.person import validate_public_person
+from lp.services.database.constants import UTC_NOW
+from lp.services.database.interfaces import IStore
+from lp.services.database.stormbase import StormBase
+from lp.services.spam.interface import ISpamObject, ISpamObjectSet
+
+__all__ = [
+    "SpamObject",
+    "SpamObjectSet",
+]
+
+
+@implementer(ISpamObject)
+class SpamObject(StormBase):
+    """An entry for a flagged spam target"""
+
+    __storm_table__ = "SpamObject"
+
+    id = Int(primary=True)
+
+    # Who flagged it
+    flagged_by_id = Int(
+        name="flagged_by", allow_none=False, validator=validate_public_person
+    )
+    flagged_by = Reference(flagged_by_id, "Person.id")
+    flagged_at = DateTime(
+        allow_none=False, default=UTC_NOW, tzinfo=timezone.utc
+    )
+
+    # Exactly one of these target pointers is non-null
+    project_id = Int(name="project")
+    project = Reference(project_id, "Product.id")
+    bug_id = Int(name="bug")
+    bug = Reference(bug_id, "Bug.id")
+    bug_comment_id = Int(name="bug_comment")
+    bug_comment = Reference(bug_comment_id, "BugMessage.id")
+    question_id = Int(name="question")
+    question = Reference(question_id, "Question.id")
+    question_comment_id = Int(name="question_comment")
+    question_comment = Reference(question_comment_id, "QuestionMessage.id")
+    team_or_user_id = Int(
+        name="team_or_user", allow_none=True, validator=validate_public_person
+    )
+    team_or_user = Reference(team_or_user_id, "Person.id")
+
+    # Spammer
+    spam_owner_id = Int(
+        name="spam_owner", allow_none=False, validator=validate_public_person
+    )
+    spam_owner = Reference(spam_owner_id, "Person.id")
+
+    def __init__(self, flagged_by, spam_owner, **target_ids):
+        super().__init__()
+        self.flagged_by = flagged_by
+        self.spam_owner = spam_owner
+        # exactly one target id should be present
+        for k, v in target_ids.items():
+            if v is not None:
+                setattr(self, k, v)
+
+    def show(self):
+        """Return a JSOn view of the spam target"""
+        data = {
+            "spam_id": self.id,
+            "flagged_at": self.flagged_at,
+            "flagged_by": getattr(self.flagged_by, "displayname", None),
+            "spam_owner": getattr(self.spam_owner, "displayname", None),
+        }
+
+        # probably better way to simplify this
+        if self.project:
+            data["type"] = "project"
+            data["target_id"] = self.project.id
+            data["name"] = self.project.name
+        elif self.bug:
+            data["type"] = "bug"
+            data["target_id"] = self.bug.id
+        elif self.bug_comment:
+            data["type"] = "bug_comment"
+            data["target_id"] = self.bug_comment.id
+            data["message"] = getattr(self.bug_comment, "text", None)
+        elif self.question:
+            data["type"] = "question"
+            data["target_id"] = self.question.id
+        elif self.question_comment:
+            data["type"] = "question_comment"
+            data["target_id"] = self.question_comment.id
+            data["message"] = getattr(self.question_comment, "text", None)
+        elif self.team_or_user:
+            data["type"] = "team_or_user"
+            data["target_id"] = self.team_or_user.id
+
+        return data
+
+    def handleSpam(self):
+        """Delete the target and remove the SpamObject entry."""
+        # to do: hide or delete the spam
+
+        # Remove entry
+        store = IStore(SpamObject)
+
+        if self.project:
+            store.remove(self.project)
+        elif self.bug:
+            store.remove(self.bug)
+        elif self.bug_comment:
+            store.remove(self.bug_comment)
+        elif self.question:
+            store.remove(self.question)
+        elif self.question_comment:
+            store.remove(self.question_comment)
+        elif self.team_or_user:
+            store.remove(self.team_or_user)
+        else:
+            raise AssertionError("SpamObject has no valid target")
+        store.remove(self)
+        store.flush()
+        return True
+
+
+@implementer(ISpamObjectSet)
+class SpamObjectSet:
+    """See ISpamObjectSet."""
+
+    def getSpamObject(self, spam_id):
+        spam_object = IStore(SpamObject).get(SpamObject, spam_id)
+        if spam_object is None:
+            raise NotFoundError(
+                "Unable to locate spam object with ID %s." % str(spam_id)
+            )
+        return spam_object
+
+    def new(self, flagged_by, spam_owner, **target_ids):
+        """See 'ISpamObjectSet'. Create a new SpamObject.
+
+        Example: set.new(flagged_by=person, spam_owner=spammer, bug_id=1234)
+
+        """
+        non_null = sum(1 for v in target_ids.values() if v is not None)
+        if non_null != 1:
+            raise AssertionError("Exactly one target id must be provided")
+        spam = SpamObject(
+            flagged_by=flagged_by, spam_owner=spam_owner, **target_ids
+        )
+
+        store = IStore(SpamObject)
+        store.add(spam)
+        store.flush()
+        # notify(ObjectCreatedEvent(spam))
+        return spam