问合并文件并将其插入BigQuery表中

EN

{"ID":"3193559","Title":"Una Familia de Diez - El secreto","Description":"Martina escucha que la Nena tiene novio y la amenaza con decirles a todos si no hace todo lo que le pida, pero despu\u00e9s la familia descubre su gran secreto.","Program":"Una Familia de Diez","Season":"1","Episode":"16","Source":"Televisa","Category":"Comedy","Syndicator":"CSv2","[CSv2] external_id":"UFDD100023004","[CSv2] pub_win_US_begin":"1657166400","[CSv2] pub_win_US_end":"1924923600","[CSv2] language":"es","[CSv2] title":"Una Familia de Diez - El secreto","[CSv2] descriptive_title":"El secreto","[CSv2] description":"Martina escucha que la Nena tiene novio y la amenaza con decirles a todos si no hace todo lo que le pida, pero despu\u00e9s la familia descubre su gran secreto.","[CSv2] supplier":"Televisa","[CSv2] categories":"Comedy","[CSv2] rating":"TV-14","[CSv2] subratings":"D,L","[CSv2] program_type":"SERIES","[CSv2] entity":"","[CSv2] exception_countries":"US ,\tUM ,PR ,\tMX ,\tAR ,\tCL ,\tCO ,\tPE ,\tEC ,\tCR ,\tSV ,\tHN ,\tBO ,\tPA ,\tDO ,\tNI ,\tPY ,\tVE ,\tUY ,\tGT","[CSv2] episode_type":"","TMS ID":"EP009112420015","external_id":"UFDD100023004","Content Type":"Entertainment","Release Year":"2007","sports_event_ID":""}

def map_keys(
    bucket_name, file_path, list_of_files
):  # pass the folder as an argument
    logging.info(f"bucket_name: {bucket_name}")
    logging.info(f"file_path: {file_path}")
    storage_client = storage.Client()
    bucket = storage_client.get_bucket(bucket_name)
    logging.info(f"list_of_files from the DAG: {list_of_files}")
    blobs = storage_client.list_blobs(
        bucket_or_name=bucket_name, prefix=mcp_source_folder
    )
    blobs = [blob for blob in blobs if "json" in blob.name]
    logging.info(f"The process found {len(blobs)} files to insert")
    if not os.path.exists("unprocessed"):
        os.makedirs("unprocessed")
    if blobs:
        for blob in blobs:
            json_content = blob.download_as_string()
            mcp_data = json.loads(json_content)
            file_name = blob.name.split("/")[-1]
            logging.info(
                f"file to store: {file_name} with {len(mcp_data)} rows"
            )
            path_unprocessed_file = f"unprocessed/{file_name}"
            unprocessed_blob = bucket.blob(path_unprocessed_file)
            with open(path_unprocessed_file, "w") as unprocessed_file:
                for datum in mcp_data:
                    model_datum = McpModel.parse_obj(datum)
                    unprocessed_file.write(json.dumps(model_datum.dict()))
                    unprocessed_file.write("\n")
            unprocessed_blob.upload_from_filename(path_unprocessed_file)