Coverage for oc_meta / run / meta / convert_citations.py: 96%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import os 

8from argparse import ArgumentParser 

9 

10from rich_argparse import RichHelpFormatter 

11 

12from oc_meta.lib.cleaner import normalize_hyphens, normalize_id 

13from oc_meta.lib.console import console, create_progress 

14from oc_meta.lib.file_manager import get_csv_data, write_csv 

15 

16CITING_COL = "citing_id" 

17CITED_COL = "cited_id" 

18 

19 

20def _csv_files_in_dir(directory: str) -> list[str]: 

21 return sorted( 

22 os.path.join(directory, f.name) 

23 for f in os.scandir(directory) 

24 if f.name.endswith(".csv") and f.is_file() 

25 ) 

26 

27 

28def _count_csv_rows(filepath: str) -> int: 

29 with open(filepath, "rb") as f: 

30 return sum(1 for _ in f) - 1 

31 

32 

33def build_id_to_omid_mapping(meta_output_dir: str) -> dict[str, str]: 

34 mapping: dict[str, str] = {} 

35 csv_files = _csv_files_in_dir(meta_output_dir) 

36 for i, csv_path in enumerate(csv_files, 1): 

37 console.print( 

38 f" Reading file {i}/{len(csv_files)}: {os.path.basename(csv_path)}" 

39 ) 

40 for row in get_csv_data(csv_path): 

41 parts = row["id"].strip().split() 

42 omids = [p for p in parts if p.startswith("omid:")] 

43 others = [p for p in parts if not p.startswith("omid:")] 

44 for omid in omids: 

45 for other in others: 

46 mapping[other.lower()] = omid 

47 return mapping 

48 

49 

50def convert_citations( 

51 meta_output_dir: str, 

52 citations_dir: str, 

53 output_dir: str, 

54) -> None: 

55 os.makedirs(output_dir, exist_ok=True) 

56 

57 citation_files = _csv_files_in_dir(citations_dir) 

58 total_cit_rows = sum(_count_csv_rows(f) for f in citation_files) 

59 

60 total_resolved = 0 

61 total_unresolved_citing = 0 

62 total_unresolved_cited = 0 

63 total_invalid = 0 

64 all_orphan_ids: set[str] = set() 

65 

66 console.print("[bold]Building ID→OMID mapping…[/bold]") 

67 mapping = build_id_to_omid_mapping(meta_output_dir) 

68 console.print(f" Loaded {len(mapping)} ID→OMID entries") 

69 

70 with create_progress() as progress: 

71 cit_task = progress.add_task("Converting citations", total=total_cit_rows) 

72 for cit_path in citation_files: 

73 out_path = os.path.join(output_dir, os.path.basename(cit_path)) 

74 

75 resolved = 0 

76 unresolved_citing = 0 

77 unresolved_cited = 0 

78 output_rows: list[dict[str, str]] = [] 

79 

80 for row in get_csv_data(cit_path): 

81 raw_citing = row[CITING_COL].strip() 

82 raw_cited = row[CITED_COL].strip() 

83 

84 citing_id = normalize_id(normalize_hyphens(raw_citing)) 

85 cited_id = normalize_id(normalize_hyphens(raw_cited)) 

86 

87 if not citing_id or not cited_id: 

88 total_invalid += 1 

89 progress.advance(cit_task) 

90 continue 

91 

92 citing_omid = mapping.get(citing_id.lower()) 

93 cited_omid = mapping.get(cited_id.lower()) 

94 

95 if citing_omid and cited_omid: 

96 output_rows.append({"citing": citing_omid, "cited": cited_omid}) 

97 resolved += 1 

98 else: 

99 if not citing_omid: 

100 unresolved_citing += 1 

101 all_orphan_ids.add(citing_id) 

102 if not cited_omid: 

103 unresolved_cited += 1 

104 all_orphan_ids.add(cited_id) 

105 progress.advance(cit_task) 

106 

107 write_csv(out_path, output_rows) 

108 total_resolved += resolved 

109 total_unresolved_citing += unresolved_citing 

110 total_unresolved_cited += unresolved_cited 

111 

112 console.print("\n[bold]Results:[/bold]") 

113 console.print(f" Total citations: {total_cit_rows}") 

114 console.print(f" Resolved: {total_resolved}") 

115 console.print(f" Invalid IDs: {total_invalid}") 

116 console.print(f" Unresolved citing: {total_unresolved_citing}") 

117 console.print(f" Unresolved cited: {total_unresolved_cited}") 

118 console.print(f" Unique orphan IDs: {len(all_orphan_ids)}") 

119 

120 

121if __name__ == "__main__": # pragma: no cover 

122 arg_parser = ArgumentParser( 

123 "convert_citations.py", 

124 description="Convert citation CSVs by replacing source IDs with OMIDs from Meta output CSVs. " 

125 "Both --meta-output and --citations take directory paths. " 

126 "Validates transitive closure and reports any unresolvable citation IDs.", 

127 formatter_class=RichHelpFormatter, 

128 ) 

129 arg_parser.add_argument( 

130 "-m", 

131 "--meta-output", 

132 required=True, 

133 help="Directory containing Meta output CSVs", 

134 ) 

135 arg_parser.add_argument( 

136 "-c", 

137 "--citations", 

138 required=True, 

139 help="Directory containing input citation CSVs", 

140 ) 

141 arg_parser.add_argument( 

142 "-o", 

143 "--output", 

144 required=True, 

145 help="Directory for output citation CSVs (with 'citing' and 'cited' columns)", 

146 ) 

147 args = arg_parser.parse_args() 

148 convert_citations( 

149 meta_output_dir=args.meta_output, 

150 citations_dir=args.citations, 

151 output_dir=args.output, 

152 )