Coverage for oc_meta / run / meta / merge_csv.py: 100%

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7from __future__ import annotations 

8 

9import argparse 

10import os 

11from typing import List 

12 

13from rich.table import Table 

14from rich_argparse import RichHelpFormatter 

15 

16from oc_meta.lib.console import console, create_progress 

17from oc_meta.lib.file_manager import get_csv_data, write_csv 

18 

19 

20def get_csv_files(directory: str) -> List[str]: 

21 if not os.path.isdir(directory): 

22 raise ValueError("The specified path '{}' is not a directory".format(directory)) 

23 return [ 

24 os.path.join(directory, f) for f in os.listdir(directory) if f.endswith(".csv") 

25 ] 

26 

27 

28def resolve_output_path(output: str) -> str: 

29 if output.endswith(".csv"): 

30 parent_dir = os.path.dirname(output) 

31 if parent_dir: 

32 os.makedirs(parent_dir, exist_ok=True) 

33 return output 

34 os.makedirs(output, exist_ok=True) 

35 return os.path.join(output, "merged.csv") 

36 

37 

38def merge_csv_files(input_dir: str, output_path: str) -> tuple[int, int, List[str]]: 

39 csv_files = get_csv_files(input_dir) 

40 if not csv_files: 

41 return 0, 0, [] 

42 

43 all_rows: List[dict[str, str]] = [] 

44 fieldnames: List[str] = [] 

45 

46 with create_progress() as progress: 

47 task = progress.add_task("Merging CSV files", total=len(csv_files)) 

48 

49 for csv_file in csv_files: 

50 data = get_csv_data(csv_file, clean_data=False) 

51 if data: 

52 if not fieldnames: 

53 fieldnames = list(data[0].keys()) 

54 all_rows.extend(data) 

55 progress.advance(task) 

56 

57 if all_rows: 

58 write_csv(output_path, all_rows, fieldnames=fieldnames) 

59 

60 return len(all_rows), len(csv_files), fieldnames 

61 

62 

63def print_merge_report(total_rows: int, files_processed: int, output_path: str) -> None: 

64 table = Table(title="Merge report") 

65 table.add_column("Metric", style="cyan") 

66 table.add_column("Value", style="green") 

67 

68 table.add_row("Files merged", str(files_processed)) 

69 table.add_row("Total rows", str(total_rows)) 

70 table.add_row("Output file", output_path) 

71 

72 console.print(table) 

73 

74 

75def main() -> int: # pragma: no cover 

76 parser = argparse.ArgumentParser( 

77 description="Merge multiple CSV files into a single file", 

78 formatter_class=RichHelpFormatter, 

79 ) 

80 parser.add_argument("input_dir", help="Directory containing input CSV files") 

81 parser.add_argument( 

82 "output", 

83 help="Output path: if ends with .csv, creates that file; otherwise, creates merged.csv in that folder", 

84 ) 

85 args = parser.parse_args() 

86 

87 if not os.path.isdir(args.input_dir): 

88 console.print( 

89 "[red]Input directory does not exist: {}[/red]".format(args.input_dir) 

90 ) 

91 return 1 

92 

93 output_path = resolve_output_path(args.output) 

94 

95 csv_files = get_csv_files(args.input_dir) 

96 if not csv_files: 

97 console.print( 

98 "[red]No CSV files found in directory: {}[/red]".format(args.input_dir) 

99 ) 

100 return 1 

101 

102 console.print("Found [green]{}[/green] CSV files to merge".format(len(csv_files))) 

103 

104 total_rows, files_processed, _ = merge_csv_files(args.input_dir, output_path) 

105 

106 print_merge_report(total_rows, files_processed, output_path) 

107 

108 return 0 

109 

110 

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

112 main()