Manage file added again to same session, data is updated.

CHANGES.rst

@@ -31,6 +31,8 @@ Changelog
   [chris-adam, sgeulette]
 - Added possibility to have elements of the same context to belong to different sessions.
   [chris-adam]
+- Manage file added again to same session, data is updated.
+  [gbastien]
 
 1.0a2 (2026-02-06)
 ------------------

checkouts.cfg

@@ -1,6 +1,7 @@
 [buildout]
 always-checkout = force
 auto-checkout +=
+    collective.iconifiedcategory
     imio.annex
     imio.helpers
     imio.pyutils
@@ -19,6 +20,8 @@ collective = https://github.com/collective
 collective_push = git@github.com:collective
 
 [sources]
+collective.iconifiedcategory = git ${remotes:collective}/collective.iconifiedcategory.git pushurl=${remotes:collective_push}/collective.iconifiedcategory.git
+imio.annex = git ${remotes:imio}/imio.annex.git pushurl=${remotes:imio_push}/imio.annex.git
 imio.annex = git ${remotes:imio}/imio.annex.git pushurl=${remotes:imio_push}/imio.annex.git
 imio.helpers = git ${remotes:imio}/imio.helpers.git pushurl=${remotes:imio_push}/imio.helpers.git
 imio.pyutils = git ${remotes:imio}/imio.pyutils.git pushurl=${remotes:imio_push}/imio.pyutils.git

src/imio/esign/configure.zcml

@@ -8,6 +8,7 @@
 
   <i18n:registerTranslations directory="locales" />
 
+  <include file="events.zcml" />
   <include file="permissions.zcml" />
 
   <include package=".browser" />

src/imio/esign/events.py

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+
+from imio.esign.utils import get_sessions_for
+
+
+def on_categorized_annex_updated(annex, event):
+    '''When an annex is modified, update check if need to update esign session.'''
+    old_values = event.old_values
+    # we are creating a new annex, not in a session
+    if not old_values:
+        return
+
+    sessions = get_sessions_for(event.parent.UID(), readonly=False)
+    if not sessions:
+        return
+
+    new_values = event.new_values
+    # if something usefull changed, we will update the session
+    update = False
+    checked_keys = ['title', 'filesize', 'relative_url']
+    for checked_key in checked_keys:
+        if new_values[checked_key] != old_values[checked_key]:
+            update = True
+            break
+    if update:
+        annex_uid = annex.UID()
+        for session_id, session in sessions.items():
+            # size
+            size_diff = new_values['filesize'] - old_values['filesize']
+            session['size'] += size_diff
+            # title and filename
+            for file_data in session['files']:
+                if file_data['uid'] == annex_uid:
+                    file_data['title'] = new_values['title']
+                    file_data['filename'] = annex.file.filename

src/imio/esign/events.zcml

@@ -0,0 +1,8 @@
+<configure
+    xmlns="http://namespaces.zope.org/zope">
+
+ <subscriber for="imio.annex.content.annex.IAnnex
+                  collective.iconifiedcategory.interfaces.ICategorizedElementUpdatedEvent"
+             handler=".events.on_categorized_annex_updated" />
+
+</configure>

src/imio/esign/tests/test_utils.py

@@ -30,6 +30,8 @@
 from plone.namedfile.file import NamedBlobFile
 from plone.namedfile.file import NamedBlobImage
 from zope.annotation import IAnnotations
+from zope.event import notify
+from zope.lifecycleevent import ObjectModifiedEvent
 
 import collective.iconifiedcategory
 import json
@@ -452,6 +454,66 @@ def test_add_files_with_duplicate_filenames(self):
         self.assertIn("same_filename-1.pdf", filenames)
         self.assertIn("same_filename-2.pdf", filenames)
 
