加载目录记录 @time: 2025-02-19 @return: 目录记录信息
(self)
| 1540 | pass |
| 1541 | |
| 1542 | def load_dir_record(self) -> Dict: |
| 1543 | """加载目录记录 |
| 1544 | @time: 2025-02-19 |
| 1545 | @return: 目录记录信息 |
| 1546 | """ |
| 1547 | # try: |
| 1548 | # if os.path.exists(self.__dir_record): |
| 1549 | # with open(self.__dir_record, 'r') as f: |
| 1550 | # return json.load(f) |
| 1551 | # return {'directories': {}, 'last_update': 0} |
| 1552 | # except Exception as e: |
| 1553 | # # logging.error("加载目录记录失败: {}".format(str(e))) |
| 1554 | # return {'directories': {}, 'last_update': 0} |
| 1555 | directories_by_hash = {} |
| 1556 | last_update_ts = 0.0 |
| 1557 | |
| 1558 | if not os.path.exists(self.__dir_record_db_path): |
| 1559 | return {'directories_by_hash': {}, 'last_update': last_update_ts} |
| 1560 | |
| 1561 | conn = self._get_db_conn() |
| 1562 | try: |
| 1563 | cursor = conn.cursor() |
| 1564 | cursor.execute("SELECT dir_path_md5_suffix_16, mtime, file_count, depth FROM directory_info") |
| 1565 | rows = cursor.fetchall() |
| 1566 | for row in rows: |
| 1567 | hash_suffix, mtime, file_count, depth = row |
| 1568 | directories_by_hash[hash_suffix] = { |
| 1569 | 'mtime': mtime, |
| 1570 | 'file_count': file_count, |
| 1571 | 'depth': depth |
| 1572 | } |
| 1573 | |
| 1574 | cursor.execute("SELECT meta_value FROM scan_metadata WHERE meta_key = ?", ('last_dir_record_update',)) |
| 1575 | row_meta = cursor.fetchone() |
| 1576 | if row_meta: |
| 1577 | try: |
| 1578 | last_update_ts = float(row_meta[0]) |
| 1579 | except ValueError: |
| 1580 | last_update_ts = 0.0 |
| 1581 | except sqlite3.Error as e: |
| 1582 | return {'directories_by_hash': {}, 'last_update': 0.0} |
| 1583 | finally: |
| 1584 | conn.close() |
| 1585 | |
| 1586 | return {'directories_by_hash': directories_by_hash, 'last_update': last_update_ts} |
| 1587 | |
| 1588 | def save_dir_record(self, record_data: Dict) -> None: |
| 1589 | """保存目录记录 |
no test coverage detected