Coverage for src/somesy/merge.py: 92%

1"""Merge harvested metadata into the canonical Somesy project model.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import re 

7from collections.abc import Iterable 

8from typing import Any 

9 

10from somesy.core.models import Entity, Person, ProjectMetadata 

11from somesy.git.models import GitAuthor, GitMetadata 

12 

13logger = logging.getLogger("somesy") 

14 

15_SCALAR_FIELDS = ( 

16 "name", 

17 "version", 

18 "description", 

19 "license", 

20 "homepage", 

21 "repository", 

22 "documentation", 

23) 

24 

25 

26def _value(value: Any) -> Any: 

27 """Convert common endpoint wrapper values to project-model values.""" 

28 if hasattr(value, "text"): 

29 return value.text 

30 if isinstance(value, (set, tuple)): 

31 return list(value) 

32 if isinstance(value, list): 

33 return [_value(item) for item in value] 

34 return getattr(value, "value", value) 

35 

36 

37def _repository(value: Any) -> str | None: 

38 """Return a repository URL acceptable to ProjectMetadata when possible.""" 

39 value = _value(value) 

40 if not isinstance(value, str): 

41 return None 

42 if match := re.fullmatch(r"git@([^:]+):(.+)", value): 

43 return f"https://{match[1]}/{match[2]}" 

44 if value.startswith("ssh://git@"): 

45 return "https://" + value.removeprefix("ssh://git@") 

46 return value 

47 

48 

49def _merge_person(existing: Person | Entity, incoming: Person | Entity): 

50 """Fill missing fields and combine roles for one matching person.""" 

51 updates: dict[str, Any] = {} 

52 for field in type(existing).model_fields: 

53 old = getattr(existing, field) 

54 new = getattr(incoming, field) 

55 if field == "contribution_types": 

56 values = list(dict.fromkeys((old or []) + (new or []))) 

57 if values != (old or []): 

58 updates[field] = values 

59 elif field in {"author", "maintainer"}: 

60 if new and not old: 

61 updates[field] = True 

62 elif field == "publication_author": 

63 if new is True and old is not True: 

64 updates[field] = True 

65 elif old is None and new is not None: 

66 updates[field] = new 

67 elif old in (None, "") and new not in (None, ""): 

68 updates[field] = new 

69 return existing.model_copy(update=updates) if updates else existing 

70 

71 

72def _merge_people( 

73 existing: list[Person | Entity], incoming: Iterable[Person | Entity] 

74) -> list[Person | Entity]: 

75 """Merge people using the model's existing identity heuristics.""" 

76 result = list(existing) 

77 for candidate in incoming: 

78 for index, person in enumerate(result): 

79 same_person = ( 

80 isinstance(person, Person) 

81 and isinstance(candidate, Person) 

82 and person.same_person(candidate) 

83 ) 

84 same_entity = ( 

85 isinstance(person, Entity) 

86 and isinstance(candidate, Entity) 

87 and person.same_person(candidate) 

88 ) 

89 if same_person or same_entity: 

90 result[index] = _merge_person(person, candidate) 

91 break 

92 else: 

93 result.append(candidate) 

94 return result 

95 

96 

97def _git_person(author: GitAuthor) -> Person | None: 

98 """Convert a Git author into a Somesy Person.""" 

99 identity = author.name 

100 if author.email: 

101 identity = f"{identity} <{author.email}>" 

102 try: 

103 return Person.from_name_email_string(identity).model_copy( 

104 update={ 

105 "author": author.author, 

106 "contribution_types": author.contribution_types, 

107 } 

108 ) 

109 except (IndexError, ValueError): 

110 logger.warning("Cannot convert Git author '%s' to Somesy metadata.", identity) 

111 return None 

112 

113 

114def merge_metadata( 

115 sources: Iterable[dict[str, Any]], git: GitMetadata | None = None 

116) -> ProjectMetadata: 

117 """Merge harvested endpoint data and Git data into ProjectMetadata.""" 

118 data: dict[str, Any] = {"people": [], "entities": []} 

119 keywords: list[str] = [] 

120 

121 candidates = list(sources) 

122 if git is not None: 

123 candidates.append(git.model_dump(exclude_none=True)) 

124 

125 for source in candidates: 

126 for field in _SCALAR_FIELDS: 

127 value = source.get(field) 

128 if field == "repository": 

129 value = _repository(value) 

130 else: 

131 value = _value(value) 

132 if value not in (None, "") and field not in data: 

133 data[field] = value 

134 

135 for keyword in _value(source.get("keywords", [])) or []: 

136 if keyword not in keywords: 

137 keywords.append(keyword) 

138 

139 people = source.get("people", []) or [] 

140 entities = source.get("entities", []) or [] 

141 data["people"] = _merge_people(data["people"], people) 

142 data["entities"] = _merge_people(data["entities"], entities) 

143 

144 for author in source.get("authors", []) or []: 

145 if isinstance(author, dict): 

146 author = GitAuthor(**author) 

147 if person := _git_person(author): 

148 data["people"] = _merge_people(data["people"], [person]) 

149 

150 if keywords: 

151 data["keywords"] = keywords 

152 return ProjectMetadata(**data)