(self, params, **kwargs)
| 2451 | return trash_restore_parser |
| 2452 | |
| 2453 | def execute(self, params, **kwargs): |
| 2454 | deleted_records = self.get_deleted_records(params) |
| 2455 | orphaned_records = self.get_orphaned_records(params) |
| 2456 | shared_folders = self.get_shared_folders(params) |
| 2457 | deleted_shared_records = shared_folders.get('records') or {} |
| 2458 | deleted_shared_folders = shared_folders.get('folders') or {} |
| 2459 | if len(deleted_records) == 0 and len(orphaned_records) == 0 and len(deleted_shared_records) == 0 and len(deleted_shared_folders) == 0: |
| 2460 | logging.info('Trash is empty') |
| 2461 | return |
| 2462 | |
| 2463 | records = kwargs.get('records') |
| 2464 | if not isinstance(records, (tuple, list)): |
| 2465 | records = None |
| 2466 | if not records: |
| 2467 | logging.info('records parameter is empty.') |
| 2468 | return |
| 2469 | |
| 2470 | records_to_restore = set() # type: Set[str] |
| 2471 | folders_to_restore = set() # type: Set[str] |
| 2472 | folder_records_to_restore = {} # type: Dict[str, List[str]] |
| 2473 | for rec in records: |
| 2474 | if rec in deleted_records: |
| 2475 | records_to_restore.add(rec) |
| 2476 | elif rec in orphaned_records: |
| 2477 | records_to_restore.add(rec) |
| 2478 | elif rec in deleted_shared_records: |
| 2479 | dsr = deleted_shared_records.get(rec) |
| 2480 | folder_uid = dsr.get('folder_uid') |
| 2481 | record_uid = dsr.get('record_uid') |
| 2482 | if folder_uid and record_uid: |
| 2483 | if folder_uid not in folder_records_to_restore: |
| 2484 | folder_records_to_restore[folder_uid] = [] |
| 2485 | folder_records_to_restore[folder_uid].append(record_uid) |
| 2486 | elif rec in deleted_shared_folders: |
| 2487 | folders_to_restore.add(rec) |
| 2488 | else: |
| 2489 | title_pattern = re.compile(fnmatch.translate(rec), re.IGNORECASE) |
| 2490 | for record_uid, del_rec in itertools.chain(deleted_records.items(), orphaned_records.items()): |
| 2491 | if record_uid in records_to_restore: |
| 2492 | continue |
| 2493 | record = vault.KeeperRecord.load(params, del_rec) |
| 2494 | if title_pattern.match(record.title): |
| 2495 | records_to_restore.add(record_uid) |
| 2496 | for record_uid, sh_rec in deleted_shared_records.items(): |
| 2497 | if record_uid in folder_records_to_restore: |
| 2498 | continue |
| 2499 | record = vault.KeeperRecord.load(params, sh_rec) |
| 2500 | if title_pattern.match(record.title): |
| 2501 | folder_uid = sh_rec.get('folder_uid') |
| 2502 | if folder_uid not in folder_records_to_restore: |
| 2503 | folder_records_to_restore[folder_uid] = [] |
| 2504 | folder_records_to_restore[folder_uid].append(record_uid) |
| 2505 | for folder_uid, sh_fol in deleted_shared_folders.items(): |
| 2506 | if folder_uid in folders_to_restore: |
| 2507 | continue |
| 2508 | try: |
| 2509 | data = json.loads(sh_fol.get('data_unencrypted')) |
| 2510 | folder_name = data.get('name') or folder_uid |
nothing calls this directly
no test coverage detected