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, PartialProjectMetadata, 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]], 

116 git: GitMetadata | None = None, 

117 *, 

118 allow_incomplete: bool = False, 

119) -> ProjectMetadata: 

120 """Merge harvested endpoint and Git data, optionally allowing missing fields.""" 

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

122 keywords: list[str] = [] 

123 

124 candidates = list(sources) 

125 if git is not None: 

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

127 

128 for source in candidates: 

129 for field in _SCALAR_FIELDS: 

130 value = source.get(field) 

131 if field == "repository": 

132 value = _repository(value) 

133 else: 

134 value = _value(value) 

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

136 data[field] = value 

137 

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

139 if keyword not in keywords: 

140 keywords.append(keyword) 

141 

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

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

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

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

146 

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

148 if isinstance(author, dict): 

149 author = GitAuthor(**author) 

150 if person := _git_person(author): 

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

152 

153 if keywords: 

154 data["keywords"] = keywords 

155 model = PartialProjectMetadata if allow_incomplete else ProjectMetadata 

156 return model(**data)