+    def test_add_files_already_exist_is_updated(self):
+        """When adding a file to the same session, it is updated."""
+        annot = get_session_annotation()
+        self.assertEqual(len(annot["sessions"]), 0)
+
+        signers = [
+            ("user1", "user1@sign.com", "User 1", "Position 1"),
+        ]
+
+        annex0_uid = self.uids[0]
+        annex0 = api.content.get(UID=annex0_uid)
+
+        sid, session = add_files_to_session(signers, (annex0_uid,))
+        self.assertEqual(sid, 0)
+        self.assertEqual(len(session["files"]), 1)
+        self.assertEqual(session["files"][0]["filename"], "annex0.pdf")
+        self.assertEqual(session["files"][0]["title"], "Annex 0")
+        self.assertEqual(session["size"], 6968)
+        # edit annex and add again, still one annex in session and data are updated
+        annex0.file.filename = u"new_annex0.pdf"
+        annex0.setTitle('New Annex 0')
+        sid, session = add_files_to_session(signers, (annex0_uid,))
+        # same session_id
+        self.assertEqual(sid, 0)
+        self.assertEqual(len(session["files"]), 1)
+        self.assertEqual(session["files"][0]["filename"], "new_annex0.pdf")
+        self.assertEqual(session["files"][0]["title"], "New Annex 0")
+        self.assertEqual(session["size"], 6968)
+        # add again exact same file
+        sid, session = add_files_to_session(signers, (annex0_uid,))
+        self.assertEqual(sid, 0)
+        self.assertEqual(len(session["files"]), 1)
+        self.assertEqual(session["files"][0]["filename"], "new_annex0.pdf")
+        self.assertEqual(session["files"][0]["title"], "New Annex 0")
+        self.assertEqual(session["size"], 6968)
+        # add second file 2 times
+        annex1_uid = self.uids[1]
+        annex1 = api.content.get(UID=annex1_uid)
+        sid, session = add_files_to_session(signers, (annex1_uid,))
+        self.assertEqual(sid, 0)
+        self.assertEqual(len(session["files"]), 2)
+        self.assertEqual(session["files"][1]["filename"], "annex1.pdf")
+        self.assertEqual(session["files"][1]["title"], "Annex 1")
+        self.assertEqual(session["size"], 13982)
+        # edit and add again
+        annex1.setTitle('New Annex 1')
+        # edit file, filename and content so size changed
+        with open(os.path.join(os.path.dirname(__file__), "annex1.pdf"), "rb") as f:
+            annex1.file = NamedBlobFile(data=f.read(), filename=u"new_annex1.pdf", contentType="application/pdf")
+        notify(ObjectModifiedEvent(annex1))
+        sid, session = add_files_to_session(signers, (annex1_uid,))
+        self.assertEqual(sid, 0)
+        self.assertEqual(len(session["files"]), 2)
+        self.assertEqual(session["files"][1]["filename"], "new_annex1.pdf")
+        self.assertEqual(session["files"][1]["title"], "New Annex 1")
+        self.assertEqual(session["size"], 13936)
+        # just to check, remove annex1
+        remove_files_from_session((annex0_uid,))
+        self.assertEqual(session["size"], 6968)
+
     def test_remove_context_from_session(self):
         """Test removing a context from a session."""
         annot = get_session_annotation()

src/imio/esign/utils.py

@@ -86,6 +86,15 @@ def add_files_to_session(
         annex = uuidToObject(uuid=uid, unrestricted=True)
         context_uid_provider = getAdapter(annex, IContextUidProvider)
         context_uid = context_uid_provider.get_context_uid()
+        # update data if adding same file to same session
+        if annot['uids'].get(uid, -1) == session_id:
+            logger.info('File with UID %s is already in session_id %s and data were updated!', uid, session_id)
+            # remove old filename to avoid filename being renamed
+            old_filename = path.splitext([fn for fn in session["files"]
+                                          if fn['uid'] == uid][0]['filename'])[0]
+            existing_files.remove(old_filename)
+            remove_files_from_session([uid], remove_empty_session=False)
+
         filename, ext = path.splitext(annex.file.filename or "no_filename.pdf")
         new_filename = get_correct_id(existing_files, filename)
         session["files"].append(
@@ -342,6 +351,17 @@ def get_session_info(session_id, portal=None):
         return annot['sessions'][session_id]
 
 
+def get_sessions_for(context_uid):
+    """ """
+    sessions = []
+    annot = get_session_annotation()
+    for f_uid in annot["c_uids"].get(context_uid, []):
+        session_id = annot["uids"].get(f_uid)
+        if session_id is not None:
+            sessions.append(annot["sessions"].get(session_id))
+    return sessions
+
+
 def remove_context_from_session(context_uids):
     """Remove all files from a session that are linked to the given context UIDs.
 
@@ -356,10 +376,11 @@ def remove_context_from_session(context_uids):
         remove_files_from_session(list(c_uids[context_uid]))
 
 
-def remove_files_from_session(files_uids):
+def remove_files_from_session(files_uids, remove_empty_session=True):
     """Remove files from their corresponding sessions.
 
     :param files_uids: list of file UIDs to remove
+    :param remove_empty_session: when the last file of a session is removed the session will be removed by default, except when False, the empty session is kept
     """
     annot = get_session_annotation()
     sessions = annot["sessions"]
@@ -389,7 +410,7 @@ def remove_files_from_session(files_uids):
             continue
 
         del session["files"][i]
-        if not session["files"]:
+        if not session["files"] and remove_empty_session:
             del sessions[session_id]
         else:
             session["last_update"] = datetime.now()

test-4.3.cfg

@@ -150,3 +150,8 @@ Pygments = 2.5.2
 # beautifulsoup
 soupsieve = 1.9.2
 backports.functools-lru-cache = 1.5
+
+# Required by:
+# mock
+mock = 3.0.5
+funcsigs = 1.0.2