handle upload-link-flags with database

Author: AlexCXC

dtable_events/tasks/dtable_upload_link_handler.py

@@ -1,121 +1,91 @@
 # -*- coding: utf-8 -*-
 import logging
-import time
-import json
 import stat
+from collections import defaultdict
 from datetime import datetime, timedelta
-from threading import Thread, Event
+from threading import Thread
 
 from apscheduler.schedulers.blocking import BlockingScheduler
 
 from seaserv import seafile_api
 
-from dtable_events.app.event_redis import RedisClient, redis_cache
 from dtable_events.db import init_db_session_class
-from dtable_events.utils import uuid_str_to_36_chars, uuid_str_to_32_chars
+from dtable_events.utils import uuid_str_to_36_chars
 
 logger = logging.getLogger(__name__)
 
 
 class DTableUploadLinkHandler(Thread):
     def __init__(self, config):
         Thread.__init__(self)
-        self._finished = Event()
-        self._redis_client = RedisClient(config)
+        self.session_class = init_db_session_class(config)
         self.interval_hours = 6
-        self.cache_timeout = 12 * 60 * 60
 
-    def get_cache_key(self, hour):
-        return f'public_form_upload_link:{hour}'
-
-    def handle_upload_link(self, event_data):
-        dtable_uuid = uuid_str_to_32_chars(event_data.get('dtable_uuid'))
-        repo_id = event_data.get('repo_id')
-        parent_dir = event_data.get('parent_dir') or ''
-        if 'public/forms' not in parent_dir:
-            return
-        cache_key = self.get_cache_key((datetime.now().hour+1)%24)
-        dtable_uuids_cache = redis_cache.get(cache_key)
-        if dtable_uuids_cache:
-            try:
-                dtable_uuids = json.loads(dtable_uuids_cache)
-            except Exception as e:
-                logger.warning('key: %s cache invalid', cache_key)
-                dtable_uuids = [(dtable_uuid, repo_id)]
-            else:
-                flag = True
-                for tmp_dtable_uuid, _ in dtable_uuids:
-                    if dtable_uuid == tmp_dtable_uuid:
-                        flag = False
-                        break
-                if flag:
-                    dtable_uuids.append((dtable_uuid, repo_id))
-        else:
-            dtable_uuids = [(dtable_uuid, repo_id)]
-        redis_cache.set(cache_key, json.dumps(dtable_uuids), self.cache_timeout)
-
-    def listen_redis(self):
-        logger.info('Starting handle dtable upload link...')
-        subscriber = self._redis_client.get_subscriber('upload-link')
-        while not self._finished.is_set():
+    def handle_flags(self, session):
+        now = datetime.now()
+        flag_time = (now - timedelta(hours=self.interval_hours)).replace(minute=0, second=0, microsecond=0)
+        offset, limit = 0, 1000
+        while True:
+            sql = "SELECT dtable_uuid, repo_id FROM dtable_form_upload_link_flags WHERE flag_time=:flag_time LIMIT :offset, :limit"
             try:
-                message = subscriber.get_message()
-                if message is not None:
-                    data = json.loads(message['data'])
-                    try:
-                        self.handle_upload_link(data)
-                    except Exception as e:
-                        logger.error('Handle dtable upload link: %s' % e)
-                else:
-                    time.sleep(0.5)
+                results = list(session.execute(sql, {'flag_time': flag_time, 'offset': offset, 'limit': limit}))
             except Exception as e:
