Coverage for src/api/indexapi_common.py: 100%

1from datetime import datetime 

2from dateutil.relativedelta import relativedelta 

3 

4 

5def lower(s): 

6 return (s.lower(),) 

7 

8 

9def oci_search(oci): 

10 citing, cited = oci.split("-") 

11 return ( 

12 f"VALUES ?citation {{ <https://w3id.org/oc/index/ci/{oci}> }} " 

13 f"VALUES ?citing_br {{ <https://w3id.org/oc/meta/br/{citing}> }} " 

14 f"VALUES ?cited_br {{ <https://w3id.org/oc/meta/br/{cited}> }}", 

15 ) 

16 

17 

18def literal_search(value, variable): 

19 return ( 

20 f'?resolved_identifier literal:hasLiteralValue "{value}" . ' 

21 f"?{variable} datacite:hasIdentifier ?resolved_identifier ." 

22 ) 

23 

24 

25def unique_citations(res): 

26 citing_groups: dict[str, str] = {} 

27 cited_groups: dict[str, str] = {} 

28 citations: dict[tuple[str, str], dict[str, str]] = {} 

29 for row in sorted(__rows(res), key=lambda row: (row["citing_br"], row["cited_br"])): 

30 key = ( 

31 __group(row["citing_br"], row["citing_ids"], citing_groups), 

32 __group(row["cited_br"], row["cited_ids"], cited_groups), 

33 ) 

34 citations.setdefault(key, row) 

35 return list(citations.values()) 

36 

37 

38def count_unique_cits(res): 

39 return [["count"], [len(unique_citations(res))]], True 

40 

41 

42def count_groups(res, side): 

43 entities = sorted({(row[f"{side}_br"], row[f"{side}_ids"]) for row in __rows(res)}) 

44 groups: dict[str, str] = {} 

45 return [["count"], [len({__group(*entity, groups) for entity in entities})]], True 

46 

47 

48def citation_row(citation, citing, cited): 

49 return [ 

50 citation["oci"], 

51 citing, 

52 cited, 

53 citation["citing_date"], 

54 cit_duration(citation["citing_date"], citation["cited_date"]), 

55 __yes_no(citation, "same_journal"), 

56 __yes_no(citation, "same_author"), 

57 ] 

58 

59 

60def get_id_val(val): 

61 return val.replace("https://w3id.org/oc/meta/br/", "") 

62 

63 

64def cit_duration(citing_complete_pub_date, cited_complete_pub_date): 

65 

66 def _contains_years(date): 

67 return date is not None and len(date) >= 4 

68 

69 def _contains_months(date): 

70 return date is not None and len(date) >= 7 

71 

72 def _contains_days(date): 

73 return date is not None and len(date) >= 10 

74 

75 consider_years = _contains_years(citing_complete_pub_date) and _contains_years( 

76 cited_complete_pub_date 

77 ) 

78 consider_months = _contains_months(citing_complete_pub_date) and _contains_months( 

79 cited_complete_pub_date 

80 ) 

81 consider_days = _contains_days(citing_complete_pub_date) and _contains_days( 

82 cited_complete_pub_date 

83 ) 

84 

85 if not consider_years: 

86 return "" 

87 citing_pub_datetime = datetime.strptime( 

88 (citing_complete_pub_date + "-01-01")[:10], "%Y-%m-%d" 

89 ) 

90 cited_pub_datetime = datetime.strptime( 

91 (cited_complete_pub_date + "-01-01")[:10], "%Y-%m-%d" 

92 ) 

93 

94 delta = relativedelta(citing_pub_datetime, cited_pub_datetime) 

95 

96 result = "" 

97 if ( 

98 delta.years < 0 

99 or (delta.years == 0 and delta.months < 0 and consider_months) 

100 or (delta.years == 0 and delta.months == 0 and delta.days < 0 and consider_days) 

101 ): 

102 result += "-" 

103 result += "P%sY" % abs(delta.years) 

104 

105 if consider_months: 

106 result += "%sM" % abs(delta.months) 

107 

108 if consider_days: 

109 result += "%sD" % abs(delta.days) 

110 

111 return result 

112 

113 

114def __rows(res): 

115 header = res[0] 

116 return [ 

117 {field: value for field, (_, value) in zip(header, typed_row)} 

118 for typed_row in res[1:] 

119 ] 

120 

121 

122def __group(br, ids, groups): 

123 identifiers = ids.split(" __ ") 

124 for identifier in identifiers: 

125 if identifier in groups: 

126 return groups[identifier] 

127 for identifier in identifiers: 

128 groups[identifier] = br 

129 return br 

130 

131 

132def __yes_no(citation, column): 

133 return "yes" if citation.get(column) == "yes" else "no"