examples/download_wav/step_1_download_wav.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import argparse
from datetime import datetime
from pathlib import Path

import pandas as pd
import requests
from tqdm import tqdm

from project_settings import project_path


def get_args():
    parser = argparse.ArgumentParser()

    parser.add_argument(
        "--excel_file_dir",
        default=(project_path / "examples/download_wav").as_posix(),
        type=str
    )
    parser.add_argument(
        "--start_date",
        default="2022-04-10 00:00:00",
        type=str
    )
    parser.add_argument(
        "--end_date",
        default="2026-04-21 00:00:00",
        type=str
    )
    parser.add_argument(
        "--output_dir",
        default=(project_path / "data/calling/358/wav_2ch").as_posix(),
        type=str
    )
    args = parser.parse_args()
    return args


excel_file_str = """
AIAgent-CallLog-20250929100824.xlsx
AIAgent-CallLog-20250929134959.xlsx
AIAgent-CallLog-20250929135030.xlsx
AIAgent-CallLog-20250929135052.xlsx
AIAgent-CallLog-20250929135122.xlsx
AIAgent-CallLog-20250929135134.xlsx
AIAgent-CallLog-20250929135209.xlsx
AIAgent-CallLog-20250929135219.xlsx
AIAgent-CallLog-20250929135247.xlsx
AIAgent-CallLog-20250929135300.xlsx
AIAgent-CallLog-20250929135311.xlsx
AIAgent-CallLog-20250929135335.xlsx
AIAgent-CallLog-20250929135344.xlsx
AIAgent-CallLog-20250929135355.xlsx
AIAgent-CallLog-20250929135443.xlsx
AIAgent-CallLog-20250929135452.xlsx
AIAgent-CallLog-20250929135501.xlsx
AIAgent-CallLog-20250929135537.xlsx
AIAgent-CallLog-20250929135544.xlsx
AIAgent-CallLog-20250929135554.xlsx
AIAgent-CallLog-20250929135630.xlsx
AIAgent-CallLog-20250929135701.xlsx
AIAgent-CallLog-20250929135710.xlsx
AIAgent-CallLog-20250929135716.xlsx
AIAgent-CallLog-20250929135755.xlsx
AIAgent-CallLog-20250929135800.xlsx
AIAgent-CallLog-20250929135809.xlsx
AIAgent-CallLog-20250929135842.xlsx
AIAgent-CallLog-20250929135849.xlsx
AIAgent-CallLog-20250929135858.xlsx
AIAgent-CallLog-20250929135909.xlsx
"""


def main():
    args = get_args()

    format_str = "%Y-%m-%d %H:%M:%S"

    start_date = datetime.strptime(args.start_date, format_str)
    end_date = datetime.strptime(args.end_date, format_str)

    excel_file_dir = Path(args.excel_file_dir)
    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    print(f"start_date: {start_date}")
    print(f"end_date: {end_date}")

    # finished
    finished = set()
    for filename in output_dir.glob("*.wav"):
        call_id = filename.stem
        finished.add(call_id)

    splits = excel_file_str.split("\n")
    for row in splits:
        name = str(row).strip()
        if len(name) == 0:
            continue
        excel_file = excel_file_dir / name

        df = pd.read_excel(excel_file.as_posix())
        for i, row in tqdm(df.iterrows()):
            call_date = row["Attempt time"]
            call_id = row["Call ID"]
            record_url = row["Recording file"]
            if pd.isna(record_url):
                continue

            if call_id in finished:
                continue
            finished.add(call_id)

            call_date = datetime.strptime(str(call_date), format_str)

            if not start_date < call_date < end_date:
                continue

            call_date_str = call_date.strftime("%Y%m%d")
            # record_url = f"https://phl-01.obs.ap-southeast-3.myhuaweicloud.com/{call_date_str}/21964/{call_id}.wav"
            # record_url = f"https://nxai-hk-1259196162.cos.ap-hongkong.myqcloud.com/{call_date_str}/3101/{call_id}.wav"
            # print(record_url)
            try:
                resp = requests.get(
                    url=record_url,
                )
            except (TimeoutError, requests.exceptions.ConnectionError):
                continue
            except Exception as e:
                print(e)
                continue

            if resp.status_code == 404:
                continue
            if resp.status_code != 200:
                raise AssertionError("status_code: {}; text: {}".format(resp.status_code, resp.text))

            filename = output_dir / "{}.wav".format(call_id)
            with open(filename.as_posix(), "wb") as f:
                f.write(resp.content)

    return


if __name__ == "__main__":
    main()