def _make_input_csv()

in fbpcs/scripts/gen_fake_data.py [0:0]

• 64 lines of code
• 8 McCabe index (conditional complexity)

def _make_input_csv(args: Dict[str, Any]) -> None:

    if args.get("<input_path>") is None:
        header_line = args.get("--from_header").split(",")
        header = [InputColumn.from_str(s) for s in header_line]

        with open(args["<output_path>"], "w") as f_out:
            f_out.write(args.get("--from_header") + "\n")

            for i in range(args["--num_records"]):
                out_line = _generate_line(
                    i,
                    "",
                    header,
                    args["--opportunity_rate"],
                    args["--test_rate"],
                    args["--purchase_rate"],
                    args["--incrementality_rate"],
                    args["--min_ts"],
                    args["--max_ts"],
                    args["--num_conversions"],
                    args["--md5_id"],
                )
                f_out.write(",".join(out_line) + "\n")
    else:
        with open(args["<input_path>"]) as f_in, open(
            args["<output_path>"], "w"
        ) as f_out:
            header_line = f_in.readline().strip().split(",")
            header = [InputColumn.from_str(s) for s in header_line]
            f_out.write(",".join(header_line) + "\n")

            if args.get("--num_records") is not None:
                for i in range(args["--num_records"]):
                    line = f_in.readline().strip()
                    out_line = _generate_line(
                        i,
                        line,
                        header,
                        args["--opportunity_rate"],
                        args["--test_rate"],
                        args["--purchase_rate"],
                        args["--incrementality_rate"],
                        args["--min_ts"],
                        args["--max_ts"],
                        args["--num_conversions"],
                        args["--md5_id"],
                    )
                    f_out.write(",".join(out_line) + "\n")
            else:
                i = 0
                line = f_in.readline().strip()
                while line != "":
                    out_line = _generate_line(
                        i,
                        line,
                        header,
                        args["--opportunity_rate"],
                        args["--test_rate"],
                        args["--purchase_rate"],
                        args["--incrementality_rate"],
                        args["--min_ts"],
                        args["--max_ts"],
                        args["--num_conversions"],
                    )
                    f_out.write(",".join(out_line) + "\n")
                    line = f_in.readline().strip()
                    i += 1