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

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 # ssh://[user@]host[:port]/path or scp-like ssh://[user@]host:path 

45 if match := re.fullmatch(r"ssh://(?:[^@/]+@)?([^:/]+)(?::\d+)?[:/](.+)", value): 

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

47 return value 

48 

49 

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

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

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

53 for field in type(existing).model_fields: 

54 old = getattr(existing, field) 

55 new = getattr(incoming, field) 

56 if field == "contribution_types": 

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

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

59 updates[field] = values 

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

61 if new and not old: 

62 updates[field] = True 

63 elif field == "publication_author": 

64 if new is True and old is not True: 

65 updates[field] = True 

66 elif old is None and new is not None: 

67 updates[field] = new 

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

69 updates[field] = new 

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

71 

72 

73def _merge_people( 

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

75) -> list[Person | Entity]: 

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

77 result = list(existing) 

78 for candidate in incoming: 

79 for index, person in enumerate(result): 

80 # same_person is False for a Person/Entity mix 

81 if person.same_person(candidate): 

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

83 break 

84 else: 

85 result.append(candidate) 

86 return result 

87 

88 

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

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

91 if author.name.endswith("[bot]"): 

92 return None 

93 identity = author.name 

94 if author.email: 

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

96 if len(author.name.split()) < 2: 

97 logger.warning( 

98 "Skipping Git author '%s': a person needs given and family names. " 

99 "Add them to the somesy input manually.", 

100 identity, 

101 ) 

102 return None 

103 try: 

104 return Person.from_name_email_string(identity).model_copy( 

105 update={ 

106 "author": author.author, 

107 "publication_author": author.author or None, 

108 "contribution_types": author.contribution_types, 

109 } 

110 ) 

111 except (IndexError, ValueError): 

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

113 return None 

114 

115 

116def merge_metadata( 

117 sources: Iterable[dict[str, Any]], 

118 git: GitMetadata | None = None, 

119 *, 

120 allow_incomplete: bool = False, 

121) -> ProjectMetadata: 

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

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

124 keywords: list[str] = [] 

125 

126 candidates = list(sources) 

127 if git is not None: 

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

129 

130 for source in candidates: 

131 for field in _SCALAR_FIELDS: 

132 value = source.get(field) 

133 if field == "repository": 

134 value = _repository(value) 

135 else: 

136 value = _value(value) 

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

138 data[field] = value 

139 

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

141 if keyword not in keywords: 

142 keywords.append(keyword) 

143 

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

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

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

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

148 

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

150 if isinstance(author, dict): 

151 author = GitAuthor(**author) 

152 if person := _git_person(author): 

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

154 

155 if keywords: 

156 data["keywords"] = keywords 

157 model = PartialProjectMetadata if allow_incomplete else ProjectMetadata 

158 return model(**data)