(self, params, **kwargs)
| 2380 | return trash_restore_parser |
| 2381 | |
| 2382 | def execute(self, params, **kwargs): |
| 2383 | deleted_records = self.get_deleted_records(params) |
| 2384 | orphaned_records = self.get_orphaned_records(params) |
| 2385 | shared_folders = self.get_shared_folders(params) |
| 2386 | deleted_shared_records = shared_folders.get('records') or {} |
| 2387 | deleted_shared_folders = shared_folders.get('folders') or {} |
| 2388 | if len(deleted_records) == 0 and len(orphaned_records) == 0 and len(deleted_shared_records) == 0 and len(deleted_shared_folders) == 0: |
| 2389 | logging.info('Trash is empty') |
| 2390 | return |
| 2391 | |
| 2392 | records = kwargs.get('records') |
| 2393 | if not isinstance(records, (tuple, list)): |
| 2394 | records = None |
| 2395 | if not records: |
| 2396 | logging.info('records parameter is empty.') |
| 2397 | return |
| 2398 | |
| 2399 | records_to_restore = set() # type: Set[str] |
| 2400 | folders_to_restore = set() # type: Set[str] |
| 2401 | folder_records_to_restore = {} # type: Dict[str, List[str]] |
| 2402 | for rec in records: |
| 2403 | if rec in deleted_records: |
| 2404 | records_to_restore.add(rec) |
| 2405 | elif rec in orphaned_records: |
| 2406 | records_to_restore.add(rec) |
| 2407 | elif rec in deleted_shared_records: |
| 2408 | dsr = deleted_shared_records.get(rec) |
| 2409 | folder_uid = dsr.get('folder_uid') |
| 2410 | record_uid = dsr.get('record_uid') |
| 2411 | if folder_uid and record_uid: |
| 2412 | if folder_uid not in folder_records_to_restore: |
| 2413 | folder_records_to_restore[folder_uid] = [] |
| 2414 | folder_records_to_restore[folder_uid].append(record_uid) |
| 2415 | elif rec in deleted_shared_folders: |
| 2416 | folders_to_restore.add(rec) |
| 2417 | else: |
| 2418 | title_pattern = re.compile(fnmatch.translate(rec), re.IGNORECASE) |
| 2419 | for record_uid, del_rec in itertools.chain(deleted_records.items(), orphaned_records.items()): |
| 2420 | if record_uid in records_to_restore: |
| 2421 | continue |
| 2422 | record = vault.KeeperRecord.load(params, del_rec) |
| 2423 | if title_pattern.match(record.title): |
| 2424 | records_to_restore.add(record_uid) |
| 2425 | for record_uid, sh_rec in deleted_shared_records.items(): |
| 2426 | if record_uid in folder_records_to_restore: |
| 2427 | continue |
| 2428 | record = vault.KeeperRecord.load(params, sh_rec) |
| 2429 | if title_pattern.match(record.title): |
| 2430 | folder_uid = sh_rec.get('folder_uid') |
| 2431 | if folder_uid not in folder_records_to_restore: |
| 2432 | folder_records_to_restore[folder_uid] = [] |
| 2433 | folder_records_to_restore[folder_uid].append(record_uid) |
| 2434 | for folder_uid, sh_fol in deleted_shared_folders.items(): |
| 2435 | if folder_uid in folders_to_restore: |
| 2436 | continue |
| 2437 | try: |
| 2438 | data = json.loads(sh_fol.get('data_unencrypted')) |
| 2439 | folder_name = data.get('name') or folder_uid |
nothing calls this directly
no test coverage detected