Coverage for src/qdrant_loader_core/graph/extractor/localfile.py: 100%

1from __future__ import annotations 

2 

3from pathlib import PurePath 

4from typing import TYPE_CHECKING 

5 

6if TYPE_CHECKING: 

7 from qdrant_loader.core.document import Document 

8 

9from ..models import ( 

10 CoreNodeLabel, 

11 GraphNode, 

12 PersonInfo, 

13) 

14from .base_extractor import BaseEntityExtractor 

15 

16 

17class LocalFileEntityExtractor(BaseEntityExtractor): 

18 """ 

19 Extract graph entities from local filesystem documents. 

20 """ 

21 

22 source_type = "localfile" 

23 

24 def _project(self, doc: Document) -> str | None: 

25 """ 

26 Local files are not associated with a project. 

27 """ 

28 return None 

29 

30 def _extract_people( 

31 self, 

32 doc: Document, 

33 ) -> list[tuple[PersonInfo, str]]: 

34 """ 

35 Local files do not expose author information by default. 

36 """ 

37 return [] 

38 

39 def _extract_container( 

40 self, 

41 doc: Document, 

42 ) -> GraphNode | None: 

43 

44 if not doc.url: 

45 return None 

46 

47 pure_path = PurePath(doc.url) 

48 

49 parent_path = str(pure_path.parent) 

50 directory = "root" if parent_path == "." else parent_path 

51 directory_name = PurePath(directory).name if directory != "root" else "root" 

52 

53 return GraphNode( 

54 id=f"dir:{directory}", 

55 label=CoreNodeLabel.CONTAINER.value, 

56 project=None, 

57 properties={ 

58 "kind": "filesystem_dir", 

59 "name": directory_name, 

60 "path": directory, 

61 }, 

62 ) 

63 

64 def _build_document_node( 

65 self, 

66 doc: Document, 

67 project: str | None, 

68 ) -> GraphNode: 

69 """ 

70 Build a document node enriched with filesystem metadata. 

71 """ 

72 

73 file_name = doc.metadata.get("file_name") 

74 

75 return GraphNode( 

76 id=doc.id, 

77 label=CoreNodeLabel.DOCUMENT.value, 

78 project=project, 

79 properties={ 

80 "title": file_name, 

81 "url": doc.url, 

82 "created_at": doc.metadata.get("created_at"), 

83 "updated_at": doc.metadata.get("updated_at"), 

84 "source_type": self.source_type, 

85 }, 

86 )