-                logger.error('Failed get message from redis: %s' % e)
-                subscriber = self._redis_client.get_subscriber('upload-link')
+                logger.error('query upload flags flag_time: %s error: %s', flag_time, e)
+                break
+            logger.debug('flag_time: %s offset: %s limit: %s query results: %s', flag_time, offset, limit, len(results))
+            repo_id_dtable_uuids_dict = defaultdict(list)
+            for dtable_uuid, repo_id in results:
+                repo_id_dtable_uuids_dict[repo_id].append(dtable_uuid)
+            for repo_id, dtable_uuids in repo_id_dtable_uuids_dict.items():
+                logger.debug('repo: %s dtable_uuids: %s', repo_id, len(dtable_uuids))
+                try:
+                    repo = seafile_api.get_repo(repo_id)
+                    if not repo:
+                        continue
+                    for dtable_uuid in dtable_uuids:
+                        public_forms_path = f'/asset/{uuid_str_to_36_chars(dtable_uuid)}/public/forms'
+                        dir_id = seafile_api.get_dir_id_by_path(repo_id, public_forms_path)
+                        if not dir_id:
+                            continue
+                        f_offset, f_limit= 0, 1000
+                        to_delete_files = []
+                        while True:
+                            dirents = seafile_api.list_dir_by_path(repo_id, public_forms_path, f_offset, f_limit)
+                            if not dirents:
+                                break
+                            for dirent in dirents:
+                                if stat.S_ISDIR(dirent.mode):
+                                    continue
+                                if (now.timestamp() - datetime.fromtimestamp(dirent.mtime)) > self.interval_hours * 60 * 60:
+                                    to_delete_files.append(dirent.obj_name)
+                            if len(dirents) < f_limit:
+                                break
+                            f_offset += f_limit
+                        logger.debug('repo: %s dtable: %s to delete files: %s', repo_id, dtable_uuid, len(to_delete_files))
+                        for file in to_delete_files:
+                            seafile_api.del_file(repo_id, public_forms_path, file, '')
+                except Exception as e:
+                    logger.exception('repo: %s handle upload flags error: %s', repo_id, e)
+            offset += limit
+        sql = "DELETE FROM dtable_form_upload_link_flags WHERE flag_time <= :flag_time"
+        try:
+            session.execute(sql, {'flag_time': flag_time})
+            session.commit()
+        except Exception as e:
+            logger.error('delete upload flags old data flag time: %s error: %s', flag_time, e)
 
-    def handle_form_timeout_images(self):
+    def run(self):
+        logger.info('start to handle upload flags...')
         sched = BlockingScheduler()
 
         @sched.scheduled_job('cron', day_of_week='*', hour='*')
         def handle():
-            handle_hour = (datetime.now() - timedelta(hours=self.interval_hours)).hour
-            cache_key = self.get_cache_key(handle_hour)
-            dtable_uuids_cache = redis_cache.get(cache_key)
-            if not dtable_uuids_cache:
-                return 
+            session = self.session_class()
             try:
-                dtable_uuids = json.loads(dtable_uuids_cache)
+                self.handle_flags(session)
             except Exception as e:
-                logger.warning('cache: %s loads dtable uuids error: %s', cache_key, e)
-                return
-            now_timestamp = datetime.now().timestamp()
-            logger.debug('start to check key: %s dtable_uuids: %s ', cache_key, len(dtable_uuids))
-            for dtable_uuid, repo_id in dtable_uuids:
-                try:
-                    public_forms_path = f'/asset/{uuid_str_to_36_chars(dtable_uuid)}/public/forms'
-                    logger.debug('start to scan repo: %s dtable: %s path: %s', repo_id, dtable_uuid, public_forms_path)
-                    dirents = []
-                    limit, offset = 1000, 0
-                    while True:
-                        step_dirents = seafile_api.list_dir_by_path(repo_id, public_forms_path, offset, limit)
-                        if not step_dirents:
-                            break
-                        dirents.extend(step_dirents)
-                        if len(step_dirents) < limit:
-                            break
-                        offset += limit
-                    logger.debug('repo: %s dtable: %s path: %s dirents: %s', repo_id, dtable_uuid, public_forms_path, len(dirents))
-                    for dirent in dirents:
-                        if stat.S_ISDIR(dirent.mode):
-                            continue
-                        if dirent.mtime and now_timestamp - dirent.mtime > self.interval_hours * 60 * 60:
-                            seafile_api.del_file(repo_id, public_forms_path, json.dumps([dirent.obj_name]), '')
-                except Exception as e:
-                    logger.exception('scan repo: %s dtable: %s path: %s error: %s', repo_id, dtable_uuid, public_forms_path, e)
-            redis_cache.delete(cache_key)
+                logger.exception('handle upload flags error: %s', e)
+            finally:
+                session.close()
 
         sched.start()
-
-    def run(self):
-        logger.info('start to listen upload-link redis and handle form timeout images...')
-        Thread(target=self.listen_redis, daemon=True).start()
-        Thread(target=self.handle_form_timeout_images, daemon=True).start()