如何在Ansible任务中调用Python对CSV按指定列排序并覆盖

问题场景 #

我有一个Ansible任务会将数据保存到/tmp路径下的CSV文件中，生成的CSV包含idm、idc、day、time四列，数据示例如下：

idm;idc;day;time
"034";3;2023-02-09;12:57:34
"056";6;2023-02-10;20:25:12
"078";2;2023-02-11;04:01:00

需要编写一个任务读取该CSV文件，按ORDER BY day, time, idm, idc的规则排序后覆盖原文件，优先通过在Playbook中调用Python实现。

生成CSV的原Ansible任务如下：

- name: Dump data into csv
  copy:
    dest: /tmp/data_{{ '%Y-%m-%d' | strftime }}.csv          
    content: |
      {{ ['idm', 'idc', 'day', 'time'] | map('trim') | join(';') }}
      {% for host in hosts_list %}
      {% set idm = host.inventory_hostname.split('_')[0].split('-')[1] %}
      {% set idm_padded = '%03d' % idm|int %}
      {% set idm_padded = '"' + idm_padded + '"' %}
      {% set idc = host.inventory_hostname.split('_')[1].upper() %}
      {% if host.grep_output.stdout_lines %}
      {% for line in host.grep_output.stdout_lines %}
      {% set day = line.strip().split(' ')[0] %}
      {% set time = line.strip().split(' ')[1] %}
      {{ [idm_padded, idc, day, time] | map('trim') | join(';') }}
      {% endfor %}
      {% endif %}
      {% endfor %}
      
  vars:
    hosts_list: "{{ ansible_play_hosts | map('extract', hostvars) | list }}"
  delegate_to: localhost
  register: csv_content
  run_once: yes

问题更新 #

已了解到可以用纯Python代码对CSV排序，示例代码如下：

with open('unsorted.csv',newline='') as csvfile:
    spamreader = csv.DictReader(csvfile, delimiter=";")
    sortedlist = sorted(spamreader, key=lambda row:(row['column_1'],row['column_2']), reverse=False)

with open('sorted.csv', 'w') as f:
    fieldnames = ['column_1', 'column_2', column_3]
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()
    for row in sortedlist:
        writer.writerow(row)

现在需要将该Python操作编写为Ansible任务，处理已生成的CSV文件。

解决方案 #

可以通过Ansible的command模块直接调用Python执行内联排序代码，复用原任务的文件路径变量确保操作目标一致，具体任务如下：

- name: 按规则排序CSV文件并覆盖原文件
  ansible.builtin.command: |
    python3 -c "
    import csv

    # 复用原任务的CSV文件路径
    csv_file = '/tmp/data_{{ '%Y-%m-%d' | strftime }}.csv'
    
    # 读取CSV内容并按指定规则排序
    with open(csv_file, newline='') as infile:
        reader = csv.DictReader(infile, delimiter=';')
        # 排序规则：day → time → idm → idc
        sorted_rows = sorted(reader, key=lambda r: (r['day'], r['time'], r['idm'], r['idc']))
    
    # 将排序后的数据写入原文件完成覆盖
    with open(csv_file, 'w', newline='') as outfile:
        writer = csv.DictWriter(outfile, fieldnames=reader.fieldnames, delimiter=';')
        writer.writeheader()
        writer.writerows(sorted_rows)
    "
  delegate_to: localhost
  run_once: yes

关键说明 #

• 复用原任务的日期格式化路径，确保操作的是当天生成的目标CSV文件
• 使用csv.DictReader自动识别表头，无需手动指定字段名
• 排序key严格对应需求的day, time, idm, idc顺序，保证排序逻辑符合要求
• 直接写入原文件实现覆盖，无需额外生成临时文件

内容的提问来源于stack exchange，提问作者Tms91